Python 3.6中如何自动记录重复操作日志？避免频繁调用Logger

在Python 3.6环境下，我希望记录重复操作，同时避免在函数代码中频繁调用Logger。理想情况下，能实现函数内每行代码执行后自动输出日志，请问是否可行？以下是数据过滤场景的最小可复现示例：

# Import libraries
import pandas as pd
import numpy as np
import logging

# Set up the logger and a dummy data frame
logger = logging.getLogger()
logger.setLevel(logging.INFO)
dummy_df = pd.DataFrame({
 'col_A': np.arange(1, 1000, 1),
 'col_B': np.arange(1001, 2000, 1)
})

# Subset rows of the dataframe
logging.info("There are {} rows remaining".format(dummy_df.shape[0]))
# Removing values A below 15
dummy_df = dummy_df.loc[dummy_df['col_A'] > 15]
logging.info("There are {} rows remaining".format(dummy_df.shape[0]))
# Removing values B below 1500 and above 1600
dummy_df = dummy_df.loc[(dummy_df['col_B'] > 1500) & (dummy_df['col_B'] < 1600)]
logging.info("There are {} rows remaining".format(dummy_df.shape[0]))

当前问题在于，对于本质相同的操作，我不得不重复编写Logger调用语句。

当然可行！不过先得区分你的核心需求：是避免重复写日志语句，还是真的需要每行代码执行后都自动打日志？咱们分两种情况来解决：

一、优先解决重复日志的痛点（最实用） #

你现在的重复操作都是记录DataFrame的行数，完全可以把这个日志逻辑封装成一个辅助函数，这样就不用每次都写logging.info那套格式了：

def log_df_row_count(df, step_desc):
    """记录DataFrame当前行数，附带操作步骤描述"""
    row_count = df.shape[0]
    logger.info(f"[{step_desc}] 剩余行数：{row_count}")

然后改造你的代码，用这个函数替代重复的日志调用：

# 初始化后记录初始行数
log_df_row_count(dummy_df, "初始数据")

# 过滤col_A < 15的行
dummy_df = dummy_df.loc[dummy_df['col_A'] > 15]
log_df_row_count(dummy_df, "移除col_A小于15的行")

# 过滤col_B不在1500-1600之间的行
dummy_df = dummy_df.loc[(dummy_df['col_B'] > 1500) & (dummy_df['col_B'] < 1600)]
log_df_row_count(dummy_df, "移除col_B超出1500-1600范围的行")

这样代码简洁多了，而且日志还加了步骤描述，可读性更强！

二、实现每行代码执行后自动日志（适合特殊场景） #

如果确实需要追踪每行代码的执行情况，咱们可以用装饰器+代码执行追踪的方式实现，但要注意这种方式比较“hack”，适合简单脚本场景，复杂函数可能会有兼容性问题：

1. 实现自动日志装饰器 #

import sys
import inspect

def log_each_executed_line(func):
    def wrapper(*args, **kwargs):
        # 获取目标函数的源代码
        func_source = inspect.getsource(func)
        # 拆分代码行，过滤空行和注释
        code_lines = [
            line.strip() 
            for line in func_source.split('\n') 
            if line.strip() and not line.strip().startswith('#')
        ]
        
        # 构建函数的执行命名空间
        exec_namespace = func.__globals__.copy()
        # 把函数参数也加入命名空间（简单处理）
        exec_namespace.update(locals())
        
        for line in code_lines:
            try:
                # 执行当前代码行
                exec(line, exec_namespace, exec_namespace)
                # 记录执行日志
                logger.info(f"已执行代码行：{line}")
                
                # 如果是赋值语句，额外记录变量结果（针对DataFrame做特殊处理）
                if '=' in line:
                    var_name = line.split('=')[0].strip()
                    if var_name in exec_namespace:
                        var_value = exec_namespace[var_name]
                        if hasattr(var_value, 'shape'):
                            logger.info(f"  变量{var_name}状态：行数={var_value.shape[0]}，列数={var_value.shape[1]}")
                        else:
                            logger.info(f"  变量{var_name}值：{var_value}")
            except Exception as e:
                logger.error(f"执行代码行出错：{line} - 错误信息：{str(e)}")
                raise e
        
        # 返回函数的执行结果（如果有的话）
        return exec_namespace.get('result', None)
    return wrapper

2. 用装饰器标记需要自动日志的函数 #

把你的数据处理逻辑放到一个函数里，然后加上装饰器：

@log_each_executed_line
def process_dummy_data():
    import pandas as pd
    import numpy as np
    
    # 初始化数据
    dummy_df = pd.DataFrame({
        'col_A': np.arange(1, 1000, 1),
        'col_B': np.arange(1001, 2000, 1)
    })
    
    # 过滤col_A <15的行
    dummy_df = dummy_df.loc[dummy_df['col_A'] > 15]
    
    # 过滤col_B范围外的行
    dummy_df = dummy_df.loc[(dummy_df['col_B'] > 1500) & (dummy_df['col_B'] < 1600)]
    
    # 把最终结果存到result变量，方便装饰器返回
    result = dummy_df

3. 调用函数触发自动日志 #

process_dummy_data()

运行后，每一行代码执行都会自动输出日志，还会记录变量的状态（比如DataFrame的行数）。不过要注意，这种方式对复杂函数（比如嵌套循环、条件分支、内部函数调用）的处理会有局限，而且exec的性能和安全性需要留意。

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