在使用subprocess时,如何在Python中复制tee行为?

88
我正在寻找一个Python解决方案,允许我将命令的输出保存到文件中,同时不隐藏它在控制台上的显示。
FYI:我在问关于Unix命令行实用程序tee(而不是Python intertools模块中同名的函数)。
细节:
- Python解决方案(不调用tee,在Windows下不可用) - 我不需要为所调用的进程提供任何输入 - 我无法控制所调用程序。我只知道它会向stdout和stderr输出一些内容,并返回退出代码。 - 在调用外部程序(subprocess)时工作 - 对于stderr和stdout都能正常工作 - 能够区分stdout和stderr,因为我可能只想将其中一个显示在控制台上,或者我可以尝试使用不同的颜色输出stderr - 这意味着stderr = subprocess.STDOUT将不起作用。 - 实时输出(渐进式) - 进程可能运行很长时间,我不能等待它完成。 - Python 3兼容代码(重要)
参考资料:
这里是我找到的一些不完整的解决方案:

Diagram http://blog.i18n.ro/wp-content/uploads/2010/06/Drawing_tee_py.png

当前代码(第二次尝试)

#!/usr/bin/python
from __future__ import print_function

import sys, os, time, subprocess, io, threading
cmd = "python -E test_output.py"

from threading import Thread
class StreamThread ( Thread ):
    def __init__(self, buffer):
        Thread.__init__(self)
        self.buffer = buffer
    def run ( self ):
        while 1:
            line = self.buffer.readline()
            print(line,end="")
            sys.stdout.flush()
            if line == '':
                break

proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdoutThread = StreamThread(io.TextIOWrapper(proc.stdout))
stderrThread = StreamThread(io.TextIOWrapper(proc.stderr))
stdoutThread.start()
stderrThread.start()
proc.communicate()
stdoutThread.join()
stderrThread.join()

print("--done--")

#### test_output.py ####

#!/usr/bin/python
from __future__ import print_function
import sys, os, time

for i in range(0, 10):
    if i%2:
        print("stderr %s" % i, file=sys.stderr)
    else:
        print("stdout %s" % i, file=sys.stdout)
    time.sleep(0.1)

stderr 1
stdout 0
stderr 3
stdout 2
stderr 5
stdout 4
stderr 7
stdout 6
stderr 9
stdout 8
--done--

预期输出应该按行排序。请注意,修改Popen以仅使用一个PIPE是不允许的,因为在实际情况下,我将希望对stderr和stdout执行不同的操作。
即使在第二种情况下,我也无法获得类似实时的输出,事实上,所有结果都是在进程完成后接收到的。默认情况下,Popen不应使用缓冲区(bufsize=0)。

1
可能是Python Popen:同时写入stdout和日志文件的重复问题。以这种方式投票是因为这是一个社区wiki :-) - Ciro Santilli OurBigBook.com
9个回答

12

我看到这是一个相当古老的帖子,但以防万一有人仍在寻找一种方法来做到这一点:

proc = subprocess.Popen(["ping", "localhost"], 
                        stdout=subprocess.PIPE, 
                        stderr=subprocess.PIPE)

with open("logfile.txt", "w") as log_file:
  while proc.poll() is None:
     line = proc.stderr.readline()
     if line:
        print "err: " + line.strip()
        log_file.write(line)
     line = proc.stdout.readline()
     if line:
        print "out: " + line.strip()
        log_file.write(line)

2
这对我有用,尽管我发现stdout,stderr = proc.communicate()更容易使用。 - Chase Seibert
31
这个解决方案对于任何可以在标准输出(stdout)或标准错误(stderr)上生成足够输出并且stdout/stderr不完全同步的子进程会导致死锁。 - jfs
@J.F.Sebastian: 是的,但是你可以通过将readline()替换为readline(size)来解决这个问题。我在其他语言中也做过类似的事情。参考:https://docs.python.org/3/library/io.html#io.TextIOBase.readline - kevinarpe
6
@kevinarpe 错了。readline(size)不能解决死锁问题。需要同时读取stdout/stderr。请参考问题下面的链接,这些链接展示了使用线程或asyncio的解决方案。 - jfs
@J.F.Sebastian,如果我只对其中一个流进行阅读,这个问题是否存在? - ThorSummoner
显示剩余2条评论

