如何实时监听并捕获Python的stdout输出？

要解决缓冲延迟问题，同时实现输出实时打印到原stdout并同步复制到目标流，你可以自定义一个流处理类替换系统sys.stdout，强制每次写入后立即刷新缓冲，避免输出被积压。

核心实现代码 #

import sys
import io

class StreamDuplicator(io.TextIOBase):
    def __init__(self, original_stream, target_stream):
        self.original = original_stream
        # 自动处理二进制流转文本流的情况
        self.target = target_stream if isinstance(target_stream, io.TextIOBase) else io.TextIOWrapper(target_stream)

    def write(self, data):
        # 同时写入原stdout和目标流
        self.original.write(data)
        self.target.write(data)
        # 强制刷新缓冲，确保实时输出
        self.flush()
        return len(data)

    def flush(self):
        self.original.flush()
        self.target.flush()

    def close(self):
        # 仅关闭目标流，原stdout由系统管理
        self.target.close()
        super().close()

两种使用方式 #

方式1：手动替换与恢复

if __name__ == "__main__":
    original_stdout = sys.stdout
    # 目标流示例：本地日志文件
    with open("output_log.txt", "w", encoding="utf-8") as log_file:
        sys.stdout = StreamDuplicator(original_stdout, log_file)
        try:
            # 测试实时输出
            import time
            for i in range(5):
                print(f"当前计数: {i}")
                time.sleep(1)
            print("测试完成")
        finally:
            # 必须恢复原stdout，避免影响后续代码
            sys.stdout = original_stdout

方式2：上下文管理器（更简洁安全）

from contextlib import contextmanager

@contextmanager
def duplicate_stdout(target_stream):
    original_stdout = sys.stdout
    try:
        sys.stdout = StreamDuplicator(original_stdout, target_stream)
        yield
    finally:
        sys.stdout = original_stdout

# 使用示例
if __name__ == "__main__":
    with open("output_log.txt", "w", encoding="utf-8") as log_file:
        with duplicate_stdout(log_file):
            print("这行内容会同时出现在控制台和日志文件")
            import time
            time.sleep(1)
            print("延迟1秒后仍实时输出")

关键说明 #

• 自定义类通过实现write和flush方法，彻底规避了Python默认的输出缓冲机制，确保每一次写入都实时生效。
• 目标流支持任意类文件对象，比如io.StringIO（内存流）、网络流等，只需保证流的读写模式匹配即可。
• 无需修改原有print或第三方库的输出代码，只要代码使用标准stdout输出，就能被自动捕获和复制。

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