春江暮客

春江暮客的个人学习分享网站

用 Python 排查 CSV 分隔符:Sniffer 候选、显式格式与全量验证

2026-10-31 技术
用 Python 排查 CSV 分隔符:Sniffer 候选、显式格式与全量验证

Excel 导出的 CSV 一读就只剩一列,或者备注里的逗号把字段拆多了,问题未必是文件损坏:它可能用分号或制表符分隔。本文面向初学者到中级读者,用 Python 标准库写一个只读分隔符诊断器,先给出候选,再核对整份文件的表头和字段数。

1. 方法:先建议,再验证

  1. 已知导出格式时直接指定 --delimiter semicolon 或 tab,不要重新猜测。
  2. 未知时,只从逗号、分号、制表符和竖线中提出一个候选。
  3. 按约定的表头顺序读取整个文件,每条数据记录都必须有相同字段数;全部通过后才输出成功 JSON。

Python 官方 Sniffer 文档说明,sniff() 根据样本文本返回 dialect,并允许限制候选分隔符。候选不是格式证明:短样本、截断的引号或不规则数据都可能让推断失败或选错。

这里不自动判断有没有表头,也不采用猜出的全部 dialect 参数;只取分隔符。示例约定首条记录一定是表头,字段用双引号包裹,字段内双引号写成两个双引号,不支持自定义转义字符。这与CSV 表头与结构校验互补:先诊断分隔符,再固定数据接口。

2. 准备:建立隔离的演示目录

使用已有的 Python 3.9 或更新版本,无第三方依赖。命令适用于 Linux 的 bash/zsh,已在 Python 3.11.4 下测试。重复实验时换一个新目录名,避免覆盖旧文件。

  1. 在同一个终端执行:
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、使用分号分隔,第一条备注含逗号,第二条备注跨两行。数据记录只有两条。

  1. 将下一节代码保存为当前目录的 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. 使用:自动候选与显式格式

  1. 显示帮助,再分别自动判断和指定分号:
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 始终作为字符串读取,不转成数字。

  1. 故意用错分隔符,观察退出码:
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 命令行参数错误,例如不支持的分隔符名称。
  1. 上游格式确认后,在自动化中固定参数,用 && 阻止失败后继续:
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. 总结

用样本推断缩小排查范围,用已知表头和整份文件的字段数检查验证候选。确认格式后固定分隔符,让退出码决定后续任务是否执行,不要把猜测当成长期的数据协议。

标签

1024 12306 2fa ablang adsense agents.md ai ai-agent ai-agents ai-seo algorithm amp antibodies apparmor argparse automation backup bash batch-processing binarycif bioinformatics biopython blockchain boltz bootstrapping boxes bubblewrap c-index cca cdn chatgpt checkpoint cli cloudflare codex cofoldarena concurrency configparser configuration copy cprofile cpu监控 cron csv cuda curl data-leakage data-processing data-quality data-validation datascience datavisualization datetime debugging decimal deployment desktop-app devtools difflib disk-space disown dns docker dovecot download electron encoding esm esm2 esm3 esmc esmfold2 etag evaluation faceswap fasta fastmcp faulthandler ffmpeg file-descriptors file-io file-management files flashppi flask folium frontend game generator git github-actions google google-research gpt-6.1-sol grep gzip harness heapq hls html http hugo imports indexnow ipaddress jailbreak javascript jev json jsonl just k-means kaggle langfuse leecode linux list litellm llm llms.txt logging logs lollipop lsof m3u8 machine-learning macos manacher matplotlib mcp memory mirror mmcif model-evaluation monitoring mp3 mp4 mpnn multiomics mutation mysql nanobert nanobodies networking networkx nginx normalize numpy ollama omegatherm pandas password pathlib pdb pep-723 performance permissions phaser pillow pip postfix preprocessing print prompt-engineering protein-alignment protein-design protein-embeddings protein-interactions protein-language-models protein-stability protein-structure proxy pydantic pyecharts pyqt python python3 r raincloud reproducibility requests reservoir-sampling rfoptimization rg ripgrep rosettafold3 roundcube rrsi rsi rsync s-tui sampling scale scikit-learn scrapy screen seaborn security selenium seo sequence-identity sha256 shell signals sitemap sklearn socket solana somaticsignatures sorting spl sqlite ssh standardize static-site subprocess symlink sysadmin system-one tarfile tcp tempfile tensorflow tkinter tls toml totp tracemalloc tron tronpy troubleshooting turtle typesafe-ai ubuntu unicode urllib usdt utf-8 uv vhhbert vite web-tools webp wordcloud wordpress workflow xml yaml zipfile zoneinfo 后台 寓言 概率 经济 贸易 迅雷解析 钱包

友情链接

其它