如何在使用subprocess时复制tee行为？

90 投票

9 回答

33205 浏览

数据工程师

提问于 2025-04-15 23:41

我在找一个Python的解决方案，想把命令的输出保存到文件里，同时又不想让它在控制台上消失。

顺便说一下，我说的是tee这个工具（Unix命令行工具），而不是Python的intertools模块里的同名函数。

具体要求

需要一个Python的解决方案（不能调用tee，因为在Windows上用不了）
我不需要给被调用的程序提供任何输入
我对被调用的程序没有控制权。我只知道它会输出一些东西到标准输出（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)

真实输出

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

错误处理 subprocess 进程管理 stderr 子进程实时输出 stdout tee

9 个回答

这段内容是把Linux中的一个叫做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（错误输出）连接到你的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 pipe的错误）。

回答于 2025-04-15 由 Python大师

分享举报

我看到这是一个比较老的帖子，但如果还有人想知道怎么做的话，以下是方法：

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)

回答于 2025-04-15 由 Python大师

分享举报

如果你不介意使用 Python 3.6，现在有一种方法可以做到这一点，使用的是 asyncio。这个方法可以让你分别捕获标准输出（stdout）和标准错误（stderr），而且还能让这两个输出同时显示在终端上，而不需要使用线程。下面是一个大致的思路：

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/

回答于 2025-04-15 由 Python大师

分享举报

如何在使用subprocess时复制tee行为？

具体要求

参考资料

当前代码（第二次尝试）

9 个回答

撰写回答