12

如果要求使用Python 3.6不是问题,现在可以使用asyncio来实现此操作。这种方法允许您单独捕获stdout和stderr,但仍然可以将两个流向tty而不使用线程。以下是大致概述:

class RunOutput:
    def __init__(self, returncode, stdout, stderr):
        self.returncode = returncode
        self.stdout = stdout
        self.stderr = stderr


async def _read_stream(stream, callback):
    while True:
        line = await stream.readline()
        if line:
            callback(line)
        else:
            break


async def _stream_subprocess(cmd, stdin=None, quiet=False, echo=False) -> RunOutput:
    if isWindows():
        platform_settings = {"env": os.environ}
    else:
        platform_settings = {"executable": "/bin/bash"}
    if echo:
        print(cmd)
    p = await asyncio.create_subprocess_shell(
        cmd,
        stdin=stdin,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        **platform_settings
    )
    out = []
    err = []

    def tee(line, sink, pipe, label=""):
        line = line.decode("utf-8").rstrip()
        sink.append(line)
        if not quiet:
            print(label, line, file=pipe)

    await asyncio.wait(
        [
            _read_stream(p.stdout, lambda l: tee(l, out, sys.stdout)),
            _read_stream(p.stderr, lambda l: tee(l, err, sys.stderr, label="ERR:")),
        ]
    )

    return RunOutput(await p.wait(), out, err)


def run(cmd, stdin=None, quiet=False, echo=False) -> RunOutput:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(
        _stream_subprocess(cmd, stdin=stdin, quiet=quiet, echo=echo)
    )

    return result

以上代码基于这篇博客文章:https://kevinmccarthy.org/2016/07/25/streaming-subprocess-stdin-and-stdout-with-asyncio-in-python/


8
这是将tee(1)直接移植到Python的过程。
import sys

sinks = sys.argv[1:]
sinks = [open(sink, "w") for sink in sinks]
sinks.append(sys.stderr)
while True:
    input = sys.stdin.read(1024)
    if input:
        for sink in sinks:
            sink.write(input)
    else:
        break

我现在正在使用Linux,但这应该适用于大多数平台。


现在谈到subprocess部分,我不知道你想如何将子进程的stdinstdoutstderr与你的stdinstdoutstderr和文件接收器“连接”起来,但我知道你可以这样做:
import subprocess

callee = subprocess.Popen(
    ["python", "-i"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
)

现在您可以像普通文件一样访问callee.stdincallee.stdoutcallee.stderr,使上述“解决方案”得以工作。如果要获取callee.returncode,则需要再调用callee.poll()

写入callee.stdin时请小心:如果进程在此时已退出,则可能会引发错误(在Linux上,我会收到IOError:[Errno 32] Broken pipe)。


3
在Linux中,这种方法并不是最优的,因为Linux提供了一个临时的tee(f_in, f_out, len, flags) API,但这不是重点,对吧? - badp
1
我更新了问题,问题是我找不到如何使用subprocess逐步获取两个管道中的数据而不是在进程结束时一次性获取所有数据。 - sorin
1
@Sorin,这意味着您必须使用两个线程。一个读取stdout,一个读取stderr。如果您要将两者都写入同一文件,则可以在开始读取时获取接收器上的锁,并在写入行终止符后释放它。 :/ - badp
使用线程来解决这个问题对我来说并不太吸引人,也许我们会找到其他的方法。奇怪的是,这是一个常见的问题,但没有人提供完整的解决方案。 - sorin
@badp 我尝试了线程的方法,但它不起作用。我更新了问题,包括新的示例。 - sorin
显示剩余2条评论

6

这就是如何完成它

import sys
from subprocess import Popen, PIPE

with open('log.log', 'w') as log:
    proc = Popen(["ping", "google.com"], stdout=PIPE, encoding='utf-8')
    while proc.poll() is None:
        text = proc.stdout.readline() 
        log.write(text)
        sys.stdout.write(text)

2
对于任何想知道的人,是的,你可以使用print()而不是sys.stdout.write()。 :-) - progyammer
@程序员 当你需要忠实地复制输出时,print会添加一个额外的换行符,这不是你想要的。 - ivan_pozdeev
是的,但 print(line, end='') 可以解决这个问题。 - Danylo Zhydyk

