如何用Python subprocess过滤并记录精简的错误回溯信息？

问题 #

编写自动化程序时，通过Python subprocess运行子进程，需要捕获并收集子进程抛出的有效错误回溯（traceback），但当前日志混入了进度输出等冗余内容。现有try-except逻辑会记录所有stderr内容，需要过滤出目标错误回溯并记录。

冗余日志示例 #

2024-03-21 10:14:44,843 - ERROR - error processing shot '20221201.033':
Traceback (most recent call last):
 File "/home/Desktop/Code/git/populate_database.py", line 76, in <module>
    raise Exception(f"n{stderr.decode()}")
Exception: n
 Fetching parlog information for shot 20221201.033 from archive: 0%|    | 0/5 [00:00<?, ?it/s]
 Fetching parlog information for shot 20221201.033 from archive: 0%|    | 0/5 [00:00<?, ?it/s]

     Mapping rates : 0%|    | 0/10 [00:00<?, ?it/s]
     Mapping rates : 100%|██████████| 10/10 [00:00<00:00, 3360.28it/s]

     Calculating impurity densities : 0%|   | 0/10 [00:00<?, ?it/s]
     Calculating impurity densities : 80%|████████ | 8/10 [00:00<00:00, 73.92it/s]
     Calculating impurity densities : 100%|██████████| 10/10 [00:00<00:00, 73.91it/s]

     Updating profile: 0%|      | 0/10 [00:00<?, ?it/s]
     Updating profile: 20%|██   | 2/10 [00:00<00:00, 8.86it/s]
     Updating profile: 100%|██████████| 10/10 [00:00<00:00, 34.81it/s]
     Updating profile: 100%|██████████| 10/10 [00:00<00:00, 29.59it/s]

         Assessing convergence: 0%|     | 0/10 [00:00<?, ?it/s]
         Assessing convergence: 100%|██████████| 10/10 [00:00<00:00, 84054.19it/s]
Traceback (most recent call last):
 File "/home/Desktop/Code/git/Examples/Program_runfiles/20221201.033.py", line 37, in <module>
    data.run(
 File "/home/Desktop/Code/git/eval.py", line 348, in run
    self.run_post_methods()
 File "/home/Desktop/Code/git/eval.py", line 469, in run_post_methods
    _store_grad_length(
 File "/home/Desktop/Code/git/utilities/write_locally.py", line 103, in _store_grad_length
    dictionary_added["tau_e"] = _get_energy_confinement_time(
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 File "/home/Desktop/Code/git/utilities/fetch_data.py", line 293, in _get_energy_confinement_time
    w_dia_spline = UnivariateSpline(t_w_dia, w_dia, k=4, s=0)
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 File "/home/anaconda3/lib/python3.11/site-packages/scipy/interpolate/_fitpack2.py", line 236, in __init__
    data = dfitpack.fpcurf0(x, y, k, w=w, xb=bbox[0],
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
dfitpack.error: (m>k) failed for hidden m: fpcurf0:m=0

需保留的目标错误回溯 #

Traceback (most recent call last):
 File "/home/Desktop/Code/git/Examples/Program_runfiles/20221201.033.py", line 37, in <module>
    data.run(
 File "/home/Desktop/Code/git/eval.py", line 348, in run
    self.run_post_methods()
 File "/home/Desktop/Code/git/eval.py", line 469, in run_post_methods
    _store_grad_length(
 File "/home/Desktop/Code/git/utilities/write_locally.py", line 103, in _store_grad_length
    dictionary_added["tau_e"] = _get_energy_confinement_time(
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 File "/home/Desktop/Code/git/utilities/fetch_data.py", line 293, in _get_energy_confinement_time
    w_dia_spline = UnivariateSpline(t_w_dia, w_dia, k=4, s=0)
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 File "/home/anaconda3/lib/python3.11/site-packages/scipy/interpolate/_fitpack2.py", line 236, in __init__
    data = dfitpack.fpcurf0(x, y, k, w=w, xb=bbox[0],
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
dfitpack.error: (m>k) failed for hidden m: fpcurf0:m=0

现有代码 #

from subprocess import Popen, PIPE, STDOUT, DEVNULL

logging.basicConfig(filename='error.log', level=logging.ERROR, format='%(asctime)s - %(levelname)s - %(message)s')

try:
    # run the file for the shot with error handling
    process = Popen(["python", dest_file], stdout=DEVNULL, stderr=PIPE)

    # wait while file is running
    process.wait()
     
    # get the console error of the subprocess
    stderr = process.communicate()[1]

    # close all figures and clear console
    plt.close()
    os.system('clear')
     
    # if we encountered an error raise it
    if stderr:
        raise Exception(f"n{stderr.decode()}")
 
except Exception as e:
    # Log the shot name along with the traceback info
    logging.error(f"error processing shot '{shot_name}':", exc_info=True)

解决方案 #

核心逻辑是从子进程的stderr输出中，提取最后一段以Traceback (most recent call last):开头的错误回溯内容，过滤掉前面的冗余进度输出。修改后的代码如下：

from subprocess import Popen, PIPE, DEVNULL
import logging
import os
import matplotlib.pyplot as plt

logging.basicConfig(filename='error.log', level=logging.ERROR, format='%(asctime)s - %(levelname)s - %(message)s')

try:
    # 运行子进程并获取输出
    process = Popen(["python", dest_file], stdout=DEVNULL, stderr=PIPE)
    _, stderr = process.communicate()

    plt.close()
    os.system('clear')
     
    if stderr:
        stderr_str = stderr.decode()
        lines = stderr_str.splitlines()
        # 定位所有Traceback起始行的索引
        traceback_indices = [i for i, line in enumerate(lines) if line.startswith('Traceback (most recent call last):')]
        
        if traceback_indices:
            # 提取最后一段完整的Traceback内容
            start_idx = traceback_indices[-1]
            filtered_content = '\n'.join(lines[start_idx:])
            raise Exception(filtered_content)
        else:
            # 未找到标准Traceback时保留原始内容
            raise Exception(stderr_str)
 
except Exception as e:
    # 仅记录过滤后的错误内容，避免混入父进程的堆栈信息
    logging.error(f"error processing shot '{shot_name}':\n{str(e)}")

关键修改说明 #

• 直接使用process.communicate()获取输出，替代wait()+communicate()的冗余操作
• 分割stderr内容为行列表，定位所有Traceback起始位置，取最后一段（子进程的真实错误回溯通常在输出末尾）
• 日志记录时直接输出过滤后的错误内容，关闭exc_info=True避免混入父进程的错误堆栈
• 增加降级处理：未找到标准Traceback时保留原始内容，防止遗漏非格式错误信息

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