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

Question

这是对here中答案的直接澄清问题。我认为它有效，但它没有!

我有以下测试 bash 脚本 (testbash.sh)，它只是创建一些输出和大量用于测试目的的错误(在 Red Hat Enterprise Linux Server 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 脚本的输出？

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='')

希望对您有所帮助: