如何为Pandas DataFrame应用自定义函数映射处理失效场景

以下是可直接运行的完整实现代码，符合你的需求：

import pandas as pd

# 初始化原始数据
df = pd.DataFrame({
    'spid': ['AB01', 'AB02', 'AB03', 'AB04'],
    'value': [15,21,1758,12],
    'target': ['fan','fan','set','lan'],
    'test': ['Yes', 'No', 'No', 'Yes']
})

mapper = pd.DataFrame({
    'check': ["row['target'] == 'fan'","row['target'] == 'set'"],
    'message' : ['fan err message for [spid]','set err message for [spid]'],
    'to_change' : [('test','Nope'),('test','Save1')]
})

# 1. 初始化空的错误信息表
err_msg = pd.DataFrame(columns=['spid', 'message'])
# 提前将规则转换为字典列表，提升遍历效率
rules = mapper.to_dict('records')

# 2. 逐行匹配规则并执行操作
for idx, row in df.iterrows():
    for rule in rules:
        # 校验是否命中规则（eval仅适用于规则为内部可控的场景）
        if eval(rule['check']):
            # 生成错误记录，替换占位符
            msg = rule['message'].replace('[spid]', row['spid'])
            err_msg.loc[len(err_msg)] = [row['spid'], msg]
            # 修改原df对应字段的值
            col, new_val = rule['to_change']
            df.loc[idx, col] = new_val
            # 若要求每行仅命中第一个匹配规则，可在此处加break跳出规则循环

# 输出结果
print("处理后的df：")
print(df)
print("\n错误信息表err_msg：")
print(err_msg)

运行结果 #

处理后的df：

错误信息表err_msg：

注意事项 #

• 如果规则来自外部不可信来源，不要使用eval执行字符串表达式，避免代码注入风险，可提前把规则转换为lambda函数存储在mapper中替代字符串格式的check条件。
• 若数据量超过10万行，不推荐用iterrows遍历，可将规则拆解为向量化操作批量处理，提升运行效率。

内容的提问来源于stack exchange，提问作者ListenSoftware Louise Ai Agent