2
根据社区维基答案,这是一个更新的版本。
- 添加了类型 - 使用`gather`代替`wait`(`wait`会产生警告) - 不必要地解码为`str` - 添加超时
这是一个完整的文件,你可以运行;超时设置为5秒,所以应该会超时。
注意:Python默认缓冲stdout,所以你需要在所有地方使用`-u`。
#!/usr/bin/env -S python3 -u

import asyncio
from typing import BinaryIO, Callable, Union
import sys

class RunOutput:
    def __init__(self, exit_code: int, stdout: list[bytes], stderr: list[bytes]):
        self.exit_code = exit_code
        self.stdout = stdout
        self.stderr = stderr


async def _read_stream(stream: asyncio.StreamReader, callback: Callable[[bytes], None]):
    while True:
        line = await stream.readline()
        if len(line) == 0:
            break
        callback(line)


async def _stream_subprocess(command: list[str]) -> RunOutput:
    p = await asyncio.create_subprocess_exec(
        *command,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )

    stdout: list[bytes] = []
    stderr: list[bytes] = []

    def tee(line: bytes, sink: list[bytes], out: BinaryIO):
        sink.append(line)
        out.write(line)

    assert p.stdout is not None
    assert p.stderr is not None

    await asyncio.gather(
        _read_stream(p.stdout, lambda l: tee(l, stdout, sys.stdout.buffer)),
        _read_stream(p.stderr, lambda l: tee(l, stderr, sys.stderr.buffer)),
    )

    exit_code = await p.wait()

    return RunOutput(exit_code, stdout, stderr)


def run(command: list[str], timeout: Union[int, float, None]) -> RunOutput:
    loop = asyncio.get_event_loop()
    return loop.run_until_complete(
        asyncio.wait_for(_stream_subprocess(command), timeout)
    )


def main():
    if "--count" in sys.argv:
        import time

        for i in range(10):
            print(f"A stdout {i}")
            print(f"B stderr {i}", file=sys.stderr)
            time.sleep(1)
            print(f"C stderr {i}", file=sys.stderr)
            print(f"D stdout {i}")
            time.sleep(1)
    else:
        run(["python3", "-u", __file__, "--", "--count"], 5)

if __name__ == "__main__":
    main()

1
从一个简单的例子开始,使用tee(稍后我会向你展示如何在没有tee的情况下完成这个操作),你可以做到以下几点:
def tee(command, **kwargs):
    p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, **kwargs)
    t_out = subprocess.Popen(['tee', '-a', '/dev/stderr'], stdin=p.stdout, stderr=subprocess.PIPE, text=True)
    t_err = subprocess.Popen(['tee', '-a', '/dev/stderr'], stdin=p.stderr, stdout=subprocess.PIPE, text=True)
    return p, t_out, t_err

在这里:
  1. 我们在一个子进程中启动命令,p,同时捕获stderrstdout
  2. 我们启动另一个子进程t_out,运行tee,只捕获stderr(允许stdout正常流出tee)。
  3. 我们对子进程t_err做同样的处理,但是将pstderr发送出去,只捕获stdout(允许stderr正常流向stderr)。
最终的结果是,您的命令的stdout和stderr会正常输出到终端,并且也会被返回的子进程捕获。
假设有一个简单的程序,它会写入到stderr和stdout:
# test.py
import sys, time
for i in range(10):
    if i % 2 == 0:
        print(i, file=sys.stderr, flush=True)
    else:
        print(i, flush=True)
    time.sleep(0.1)

你可以这样做:
print('starting')
process, t_out, t_err = tee([sys.executable, 'test.py'])
while process.poll() is None:
    time.sleep(0.1)  # wait for process to finish
