Python实现两程序修改后文件名的高效匹配方案

两个独立程序处理同一批输入文件，输出文件名均基于原始文件名生成，忽略文件路径与扩展名的前提下，命名规则如下：

• 程序1：仅在原始文件名末尾追加符合(_[[:alnum:]]+)*规则的内容，即文件名结构为原始文件名 + 后缀
• 程序2：在原始文件名末尾追加符合(_[[:alnum:]]+)*规则的后缀，同时在文件名开头追加符合([[:alnum:]]+_)*规则的前缀，即文件名结构为前缀 + 原始文件名 + 后缀

现有来自程序2的千万级文件路径（去重后约1万条唯一项），需要将每条路径唯一匹配到程序1生成的对应文件路径（程序1共约1万条唯一文件路径），要求实现高效。

已实现的基础代码如下：

#!/usr/bin/env python3
import os

##### DATA #####

program1_files = [
    # Around 10 thousand of unique items (parsed from a file).
    'path1/jobxxx/n1_file_001_xxx.tiff',
    'path1/jobxxx/n1_file_002_yyy_xxx.tiff',
    'path1/jobxxx/n2_file_001_yyy.tiff'
]

program2_files = [
    # Around 10 millions of items (parsed from a file),
    # with only around 10 thousand of unique items
    # IMO the items should be directly processed/matched while parsing the file
    'path2/JXX/00001_n1_file_001_XXX_yyy_zz.mrc',
    'path2/JXX/00001_n1_file_001_XXX_yyy_zz.mrc',
    'path2/JXX/00002_n1_file_002_XXX_yyy_zz.mrc',
    'path2/jXX/00003_n2_file_001_XXX_yyy_zz.mrc',
    'path2/JZZ/00101_YYY_n1_file_001_xx_zzz.mrc',
    'path2/JZZ/00102_YYY_n2_file_001_xx_zzz.mrc'
]

################

def get_parts(str):
    return os.path.splitext(os.path.basename(str))[0].split('_')

program1_fileparts = dict(zip(program1_files, map(get_parts,program1_files)))

file2_to_file1 = {}

for file2 in program2_files:

    if file2 in file2_to_file1: continue   # skip already processed file2

    file2_parts = get_parts(file2)

    for file1 in program1_files:

        file1_parts = program1_fileparts[file1]

        # here I'm a little lost on what to do

原代码采用双重循环暴力匹配的时间复杂度为O(n*m)（n、m均为1万量级时约1亿次循环，Python执行效率偏低），可以用**前缀字典树（Trie）**优化，将时间复杂度降到线性级别，全程仅需数万次操作，千万级数据也能秒级处理。

核心逻辑 #

两个程序生成的文件名仅公共部分为原始文件名，按_拆分后：

1. 程序1的文件名拆分段 = 原始文件名拆分段 + 后缀拆分段
2. 程序2的文件名拆分段 = 前缀拆分段 + 原始文件名拆分段 + 后缀拆分段

匹配时只需找到：程序2拆分段中存在一个连续子段，是某个程序1拆分段的前缀，且这个公共前缀长度最长；若公共前缀长度相同，优先选择公共前缀等于程序1整个拆分段的结果（对应原始文件名完全匹配程序1文件名的场景）。

优化步骤 #

1. 预处理程序1文件：将所有程序1文件名拆分后统一转小写（适配大小写不敏感场景，若大小写敏感可省略），构建前缀字典树，每个节点记录当前路径下的最优匹配文件和对应得分。
2. 逐行读取程序2文件列表，遇到已匹配的路径直接跳过（利用千万级数据重复率高的特点减少计算），对每个唯一的程序2文件拆分后，在字典树中遍历所有可能的前缀起始位置，记录得分最高的匹配结果即可。

完整实现代码 #

#!/usr/bin/env python3
import os
from typing import Dict, Tuple, Optional

def get_parts(file_path: str) -> list[str]:
    """提取文件名（无路径无扩展名）并按下划线拆分，统一转小写实现大小写不敏感匹配"""
    basename = os.path.basename(file_path)
    name_without_ext = os.path.splitext(basename)[0]
    return [seg.lower() for seg in name_without_ext.split('_')]

def build_trie(program1_files: list[str]) -> Dict:
    """构建程序1文件名的前缀字典树"""
    # 根节点结构：{'__best__': (匹配的文件路径, 得分), '段名': 子节点}
    root: Dict = {'__best__': (None, 0)}
    for f1 in program1_files:
        parts = get_parts(f1)
        f1_len = len(parts)
        current_node = root
        for depth, seg in enumerate(parts, start=1):
            if seg not in current_node:
                current_node[seg] = {'__best__': (None, 0)}
            current_node = current_node[seg]
            # 得分规则：长度权重*2，完全匹配整个f1段额外加1，保证长匹配优先、等长下完全匹配优先
            score = depth * 2 + (1 if depth == f1_len else 0)
            if score > current_node['__best__'][1]:
                current_node['__best__'] = (f1, score)
    return root

def match_files(program1_files: list[str], program2_files_iter) -> Dict[str, str]:
    """
    匹配程序2文件到程序1文件
    program2_files_iter 支持传入可迭代对象（比如文件句柄），逐行读取避免一次性加载千万级数据到内存
    """
    trie = build_trie(program1_files)
    file2_to_file1: Dict[str, str] = {}
    
    for f2 in program2_files_iter:
        f2 = f2.strip()  # 去掉行尾换行符
        if not f2 or f2 in file2_to_file1:
            continue  # 跳过空行和已处理的重复文件
        
        parts2 = get_parts(f2)
        n = len(parts2)
        best_f1: Optional[str] = None
        best_score: int = 0
        
        for start in range(n):
            # 剩余段最多能拿到的得分，如果低于当前最佳得分直接提前终止遍历
            max_possible_score = 2 * (n - start) + 1
            if max_possible_score <= best_score:
                break
            
            current_node = trie
            for i in range(start, n):
                seg = parts2[i]
                if seg not in current_node:
                    break
                current_node = current_node[seg]
                candidate_f1, candidate_score = current_node['__best__']
                if candidate_score > best_score:
                    best_score = candidate_score
                    best_f1 = candidate_f1
        
        if best_f1 is not None:
            file2_to_file1[f2] = best_f1
    
    return file2_to_file1

# 测试
if __name__ == '__main__':
    # 实际使用时将program2_files换成open('program2_list.txt', 'r')即可，无需一次性加载所有内容
    result = match_files(program1_files, program2_files)
    for f2, f1 in result.items():
        print(f"{f2} -> {f1}")

测试输出 #

path2/JXX/00001_n1_file_001_XXX_yyy_zz.mrc -> path1/jobxxx/n1_file_001_xxx.tiff
path2/JXX/00002_n1_file_002_XXX_yyy_zz.mrc -> path1/jobxxx/n1_file_002_yyy_xxx.tiff
path2/jXX/00003_n2_file_001_XXX_yyy_zz.mrc -> path1/jobxxx/n2_file_001_yyy.tiff
path2/JZZ/00101_YYY_n1_file_001_xx_zzz.mrc -> path1/jobxxx/n1_file_001_xxx.tiff
path2/JZZ/00102_YYY_n2_file_001_xx_zzz.mrc -> path1/jobxxx/n2_file_001_yyy.tiff

完全符合预期匹配结果。

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