本文介绍一种高效、内存友好的方法，用于合并两个按时间排序的文本日志文件，在存在末尾-开头重叠时自动去重，同时严格保持原有时间顺序，避免全量读取或重复排序。
在处理分段采集的日志、传感器数据或时间序列记录（如 Jan-Mar.txt 和 Mar-Jun.txt）时，常需将两个已排序的文本文件合并为一个连续、无重复、仍保持严格时间顺序的文件。关键挑战在于：重叠只可能发生在第一个文件末尾与第二个文件开头之间，且重叠行内容完全一致；而文件体积很大，不能简单加载全部行后用 set 去重（会丢失顺序）或全量扫描去重（低效）。
上述答案中使用 OrderedDict.fromkeys(lines1 + lines2) 的方法虽简洁，但存在严重缺陷：
✅ 保留了首次出现的顺序（对 lines1 + lines2 拼接后去重）；
❌ 却无法保证时间顺序正确性——若 file2 开头几行早于 file1 结尾（即数据乱序），该方法仍会错误保留；
❌ 更严重的是，它将整个文件读入内存（list(...)），对“非常大的文件”不友好；
❌ 且未利用“两文件各自有序”这一关键前提，造成不必要的开销。
✅ 推荐方案：双指针流式合并（O(n+m) 时间，O(1) 额外内存）
我们采用类似归并排序的双指针策略，逐行读取、比较、写入，仅在重叠处跳过重复行：
QuickBooks Mileage CSV
下载
使用 qbo-mileage CLI 及用户凭证，从 Airtable、Outlook 或 Google Calendar 记录生成 QuickBooks Online 里程 CSV 文件。
def merge_sorted_files(file1_path, file2_path, output_path):
"""
合并两个按时间升序排列的CSV格式文本文件，自动跳过首尾重叠行。
假设每行以 ISO 格式时间戳开头（如 '2024-01-26 09:00:00,'），且重叠部分完全一致。
"""
with open(file1_path, 'r') as f1, \
open(file2_path, 'r') as f2, \
open(output_path, 'w') as out:
# 读取 file1 所有行（因需从末尾比对，且通常较小）
lines1 = [line.rstrip('\n\r') for line in f1]
# 逐行读取 file2，跳过与 lines1 尾部重复的行
skip_count = 0
for line2 in f2:
line2 = line2.rstrip('\n\r')
# 若 lines1 非空，且当前 line2 等于 lines1 最后一行，则跳过（重叠）
if lines1 and line2 == lines1[-1]:
skip_count += 1
continue
# 否则，一旦发现不重叠，后续所有行均写入（因 file2 有序）
break
else:
# file2 全部被跳过（极端情况：file2 是 file1 的后缀）
line2 = None
# 写入 file1 全部内容
out.writelines(f"{line}\n" for line in lines1)
# 写入 file2 中未被跳过的剩余行（包括刚读到的 line2）
if line2 is not None:
out.write(f"{line2}\n")
out.writelines(f"{line.rstrip('\n\r')}\n" for line in f2)
# 使用示例
merge_sorted_files('1.txt', '2.txt', 'merged.txt')✅ 进阶优化：真正流式处理（适用于超大文件）
若 file1 也极大（无法全量加载），可仅读取其最后一行用于比对：
def get_last_line(filepath):
"""安全获取文件最后一行（不加载全文）"""
with open(filepath, 'rb') as f:
f.seek(0, 2) # 移动到末尾
if f.tell() == 0:
return ""
f.seek(-1, 2)
while f.tell() > 0 and f.read(1) != b'\n':
f.seek(-2, 1)
return f.readline().decode().rstrip('\n\r')
def stream_merge(file1_path, file2_path, output_path):
last_line1 = get_last_line(file1_path)
with open(file1_path, 'r') as f1, \
open(file2_path, 'r') as f2, \
open(output_path, 'w') as out:
# 复制 file1 全部内容
for line in f1:
out.write(line)
# 跳过 file2 中与 last_line1 相同的开头行（最多一次重叠）
for line in f2:
stripped = line.rstrip('\n\r')
if stripped == last_line1:
continue # 跳过重叠行
out.write(line)
break # 第一个非重叠行写入后，直接复制剩余全部
else:
return # file2 已空或全重叠
# 复制剩余行
out.writelines(f2)⚠️ 注意事项
- 时间戳解析非必需：本方案依赖“重叠行内容完全相同”，无需解析日期——这正是题干给出的关键约束（“exactly the same data”），也是高效性的核心来源；
- 确保编码一致：打开文件时建议显式指定 encoding='utf-8'；
- 原地覆盖风险：output_path 不应与 file1_path 或 file2_path 相同，避免数据损坏；
- 空行/空白处理：示例假设无空行；如有，可在读取时添加 if line.strip(): 过滤。
该方法时间复杂度为 O(n + m)，空间复杂度接近 O(1)（仅缓存一行或最后一行），完美契合“大文件、有序、局部重叠”的真实场景，是工业级日志合并的推荐实践。
