Python3:打印子进程标准输出行输出

问题描述 投票:0回答:1

我创建了一个简单的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,我想知道您是否可以提供一些可能出现问题的见解。禁用调试后,我得到了正确的文本输出。谢谢您的帮助。

编辑:根据下面的评论,修复非常简单:

        if debug:
            print(process.stdout.rstrip())
python python-3.x subprocess
1个回答
0
投票

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

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)
    
© www.soinside.com 2019 - 2024. All rights reserved.