库存调整 CSV 看起来有三列，导入后却只剩两个字段；另一份文件的 sku 明明存在，程序却报 KeyError。这类问题先查文件边界，别急着给字典取值加默认值。默认值会把格式错误变成库存错误。
本文约定一个具体接口：上传 UTF-8 的 CSV，列为 sku,quantity,note，SKU 只能包含大写字母、数字和连字符，数量为 0 到 1000000 的整数，备注允许引号内换行。允许 UTF-8 BOM 和列顺序变化，拒绝重复、缺失及额外列，整批验证成功后才允许写入数据库。代码适用于 Python 3.12 及以上，只用标准库。
一、先复现“没有异常，但字段已经丢了”
import csv
import io
source = io.StringIO("sku,quantity,quantity\nA-01,3,900\n", newline="")
print(next(csv.DictReader(source)))
# 输出：{'sku': 'A-01', 'quantity': '900'}
重复表头映射到同一个字典键，前一个值被覆盖。等业务层拿到字典，再检查字段数量，已经无法恢复被覆盖的数据。
另外两类现象需要分别定位：
- UTF-8 BOM 被普通
utf-8解码保留，首列表头可能是\ufeffsku。检查表头的repr()，肉眼看表格通常发现不了。 DictReader遇到较长行会把剩余值放到restkey对应的键中，默认键是None；较短行使用restval，默认值也是None。这些情况不会自动成为业务校验错误。该行为见 Python csv 官方文档。
排查时记录导入批次、记录序号、物理行号、预期列数和实际列数即可。不要直接打印上传文件、备注和整行内容。表头也是外部输入，生产日志更适合记录校验错误类型，而不是无限制输出原文。
二、修复顺序：先检查原始列，再构造字典
这里采用 csv.reader，在表头校验成功且每行长度一致后才执行 zip。strict=True 用于识别部分 CSV 语法错误，它不会替你检查业务字段、重复表头或列数。
将下面代码保存为 inventory_csv.py。文件由调用方打开，解析器只接收文本流，测试无需真实网络或数据库。
import csv
import re
from dataclasses import dataclass
from typing import TextIO
class ImportFormatError(ValueError):
"""上传文件不符合约定时，阻止整批数据进入写入阶段。"""
@dataclass(frozen=True)
class InventoryRow:
"""保留通过校验的库存调整数据，不携带原始上传行。"""
sku: str
quantity: int
note: str
def parse_inventory(source: TextIO) -> list[InventoryRow]:
"""先完整校验小批量文件，避免解析到一半已经更新库存。"""
expected = {"sku", "quantity", "note"}
reader = csv.reader(source, strict=True)
result: list[InventoryRow] = []
seen_skus: set[str] = set()
try:
headers = next(reader, None)
if headers is None:
raise ImportFormatError("文件为空")
if len(headers) != len(set(headers)):
raise ImportFormatError("表头存在重复列")
if set(headers) != expected:
raise ImportFormatError("表头必须包含且仅包含 sku、quantity、note")
for record_number, values in enumerate(reader, start=1):
location = f"第 {record_number} 条记录（结束于物理行 {reader.line_num}）"
if record_number > 10000:
raise ImportFormatError("记录数超过 10000 条")
if len(values) != len(headers):
raise ImportFormatError(f"{location}的列数不匹配")
row = dict(zip(headers, values, strict=True))
sku = row["sku"]
raw_quantity = row["quantity"]
note = row["note"]
if re.fullmatch(r"[A-Z0-9][A-Z0-9-]{0,31}", sku) is None:
raise ImportFormatError(f"{location}的 SKU 格式无效")
if sku in seen_skus:
raise ImportFormatError(f"{location}存在重复 SKU")
if re.fullmatch(r"(?:0|[1-9][0-9]{0,6})", raw_quantity) is None:
raise ImportFormatError(f"{location}的数量必须为无前导零的非负整数")
quantity = int(raw_quantity)
if quantity > 1000000:
raise ImportFormatError(f"{location}的数量超过上限")
if len(note) > 200 or "\x00" in note:
raise ImportFormatError(f"{location}的备注过长或包含空字符")
seen_skus.add(sku)
result.append(InventoryRow(sku, quantity, note))
except csv.Error as exc:
raise ImportFormatError(
f"物理行 {reader.line_num} 附近存在 CSV 语法错误"
) from exc
if not result:
raise ImportFormatError("文件没有数据记录")
return result
表头使用精确比较， sku、SKU 和 sku 不视为同一字段。不要对所有字段一律 strip()：那会改变 SKU 和备注内容，也可能把坏模板悄悄变成“能导入的模板”。如果业务确实允许别名或空格，先定义映射，再对映射后的表头检查重复。
数量先检查长度和 ASCII 十进制语法，再调用 int。这样 +3、3.0、03、全角数字以及超长数字都不会混入整数边界。数量 0 合法；库存是否允许归零，需要权限和业务规则另行控制。
三、把编码处理放在文件入口
将下面代码放在调用方。utf-8-sig 能读取带 BOM 和不带 BOM 的 UTF-8 文件；它不是 GBK 自动检测器。签名处理见 Python 编码文档。
from pathlib import Path
from inventory_csv import ImportFormatError, InventoryRow, parse_inventory
def load_inventory(path: Path) -> list[InventoryRow]:
"""只接受明确约定的编码，不通过丢弃字节掩盖坏文件。"""
try:
with path.open(encoding="utf-8-sig", errors="strict", newline="") as source:
return parse_inventory(source)
except UnicodeDecodeError as exc:
raise ImportFormatError("文件编码不是有效的 UTF-8，请按模板重新导出") from exc
newline="" 让 CSV 模块处理字段内部换行。不要使用 splitlines() 后逐行 split(",")：备注中的逗号、双引号和换行都有 CSV 自己的转义规则。
上述函数返回的是验证后的列表，此时还没有修改库存。数据落库仍须参数化 SQL、验证操作者和每个 SKU 所属租户，并在事务中写入。解析通过不代表 SKU 存在，也不代表调用者有权修改它。
四、用回归测试守住格式边界
保存为 test_inventory_csv.py。测试直接检查原始字节入口的 BOM、非法编码，以及解析器的正常、边界和失败路径。
import io
import unittest
from inventory_csv import ImportFormatError, InventoryRow, parse_inventory
def parse_bytes(payload: bytes) -> list[InventoryRow]:
"""模拟文件入口，让编码与换行行为参与回归测试。"""
with io.TextIOWrapper(
io.BytesIO(payload), encoding="utf-8-sig", errors="strict", newline=""
) as source:
return parse_inventory(source)
class InventoryCsvTests(unittest.TestCase):
"""验证结构错误不会转化为可写入的库存数据。"""
def test_valid_bom_and_reordered_headers(self):
payload = b'\xef\xbb\xbfnote,sku,quantity\r\n"a,b",A-01,0\r\n'
rows = parse_bytes(payload)
self.assertEqual((rows[0].sku, rows[0].quantity, rows[0].note),
("A-01", 0, "a,b"))
def test_quoted_multiline_note(self):
rows = parse_bytes(b'sku,quantity,note\nA-01,2,"a\nb"\n')
self.assertEqual(rows[0].note, "a\nb")
def test_reject_invalid_headers(self):
for header in ("sku,quantity,quantity", "sku,quantity",
"sku,quantity,note,extra", " sku,quantity,note"):
with self.subTest(header=header), self.assertRaises(ImportFormatError):
parse_bytes((header + "\nA-01,2,x\n").encode())
def test_reject_wrong_width_and_blank_record(self):
for line in ("A-01,2", "A-01,2,x,extra", ""):
with self.subTest(line=line), self.assertRaises(ImportFormatError):
parse_bytes(("sku,quantity,note\n" + line + "\n").encode())
def test_quantity_boundaries(self):
self.assertEqual(parse_bytes(b"sku,quantity,note\nA-01,1000000,\n")[0].quantity,
1000000)
for quantity in ("", "-1", "+1", "01", "1.0", "1000001", "１"):
with self.subTest(quantity=quantity), self.assertRaises(ImportFormatError):
parse_bytes(f"sku,quantity,note\nA-01,{quantity},\n".encode())
def test_reject_duplicate_sku_and_bad_sku(self):
for body in ("A-01,2,\nA-01,3,\n", "a-01,2,\n", ",2,\n"):
with self.subTest(body=body), self.assertRaises(ImportFormatError):
parse_bytes(("sku,quantity,note\n" + body).encode())
def test_empty_and_header_only(self):
for payload in (b"", b"sku,quantity,note\n"):
with self.subTest(payload=payload), self.assertRaises(ImportFormatError):
parse_bytes(payload)
def test_malformed_quote_and_note_limit(self):
for body in ('A-01,2,"unfinished', "A-01,2," + "x" * 201,
"A-01,2,a\x00b"):
with self.subTest(body=body), self.assertRaises(ImportFormatError):
parse_bytes(("sku,quantity,note\n" + body).encode())
def test_invalid_utf8(self):
with self.assertRaises(UnicodeDecodeError):
parse_bytes(b"sku,quantity,note\nA-01,2,\xff\n")
def test_record_limit(self):
body = "".join(f"A-{number},1,\n" for number in range(10001))
with self.assertRaisesRegex(ImportFormatError, "10000"):
parse_bytes(("sku,quantity,note\n" + body).encode())
if __name__ == "__main__":
unittest.main()
在两个文件所在目录运行：
python -m unittest -v test_inventory_csv.py
十个测试方法必须全部通过。重复表头测试特别重要：如果改回“不校验表头直接使用 DictReader”，这份坏数据会重新被接受。
五、上线时还需要补哪些边界
本例适合最多 10000 条的小批量导入，调用方还应在上传入口限制请求体，例如最大 5 MiB，并对每条记录施加业务授权。备注长度检查发生在 CSV 解析之后，不能替代请求体大小限制。csv.field_size_limit() 是进程共享设置，不宜每个请求临时调高再还原；默认解析上限触发时，本例会转换为格式错误。
大文件可先完整解析到隔离的暂存表，校验成功再事务性合并。不要边解析边提交正式表，否则末行错误会留下半批写入。批次幂等键用于防止重复提交，同一个文件内的重复 SKU 校验不能替代批次幂等。
保留批次状态、错误类别、记录序号及结束物理行号。reader.line_num 统计物理行，因此多行备注会使它大于记录序号；给用户定位时应同时展示两者。重新导出为电子表格的内容还要单独评估公式执行风险，CSV 解析成功并不提供该项防护。
本文示例在本地标准库环境验证，不包含真实上传网关和数据库事务测试。接入系统后需要补充上传大小、租户权限、失败回滚及重复批次测试。
六、总结
CSV 导入的关键是保留检查顺序：明确编码和方言，检查原始表头，检查每行列数，再校验业务字段，最后进入事务写入。这样才能在信息被字典覆盖之前拒绝坏文件，也能让“模板错误”和“库存规则错误”分别得到可定位的反馈。
Discussion
评论