Python中基于现有映射文件高效创建新映射的通用方法

核心思路 #

这个需求本质是两段关联映射的透传，纯Python环境下用哈希字典做中间映射查询是效率最高的方案，整体时间复杂度为O(M+N)（M、N分别是两份映射文件的行数），支持百万行级文件快速处理，且可以通过参数配置适配不同列位置、不同分隔符、不同缺失值填充规则的同类场景，无需重复编写逻辑。
处理规则完全匹配示例要求：

• 先加载中间键到目标值的映射（即示例中B->C的关系）
• 遍历第一份映射的A->B关系，直接查询中间映射字典，不存在对应B值时填充默认值0
• 自动处理行尾多余分隔符、空字段的异常情况

通用实现代码 #

def generate_chain_map(
    map1_path: str,
    map2_path: str,
    output_path: str,
    map1_key_col: int = 0,
    map1_relate_col: int = 1,
    map2_key_col: int = 0,
    map2_target_col: int = 1,
    default_val = 0,
    sep: str = ','
) -> None:
    """
    链式映射生成通用函数
    :param map1_path: 第一段映射文件路径（A->B）
    :param map2_path: 第二段映射文件路径（B->目标值）
    :param output_path: 结果输出路径
    :param map1_key_col: 第一段映射中起始键（A）的列索引，从0开始计数
    :param map1_relate_col: 第一段映射中关联中间键（B）的列索引，从0开始计数
    :param map2_key_col: 第二段映射中中间键（B）的列索引，从0开始计数
    :param map2_target_col: 第二段映射中目标值（C）的列索引，从0开始计数
    :param default_val: 中间键不存在时的填充默认值
    :param sep: 文件字段分隔符
    """
    # 加载中间键到目标值的映射字典，单值查询复杂度为O(1)
    b_to_c = {}
    with open(map2_path, 'r', encoding='utf-8') as f2:
        for line in f2:
            line = line.strip()
            if not line:
                continue
            parts = line.split(sep)
            # 跳过字段数不足的无效行
            if len(parts) <= max(map2_key_col, map2_target_col):
                continue
            b_val = parts[map2_key_col].strip()
            c_val = parts[map2_target_col].strip()
            if b_val:
                b_to_c[b_val] = c_val
    # 遍历第一段映射逐行生成结果
    with open(map1_path, 'r', encoding='utf-8') as f1, open(output_path, 'w', encoding='utf-8') as f_out:
        for line in f1:
            line = line.strip()
            if not line:
                continue
            parts = line.split(sep)
            if len(parts) <= max(map1_key_col, map1_relate_col):
                continue
            a_val = parts[map1_key_col].strip()
            b_val = parts[map1_relate_col].strip()
            # 查映射，不存在对应键则返回默认值
            res_c = b_to_c.get(b_val, default_val)
            f_out.write(f"{a_val}{sep}{res_c}\n")

# 示例场景调用方式
if __name__ == "__main__":
    generate_chain_map(
        map1_path="map1.txt",
        map2_path="map2.txt",
        output_path="map3.txt"
    )

示例运行结果 #

用给出的测试文件运行后，生成的map3.txt内容完全符合预期：

1,25
2,0
3,300

复用说明 #

遇到同类场景不需要修改核心逻辑，只需要调整传入参数即可：

• 如果需要提取B映射的D列（即第二份文件的第3列），只需要把map2_target_col参数设为2
• 如果文件是制表符分隔的TSV格式，把sep参数设为'\t'
• 如果缺失值需要填充'N/A'而非0，修改default_val参数为'N/A'即可
• 如果是三段、四段的长链式映射，只需要重复加载中间映射字典的逻辑，逐层查询即可

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