用 Python 标准库检查 CSV：重复表头、错列行与错误汇总

CSV 看起来像“用逗号分隔的文本”，但合法字段也可以包含逗号、换行和引号。例如 "北京,上海" 是一个字段，直接调用 line.split(",") 会把它拆成两列。更稳妥的做法是交给 Python 标准库 csv 解析，再单独验证业务规则。本文用一个任务清单演示：先检查表头，再逐行报告错列、空值和无效数字。

示例数据

把下面内容保存为 tasks.csv：

id,title,minutes
1,备份,30
2,检查,abc
3,发布
4,复盘,40,多余
5,总结,45

这里第 2 条记录的时长不是整数，第 3 条少一列，第 4 条多一列。我们希望继续检查后续行，最后一起列出问题，而不是在遇到第一个错误时就丢失其余诊断信息。

完整脚本

将下面代码保存为 inspect_csv.py，用 Python 3 运行，无需安装第三方包：

#!/usr/bin/env python3
"""Validate a small task CSV with Python's standard library."""

import csv
import sys
from collections import Counter
from pathlib import Path


REQUIRED = ("id", "title", "minutes")


def inspect(path: Path) -> tuple[int, list[str]]:
    with path.open(encoding="utf-8-sig", newline="") as handle:
        reader = csv.reader(handle, strict=True)
        header = next(reader, None)
        if header is None:
            raise ValueError("CSV 文件为空")

        duplicates = [name for name, count in Counter(header).items() if count > 1]
        if duplicates:
            raise ValueError(f"重复表头：{', '.join(duplicates)}")
        missing = [name for name in REQUIRED if name not in header]
        if missing:
            raise ValueError(f"缺少必需列：{', '.join(missing)}")

        accepted = 0
        errors: list[str] = []
        for row in reader:
            line_number = reader.line_num
            if not row or all(not cell.strip() for cell in row):
                continue
            if len(row) != len(header):
                errors.append(
                    f"第 {line_number} 行：应有 {len(header)} 列，实际 {len(row)} 列"
                )
                continue

            record = dict(zip(header, row))
            if not record["id"].strip() or not record["title"].strip():
                errors.append(f"第 {line_number} 行：id 和 title 不能为空")
                continue
            try:
                minutes = int(record["minutes"])
            except ValueError:
                errors.append(f"第 {line_number} 行：minutes 必须是整数")
                continue
            if minutes <= 0:
                errors.append(f"第 {line_number} 行：minutes 必须大于 0")
                continue
            accepted += 1

    return accepted, errors


def main() -> None:
    try:
        accepted, errors = inspect(Path(sys.argv[1]))
    except (IndexError, OSError, ValueError, csv.Error) as exc:
        raise SystemExit(f"无法检查 CSV：{exc}") from exc

    print(f"有效记录：{accepted}")
    for error in errors:
        print(error)
    if errors:
        raise SystemExit(1)


if __name__ == "__main__":
    main()

执行 python3 inspect_csv.py tasks.csv，我在本地运行得到：

有效记录：2
第 3 行：minutes 必须是整数
第 4 行：应有 3 列，实际 2 列
第 5 行：应有 3 列，实际 4 列

有效的两条记录来自第 2 行和第 6 行。脚本在发现行级错误后以退出码 1 结束，方便在自动化流程中阻止有问题的文件继续流转。没有错误时退出码为 0。

为什么要先检查表头？

csv.reader 返回字段列表，不会替业务判断 id、title 和 minutes 是否存在。脚本先用 Counter 找重复表头，再检查必需列。重复表头尤其容易被忽视：如果直接用 dict(zip(header, row)) 转字典，同名列的前一个值会被后一个覆盖。先拒绝重复列名，才能避免静默丢数据。脚本允许额外的非重复表头，但每条数据仍须与表头的总列数一致。

文件以 encoding="utf-8-sig" 打开，是为了兼容开头带 UTF-8 BOM 的 CSV；否则第一个列名可能变成 \ufeffid，看起来是 id，实际却匹配不上。newline="" 则让 csv 模块自己处理换行，尤其适合含引号换行的字段。

行号和错误边界

reader.line_num 计数的是读取到的物理行，本例每条记录占一行，因此提示与编辑器行号一致。若某个被引号包裹的字段跨了多行，它指向该记录读取结束时的行号，不应误写成“第几条记录”。空行会跳过；缺列或多列会报告但不计为有效记录；字段数正确后才检查必填值及正整数时长。

csv.reader(..., strict=True) 可以帮助发现部分格式错误，但它不会替代业务校验，所以列数和字段内容仍需显式检查。实际导入前，还可以按业务需要增加 ID 去重、时长上限、日期格式等规则；保持“解析”和“校验”两层分开，错误才容易定位和测试。