如何从Python中运行bash脚本并获取所有输出？

Question

这是一个直接针对这里的回答进行澄清的问题，我以为它能够起作用，但事实并非如此！

我有以下测试bash脚本（testbash.sh），它只是为了测试目的创建一些输出和许多错误（在Red Hat Enterprise Linux Server release 7.6 (Maipo)和Ubuntu 16.04.6 LTS上运行）:

export MAX_SEED=2
echo "Start test"
pids=""

for seed in `seq 1 ${MAX_SEED}`
do
  python -c "raise ValueError('test')" &
  pids="${pids} $!"
done
echo "pids: ${pids}"
wait $pids
echo "End test"

如果我运行这个脚本，我会得到以下输出：

Start test
pids:  68322 68323
Traceback (most recent call last):
  File "<string>", line 1, in <module>
ValueError: test
Traceback (most recent call last):
  File "<string>", line 1, in <module>
ValueError: test
[1]-  Exit 1                  python -c "raise ValueError('test')"
[2]+  Exit 1                  python -c "raise ValueError('test')"
End test

这是预期的结果。没问题。我想要获得错误！

现在这里是应该捕获所有输出的 Python 代码：

from __future__ import print_function

import sys
import time
from subprocess import PIPE, Popen, STDOUT
from threading  import Thread

try:
    from queue import Queue, Empty
except ImportError:
    from Queue import Queue, Empty  # python 2.x    

ON_POSIX = 'posix' in sys.builtin_module_names

def enqueue_output(out, queue):
    for line in iter(out.readline, b''):
        queue.put(line.decode('ascii'))
    out.close()

p = Popen(['. testbash.sh'], stdout=PIPE, stderr=STDOUT, bufsize=1, close_fds=ON_POSIX, shell=True)
q = Queue()
t = Thread(target=enqueue_output, args=(p.stdout, q))
t.daemon = True # thread dies with the program
t.start()

# read line without blocking
while t.is_alive():
    #time.sleep(1)
    try:
        line = q.get(timeout=.1)
    except Empty:
        print(line)
        pass
    else:
        # got line
        print(line, end='')

p.wait()
print('returncode = {}'.format(p.returncode))

但是当我运行这段代码时，只会得到以下输出：

Start test
pids:  70191 70192
Traceback (most recent call last):
returncode = 0

或者输出以下内容（不包括行End test）:

Start test
pids:  10180 10181
Traceback (most recent call last):
  File "<string>", line 1, in <module>
ValueError: test
Traceback (most recent call last):
  File "<string>", line 1, in <module>
ValueError: test
returncode = 0

大部分输出都不见了！我该怎么解决？还有，我需要一些方法来检查bash脚本中是否有任何命令没有成功执行。在这个例子中就是这种情况，但打印出的错误代码仍为0。我期望一个错误代码 != 0。

重要的是不必立即获得输出。几秒钟的延迟是可以接受的。如果输出顺序有点混乱也不要紧。重要的是获取所有输出(stdout和stderr)。

也许有一种更简单的方法只从python启动bash脚本并获取其输出？

- Alex

---

我已将您的“'. testbash.sh'”更改为“./testbash.sh”，并在Python3上进行了测试，它可以正常工作。希望您首先澄清Python版本并尝试修复命令问题。 - Sraw

当我尝试这样做时，我会收到一个错误信息：/bin/sh: ./testbash.sh: 没有那个文件或目录！ - Alex

那么你正在使用Python来启动一个Bash脚本来启动Python脚本...？在我看来并不是最好的解决方案... - Chris Maes

这是非常正确的。但目前似乎这是唯一的方法（因为bash脚本在启动并行python脚本等时运行在集群上）。对于这个问题来说，这并不重要。可以是任何创建多行错误消息的shell命令。 - Alex

Answer 1

要在Python3中运行

from __future__ import print_function
import os
import stat
import sys
import time
from subprocess import PIPE, Popen, STDOUT
from threading  import Thread
try:
    from queue import Queue, Empty
except ImportError:
    from Queue import Queue, Empty  # python 2.x
ON_POSIX = 'posix' in sys.builtin_module_names
TESTBASH = '/tmp/testbash.sh'
def create_bashtest():
    with open(TESTBASH, 'wt') as file_desc:
        file_desc.write("""#!/usr/bin/env bash
export MAX_SEED=2
echo "Start test"
pids=""
for seed in `seq 1 ${MAX_SEED}`
do
  python -c "raise ValueError('test')" &
  pids="${pids} $!"
  sleep .1 # Wait so that error messages don't get out of order.
done
wait $pids; return_code=$?
sleep 0.2 # Wait for background messages to be processed.
echo "pids: ${pids}"
echo "End test"
sleep 1 # Wait for main process to handle all the output
exit $return_code
""")
    os.chmod(TESTBASH, stat.S_IEXEC|stat.S_IRUSR|stat.S_IWUSR)

def enqueue_output(queue):
    pipe = Popen([TESTBASH], stdout=PIPE, stderr=STDOUT,
                 bufsize=1, close_fds=ON_POSIX, shell=True)
    out = pipe.stdout
    while pipe.poll() is None:
        line = out.readline()
        if  line:
            queue.put(line.decode('ascii'))
        time.sleep(.1)
    print('returncode = {}'.format(pipe.returncode))

