Python3：逐行打印子进程标准输出

Question

我创建了一个简单的Python函数：

import subprocess
from io import TextIOWrapper


def run_shell_command(command: list, debug: bool = False):
    '''
    Run shell command

    :param command: Shell command
    :param debug: Debug mode
    :return: Result code and message
    '''
    try:
        process = subprocess.run(
            command, check=True, text=True, timeout=5,
            stdout=subprocess.PIPE, stderr=subprocess.STDOUT
        )
        if debug:
            for line in TextIOWrapper(process.stdout, encoding='utf-8'):
                print(line)
        message = 'Shell command executed sucessfully'
        return ({'code': 200, 'msg': message, 'stdout': process.stdout})
    except subprocess.CalledProcessError as e:
        return ({'code': 500, 'msg': e.output})


if __name__ == "__main__":
    command = run_shell_command(['ls', '-lah'], True)
    print(command)

当我在调试模式下运行它时，出现以下错误：

Traceback (most recent call last):
  File "/tmp/command.py", line 28, in <module>
    command = run_shell_command(['ls', '-lah'], True)
  File "/tmp/command.py", line 19, in run_shell_command
    for line in TextIOWrapper(process.stdout, encoding="utf-8"):
AttributeError: 'str' object has no attribute 'readable'

在 Linux 服务器上运行 Python 3.9，我想知道您是否可以提供一些可能出现问题的见解。禁用调试后，我得到了正确的文本输出。谢谢您的帮助。

Answer 1

不幸的是，“简单”并不能解决所有涉及的极端情况——您需要在流式传输时读取子流程

stdout

，将其打印出来并将其累积在缓冲区中，并跟踪时间，以便您可以正确超时。请注意，如果 4096 字节读取恰好以多行字符结束，这也可能存在错误（尽管不是很危险）。

import subprocess
import time


def run_shell_command(command: list, debug: bool = False, timeout: float = 5):
    """
    Run shell command

    :param command: Shell command
    :param debug: Debug mode
    :return: Result code and message
    """
    process = subprocess.Popen(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
    )
    start_time = time.time()
    output = b""
    while True:
        buf = process.stdout.read(4096)
        if debug:
            # could fail if `buf` happens to end in a multi-byte character
            print(buf.decode("utf-8", "ignore"))
        output += buf

        if time.time() - start_time > timeout:
            process.kill()
            message = "Shell command timed out"
            return {"code": 500, "msg": message, "stdout": output}

        if process.poll() is not None:
            break
    if process.returncode != 0:
        message = "Shell command failed"
        return {"code": 500, "msg": message, "stdout": output}

    message = "Shell command executed successfully"
    return {"code": 200, "msg": message, "stdout": output}


if __name__ == "__main__":
    command = run_shell_command(["ls", "-lah"], True)
    print(command)