使用子进程时如何在 Python 中复制 tee 行为？

Question

我正在寻找一个 Python 解决方案，它允许我将命令的输出保存在文件中，而不将其从控制台隐藏。

仅供参考：我正在询问 tee （作为 Unix 命令行实用程序）和不是 Python intertools 模块中的同名函数。

详细信息

• Python 解决方案（不调用 tee，它在 Windows 下不可用）
• 我不需要为被调用进程向 stdin 提供任何输入
• 我无法控制被调用程序。我所知道的是它会向 stdout 和 stderr 输出一些内容并返回退出代码。
• 在调用外部程序（子进程）时工作
• 对 stderr 和 stdout 都
• 适用 能够区分 stdout 和 stderr，因为我可能只想显示其中之一控制台或者我可以尝试使用不同的颜色输出 stderr - 这意味着 stderr = subprocess.STDOUT 将不起作用。
• 实时输出（渐进式）- 该过程可以运行很长时间，而且我无法等待它完成。
• Python 3 兼容代码（重要）

参考资料

以下是我迄今为止发现的一些不完整的解决方案：

• http://devlishgenius.blogspot.com/2008/10/logging-in-real-time-in-python.html（mkfifo 仅适用于 Unix）
• http://blog.kagesenshi.org/2008/02/teeing-python-subprocesspopen- output.html （根本不起作用）

图 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)

Real output

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)。

Answer 1

如果需要 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/

Answer 2

我发现这是一篇相当旧的帖子，但以防万一有人仍在寻找一种方法来做到这一点：

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)

Answer 3

这是 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 部分，我不知道如何“连接”子进程的 stdin、stdout 和 stderr< /code> 到您的 stdin、stdout、stderr 和文件接收器，但我知道您可以这样做：

import subprocess

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

现在您可以访问 callee.stdin，callee.stdout 和 callee.stderr 像普通文件一样，启用上述功能“解决方案”发挥作用。如果您想获取 callee.returncode ，您需要额外调用 <代码>callee.poll()。

写入callee.stdin时要小心：如果执行此操作时进程已退出，可能会出现错误（在 Linux 上，我收到IOError: [Errno 32] Broken pipeline代码>）。

Answer 4

这是可以做到的

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)

Answer 5

根据社区 wiki 的回答，这里有一个更新版本。

• 添加类型
• 使用 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()

Answer 6

从使用 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. 在子进程中启动我们的命令，< code>p，捕获 stderr 和 stdout。
2. 我们启动另一个子进程t_out，运行tee，并且我们仅捕获stderr（允许stdout从tee流出） > 通常）
3. 我们对子进程 t_err 执行相同的操作，但从 p 发送 stderr 并仅捕获 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())

除了程序的输出之外，您还可以看到 Python 脚本可以读取 stdout 和 stderr：

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 程序，它读取 stdin 并像 tee 一样读取输出。

例如，可以改用以下 Python 脚本。它所做的只是读取 stdin 并将其打印到 stdout 和 stderr。

# 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 大约同时写入，则需要进行一些重大更改才能保证消息到达终端的正确顺序。

Answer 7

如果您不想与进程交互，可以使用 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()

test.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 写入文件。您也可以通过通信向进程发送参数，尽管我无法弄清楚如何持续与子进程交互。

Answer 8

在 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< /code> 以同样的方式获得splice）。请参阅 tee(2) 手册页< /a> 为例。

Answer 9

我的解决方案并不优雅，但它有效。

您可以使用 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()