Python CSV导入校验：坏行报告与安全导出（附源码）

把库存CSV转成JSON不难，难的是知道哪些行不能用、为什么被拒绝，以及重跑时会不会覆盖上一次结果。本文用14条虚构库存记录做一套可复现的预检流程：默认整批拦截坏数据，显式允许时才导出合格子集，并把错误位置和源文件校验值一起保留下来。

适合正在写小批量导入脚本的Python开发者。代码使用标准库，最低Python 3.11；文件发布步骤面向支持硬链接的macOS/Linux本地文件系统。本文实测环境是macOS与Python 3.14，示例没有连接数据库。

下载完整实战包：脚本、两份CSV样本、20项测试与运行说明（ZIP，约8KB）

先看一次失败：程序正常运行，不代表数据可以导入

下载解压后进入csv-gate-v1目录。包中dirty.csv有14条数据记录，刻意包含重复编号、空名称、负数量、三位小数、NaN、多一列和少一列。另有带逗号及换行的合法名称，用来检查解析器有没有误拆记录。

python3 inventory_gate.py fixtures/dirty.csv

这条命令只打印预览，不创建结果文件；本地实测退出码为2。报告的summary如下。代码跑完了，质量关仍然没有放行，这两个状态必须分开。

{
  "records": 14,
  "accepted": 5,
  "rejected": 9
}

本例通过的SKU是001、002、003、004和abc。001保留前导零，名称中的逗号和换行保留，4.2输出为字符串“4.20”。表格按拒绝记录计数；如果某行同时犯两个错误，错误码总数可以大于拒绝记录数。

把规则写成契约，别让清洗脚本替业务猜答案

表头必须恰好包含sku、name、quantity、unit_price四列，允许换顺序，但重复表头和未知列都拒绝。所有字段只去掉两端空白。规则并非对所有库存系统通用；它们是本包明确采用、能够被测试的输入契约。

为什么金额输出字符串？这是本例接口契约的选择，用来保留十进制表示，避免经过float。下游如果只接受“分”为单位的整数，应另定转换契约和范围；不能一边输出字符串，一边假定接收方会自动理解。包中的Decimal只格式化已经满足位数限制的值，不承担隐式四舍五入。

重复SKU全部拦截，避免“谁排在前面谁算数”

按读取顺序保留第一条很容易写，却会把文件排序变成隐藏的业务规则。样本中DUP的两条名称、数量不同，脚本没有证据选哪条，因此两条都进入拒绝列表。即使第一条还有负数量，第二条格式正确，也仍然一起拦截。

frequencies = Counter(sku for _, _, _, sku in records if sku)
# 校验每条记录时：
if sku and frequencies[sku] > 1:
    errors.append("DUPLICATE_SKU")

代价是需要先读取记录、统计全文件SKU，再决定每行状态；不能发现第一条合格记录就立即写入目标系统。这个取舍适合本例的小批量预检，也解释了它为什么不是海量数据的流式处理器。若业务明确允许合并数量或覆盖旧值，应写出合并规则和相应测试，再修改这一段。

一条CSV记录可能占多行，错误报告要能找回原位置

示例的第3条合法记录在名称里包含换行，因此第5条记录实际位于文件第7行。报告同时保存record、line_start和line_end，避免让运营同事按错误的“第5行”反复查找。下面是报告中的真实片段。

{
  "record": 5,
  "line_start": 7,
  "line_end": 7,
  "sku": "DUP",
  "errors": [
    "DUPLICATE_SKU"
  ]
}

定位来自csv.reader读取前后的line_num，而不是简单对文本splitlines再计数。源码使用UTF-8-SIG及newline空字符串处理文件；BOM不会进入首列表头，引号内换行也不会被当成新记录。CSV语法错误、无效编码和表头错误会终止整文件处理，不会在“允许部分导出”时被静默跳过。

预览、严格导出、部分导出：明确三种操作

先运行严格导出。因为dirty.csv存在9条拒绝记录，进程退出2，strict.json不会生成。这样后续流程无法误用一个外表正常、实际已经丢掉数据的结果文件。

python3 inventory_gate.py fixtures/dirty.csv --commit strict.json

只有你已经决定接受这批合格子集时，才使用下面的明确开关。本地运行后partial.json包含5条items、9条rejected、summary及outcome。部分导出返回3，故意不伪装成全量成功。

python3 inventory_gate.py fixtures/dirty.csv --commit partial.json --allow-rejected

clean.csv保留那5条符合规则的输入，用来验证完整成功路径：

python3 inventory_gate.py fixtures/clean.csv --commit clean.json

逐条运行这些命令，不要直接放进set -e的脚本后误以为每步都会执行。前两次故意返回非零；再次练习时换新的结果文件名。预览本身只写stdout，使用者自行添加的shell重定向不在本程序的防覆盖保护范围内。

最后一步保护旧结果：先写临时文件，再发布新路径

直接用write_text写目标文件，一旦路径已存在就可能覆盖旧结果；只做exists检查再写，也留下了检查与写入之间的竞争窗口。本例把完整JSON写入目标目录中的临时文件，flush和fsync后，再用硬链接创建最终路径。最终路径已存在时，创建失败。

with tempfile.NamedTemporaryFile(dir=target.parent, delete=False) as tmp:
    temp_path = tmp.name
    tmp.write(payload)
    tmp.flush()
    os.fsync(tmp.fileno())
os.link(temp_path, target)  # 已存在则失败；完整代码在finally中清理临时名

再次运行刚才的clean.json命令，本地得到FileExistsError和退出码1。测试还检查了旧文件字节完全不变、临时文件清理、目标为符号链接时不会被替换。文件系统不支持硬链接就报错，不自动退回可能覆盖目标的做法。

这一步只保护单个本地结果文件的发布。它不是数据库事务，也不承诺掉电后的持久性；目标目录应事先存在且可信。网络盘、Windows以及并发业务消费者未做兼容验证。

20项测试覆盖了什么，仍没覆盖什么

python3 -m unittest -v

在解压后的目录执行上面命令，本次结果为20项通过。测试文件可以直接查看和修改；它验证的不只是一个正常输入。

上限为32MiB、100000条数据记录、单字段65536字符，完整输入和记录会放在内存。上限测试通过降低测试阈值触发相应分支，不是一次十万条数据的性能压测。拒绝率低也不意味着事实正确：某商品库存本来是8而CSV写10，若两者都满足契约，这段程序无法发现。

接到真实导入流程时，再补最后一段闭环

把生成的JSON交给正式导入器之前，先核对source.sha256是否对应本次原始文件，再由目标系统处理权限、唯一约束、事务和幂等。若失败后重试，不能只依赖“新文件名”推断不会重复写库；本包从未连接数据库，不能替你承诺业务幂等。

与运营协作时，保留原文件不改，按rejected中的位置修正一份新文件，重新预览，再用新输出名发布。这样能解释每次丢了哪些行、为什么改、哪份结果来自哪份输入。代码包中的README把输入契约、错误码、命令和边界放在一起，可以作为你自己导入流程的起点。

代码包SHA-256：0154dbd47ef4146861682043ca5356bb8ac60962ce2b4f9a6b430a8701005aff。用于核对下载内容与本次交付一致；代码采用MIT许可。

参考依据与相关旧稿

• Python csv官方文档：newline、reader与line_num
• Python os官方文档：link与fsync
• Python Decimal官方文档
• 原来的CSV清洗小例子
• DictReader的异常列处理
• JSON小数与Decimal的类型取舍

资料查阅与编辑日期：2026年10月16日。作者：玖柒。