用 Python 校验 CSV 结构：固定表头顺序，拦截缺列与多列

上游导出的 CSV 调换了两列，导入脚本却仍按旧位置读取；也可能某条备注里多了一个没加引号的逗号，让后面的值整体错位。文件能打开，并不代表结构符合下游约定。

本文面向初学者到中级读者，用 Python 标准库做一个固定表头、固定顺序、固定列数的命令行检查器。它逐条读取文件，遇到首个问题立即返回非零；只有检查完整个文件才输出成功 JSON，适合放在批处理导入前。

1. 方法：把表头当作接口约定

本例约定表头依次为 sample_id,label,note：

列名按原文精确比较，顺序也必须相同；增删列、重复列名、首尾空白都会被拒绝。
每条数据记录的字段数必须与表头相等；空白行也算失败。
文件至少有一条数据记录。正确列数下的空字段允许通过，字段类型和业务含义留给后续检查。
默认使用逗号和双引号，接受 UTF-8 或带 BOM 的 UTF-8；分号文件要显式传参。

这里使用 csv.reader 取得字段列表。Python 官方 CSV 文档说明，DictReader 会为缺少的字段填默认值，把多出的字段放进额外项；直接拿到字典，不代表列数正确。本例先比较列表长度，避免依赖这种补齐行为。

结构检查只负责这份约定；标签取值、ID 是否为空等规则，可在通过后接上 CSV 流式处理与标签检查。它不是通用的完整 CSV 标准合规检查器。

2. 准备环境与测试文件

需要 Python 3.9 或更新版本，无第三方依赖。以下示例已在 Linux、Python 3.11.4 上测试，shell 命令可在 Bash 或 zsh 执行。

新建演示目录，后续命令都在这个目录执行；重复练习时换一个新目录名：

python3 --version
mkdir csv-schema-demo
cd csv-schema-demo

生成测试数据。valid.csv 带 UTF-8 BOM，使用 CRLF 换行，同时包含引号包住的逗号和跨两行的字段：

python3 - <<'PY_DATA'
import csv
from pathlib import Path

with open("valid.csv", "w", encoding="utf-8-sig", newline="") as target:
    writer = csv.writer(target)
    writer.writerow(["sample_id", "label", "note"])
    writer.writerow(["001", "1", "quoted, comma"])
    writer.writerow(["002", "0", "two\nlines"])

fixtures = {
    "reordered.csv": "label,sample_id,note\n1,001,ok\n",
    "duplicate.csv": "sample_id,label,label\n001,1,ok\n",
    "short.csv": "sample_id,label,note\n001,1\n",
    "extra.csv": "sample_id,label,note\n001,1,ok,extra\n",
    "broken.csv": 'sample_id,label,note\n001,1,"unfinished\n',
    "blank.csv": "sample_id,label,note\n\n",
    "header-only.csv": "sample_id,label,note\n",
    "empty.csv": "",
    "semicolon.csv": "sample_id;label;note\n001;1;ok\n",
}
for name, text in fixtures.items():
    Path(name).write_text(text, encoding="utf-8")
PY_DATA

这只创建当前演示目录里的文件，不读取生产数据。出现 UnicodeDecodeError 时，先参考 CSV 编码确认与 UTF-8 转换，不要用 errors="ignore" 丢掉原始字节。

3. 完整实现：表头与记录宽度校验

把下面的代码保存为当前目录的 csv_schema.py：

"""Validate an exact CSV header and row widths. Python 3.9+, standard library."""
import argparse
import csv
import json
from pathlib import Path
import sys