create_bashtest()
C_CHANNEL = Queue()

THREAD = Thread(target=enqueue_output, args=(C_CHANNEL,))
THREAD.daemon = True
THREAD.start()

while THREAD.is_alive():
    time.sleep(0.1)
    try:
        line = C_CHANNEL.get_nowait()
    except Empty:
        pass # print("no output")
    else:
        print(line, end='')

希望这有所帮助：

- Philippe

Answer 2

如果你正在寻找这些代码行：

[1]-  Exit 1                  python -c "raise ValueError('test')"
[2]+  Exit 1                  python -c "raise ValueError('test')"

这是bash shell的一个函数，通常只在交互模式下可用，即当您在终端中键入命令时。如果您检查bash源代码，您可以看到它在将内容打印到stdout/stderr之前明确检查模式。

在较新版本的bash中，您无法在脚本内部设置此选项：请参见https://unix.stackexchange.com/a/364618。但是，在启动脚本时，您可以自己设置此选项：

p = Popen(['/bin/bash -i ./testbash.sh'], stdout=PIPE, stderr=STDOUT, bufsize=1, close_fds=ON_POSIX, shell=True)

我要注意的是，这只在Python3上适用 - Python2只能获得部分输出。不清楚您使用的Python版本，但考虑到Python2已经停止维护，我们应该尝试转向Python3。

至于bash脚本，即使设置了交互模式，似乎你仍然需要改变如何等待以获取输出：wait。

#!/bin/bash
export MAX_SEED=2
echo "Start test"
pids=""

for seed in `seq 1 ${MAX_SEED}`
do
    python -c "raise ValueError('test')" &
    pids="${pids} $!"
done
echo "pids: ${pids}"
wait -n $pids
wait -n $pids
ret=$?
echo "End test"
exit $ret

正常的wait对我来说不起作用（Ubuntu 18.04），但wait -n似乎可以工作，但由于它只等待下一个作业完成，我调用它一次就会产生不一致的输出。为每个启动的作业调用wait -n似乎可以解决问题，但程序流应该重新设计为循环等待与启动作业数量相同的次数。

还要注意更改脚本的返回代码，Philippe的答案是正确的方法-$?变量具有最后一个失败的命令的返回代码，你可以将其传递给exit。（Python版本的另一个差异：对我来说，Python2返回127，而Python3返回1）。如果您需要每个作业的返回值，则可能的一种方法是解析交互式作业退出行中的值。

- Mostly Harmless

Answer 3

首先，看起来缓冲区没有被刷新。将stdout/stderr重定向（并为了安全起见，附加）到文件而不是终端，可能会有所帮助。如果您确实需要两者，您总是可以使用tee（或tee -a）。使用上下文管理器可能会有所帮助。

至于零返回代码，$!https://unix.stackexchange.com/questions/386196/doesnt-work-on-command-line!可能会调用历史记录，从而导致$!的值为空。

如果您最终只得到一个裸的wait，则返回代码将为零。无论如何，返回代码可能很棘手，您可能正在从其他地方选择成功的返回代码。

查看stdbuf命令以更改stdout和stderr的缓冲区大小： Is there a way to flush stdout of a running process 这也可能有助于获得其余预期输出。

- ShpielMeister

Answer 4

将 while 块重写为以下方式：

# read line without blocking
while t.is_alive():
    try:
        line = q.get(block=False)
    except Empty:
        # print(line)
        pass
    else:
        # got line
        print(line, end='')

当队列中没有数据时，您不希望在获取Queue中的一行时被阻塞，也不需要超时，因为只有在需要阻塞线程时才会使用。因此，如果Queue.get()抛出Empty，则没有要打印的行，我们只需pass。

===

另外，让我们澄清脚本执行逻辑。

由于您正在使用 Bash 表达式，并且 Popen 使用的默认 shell 是 /bin/sh，因此您可能希望以以下方式重写调用行：

p = Popen(['/usr/bin/bash','-c', './testbash.sh'], stdout=PIPE, stderr=STDOUT, bufsize=1, close_fds=ON_POSIX)

给你的Shell脚本添加一个shebang也不会有什么坏处：

#!/usr/bin/env bash
<... rest of the script ...>

- KMZ

---

最后一件事：有时会出现ValueError: test行被覆盖或叠加的情况，我无法弄清楚为什么会发生这种情况，但显然这是一个shell/管道问题，与你的Python代码无关。 - KMZ

Answer 5

只是猜测 - 可能是因为空字符/空格开头的行未被您的逻辑识别为一行。

也许这个缩进是问题所在。另一个选项是，可能存在制表符或类似的字符，导致 ASCII 解码失败。

- x-of Rezbach

Answer 6

这是我通常使用 subprocess 的方式：

import subprocess

with subprocess.Popen(["./test.sh"], shell=True, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.PIPE) as p:
    error = p.stderr.read().decode()
    std_out = p.stdout.read().decode()
    if std_out:
        print(std_out)
    if error:
        print("Error message: {}".format(error))

在这里，您可以解码并读取标准输出和标准错误流。您可以获取所有内容，但不一定按照相同的顺序，我不知道这是否是一个问题。

- Christoffer

---

这段代码是阻塞的。我需要一个非阻塞的代码。 - Alex