如何在Python中高效合并有序且可能重叠的文本文件？

文件内容示例 #

第一个文件（file1.txt） #

2024-01-26 09:00:00, Alice, 4, Win
2024-01-27 09:00:00, Bob, 6, Loss
2024-01-28 09:00:00, Charlie, 2, Loss
2024-01-29 09:00:00, Denise, 3, Loss

第二个文件（file2.txt） #

2024-01-29 09:00:00, Denise, 3, Loss
2024-01-30 09:00:00, Eve, 3, Win
2024-01-31 09:00:00, Frank, 15, Win

期望合并结果 #

2024-01-26 09:00:00, Alice, 4, Win
2024-01-27 09:00:00, Bob, 6, Loss
2024-01-28 09:00:00, Charlie, 2, Loss
2024-01-29 09:00:00, Denise, 3, Loss
2024-01-30 09:00:00, Eve, 3, Win
2024-01-31 09:00:00, Frank, 15, Win

数据特性 #

• 两个文件的数据均按时间顺序排列
• 可能存在重叠，且重叠部分数据完全一致，需删除重复内容
• 文件体积较大，效率优先
• 必须保持输出的时间顺序

（本质是合并同一份时序数据的两个片段，比如1-3月和3-6月合并为1-6月；也可能无重叠，比如1-3月和6-8月）

已尝试/考虑的方法 #

• 直接追加后删重复行：效率低，因为重复仅出现在第一个文件末尾和第二个文件开头，无需遍历整个文件
• 使用set去重：不合适，会破坏原有顺序，且排序后效率低下
• 人工逻辑：取第一个文件最后一行，判断是否晚于第二个文件首行，若否则跳过第二个文件中早于该时间的行，但提取和转换时间实现复杂
• 使用shutil.copyfileobj：仅能直接追加，无法处理重叠

最优实现方案 #

利用两个文件按时间有序、重叠仅出现在首尾的特性，采用流式处理方案，无需加载整个文件到内存，效率拉满：

基础版（基于完整行匹配） #

适用于重叠行内容完全一致的场景，直接比对整行内容：

def merge_ordered_files(file1_path, file2_path, output_path):
    # 写入第一个文件并记录最后一行
    last_line = None
    with open(file1_path, 'r', encoding='utf-8') as f1, open(output_path, 'w', encoding='utf-8') as out:
        for line in f1:
            out.write(line)
            last_line = line.strip()
    
    # 处理第二个文件，跳过开头重复行
    if not last_line:
        # 第一个文件为空，直接写入第二个文件全部内容
        with open(file2_path, 'r', encoding='utf-8') as f2, open(output_path, 'a', encoding='utf-8') as out:
            for line in f2:
                out.write(line)
        return
    
    with open(file2_path, 'r', encoding='utf-8') as f2, open(output_path, 'a', encoding='utf-8') as out:
        skip_duplicates = True
        for line in f2:
            current_line = line.strip()
            if skip_duplicates:
                if current_line == last_line:
                    continue
                skip_duplicates = False
            out.write(line)

通用版（基于时间戳匹配） #

如果需要仅比对时间戳（比如重叠行其他字段可能有差异，但时间在重叠范围内），可以提取时间部分判断：

def extract_timestamp(line):
    # 提取行中第一个逗号前的时间戳
    return line.split(',')[0].strip()

def merge_ordered_files_by_timestamp(file1_path, file2_path, output_path):
    last_timestamp = None
    with open(file1_path, 'r', encoding='utf-8') as f1, open(output_path, 'w', encoding='utf-8') as out:
        for line in f1:
            out.write(line)
            last_timestamp = extract_timestamp(line)
    
    if not last_timestamp:
        with open(file2_path, 'r', encoding='utf-8') as f2, open(output_path, 'a', encoding='utf-8') as out:
            for line in f2:
                out.write(line)
        return
    
    with open(file2_path, 'r', encoding='utf-8') as f2, open(output_path, 'a', encoding='utf-8') as out:
        skip_older = True
        for line in f2:
            current_timestamp = extract_timestamp(line)
            if skip_older:
                if current_timestamp <= last_timestamp:
                    continue
                skip_older = False
            out.write(line)

方案优势 #

• 内存高效：流式逐行处理，大文件也不会占满内存
• 时间高效：仅遍历两个文件各一次，去重逻辑仅在第二个文件开头执行
• 逻辑简洁：无需复杂的时间解析或排序，完全贴合数据特性

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