def validate(path, columns, delimiter):
    with path.open(encoding="utf-8-sig", newline="") as source:
        reader = csv.reader(source, delimiter=delimiter, strict=True)
        try:
            header = next(reader, None)
            if header != columns:
                raise ValueError(f"header mismatch: expected {columns!r}, got {header!r}")
            records = 0
            for records, row in enumerate(reader, start=1):
                if len(row) != len(columns):
                    raise ValueError(
                        f"record {records}, ending line {reader.line_num}: "
                        f"expected {len(columns)} fields, got {len(row)}"
                    )
            if records == 0:
                raise ValueError("no data records")
        except csv.Error as exc:
            raise ValueError(f"CSV parse error near line {reader.line_num}: {exc}") from exc
    return {"file": str(path), "columns": header, "records": records, "status": "ok"}


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("file", type=Path)
    parser.add_argument("--columns", nargs="+", required=True,
                        help="exact column names, in the required order")
    parser.add_argument("--delimiter", choices=[",", ";"], default=",")
    args = parser.parse_args()
    if any(not name or name != name.strip() for name in args.columns):
        parser.error("column names must be nonempty with no surrounding whitespace")
    if len(set(args.columns)) != len(args.columns):
        parser.error("expected column names must be unique")
    try:
        report = validate(args.file, args.columns, args.delimiter)
    except (OSError, UnicodeError, ValueError) as exc:
        print(f"ERROR: {args.file}: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(report, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

文件以 newline="" 打开，把 CSV 换行处理交给解析器；strict=True 让解析器对它识别出的坏 CSV 抛出异常。相关行为见官方文档中的 csv.reader 和 Dialect.strict。严格模式不会替你检查表头约定、数据类型或所有引号书写问题，所以代码仍需显式比较表头和列数。

检查器不修改输入，也不会把全部记录放进列表；内存主要由当前记录与表头占用，单个超大字段仍可能触发解析器的字段大小限制。预期列名也必须非空、唯一且没有首尾空白，避免把错误约定当成合法配置。

4. 使用方法与错误定位

查看帮助并检查正常文件：

python3 csv_schema.py --help
python3 csv_schema.py valid.csv --columns sample_id label note

预期输出：

{"file": "valid.csv", "columns": ["sample_id", "label", "note"], "records": 2, "status": "ok"}

records 为 2，而不是文件的物理行数。第二条数据记录的 note 含有换行，仍然只算一条 CSV 记录。

检查多列文件，并在 shell 中处理这个预期失败：

if python3 csv_schema.py extra.csv --columns sample_id label note; then
  printf 'Schema accepted\n'
else
  printf 'Schema rejected; check stderr above\n'
fi

标准错误包含：

ERROR: extra.csv: record 1, ending line 2: expected 3 fields, got 4

record 1 从表头后开始计数；ending line 2 是解析器消耗到的物理结束行。官方 csvreader.line_num 文档明确区分行数与记录数。对于跨行记录，这两个数字可能不同；解析错误的 near line 是诊断位置，不是精确到字符的定位。

对明确使用分号的文件指定分隔符：

python3 csv_schema.py semicolon.csv \
  --columns sample_id label note --delimiter ';'

不要靠删除逗号或按 split(",") 拆文本来修复记录，那会破坏正常的引号和多行字段。

退出码	含义
0	整个文件通过本例结构约定，标准输出是一行 JSON。
1	结构、解析、解码或文件访问失败，只在标准错误输出原因。
2	命令行参数错误，例如预期列名重复或不支持的分隔符。

5. 验证成功与失败分支

运行下面的自检，不需要制造真实的线上故障。它断言退出码、错误原因和成功报告，同时允许被测子进程返回非零：

python3 - <<'PY_CHECK'
import json
import subprocess
import sys

columns = ["--columns", "sample_id", "label", "note"]
cases = [
    (["valid.csv", *columns], 0, None),
    (["reordered.csv", *columns], 1, "header mismatch"),
    (["duplicate.csv", *columns], 1, "header mismatch"),
    (["short.csv", *columns], 1, "expected 3 fields, got 2"),
    (["extra.csv", *columns], 1, "expected 3 fields, got 4"),
    (["broken.csv", *columns], 1, "CSV parse error"),
    (["blank.csv", *columns], 1, "expected 3 fields, got 0"),
    (["header-only.csv", *columns], 1, "no data records"),
    (["empty.csv", *columns], 1, "header mismatch"),
    (["missing.csv", *columns], 1, "ERROR:"),
    (["semicolon.csv", *columns], 1, "header mismatch"),
    (["semicolon.csv", *columns, "--delimiter", ";"], 0, None),
    (["valid.csv", "--columns", "sample_id", "sample_id"], 2, "must be unique"),
]
for arguments, code, message in cases:
    result = subprocess.run(
        [sys.executable, "csv_schema.py", *arguments],
        capture_output=True, text=True, check=False,
    )
    assert result.returncode == code, (arguments, result)
    if code == 0:
        report = json.loads(result.stdout)
        assert report["status"] == "ok", report
        assert report["records"] == (2 if arguments[0] == "valid.csv" else 1), report
        assert result.stderr == "", result
    else:
        assert result.stdout == "" and message in result.stderr, (arguments, result)
print("OK: 13 CSV schema checks")
PY_CHECK

预期输出：

OK: 13 CSV schema checks

这些检查覆盖表头换序、重复列名、缺列、多列、未闭合引号、空白行、只有表头、空文件、文件缺失、分隔符不匹配与错误参数。正常样例同时验证 BOM、CRLF、字段内逗号和字段内换行，避免把合法 CSV 误报为坏文件。

6. 接入批处理与排错

用 && 控制下游步骤，只在结构检查成功后继续：

python3 csv_schema.py valid.csv --columns sample_id label note \
  && python3 -c 'print("Downstream step started")'

演示中的下游命令只打印一行文本，实际使用时换成已有的导入命令。检查和导入期间应保持输入文件不变；如果生产者仍在写入，先让它完成，再检查一份固定的产物，否则两步可能读到不同内容。

按报错修复来源，再重新检查：

报错或现象	具体处理
`header mismatch`	对照输出的 expected 与 got，核对上游字段名、顺序和分隔符。确认接口确实变更后，才同步修改预期列名和下游读取逻辑。
`expected 3 fields, got 4`	查看相应记录，确认含逗号的字段是否正确加引号；修复导出逻辑后重新生成，不要直接截掉最后一列。
`got 2` 或 `got 0`	检查导出是否截断、遗漏字段或插入空白行。正常空值应保留分隔符，例如 `001,1,`。
`CSV parse error`	核对引号是否闭合以及输入是否完整；单字段过大也可能触发解析器限制，应先核实数据和所需上限。
`no data records`	本例拒绝只有表头的文件。若业务允许空批次，应明确修改约定和测试，不要把失败一律当作成功。
`Permission denied` / 文件缺失	使用执行任务的账号核对实际路径、读权限和父目录访问权限。

结构通过仍不代表 ID 唯一。需要防止批次合并后重复导入时，再用 Python 与 SQLite 检查 CSV 重复 ID补充唯一性检查。

7. 小结

固定表头名称、顺序与每条记录的列数，可以在导入前发现上游结构漂移和部分坏记录。先用包含引号、多行字段与失败样例的小文件验证，再接入批处理；保持结构检查、字段取值和 ID 唯一性各自有明确的规则与退出码。