csv-excel-merger

CSV/Excel 合并器

将表格文件合并为一份干净的输出，不会在不知不觉中丢失、重复或损坏行。

工作流

复制此检查清单并跟踪进度：

- [ ] 1. Profile inputs
- [ ] 2. Choose append vs join
- [ ] 3. Map columns and normalize keys
- [ ] 4. Merge and resolve conflicts
- [ ] 5. Verify row math
- [ ] 6. Write output and report

1. 分析输入文件。 先运行随附的分析器；它会报告编码、分隔符、行数、表头、候选键和表头重合情况，且不会更改任何内容：

   python scripts/profile_inputs.py file1.csv file2.xlsx
   

   Excel 文件会按工作表分别分析。如果工作簿包含多个工作表，请与用户确认应纳入哪些工作表。

2. 选择操作。 这是最容易出错的决定：

   • 追加（堆叠）——来自不同来源或时期的同类记录（1 月和 2 月的导出文件、三个潜在客户名单）。使用 pd.concat，然后去重。
   • 联接（补充信息）——关于同一实体的不同信息（联系人及其交易金额）。使用 pd.merge，按键联接。
   • 不确定？如果文件的大多数列相同，就追加；如果它们只有一个 ID 列相同，就联接。

3. 映射列并规范化键。 为每个文件构建明确的 {original: unified} 重命名映射（常见变体见 references/merge_strategies.md）；如果匹配不明确，请向用户展示映射。去重或联接之前，先规范化键列：去除空白、将电子邮件地址转换为小写、去掉电话号码中的非数字字符、统一日期格式。否则，[email protected] 和 [email protected] 会作为两个人保留下来。

4. 合并。 读取每个文件时使用 dtype=str，这样 ID、邮政编码和电话号码的前导零就能保留，然后再转换特定列。

   import pandas as pd
   
   frames = []
   for path, rename in [("jan.csv", {"E-mail": "email"}), ("feb.xlsx", {"Email Address": "email"})]:
       df = (pd.read_excel(path, dtype=str) if path.endswith((".xlsx", ".xls"))
             else pd.read_csv(path, dtype=str, encoding="utf-8-sig",  # use the profiler's encoding
                           keep_default_na=False))
       df = df.rename(columns=rename)
       df["email"] = df["email"].str.strip().str.lower()
       df["source_file"] = path          # lineage for every row
       frames.append(df)
   
   combined = pd.concat(frames, ignore_index=True, sort=False)
   # Blank keys are not duplicates of each other: set them aside before deduping.
   has_key = combined["email"].fillna("") != ""
   # Later files win: list the most recent source last, then keep="last".
   deduped = combined[has_key].drop_duplicates(subset=["email"], keep="last")
   no_key = combined[~has_key]
   merged = pd.concat([deduped, no_key], ignore_index=True)
   

   联接时，让 pandas 强制检查预期的关系，这样重复键会触发错误，而不是让行数倍增：

   out = pd.merge(contacts, deals, on="email", how="left",
                  validate="one_to_one", indicator=True)
   unmatched = out[out["_merge"] == "left_only"]
   

   冲突处理策略（保留第一条/最后一条/信息最完整的记录、合并字段、标记以供审核）见 references/merge_strategies.md。

5. 报告之前先验证。 未经检查，绝不交付合并结果：

   rows_in = sum(len(f) for f in frames)
   assert len(merged) > 0, "merge produced an empty frame"
   assert len(merged) <= rows_in, "more rows out than in: check the join keys"
   assert deduped["email"].is_unique, "duplicate keys remain after dedupe"
   print(f"in={rows_in} out={len(merged)} removed={rows_in - len(merged)} blank_keys={len(no_key)}")
   print(merged["source_file"].value_counts())
   

   对照源文件，手动抽查三条被删除的重复记录；断言只能证明计算结果正确，无法证明保留的是正确行。

6. 写入输出并报告。 使用 references/output_template.md 中的布局。

   • 面向 Excel 用户的 CSV：to_csv(path, index=False, encoding="utf-8-sig")（BOM 可确保 Excel 正确读取重音字符）。
   • Excel：安装 openpyxl 后使用 to_excel(path, index=False)。一个工作表最多容纳 1,048,576 行；超过此数量时，请拆分文件或使用 CSV/Parquet。
   • 如果存在非空的冲突集或未匹配集，也要写入 conflicts_review.csv 或 unmatched.csv。

pandas 版本说明

当前 pandas 版本为 3.x（Python 3.11+）。以下差异会影响合并操作：

• 文本列默认使用 str dtype，而非 object。请检查 pd.api.types.is_string_dtype(col)，不要使用 dtype == object。
• Copy-on-Write 始终启用。诸如 df[col][mask] = x 的链式赋值永远不会更新 df（pandas 只会发出警告）；请使用 df.loc[mask, col] = x。
• 解析后的日期时间默认采用微秒精度。如果下游需要纳秒精度，请在转换为整数之前调用 .dt.as_unit("ns")。
• pd.read_excel(..., engine="calamine")（需要 python-calamine）读取大型工作簿的速度比 openpyxl 快得多。

此技能中的代码也兼容 pandas 2.2。

需要检查的故障模式

• 连接时行数激增 - 两侧存在重复键会导致行数成倍增加。validate= 可以捕获此问题。
• 前导零丢失 - 读取时未设置 dtype=str 会将 01234 转成 1234。
• 被 Excel 篡改的值 - 如果源文件中的长 ID 已显示为 1.23E+15，或日期已被重新格式化，pandas 就无法还原；应标记这些值。
• 标题行不在第 1 行 - 带有标题块的导出文件需要设置 skiprows= 或 header=。
• 编码混杂 - UTF-8 文件中混有一个 cp1252 文件时，会出现 é 乱码。分析器会报告每个文件的编码。
• 列被静默丢弃 - 仅存在于一个文件中的列在追加后会大部分为空。保留该列并报告其完整度；切勿在不说明的情况下丢弃数据。
• 大文件（超过几百 MB）- 使用 chunksize= 读取，或使用 Polars/DuckDB，并通过键集合去重，而不要将所有数据加载到内存中。