我明白你的痛点——用subprocess.run虽然能完整捕获输出，但长命令执行时看不到实时进度，调试阶段特别不方便。下面给你一套实用的实现方案，既能按日志级别（比如DEBUG及以上）触发实时输出，又能同时捕获输出供后续处理。

核心思路 #

当日志级别满足预设条件（比如DEBUG）时，我们用subprocess.Popen替代subprocess.run——因为Popen不会阻塞等待命令结束，允许我们逐行读取子进程的stdout/stderr流，一边实时打印到控制台（或日志），一边把输出内容保存下来。如果日志级别不满足，就回到原来的subprocess.run模式，只静默捕获输出不实时显示。

完整代码实现 #

import subprocess
import logging

# 先配置日志（可根据实际场景调整级别和格式）
logging.basicConfig(
    level=logging.INFO,  # 默认设为INFO，调试时可改为DEBUG
    format='%(levelname)s: %(message)s'
)

def run_command(command_line, live_output_log_level=logging.DEBUG):
    """
    执行外部命令，根据日志级别决定是否实时显示输出，同时始终捕获完整输出
    :param command_line: 要执行的命令字符串
    :param live_output_log_level: 触发实时输出的最低日志级别
    :return: (stdout_str, stderr_str, returncode)
    """
    logger = logging.getLogger()
    # 判断当前日志级别是否满足实时输出要求
    enable_live_output = logger.isEnabledFor(live_output_log_level)

    if enable_live_output:
        stdout_lines = []
        stderr_lines = []
        # 启动子进程，开启行缓冲保证实时输出
        with subprocess.Popen(
            command_line,
            shell=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,  # 文本模式读取，无需手动解码
            bufsize=1,  # 行缓冲模式
            universal_newlines=True
        ) as proc:
            # 实时读取并打印stdout
            for line in proc.stdout:
                stripped_line = line.rstrip('\n')
                stdout_lines.append(stripped_line)
                logger.log(live_output_log_level, stripped_line)
            
            # 实时读取并打印stderr
            for line in proc.stderr:
                stripped_line = line.rstrip('\n')
                stderr_lines.append(stripped_line)
                logger.log(live_output_log_level, stripped_line)
            
            # 等待命令执行完成，获取返回码
            proc.wait()
        
        stdout_str = '\n'.join(stdout_lines)
        stderr_str = '\n'.join(stderr_lines)
        returncode = proc.returncode
    else:
        # 无需实时输出，用原方式静默捕获
        proc = subprocess.run(
            command_line,
            shell=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True
        )
        stdout_str = proc.stdout.strip()
        stderr_str = proc.stderr.strip()
        returncode = proc.returncode

    # 衔接你原来的错误处理逻辑
    if returncode != 0:
        logger.error(f"命令执行失败，返回码: {returncode}")
        logger.error(f"错误输出: {stderr_str}")
        # 可根据需求抛出异常或做其他处理
        # raise RuntimeError(f"Command failed with return code {returncode}: {stderr_str}")
    
    return stdout_str, stderr_str, returncode

# 示例测试
if __name__ == "__main__":
    # 测试一个长耗时命令，比如连续ping 10次
    test_cmd = "ping -c 10 google.com"
    # 日志级别为DEBUG时实时显示输出；设为INFO时仅静默捕获
    stdout, stderr, rc = run_command(test_cmd)
    print("\n=== 最终捕获的完整输出 ===")
    print(f"STDOUT:\n{stdout}")
    print(f"STDERR:\n{stderr}")

关键细节说明 #

• 日志级别判断：用logger.isEnabledFor()动态检查当前日志级别，生产环境设为INFO时自动关闭实时输出，调试时切到DEBUG即可开启，非常灵活。
• 实时输出保障：设置bufsize=1开启行缓冲，配合text=True，确保子进程的每一行输出都能立即被读取和打印，不会因为缓冲区积压延迟显示。
• 输出同步处理：读取每一行输出时，既添加到列表保存完整内容，又用对应日志级别打印，实现“实时查看+事后追溯”两不误。
• 兼容性：代码适配Python 3.7+，text=True是universal_newlines=True的直观别名，可读性更强。

测试方式 #

1. 将日志级别改为logging.DEBUG，运行代码，你会看到ping的每一行输出实时打印在控制台。
2. 改回logging.INFO，运行代码，控制台只会显示错误信息（如果有），不会实时输出ping的过程，但完整输出依然会被捕获。

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