几十个CSV散在一个文件夹里？一个脚本扫全目录、按手机号去重合并

2026-10-06 · 本文所有代码均在真实测试数据上跑通后发布

之前的文章讲过两份销售流水的合并，这次是更常见的变体：文件不是两三个，而是几十个，散在一个（甚至多层嵌套的）文件夹里——每月导出一份、每个渠道导出一份，想按手机号合并成一份去重后的名单。逐个文件改路径太累，本文给一个递归扫目录的版本。

测试数据（真实跑过，故意埋了4个坑）

• a.csv：UTF-8，张三/李四两行
• b.csv：GBK 编码，张三（与 a.csv 重复，应去重）、王五
• c.csv：表头带空格（"姓名 ,手机号"），Excel 手改过的文件常这样
• d.csv：有一行列数不足（钱七缺城市列）
• e.csv：完全空文件

脚本

import os, csv

SRC = "dir_test"   # 要扫描的目录
OUT = os.path.join(SRC, "merged.csv")
seen = {}   # 手机号 -> [姓名]
files = rows = 0
bad = []

def read_rows(p):
    for enc in ("utf-8-sig", "gbk"):
        try:
            with open(p, encoding=enc, newline="") as f:
                return list(csv.reader(f))
        except UnicodeDecodeError:
            continue
    raise UnicodeDecodeError

for root, _, names in os.walk(SRC):
    for n in sorted(names):
        if not n.lower().endswith(".csv"):
            continue
        p = os.path.join(root, n)
        if os.path.abspath(p) == os.path.abspath(OUT):
            continue                          # 别把上次的结果当输入
        files += 1
        try:
            data = read_rows(p)
        except UnicodeDecodeError:
            bad.append(f"{n}: 编码无法识别，跳过"); continue
        if not data:
            bad.append(f"{n}: 空文件"); continue
        idx = {h.strip(): i for i, h in enumerate(data[0])}   # 表头去空格
        if "手机号" not in idx or "姓名" not in idx:
            bad.append(f"{n}: 表头缺姓名/手机号，实际={data[0]}"); continue
        for ln, row in enumerate(data[1:], 2):
            if len(row) < len(data[0]):
                bad.append(f"{n}:{ln} 列数不足，跳过"); continue
            key = row[idx["手机号"]].strip()
            if not key:
                bad.append(f"{n}:{ln} 手机号为空，跳过"); continue
            if key in seen:
                continue                      # 先到先得
            seen[key] = [row[idx["姓名"]].strip(), key]
            rows += 1

print(f"扫描文件 {files} 个，去重后 {rows} 行")
for b in bad:
    print("⚠", b)
with open(OUT, "w", encoding="utf-8-sig", newline="") as f:
    w = csv.writer(f)
    w.writerow(["姓名", "手机号"])
    for v in seen.values():
        w.writerow(v)

实测输出（逐项核对过）

扫描文件 5 个，去重后 5 行
⚠ d.csv:2 列数不足，跳过
⚠ e.csv: 空文件
姓名,手机号
张三,13800000001
李四,13800000002
王五,13800000003
赵六,13800000004
孙八,13800000006

核对：张三重复只留一条、GBK 的 b.csv 正常读出、带空格表头的 c.csv 正常识别、坏行和空文件被点名跳过而不是静默消失——结果与预期完全一致。

实跑抓到的四个坑（都是真踩的）

1. 试编码不能用「打开后不读」的方式判断。open(..., encoding=...) 是惰性的，文件打开成功不代表能解码——必须真的读出第一行内容，UnicodeDecodeError 才会抛出来。想在一个 try 里完成「试开+试读」，把整个读文件放进循环里最稳。
2. 输出文件必须排除在扫描之外。结果 merged.csv 就写在同一个目录里，重跑时如果没排除，它会被当输入再次扫进来。这次是靠绝对路径比对 os.path.abspath 挡住的。
3. 表头要 h.strip() 再匹配。人工编辑过的 CSV 表头经常带不可见空格，直接 "手机号" in header 会匹配失败，整个文件被误判成"表头错误"跳过。
4. 坏行要点名跳过，不能静默。列数不足、手机号为空的行如果直接吞掉，你到最后都不知道少了谁——所有异常都收集进 bad 列表最后统一打印，带文件名和行号。

如果要去重的键不是手机号而是订单号/身份证号，把 idx["手机号"] 换掉即可；要"保留最新一条"而不是"先到先得"，见之前那篇 按手机号保留最新一条。

虚拟物品交付后不退款；未收到或未达承诺免费补发/重做。