CSV字段数不一致应分层处理：用csv.Sniffer+异常捕获跳过偶发错误行；pandas.read_csv通过on_bad_lines等参数容错对齐；预处理清洗混乱行；最终以Pydantic等Schema校验实现构建时约束。
CSV 字段数量不一致是常见但棘手的问题，直接用 csv.reader 读取可能报 Error: line contains NULL byte 或 csv.Error: expected X fields, saw Y。优雅处理的核心不是强行“修”数据，而是**明确意图、分层应对、保留上下文**。
用 csv.Sniffer + 异常捕获识别并跳过异常行
适用于：数据整体规范，仅偶发格式错误（如某行多了一个逗号、换行符未转义）。
- 启用
csv.reader的strict=False（Python 3.12+），或手动捕获csv.Error - 结合
csv.Sniffer推断分隔符和引号规则，提升鲁棒性 - 记录被跳过的行号和原始内容，便于后续人工核查
import csv
with open("data.csv") as f:
reader = csv.reader(f, skipinitialspace=True)
for i, row in enumerate(reader, 1):
try:
# 假设期望 5 列
if len(row) != 5:
raise csv.Error(f"expected 5 fields, got {len(row)}")
# 正常处理
process_row(row)
except csv.Error as e:
print(f"Warning: line {i} skipped — {e}. Raw: {row!r}")
用 pandas.read_csv 的容错参数自动对齐列
适用于：批量处理、字段语义明确、可接受填充缺失值（NaN/None）。
-
on_bad_lines='skip'：跳过问题行（默认行为在旧版是报错） -
on_bad_lines='warn'：警告并跳过（推荐，兼顾安全与可观测性） -
usecols+names显式指定列名，避免因首行缺失导致列名错位 -
keep_default_na=False和na_values=[]防止把合法字符串（如 "NULL"）误判为缺失
import pandas as pd
df = pd.read_csv(
"data.csv",
on_bad_lines="warn",
names=["id", "name", "email", "phone", "city"], # 强制列名
dtype=str, # 避免自动类型推断干扰字段数判断
keep_default_na=False
)
# 自动用 NaN 填充缺失字段，保持 DataFrame 结构完整
预处理：用生成器按行清洗，再交给标准解析器
适用于：格式混乱但有规律（如字段内含换行、引号不闭合、尾部多余逗号）。
立即学习“Python免费学习笔记（深入）”；
- 不依赖
csv模块逐行读取原始文本，用正则或状态机做轻量清洗 - 例如：修复未闭合双引号 → 把后续行拼接到当前行，直到遇到匹配的结束引号
- 清洗后输出“干净行流”，再喂给
csv.reader(StringIO(clean_line)) - 比全量加载内存更省内存，适合大文件
根本解法：定义 Schema 并校验，而非被动适配
真正优雅，是把“字段数不一致”从运行时错误变成构建时约束。
- 用
pydantic或dataclass定义每行结构，带字段类型和必填标记 - 解析后逐行实例化，失败时抛出清晰错误（哪个字段缺失/类型不符）
- 配合
csv.DictReader，用字段名而非位置索引，降低错位风险 - 把校验逻辑封装成可复用函数，统一项目内 CSV 处理风格
from pydantic import BaseModel, ValidationError
class UserRow(BaseModel):
id: int
name: str
email: str
<h1>用 DictReader 读取 → 转 dict → 实例化</h1><p>for i, row_dict in enumerate(csv.DictReader(f)):
try:
user = UserRow(**row_dict)
except ValidationError as e:
print(f"Line {i+1} invalid: {e}")