print('done')
print('stdout:', t_out.stderr.read())
print('stderr:', t_err.stdout.read())

除了程序的输出之外,你可以看到标准输出(stdout)和标准错误(stderr)可以被Python脚本读取:
starting
0
1
2
3
4
5
6
7
8
9
done
stdout: 1
3
5
7
9

stderr: 0
2
4
6
8

不使用tee

请注意,实际上并不需要使用tee程序。这完全可以是一个纯Python程序,它读取标准输入并像tee一样将输出复制到标准输出。

例如,可以使用以下Python脚本代替。它只是读取标准输入并将其打印到标准输出和标准错误输出。

# tee.py
import sys
for line in sys.stdin:
    print(line, file=sys.stdout, flush=True, end='')
    print(line, file=sys.stderr, flush=True, end='')

然后,第一个例子可以这样修改:
def tee(command, **kwargs):
    p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, **kwargs)
    t_out = subprocess.Popen([sys.executable, 'tee.py'], stdin=p.stdout, stderr=subprocess.PIPE, text=True)
    t_err = subprocess.Popen([sys.executable, 'tee.py'], stdin=p.stderr, stdout=subprocess.PIPE, text=True)
    return p, t_out, t_err

最终结果与第一个示例类似,但不需要使用程序tee
这个解决方案并不一定需要使用额外的子进程。这只是一种方法。同样的解决方案也可以在两个线程中完成,这两个线程消耗第一个子进程的stderr/stdout。
注意:如果stdout和stderr在大致相同的时间写入,为了确保消息按正确的顺序到达终端,可能需要进行一些重大的更改。

0
如果您不想与进程进行交互,那么使用subprocess模块就可以了。
例如:

tester.py

import os
import sys

for file in os.listdir('.'):
    print file

sys.stderr.write("Oh noes, a shrubbery!")
sys.stderr.flush()
sys.stderr.close()

testing.py

import subprocess

p = subprocess.Popen(['python', 'tester.py'], stdout=subprocess.PIPE,
                     stdin=subprocess.PIPE, stderr=subprocess.PIPE)

stdout, stderr = p.communicate()
print stdout, stderr

在您的情况下,您可以先将stdout/stderr写入文件。您也可以使用communicate向进程发送参数,但我无法弄清如何持续与子进程交互。

2
这不会在标准输出的上下文中显示标准错误的错误消息,这可能使调试shell脚本等几乎不可能。 - RobM
意思是……?在这个脚本中,通过 STDERR 传递的任何内容都会与 STDOUT 一起打印到屏幕上。如果您正在引用返回代码,请使用p.poll()来检索它们。 - Wayne Werner
3
这不满足“渐进式”的条件。 - ivan_pozdeev

-1

在Linux上,如果你真的需要像tee(2)系统调用这样的东西,你可以像这样获取它:

import os
import ctypes

ld = ctypes.CDLL(None, use_errno=True)

SPLICE_F_NONBLOCK = 0x02


def tee(fd_in, fd_out, length, flags=SPLICE_F_NONBLOCK):
    result = ld.tee(
        ctypes.c_int(fd_in),
        ctypes.c_int(fd_out),
        ctypes.c_size_t(length),
        ctypes.c_uint(flags),
    )

    if result == -1:
        errno = ctypes.get_errno()
        raise OSError(errno, os.strerror(errno))

    return result

如果您想使用此功能,您可能需要使用Python 3.10和带有os.splice的某些内容(或者以相同方式使用ctypes来获取splice)。请参见{{link1:tee(2)手册}}中的示例。


-2

我的解决方案不够优雅,但它有效。

你可以使用PowerShell在WinOS下访问“tee”。

import subprocess
import sys

cmd = ['powershell', 'ping', 'google.com', '|', 'tee', '-a', 'log.txt']

if 'darwin' in sys.platform:
    cmd.remove('powershell')

p = subprocess.Popen(cmd)
p.wait()

在MacOS中,ping命令会给出无效的命令行错误消息。 - ivan_pozdeev

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接