实时打印使用 Python 中的子进程启动的 bash 命令的结果

Posted 2023-02-23

【中文标题】实时打印使用 Python 中的子进程启动的 bash 命令的结果

【英文标题】：Print in real time the result of a bash command launched with subprocess in Python

【发布时间】：2019-05-22 15:29:56

【问题描述】：

我正在使用 subprocess 模块来运行 bash 命令。我想实时显示结果，包括在没有添加新行的情况下，但输出仍然被修改。

我正在使用 python 3。我的代码使用子进程运行，但我对任何其他模块都是开放的。我有一些代码为添加的每个新行返回一个生成器。

import subprocess
import shlex

def run(command):
    process = subprocess.Popen(shlex.split(command), stdout=subprocess.PIPE)
    while True:
        line = process.stdout.readline().rstrip()
        if not line:
            break
        yield line.decode('utf-8')

cmd = 'ls -al'
for l in run(cmd):
     print(l)   

例如，rsync -P file.txt file2.txt 形式的命令会出现问题，该命令会显示进度条。

例如，我们可以先在 bash 中创建一个大文件：

base64 /dev/urandom | head -c 1000000000 > file.txt 

然后尝试使用python显示rsync命令：

cmd = 'rsync -P file.txt file2.txt'
for l in run(cmd):
    print(l)

有了这段代码，进度条只在流程结束时打印，但我想实时打印进度。

【参考方案1】：

来自this answer，您可以在 python 中打印时禁用缓冲：

您可以使用“python -u”跳过整个 python 进程的缓冲 （或#!/usr/bin/env python -u 等）或通过设置环境 变量PYTHONUNBUFFERED。

您还可以将sys.stdout 替换为其他一些流，例如包装器 每次调用后都会刷新。

这样的东西（没有真正测试过）可能会起作用......但是有 可能会出现的问题。例如，我不认为 将在 IDLE 中工作，因为 sys.stdout 已经被替换为一些 那里不喜欢被冲洗的有趣物体。 （这可能是 虽然被认为是 IDLE 中的错误。）

>>> class Unbuffered:
..     def __init__(self, stream):
..         self.stream = stream
..     def write(self, data):
..         self.stream.write(data)
..         self.stream.flush()
..     def __getattr__(self, attr):
..         return getattr(self.stream, attr)
..
>>> import sys
>>> sys.stdout=Unbuffered(sys.stdout)
>>> print 'Hello'
Hello

【讨论】：

感谢您的帮助。但是，虽然 -u 肯定有用，但我想通过 python 显示 bash 脚本的结果。例如，我希望能够在 jupyter 笔记本中使用 run 命令并打印结果，或者等待写入特定行来执行操作，或者漂亮地打印 bash 命令的结果。如果我只使用 -u 和（例如）os.system(cmd)，我将无法做到这一点。