合并DataFrame字典匹配函数及解析应用给定伪代码的技术咨询

问题背景 #

我正在开展一个项目，需将DataFrame（df）的值与多个参考字典的标准值交叉匹配。现有示例数据、两个参考字典（lang_dict、states_dict）和两个逻辑重复的匹配函数（lang、states），希望合并为单个函数。我已获得一段伪代码但无法理解其原理，现需理解该伪代码，并将其应用到现有场景，通过parsing_map调用函数处理对应df列。

示例数据 #

import pandas as pd
import numpy as np

data= [["English",np.nan,"smith","Alabama","","","","manufacturing","Access","Commercial","","",""]]

df= pd.DataFrame(data,columns=['Communication_Language__c','firstName', 'lastName', 'state', 'country', 'company', 'email', 'industry', 'System_Type__c', 'AccountType', 'customerSegment', 'Existing_Customer__c', 'GDPR_Email_Permission__c'])

error_dict={}

参考字典 #

lang_dict= {"English":"ENG", "French":"FR"}
states_dict= {"Alabama": "AL", "Alaska": "AK"}

现有匹配函数 #

def lang(df_lang, column="lang", lang_dict=lang_dict):
    for k, v in lang_dict.items():
       df_lang[column] = df[column].replace(k,v)
    return df_lang

def states(df_states, column="state", states_dict=states_dict):
    for i,j in states_dict.items():
        df_states[column]=df[column].replace(i,j)
    return df_states

parsing_map={
"state":[states],
"Communication_Language__c": [lang]
}

给定伪代码 #

def cross_reference(dataframe, error_dict, column_name, reference_dict_path = None):
    if reference_dict_path == None:
       reference_dict = Load_Json(column_name + '.json')
    else:
        reference_dict = Load_Json(reference_dict_path)
        relevent_column = dataframe[column_name]

#  SOME OPERATION ON dataframe AND error_dict
    return dataframe, error_dict

parsing_map={
"state":[states],
"Communication_Language__c": [lang]
}

for i in parsing_map.keys():
   df, error_dict = parsing_map[i](df, error_dict, i)

伪代码原理解析 #

这段伪代码的核心是抽离重复逻辑，实现通用化的字典匹配处理：

• cross_reference是一个通用函数，负责处理任意列的字典匹配需求，避免重复编写lang、states这类逻辑一致的函数
• 字典加载逻辑：支持两种方式——按列名自动加载对应JSON，或指定路径加载JSON（你的场景里可以直接传入内存中的字典，不需要JSON加载）
• 核心操作（伪代码中的SOME OPERATION）：用参考字典替换目标列的原始值，同时可以将未匹配到标准值的记录存入error_dict，方便后续排查
• 最后通过遍历parsing_map，批量为每个列调用处理函数，实现高效的批量数据标准化

适配现有场景的实现代码 #

1. 合并为通用匹配函数 #

替换原来的lang、states函数，实现一个支持错误记录的通用函数：

def cross_reference(dataframe, error_dict, column_name, reference_dict):
    # 获取目标列数据
    col_data = dataframe[column_name]
    # 用参考字典批量替换值（比循环更高效）
    dataframe[column_name] = col_data.replace(reference_dict)
    # 记录未匹配到标准值的非空内容到error_dict
    unmatched_values = col_data[~col_data.isin(reference_dict.keys()) & ~col_data.isna()]
    if not unmatched_values.empty:
        error_dict[column_name] = unmatched_values.unique().tolist()
    return dataframe, error_dict

2. 更新parsing_map #

调整parsing_map，将列名与通用函数、对应参考字典绑定：

parsing_map = {
    "state": [cross_reference, states_dict],
    "Communication_Language__c": [cross_reference, lang_dict]
}

3. 批量处理所有目标列 #

遍历parsing_map，调用通用函数完成所有列的匹配替换：

for col_name in parsing_map.keys():
    process_func, ref_dict = parsing_map[col_name]
    df, error_dict = process_func(df, error_dict, col_name, ref_dict)

4. 验证结果 #

处理后查看DataFrame和错误字典：

# 查看处理后的目标列
print(df[["Communication_Language__c", "state"]])
# 输出：
#   Communication_Language__c state
# 0                        ENG    AL

# 查看错误记录（示例数据无未匹配值，所以为空）
print(error_dict)
# 输出：{}

5. 测试未匹配场景 #

如果DataFrame中存在不在参考字典的值，会被自动记录到error_dict：

# 模拟未匹配的state值
df.loc[0, "state"] = "Texas"
# 重新处理该列
df, error_dict = cross_reference(df, error_dict, "state", states_dict)

print(error_dict)
# 输出：{'state': ['Texas']}

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