用 Python 排查 CSV 分隔符:Sniffer 候选、显式格式与全量验证
Excel 导出的 CSV 一读就只剩一列,或者备注里的逗号把字段拆多了,问题未必是文件损坏:它可能用分号或制表符分隔。本文面向初学者到中级读者,用 Python 标准库写一个只读分隔符诊断器,先给出候选,再核对整份文件的表头和字段数。
1. 方法:先建议,再验证
- 已知导出格式时直接指定
--delimiter semicolon或tab,不要重新猜测。 - 未知时,只从逗号、分号、制表符和竖线中提出一个候选。
- 按约定的表头顺序读取整个文件,每条数据记录都必须有相同字段数;全部通过后才输出成功 JSON。
Python 官方 Sniffer 文档说明,sniff() 根据样本文本返回 dialect,并允许限制候选分隔符。候选不是格式证明:短样本、截断的引号或不规则数据都可能让推断失败或选错。
这里不自动判断有没有表头,也不采用猜出的全部 dialect 参数;只取分隔符。示例约定首条记录一定是表头,字段用双引号包裹,字段内双引号写成两个双引号,不支持自定义转义字符。这与CSV 表头与结构校验互补:先诊断分隔符,再固定数据接口。
2. 准备:建立隔离的演示目录
使用已有的 Python 3.9 或更新版本,无第三方依赖。命令适用于 Linux 的 bash/zsh,已在 Python 3.11.4 下测试。重复实验时换一个新目录名,避免覆盖旧文件。
- 在同一个终端执行:
python3 --version
mkdir csv-delimiter-demo
cd csv-delimiter-demo
python3 - <<'PY_DATA'
from pathlib import Path
Path("samples.csv").write_bytes(
b'\xef\xbb\xbfid;note\n001;"alpha,beta"\n002;"two\nlines"\n'
)
PY_DATA
文件带 UTF-8 BOM、使用分号分隔,第一条备注含逗号,第二条备注跨两行。数据记录只有两条。
- 将下一节代码保存为当前目录的
inspect_csv.py,也可下载完整脚本。
3. 完整实现:重置游标,核对整份文件
"""Inspect completed local CSV files; guess only the delimiter. Python 3.9+."""
import argparse
import csv
import json
from pathlib import Path
import sys
DELIMITERS = {"comma": ",", "semicolon": ";", "tab": "\t", "pipe": "|"}
def inspect_csv(path, mode, columns):
if len(columns) < 2 or len(set(columns)) != len(columns):
raise ValueError("expected columns must be unique and number at least two")
if any(not name.strip() for name in columns):
raise ValueError("expected column names must not be blank")
with path.open(encoding="utf-8-sig", newline="") as stream:
if mode == "auto":
sample = stream.read(8192)
try:
delimiter = csv.Sniffer().sniff(sample, delimiters=",;\t|").delimiter
except csv.Error as exc:
raise ValueError("cannot suggest delimiter; specify --delimiter") from exc
stream.seek(0)
else:
delimiter = DELIMITERS[mode]
reader = csv.reader(stream, delimiter=delimiter, quotechar='"',
doublequote=True, skipinitialspace=False, strict=True)
try:
header = next(reader, None)
if header != columns:
raise ValueError(f"header mismatch: expected {columns!r}, got {header!r}")
records = 0
for row in reader:
if len(row) != len(columns):
raise ValueError(
f"record {records + 1}, ending line {reader.line_num}: "
f"expected {len(columns)} fields, got {len(row)}"
)
records += 1
except csv.Error as exc:
raise ValueError(f"CSV error near physical line {reader.line_num}: {exc}") from exc
return {"delimiter": delimiter, "columns": columns, "records": records,
"physical_lines": reader.line_num, "selection": mode}
def main():
parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
parser.add_argument("file", type=Path)
parser.add_argument("--delimiter", choices=["auto", *DELIMITERS], default="auto")
parser.add_argument("--columns", nargs="+", required=True)
args = parser.parse_args()
try:
report = inspect_csv(args.file, args.delimiter, args.columns)
except (OSError, UnicodeError, ValueError) as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 1
print(json.dumps(report, ensure_ascii=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
采样最多读取 8192 个字符,随后 seek(0) 回到开头;忘记重置会跳过表头。工具以 utf-8-sig 读取 UTF-8 文件并兼容开头的 BOM,不猜测其他编码。遇到解码错误,先按CSV 转 UTF-8 教程核对来源编码。
官方 csv.reader 文档要求文件对象用 newline="" 打开;line_num 文档区分物理行与记录数。代码让 CSV 解析器处理引号内换行,而不是逐行 split()。
这是小型、已完成且检查期间不变的本地普通文件工具,不要传入管道、设备或持续追加的文件。全量验证按记录迭代,不保存全部记录,但单个超长字段仍受解析器限制,不能把它当作处理不可信巨型输入的资源隔离方案。
4. 使用:自动候选与显式格式
- 显示帮助,再分别自动判断和指定分号:
python3 inspect_csv.py --help
python3 inspect_csv.py samples.csv --columns id note
python3 inspect_csv.py samples.csv --delimiter semicolon --columns id note
自动模式的预期结果:
{"delimiter": ";", "columns": ["id", "note"], "records": 2, "physical_lines": 4, "selection": "auto"}
显式模式的计数相同,但 selection 为 semicolon。physical_lines 为 4,而 records 为 2,因为一条备注跨两行。001 始终作为字符串读取,不转成数字。
- 故意用错分隔符,观察退出码:
python3 inspect_csv.py samples.csv --delimiter comma --columns id note
printf 'exit=%s\n' "$?"
输出到 stderr 的错误如下,随后 shell 打印 exit=1:
ERROR: header mismatch: expected ['id', 'note'], got ['id;note']
| 退出码 | 含义 |
|---|---|
| 0 | 全部记录通过指定表头和字段数检查,stdout 输出一份 JSON。 |
| 1 | 文件、编码、推断或结构检查失败,stderr 输出错误,stdout 为空。 |
| 2 | 命令行参数错误,例如不支持的分隔符名称。 |
- 上游格式确认后,在自动化中固定参数,用
&&阻止失败后继续:
python3 inspect_csv.py samples.csv --delimiter semicolon --columns id note \
&& printf 'Input structure accepted\n'
工具不会重写输入,也不会打印数据记录,但错误信息可能包含实际表头;不要拿含敏感表头的数据直接写公共日志。通过检查不代表字段值、ID 唯一性或业务规则正确。
5. 验证:正常与故障样例都要测
保存为 test_inspect.py,或下载测试脚本,然后在演示目录执行 python3 test_inspect.py:
"""Run from the directory containing inspect_csv.py. No third-party packages."""
import csv
import hashlib
import io
from pathlib import Path
import subprocess
import sys
import tempfile
script = Path("inspect_csv.py").resolve()
cases = [
("semicolon", b'id;note\n001;"alpha,beta"\n', "auto", 0, ";"),
("comma", b'id,note\n001,"alpha,beta"\n', "auto", 0, ","),
("tab", b'id\tnote\n001\talpha\n', "tab", 0, "\t"),
("pipe", b'id|note\n001|alpha\n', "pipe", 0, "|"),
("multiline", b'\xef\xbb\xbfid;note\r\n001;"a\nb"\r\n', "semicolon", 0, ";"),
("wrong", b'id;note\n001;alpha\n', "comma", 1, None),
("ragged", b'id;note\n001;alpha;extra\n', "semicolon", 1, None),
("blank", b'id;note\n\n', "semicolon", 1, None),
("quotes", b'id;note\n001;"unfinished\n', "semicolon", 1, None),
("encoding", b'id;note\n001;\xff\n', "semicolon", 1, None),
("empty", b'', "auto", 1, None),
]
with tempfile.TemporaryDirectory() as directory:
for name, data, mode, expected, delimiter in cases:
path = Path(directory) / (name + ".csv")
path.write_bytes(data)
before = hashlib.sha256(data).hexdigest()
result = subprocess.run(
[sys.executable, str(script), str(path), "--delimiter", mode,
"--columns", "id", "note"], capture_output=True, text=True
)
assert result.returncode == expected, (name, result.stderr)
assert hashlib.sha256(path.read_bytes()).hexdigest() == before
if expected == 0:
import json
report = json.loads(result.stdout)
assert report["delimiter"] == delimiter
assert report["records"] == 1 and not result.stderr
else:
assert not result.stdout and result.stderr.startswith("ERROR:")
print(f"PASS {name}: exit={expected}")
# A naive split breaks the quoted comma; csv.reader preserves it.
assert '001,"alpha,beta"'.split(",") != ["001", "alpha,beta"]
assert next(csv.reader(io.StringIO('001,"alpha,beta"'))) == ["001", "alpha,beta"]
print("PASS quoted comma: field values preserved")
预期输出 11 条用例通过信息,最后一行是 PASS quoted comma: field values preserved。测试同时核对退出码、输出通道及输入哈希,覆盖 BOM、CRLF、字段内换行、错误分隔符、不齐的字段、空白记录、未闭合引号、非法 UTF-8 和空文件。
6. 排错:不要靠替换字符修复 CSV
| 现象 | 具体处理 |
|---|---|
cannot suggest delimiter |
确认文件非空,向导出方核对格式,然后使用 --delimiter;不要失败后默认改成逗号。Python 3.11.4 实测中,CRLF 文件内的多行引号字段也可能让自动推断失败,显式指定后仍能正确解析。 |
header mismatch |
核对分隔符、表头顺序、大小写和空格;代码不会自动去掉列名空格。 |
expected 2 fields, got ... |
按结束物理行检查缺失字段、额外分隔符或空白行,回到导出端修正。 |
CSV error |
核对引号是否闭合、字段内引号是否成对;strict=True 不能替代业务校验。 |
field larger than field limit |
先确认是否因分隔符错误把整行读成一个字段,再参考CSV 大字段限制排查。 |
不要用 replace(";", ",") 转换格式,也不要用 split(",") 读取 CSV:它们不理解引号内的分隔符。若要正式转换,先固定已确认的格式,再使用 csv.reader 和 csv.writer 写入新的输出路径。
7. 总结
用样本推断缩小排查范围,用已知表头和整份文件的字段数检查验证候选。确认格式后固定分隔符,让退出码决定后续任务是否执行,不要把猜测当成长期的数据协议。
- 原文作者:春江暮客
- 原文链接:https://www.bobobk.com/python-csv-delimiter-sniffer.html
- 版权声明:本作品采用 知识共享署名-非商业性使用-禁止演绎 4.0 国际许可协议 进行许可,非商业转载请注明出处(作者,原文链接),商业转载请联系作者获得授权。