如何通过logging模块记录外部进程进度条的最终输出

问题描述 #

编写Python脚本实现两个功能：

• 执行应用程序mcu_flash.exe
• 使用logging模块将程序输出同时记录到日志文件和终端

当前脚本已实现基本功能，但mcu_flash.exe的进度条输出导致日志文件中产生大量冗余行，示例如下：

......
[2023-01-19 17:51:11,235] - [INFO] : Start flash code:  
[2023-01-19 17:52:54,818] - [INFO] : [                                                  ][0%]
[                                                  ][0%]
[                                                  ][1%]
[                                                  ][1%]
......

而终端输出正常（进度条在同一行更新）。需要修改脚本，让日志文件仅记录进度条的最终状态，且保持一行展示。

原代码如下：

import os
import subprocess
import logging
import sys

def ToDoBat(dir, command):
    os.environ["PYTHONUNBUFFERED"] = "1"
    p = subprocess.Popen(command, shell=True, cwd=dir, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    curline = p.stdout.readline()
    while (curline != b''):
        try:
            curline = curline.decode(encoding='utf-8')
        except UnicodeDecodeError:
            try:
                curline = curline.decode(encoding='gbk')
            except UnicodeDecodeError:
                curline = "[Error] can't decode the command line"
        logging.info(curline.strip("\r\n"))
        curline = p.stdout.readline()
    p.wait()
    return p.returncode

def main():
    logging.basicConfig(level=logging.DEBUG,
                        format="[%(asctime)s] - [%(levelname)s] : %(message)s",
                        encoding='uft-8',
                        handlers=[
                            logging.FileHandler(
                                filename=os.path.join(os.path.dirname(os.path.abspath(__file__)), "log.log"),
                                mode='a',
                                encoding='utf-8'),
                            logging.StreamHandler(sys.stdout)
                        ])
    ToDoBat(os.path.join(os.sep, os.path.dirname(os.path.abspath(__file__)), "mcu_flash.exe"))

if __name__ == "__main__":
    main()

解决方案 #

进度条输出通常通过\r（回车符）覆盖当前行实现，而非换行。我们可以通过缓存进度条中间内容，仅在遇到非进度条行或进程结束时记录最终的进度条状态；同时保持终端的实时进度更新。

修改后的代码 #

import os
import subprocess
import logging
import sys
import re

# 匹配进度条行的正则表达式，适配[====...][xx%]格式
PROGRESS_PATTERN = re.compile(r'^\[.*\]\[\d+%\]$')

def ToDoBat(dir, command):
    os.environ["PYTHONUNBUFFERED"] = "1"
    # 使用text=True自动处理解码，简化编码逻辑
    p = subprocess.Popen(command, shell=True, cwd=dir, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
    progress_buffer = ""
    
    for line in p.stdout:
        # 移除换行/回车符，避免空行干扰
        line = line.rstrip("\r\n")
        if not line:
            continue
        
        # 判断是否为进度条行
        if PROGRESS_PATTERN.match(line):
            progress_buffer = line
            # 终端实时更新进度条（覆盖当前行）
            print(f"[INFO] : {progress_buffer}", end="\r", file=sys.stdout)
            sys.stdout.flush()
        else:
            # 先输出缓存的最终进度条（如果存在）
            if progress_buffer:
                logging.info(progress_buffer)
                progress_buffer = ""
            # 记录非进度条内容
            logging.info(line)
    
    # 进程结束后，输出最后缓存的进度条
    if progress_buffer:
        logging.info(progress_buffer)
        # 终端输出最终进度条并换行，避免覆盖后续内容
        print(f"[INFO] : {progress_buffer}", file=sys.stdout)
    
    p.wait()
    return p.returncode

def main():
    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)
    
    # 文件日志处理器：记录完整内容（含最终进度条）
    file_handler = logging.FileHandler(
        filename=os.path.join(os.path.dirname(os.path.abspath(__file__)), "log.log"),
        mode='a',
        encoding='utf-8'
    )
    file_formatter = logging.Formatter("[%(asctime)s] - [%(levelname)s] : %(message)s")
    file_handler.setFormatter(file_formatter)
    logger.addHandler(file_handler)
    
    # 终端日志处理器：输出非进度条的日志内容
    stream_handler = logging.StreamHandler(sys.stdout)
    stream_formatter = logging.Formatter("[%(asctime)s] - [%(levelname)s] : %(message)s")
    stream_handler.setFormatter(stream_formatter)
    logger.addHandler(stream_handler)
    
    # 修正函数调用参数错误
    dir_path = os.path.dirname(os.path.abspath(__file__))
    command = os.path.join(dir_path, "mcu_flash.exe")
    ToDoBat(dir_path, command)

if __name__ == "__main__":
    main()

关键修改说明 #

1. 正则识别进度条：用PROGRESS_PATTERN精准匹配进度条格式，确保只缓存符合要求的内容。
2. 缓存与输出分离：
   • 进度条行仅更新缓存，终端通过print(end="\r")实时覆盖展示。
   • 遇到非进度条行时，先将缓存的最终进度条写入日志，再处理当前行。
   • 进程结束后强制写入最后缓存的进度条，避免遗漏。
3. 代码优化：
   • 使用text=True自动处理子进程输出解码，简化编码异常处理。
   • 修复原代码中encoding拼写错误、函数调用参数缺失等问题。

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