别等脏数据进库：用 Python 标准库做 CSV 导入前的契约审计

CSV 导入常见的问题不是“文件打不开”，而是文件能读、任务也显示成功，直到下游才发现主键重复、日期格式错误、数值越界或关键字段为空。此时回滚、清理和重新对账的成本，通常远高于在入口处先做一次确定性检查。

本文基于一个只用 Python 标准库实现的 CSV 契约审计工具，说明如何把“数据能不能导入”从口头约定变成可执行的导入前检查。

一、为什么先做契约审计

传统做法往往是在业务代码里零散地写 if not value、try: int(value)，检查规则分散且结果难复核。更稳妥的方式是把规则放进 JSON 契约，由单一入口输出结构化报告：

• 必填列是否存在；
• 主键是否为空或重复；
• 字段类型、日期格式、数值上下界是否合法；
• 枚举值是否在白名单中；
• 空值比例是否超过阈值。

审计只读取文件，不改写原始 CSV。这样既能在 CI 中阻断错误导入，也便于业务人员根据报告修复源文件。

二、把规则写成 JSON

示例契约如下：

{
  "required_columns": ["id", "email", "signup_date", "score", "status"],
  "primary_key": "id",
  "columns": {
    "email": {"type": "string", "required": true},
    "signup_date": {"type": "date", "format": "%Y-%m-%d"},
    "score": {"type": "float", "min": 0, "max": 100},
    "status": {
      "type": "string",
      "allowed": ["active", "disabled"],
      "max_null_ratio": 0.1
    }
  }
}

规则表达的是数据契约，而不是业务清洗逻辑。比如 score 可以被判定为“在范围内”或“越界”，但工具不会替业务猜测错误值应当修正成多少。

三、核心检查顺序

工具先读取表头，再逐行检查：

1. 必需列缺失时直接记录 missing_column；
2. 主键为空或重复时记录 blank_primary_key / duplicate_primary_key；
3. 按字段规则检查字符串、整数、浮点数和日期；
4. 数值检查 min、max；
5. 字符串检查 allowed 白名单；
6. 检查空值比例是否超过字段阈值；
7. 将问题压缩成 JSON，并在达到样例上限时给出被截断的剩余数量。

日期通过 datetime.strptime 按契约指定格式解析，浮点数额外拒绝 NaN 和 Infinity，避免“看似能转换、实际污染后续计算”的值。

四、结构化报告与退出码

干净文件会输出：

{
  "ok": true,
  "rows": 2,
  "issue_count": 0,
  "issues": [],
  "issues_truncated": 0
}

发现问题时，报告会保留行号、列名、错误码和说明。例如重复主键会指出首次出现的行号，方便定位；详细问题默认最多保留 20 条，但 issue_count 仍统计全部问题，避免因截断而低估数据风险。

命令行退出码明确区分结果：

• 0：通过；
• 1：存在契约问题；
• 2：参数或契约配置错误。

这样在定时同步任务中可以只依赖退出码决定是否继续导入，同时保留 JSON 报告用于审计。

五、实际验证

项目使用 unittest 覆盖了五类场景：

test_clean_file_passes                    ok
test_duplicate_primary_key_is_reported    ok
test_invalid_date_and_numeric_range        ok
test_null_ratio_and_allowed_values         ok
test_bad_contract_is_rejected              ok
Ran 5 tests in 0.007s

示例 CSV 的实际审计结果：

{
  "ok": true,
  "rows": 2,
  "issue_count": 0
}

六、适用边界

这个工具适合做导入前的确定性门槛，但不等于完整的数据治理能力。跨表引用、业务状态机一致性、金额精度和历史记录对账仍需要更上层规则。它的价值在于先把低层、可重复、可解释的结构性问题挡在入口外，让后续业务校验专注于真正需要上下文判断的部分。

对于批量导入系统，最省事的设计往往不是写更多补偿脚本，而是让不合格数据在进入系统之前就被清楚地拒绝。