批量合并表格时,不要只把文件读进来拼接。本例让 DeepSeek 协助写一个 CSV 合并脚本:先检查每个文件列名一致,再附加来源文件名,最后核对输入行数之和与输出行数。
准备输入与运行环境
Python 3.11,安装 pandas。创建 incoming 目录和两个虚构 CSV:a.csv 为 id,amount_cents 加 001,100,b.csv 同表头加 002,200。所有文件用 UTF-8;输出写到 incoming 之外。

给 DeepSeek 的任务说明
批量读取 incoming 中的 CSV,固定列为 id 和 amount_cents,全部先按字符串读取以保留编号。每个文件列名顺序必须一致;加入 source_file 列。输出 merged.csv,输入行数和等于输出行数。不自动去重,不覆盖已存在输出。
在本地运行与检查
将下面代码保存为 example_10.py。进入保存该文件的目录,在终端执行 python example_10.py。代码应由你检查后执行,按实际字段修改。
from pathlib import Path
import pandas as pd
files = sorted(Path("incoming").glob("*.csv"))
out = Path("merged.csv")
if not files:
raise ValueError("没有输入 CSV")
if out.exists():
raise FileExistsError("输出已存在")
frames = []
expected = ["id", "amount_cents"]
total = 0
for file in files:
frame = pd.read_csv(file, dtype="string", encoding="utf-8-sig", keep_default_na=False)
if frame.columns.tolist() != expected:
raise ValueError(f"表头不一致: {file.name}")
total += len(frame)
frame["source_file"] = file.name
frames.append(frame)
merged = pd.concat(frames, ignore_index=True)
assert len(merged) == total
merged.to_csv(out, index=False, encoding="utf-8-sig")
back = pd.read_csv(out, dtype="string", encoding="utf-8-sig", keep_default_na=False)
assert len(back) == total and back.columns.tolist() == expected + ["source_file"]
print("files", len(files), "input_rows", total, "output_rows", len(back))
print(back.groupby("source_file").size().to_dict())
拼接、关联和去重不能混做
本例按行拼接同结构文件。如果你希望按订单号把两张不同列的表关联,要另用 merge 并检查一对一或一对多关系。编号跨文件重复可能是合法历史记录,先保留 source_file 供审查,不能悄悄删掉。
扩大到自己的目录
先列出将读取的文件并确认都属于本批。表头差异会直接停止,应该检查是否多了空格、列顺序或数据口径不同。大量文件可以分批写出;单次 concat 会占用内存,不适合无限大的目录。
怎样确认结果正确
示例应打印 files 2 input_rows 2 output_rows 2,来源行数分别为 a.csv 1、b.csv 1。打开 merged.csv,检查编号 001、002 和 source_file;把 b.csv 的金额列改名,应停止并显示表头不一致。
适用边界与常见问题
这是同结构虚构 CSV 的合并,不做金额清洗、去重或跨表关联。不同币种、不同时间粒度或不同含义的同名列必须先统一口径;程序看到表头一致也不能证明业务含义一致。
资料依据
接口说明核验日期:2026 年 10 月 1 日。pandas CSV I/O。
Ai菜鸟网。发布者:AI小管家,转载请注明出处:https://www.alyyhw.com/30240.html