Python 读取 CSV 的四个边界:引号、重复列、缺字段和类型转换
为什么不能直接按逗号拆分
处理简单表格时,CSV 很方便,但逗号也可能出现在被引号包住的字段里。例如 "Li, Ming",85 是两个字段,不是三个。本文用 Python 标准库 csv 做一个小型导入检查:既解析引号,又在数据不符合预期时给出错误,而不是悄悄把错误数据写入后续流程。
一份可运行的示例
假设输入至少包含 name 和 score 两列,其中 score 必须是整数。代码用 DictReader 按列名取值,并检查重复列名、列数不匹配与类型转换失败:
import csv
from io import StringIO
def read_rows(text):
reader = csv.DictReader(StringIO(text, newline=''))
required = {'name', 'score'}
if not reader.fieldnames or not required.issubset(reader.fieldnames):
raise ValueError('missing required columns')
if len(reader.fieldnames) != len(set(reader.fieldnames)):
raise ValueError('duplicate column names')
result = []
for record, row in enumerate(reader, start=2):
if None in row or any(v is None for v in row.values()):
raise ValueError(f'record {record}: wrong field count')
try:
score = int(row['score'])
except ValueError as exc:
raise ValueError(f'record {record}: score must be an integer') from exc
result.append((row['name'], score))
return result
rows = read_rows('name,score\nAlice,90\n"Li, Ming",85\n')
print(rows)
# [('Alice', 90), ('Li, Ming', 85)]
四个检查分别解决什么问题
第一,引号中的逗号。 csv 解析器会把引号内的逗号保留在字段值中,避免手写 split(',') 把姓名拆开。实际文件还可能包含转义引号、字段内换行等情况,不能只靠简单分割。
第二,重复列名。 DictReader 返回字典;同名列会互相覆盖,原始信息可能丢失。因此解析数据前先比较列名数量与去重后的数量,发现重复直接拒绝。
第三,字段数量。 如果一行比表头多字段,DictReader 默认把多余值放在 None 键下;如果少字段,缺少的值默认为 None。示例对这两种情况都报错,而允许明确给出的空字符串继续由业务规则决定如何处理。
第四,数字类型。 CSV 的值读出来是字符串,score 需要显式转为整数。转换失败时重新抛出 ValueError,同时用异常链保留底层原因;实际项目可以再检查分数范围,例如 0 到 100。
如何使用与验证
示例使用 StringIO 方便复制运行。读取本地文件时,可使用 open(path, encoding='utf-8-sig', newline='') 并将文件对象交给 DictReader;utf-8-sig 能兼容带 UTF-8 BOM 的文件,具体编码仍应与数据来源约定。不要把未知编码的文件一律当作 UTF-8。
我用一组自建数据验证:正常记录、引号内逗号可解析;缺少必需列、重复列名、非整数分数以及额外字段都会报错。示例中的 record 是记录序号加表头偏移,不保证等于物理行号:字段含换行时,一条记录可能跨多行。要定位物理行可结合 reader.line_num,并明确它指向解析后的行位置。
进一步完善
生产导入还要按业务约定处理空值、空白字符、日期格式、数值范围和文件大小;如果需要一次展示全部错误,可以收集错误列表,确认整批数据通过后再入库。这样可以避免导入到一半才发现坏数据。此示例偏重清晰地展示检查点,并没有实现并发导入、流式报错汇总或数据库事务。
- 点赞
- 收藏
- 关注作者
评论(0)