使用Pexpect和Logging为Python子进程输出添加时间戳的实践指南

import logging
import pexpect
import sys

# 配置日志系统
# 日志将同时输出到文件 'subprocess_output.log' 和控制台
# 文件模式设置为 'a' (append)，表示追加写入
# format 定义了日志的输出格式：时间戳、日志级别、消息
# level=logging.INFO 设置最低记录级别为INFO，即只记录INFO及以上级别的信息
# encoding='utf-8' 确保日志文件和控制台输出的编码正确，避免乱码
logging.basicConfig(
    filename='subprocess_output.log',  # 日志文件路径
    filemode='a',                      # 追加模式写入文件
    format='%(asctime)s %(levelname)-8s %(message)s',
    level=logging.INFO,
    encoding='utf-8'
)

# 添加一个StreamHandler，将日志同时输出到控制台
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)-8s %(message)s'))
logging.getLogger().addHandler(console_handler)

def run_command_with_timestamp_logging(command: str):
    """
    运行指定的命令行命令，并使用logging模块为每行输出添加时间戳。

    Args:
        command (str): 要执行的命令行字符串。
    """
    logging.info(f"--- 开始执行命令: {command} ---")
    try:
        # 使用 pexpect.spawn 启动命令
        # encoding="utf-8" 确保正确处理各种字符编码
        # timeout=None 表示不设置超时，等待命令自然完成
        p = pexpect.spawn(command, encoding="utf-8", timeout=None)

        # 逐行读取子进程的输出
        # p.readline() 会阻塞直到读取到一行或EOF
        while True:
            try:
                line = p.readline()
                if not line:  # 如果读取到空行，表示子进程已关闭其stdout，即EOF
                    break
                # 移除行末的换行符和回车符，然后记录
                logging.info(line.strip())
            except pexpect.EOF:
                # 捕获 pexpect.EOF 异常，表示子进程已退出
                break
            except pexpect.TIMEOUT:
                # 捕获 pexpect.TIMEOUT 异常，如果设置了超时且超时发生
                logging.warning(f"命令 '{command}' 执行超时。")
                break
            except Exception as e:
                logging.error(f"读取命令 '{command}' 输出时发生未知错误: {e}")
                break

        # 等待子进程结束并获取退出状态
        p.wait()
        if p.exitstatus is not None and p.exitstatus != 0:
            logging.error(f"命令 '{command}' 执行失败，退出码: {p.exitstatus}")
        else:
            logging.info(f"命令 '{command}' 执行成功，退出码: {p.exitstatus}")

    except pexpect.ExceptionPexpect as e:
        logging.critical(f"执行命令 '{command}' 时发生pexpect异常: {e}")
    except Exception as e:
        logging.critical(f"执行命令 '{command}' 时发生未知异常: {e}")
    finally:
        logging.info(f"--- 命令执行结束: {command} ---")

# 示例用法
if __name__ == "__main__":
    print("请查看 'subprocess_output.log' 文件和控制台输出。")

    # 示例 1: 列出当前目录文件
    run_command_with_timestamp_logging("ls -l")

    print("\n--- 运行另一个示例命令 (可能需要Docker环境) ---")
    # 示例 2: 模拟一个可能长时间运行的命令，如 Docker 构建
    # 注意：如果你的环境中没有docker，此命令会报错，但日志会记录错误
    run_command_with_timestamp_logging("docker build .")

    print("\n--- 运行一个会出错的示例命令 ---")
    # 示例 3: 运行一个不存在的命令，观察错误日志
    run_command_with_timestamp_logging("non_existent_command_xyz123")