如何在Python中封装进程运行：实时输出+捕获返回码与输出

需求概述 #

需要实现一个函数，替代subprocess.run(args, capture_output=True)，满足以下要求：

• 实时显示子进程的stdout和stderr，效果等同于直接在Shell中运行
• 同时捕获这两个流的内容（字节格式），供后续处理
• 获取进程的返回码
• 返回的对象与subprocess.run返回的CompletedProcess结构兼容，原有代码无需修改

实现代码 #

import subprocess
import threading
from types import SimpleNamespace

def run_and_output_realtime(args):
    # 启动子进程，获取管道
    process = subprocess.Popen(
        args,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=False  # 保持字节流，与capture_output=True行为一致
    )

    # 定义缓冲区存储捕获的流内容
    stdout_buffer = []
    stderr_buffer = []

    def read_stream(stream, buffer, is_stderr=False):
        """读取流内容，实时输出并写入缓冲区"""
        for chunk in iter(lambda: stream.read(1024), b''):
            buffer.append(chunk)
            # 实时输出到对应流：stdout到控制台标准输出，stderr到控制台错误输出
            if is_stderr:
                print(chunk.decode(), end='', flush=True)
            else:
                print(chunk.decode(), end='', flush=True)
        stream.close()

    # 启动线程处理stdout和stderr
    stdout_thread = threading.Thread(target=read_stream, args=(process.stdout, stdout_buffer, False))
    stderr_thread = threading.Thread(target=read_stream, args=(process.stderr, stderr_buffer, True))
    
    stdout_thread.start()
    stderr_thread.start()

    # 等待线程和进程结束
    stdout_thread.join()
    stderr_thread.join()
    returncode = process.wait()

    # 拼接缓冲区内容，模拟CompletedProcess的结构
    result = SimpleNamespace(
        returncode=returncode,
        stdout=b''.join(stdout_buffer),
        stderr=b''.join(stderr_buffer)
    )

    return result

使用示例 #

直接替换原有代码的第一行即可，其余逻辑完全兼容：

# 替换前
# run_tool_result = subprocess.run(args, capture_output=True)
# 替换后
run_tool_result = run_and_output_realtime(args)

if run_tool_result.returncode != 0:
    if "Unauthorized" in run_tool_result.stderr.decode():
        print("Please check authorization")
else:
    if "Duration" in run_tool_result.stdout.decode():
        # 解析输出中的时长信息
        pass

关键说明 #

1. 实时输出：通过线程逐块读取子进程的输出流，每读取一块就立即打印，保证实时性
2. 流捕获：将读取到的字节块存入缓冲区，最后拼接成完整的字节流，与subprocess.run的capture_output=True返回格式一致
3. 兼容性：使用SimpleNamespace创建返回对象，包含returncode、stdout、stderr属性，完全适配原有代码的访问方式
4. 编码处理：读取字节流后解码打印，保持与Shell输出一致的显示效果；捕获的内容保留字节格式，方便后续按需解码

内容的提问来源于stack exchange，提问作者goose_lake