Agent 读取 CSV 前，先留下一份输入证据清单
在字段映射之前保存原始输入证据，让结构异常和空值待确认状态有明确出口。
1. 表格出现之前，输入已经发生了变化
Agent 接收一份 CSV 后，常见的第一步是把它读成 DataFrame，再交给后面的规则或模型解释。到了这个位置，某些输入区别已经看不见了：原文中的 NA 可能变成缺失值，少写的一列可能被补成空字符串，重复表头可能获得自动追加的后缀。后续程序看到的表格很整齐，却未必能回答“文件里原来写了什么”。
本文在表格映射之前增加一份 CSV 输入证据清单，下面称为 manifest。它绑定保留下来的原始文件，逐条记录解析出的字段字符串、字段数、逻辑记录编号和物理起止行；结构不符合约定时保留证据并阻断后续处理。这个入口适合放在 Agent 的文件接收工具中，先完成确定性的验收，再允许业务解释继续。
示例全部使用固定合成输入。本次离线运行环境是 Python 3.12.14、pandas 2.2.3，没有安装依赖、调用模型或读取真实用户文件。manifest 程序与主体测试只用 Python 标准库；pandas 仅用于复现几种读取行为。完整源码、输入定义和手写预期都在正文中，复制到一个空目录即可重建实验。
2. 两项 pandas 选项解决的是不同问题
固定主样本包含八条数据记录，表头为 item_id,label,note。前两个 label 分别写成 NA 和 NULL，随后各放一个未加引号和加引号的空字段，再放空格、带空格的文字及字符串 0。第一条 note 内含一个 CRLF，方便同时检验记录定位。
默认 read_csv 在这份样本中把前四个 label 转成缺失值。只设置 dtype=str，缺失值推断仍然发生；只设置 keep_default_na=False，item_id 仍被推断为整数。两项合用后，主样本中的解码字段字符串得到保留。这是两类独立的转换开关，具体含义可查 pandas 2.2.3 read_csv 文档。
样本中的一个空格、引号内的一个空格和 NA 在默认读取中都没有成为缺失值，因此不能把结果概括成“空白都会被吞掉”。另一个实验显式给 label 设置 na_values=['']，只把两个空字符串解释成缺失值。这是在演示选择一条规则会产生什么结果，本文的入口不会自动采用它。
3. 字段列表能留下表格隐藏的结构异常
把 a,b,c 后面的数据写成 1,2，两项选项合用的 pandas 读取结果仍是三个字段，最后补了一个空字符串。源文件少写一列，与源文件写了第三个空字段，在这里变成相同的单元格值。csv.reader 分别返回两个和三个字符串，可以在补齐发生之前留下这个区别。
另一份微型样本的表头为 a,b，第一条数据是 1,2,3。本次 pandas 把第一个字段推断为索引，列 a、b 获得后面的 2、3；如果先出现一条正常的两字段记录，再出现三字段记录，同一读取方式则抛出 ParserError。不能把“列数超出表头总会报错”作为入口保证。
重复的 id,id 表头在 pandas 中成为 id,id.1，而直接使用 csv.DictReader 会让后一个 id 覆盖前一个。本文先保留表头列表，验收后才映射成字典。标准库的 strict=True 在未闭合引号样本中报错，但仍允许不等宽的字段列表；列数需要程序自己检查。Python csv 方言说明和 DictReader 说明可与后面的负控源码一起核对。
4. 给这个入口写一份有限的约定
这里采用固定 UTF-8 严格解码、逗号分隔、双引号引用、双引号加倍转义，不猜编码和分隔符，也不去除字段空格。表头必须按顺序等于 item_id,label,note，不接受重复、缺失、额外或调换顺序的表头。每条数据必须正好有三个字段，独立空记录会阻断处理。CSV 解析错误发生后，只保留此前已经返回的完整记录，并把 parse_complete 设为 false；错误区间内的残片不会冒充完整记录。
在这组教学输入里，NA 和 NULL 明确作为普通 label 文字。空 label 暂不赋予缺失、未知或零的业务含义，而是产生 empty_semantics_pending，使整个文件停在 semantics_pending。这个标记专门针对 label；空 note 允许通过。空格原样保留，本文不声称它与空 label 等价。只有表头的文件允许通过并产生零条数据，完全空文件则因没有表头而阻断。这些都是示例自己的规则，接入实际数据源前需要与其约定逐项核对。
manifest 的 ready 只表示通过上述有限契约。它没有检查业务编号是否合法、编号是否重复、note 是否可信，也没有授予任何外部行动权限。CSV 字段进入 Agent 后仍是待处理数据，不能因为输入验收通过就成为工具调用指令。
5. 记录定位必须绑定原文件
程序直接把解码后的完整文本传给 StringIO(newline='')，没有先过滤空行、注释或其他内容。每次 reader 返回一条记录，就取 reader.line_num 作为结束行；起始行是上一条结束行加一。Python 的 line_num 文档明确区分已经读取的源行数量和已返回的记录数量。主样本的 header 在第 1 行，R0001 横跨第 2—3 行，R0002 在第 4 行。
这些是物理行号，不能拿去对原始字节做切片。R0001 也是当前源哈希和解析约定下的顺序编号；在文件前方插入记录后，它可能指向别的数据。需要长期稳定身份时，应另外设计业务标识，不能把这个局部定位当成业务主键。
程序把输入一次读入字节对象，对这份对象计算 SHA-256，并把相同字节保存为输出目录里的 source.csv。manifest 中的文件名相对于该目录，复核时需要同时保留这两个文件。哈希能核对字节内容有没有变化，不能证明文件来自谁、获得过授权或内容为真。
csv.reader 返回的是解析后的字符串，并不保存字段的原始拼写。未加引号的空字段与 "" 都成为空字符串；引号和转义写法也不能从字段列表中逐字重建。这里保留了字段内的 CRLF，但这不构成通用的“无损 CSV”保证。原始文件的副本承担原文证据，manifest 承担结构定位，两者要一起使用。
6. 在空目录运行
把下面五个完整文件分别保存为所标注的文件名。核心实验只需现有 Python 3.12。先执行 python -m unittest -v test_intake.py，再执行 python fixtures.py 生成 14 份固定输入。fixtures.py 中的字节字面量明确写出 CRLF，避免从网页手动复制 CSV 时换行被改变。该脚本会重写当前目录的 inputs 内同名合成文件，只应在专用实验目录运行。
python -m unittest -v test_intake.py
python fixtures.py
python intake.py inputs/main.csv demo-main
python intake.py inputs/ready.csv demo-ready第三条命令打印 semantics_pending 并返回退出码 2，这是预期的阻断结果；随后可单独运行第四条。它打印 ready 并返回 0。每个输出目录都包含 source.csv 和 manifest.json。输出目录已存在时程序报错，重新演示应换一个新目录名。代码未实现原子写入、并发锁或失败回滚，操作系统 I/O 错误会原样抛出；输出中只出现一个文件时应视为不完整，不得交给下游。
如果本机已经有 pandas 2.2.3，可额外运行 python probe_pandas.py。它不安装依赖，其他 pandas 版本会退出并说明版本不匹配。核心 manifest 与主体测试不需要 pandas。为避免优化模式去掉负控断言，运行 probe_pandas.py 时不要加 -O。
7. intake.py 完整文件
"""Small, explicit CSV intake contract. Standard library only."""
import csv
import hashlib
import io
import json
from pathlib import Path
import sys
HEADER = ['item_id', 'label', 'note']
DIALECT = dict(delimiter=',', quotechar='"', doublequote=True,
escapechar=None, skipinitialspace=False,
quoting=csv.QUOTE_MINIMAL, strict=True)
def inspect_bytes(raw):
manifest = {
'version': 1,
'contract': 'item-label-note-v1',
'source': {'file': 'source.csv', 'sha256': hashlib.sha256(raw).hexdigest(),
'byte_count': len(raw)},
'parse': {'encoding': 'utf-8', 'errors': 'strict',
'newline': '', 'dialect': dict(DIALECT)},
'expected_header': list(HEADER),
'records': [], 'errors': [], 'pending': [],
'parse_complete': False, 'status': 'blocked',
}
errors = manifest['errors']
try:
text = raw.decode('utf-8', errors='strict')
except UnicodeDecodeError as exc:
errors.append({'code': 'invalid_utf8', 'byte_start': exc.start,
'byte_end': exc.end})
return manifest
reader = csv.reader(io.StringIO(text, newline=''), **DIALECT)
previous = 0
try:
for ordinal, fields in enumerate(reader):
record = {'record_id': 'header' if ordinal == 0 else f'R{ordinal:04d}',
'physical_lines': [previous + 1, reader.line_num],
'field_count': len(fields), 'fields': fields}
manifest['records'].append(record)
previous = reader.line_num
rid = record['record_id']
if ordinal == 0:
if len(set(fields)) != len(fields):
errors.append({'code': 'duplicate_header', 'record_id': rid})
if fields != HEADER:
errors.append({'code': 'header_mismatch', 'record_id': rid})
elif not fields:
errors.append({'code': 'blank_record', 'record_id': rid})
elif len(fields) != len(HEADER):
errors.append({'code': 'field_count', 'record_id': rid,
'expected': len(HEADER), 'actual': len(fields)})
elif manifest['records'][0]['fields'] == HEADER and fields[1] == '':
manifest['pending'].append({'code': 'empty_semantics_pending',
'record_id': rid, 'field': 'label'})
manifest['parse_complete'] = True
except csv.Error:
errors.append({'code': 'csv_syntax',
'physical_lines': [previous + 1, reader.line_num]})
if not manifest['records'] and manifest['parse_complete']:
errors.append({'code': 'missing_header'})
if not errors:
manifest['status'] = 'semantics_pending' if manifest['pending'] else 'ready'
return manifest
def accepted_rows(raw):
"""Recompute from bytes; never trust a caller-edited manifest status."""
manifest = inspect_bytes(raw)
if manifest['status'] != 'ready':
raise ValueError(manifest['status'])
return [{'source_sha256': manifest['source']['sha256'],
'record_id': r['record_id'], 'physical_lines': r['physical_lines'],
'values': dict(zip(HEADER, r['fields']))}
for r in manifest['records'][1:]]
def main():
if len(sys.argv) != 3:
raise SystemExit('usage: python intake.py INPUT.csv NEW_OUTPUT_DIR')
raw = Path(sys.argv[1]).read_bytes()
manifest = inspect_bytes(raw)
output = Path(sys.argv[2])
output.mkdir(parents=True, exist_ok=False)
(output / 'source.csv').write_bytes(raw)
(output / 'manifest.json').write_text(
json.dumps(manifest, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
print(manifest['status'])
raise SystemExit(0 if manifest['status'] == 'ready' else 2)
if __name__ == '__main__':
main()
8. fixtures.py 完整文件
"""Fixed synthetic input bytes. No real user data."""
from pathlib import Path
MAIN = (b'item_id,label,note\r\n'
b'0007,NA,"first line\r\nsecond line"\r\n'
b'0042,NULL,plain\r\n'
b'0000,,unquoted-empty\r\n'
b'0001,"",quoted-empty\r\n'
b'0002, ,one-space\r\n'
b'0003," ",quoted-one-space\r\n'
b'0004, NA ,padded-token\r\n'
b'0005,0,literal-zero\r\n')
H = b'item_id,label,note\n'
CASES = {
'main': MAIN,
'ready': H + b'0007,NA,plain\n',
'short': H + b'0007,NA\n',
'present_empty_note': H + b'0007,NA,\n',
'long_first': H + b'0007,NA,plain,extra\n',
'long_later': H + b'0007,NA,plain\n0042,NULL,plain,extra\n',
'duplicate_header': b'item_id,label,label\n0007,NA,plain\n',
'wrong_header': b'label,item_id,note\nNA,0007,plain\n',
'missing_header_column': b'item_id,label\n0007,NA,plain\n',
'blank': H + b'0007,NA,plain\n\n0042,NULL,plain\n',
'unclosed': H + b'0007,NA,"unfinished\n',
'empty_file': b'',
'header_only': H,
'invalid_utf8': H + b'0007,NA,\xff\n',
}
if __name__ == '__main__':
directory = Path('inputs')
directory.mkdir(exist_ok=True)
for name, raw in CASES.items():
(directory / f'{name}.csv').write_bytes(raw)
print(f'{len(CASES)} synthetic inputs written')
9. expected.json 完整文件
{
"main": {"status": "semantics_pending", "errors": [], "records": 9, "parse_complete": true, "pending": ["R0003", "R0004"]},
"ready": {"status": "ready", "errors": [], "records": 2, "parse_complete": true, "pending": []},
"short": {"status": "blocked", "errors": ["field_count"], "records": 2, "parse_complete": true, "pending": []},
"present_empty_note": {"status": "ready", "errors": [], "records": 2, "parse_complete": true, "pending": []},
"long_first": {"status": "blocked", "errors": ["field_count"], "records": 2, "parse_complete": true, "pending": []},
"long_later": {"status": "blocked", "errors": ["field_count"], "records": 3, "parse_complete": true, "pending": []},
"duplicate_header": {"status": "blocked", "errors": ["duplicate_header", "header_mismatch"], "records": 2, "parse_complete": true, "pending": []},
"wrong_header": {"status": "blocked", "errors": ["header_mismatch"], "records": 2, "parse_complete": true, "pending": []},
"missing_header_column": {"status": "blocked", "errors": ["header_mismatch"], "records": 2, "parse_complete": true, "pending": []},
"blank": {"status": "blocked", "errors": ["blank_record"], "records": 4, "parse_complete": true, "pending": []},
"unclosed": {"status": "blocked", "errors": ["csv_syntax"], "records": 1, "parse_complete": false, "pending": []},
"empty_file": {"status": "blocked", "errors": ["missing_header"], "records": 0, "parse_complete": true, "pending": []},
"header_only": {"status": "ready", "errors": [], "records": 1, "parse_complete": true, "pending": []},
"invalid_utf8": {"status": "blocked", "errors": ["invalid_utf8"], "records": 0, "parse_complete": false, "pending": []}
}
10. test_intake.py 完整文件
import hashlib
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from fixtures import CASES, MAIN
from intake import accepted_rows, inspect_bytes
ROOT = Path(__file__).resolve().parent
class IntakeTests(unittest.TestCase):
def test_fixed_cases(self):
expected = json.loads((ROOT / 'expected.json').read_text(encoding='utf-8'))
self.assertEqual(set(CASES), set(expected))
for name, raw in CASES.items():
with self.subTest(name=name):
m = inspect_bytes(raw)
actual = dict(status=m['status'], errors=[e['code'] for e in m['errors']],
records=len(m['records']), parse_complete=m['parse_complete'],
pending=[p['record_id'] for p in m['pending']])
self.assertEqual(actual, expected[name])
def test_main_exact_evidence(self):
m = inspect_bytes(MAIN)
self.assertEqual(m['source']['byte_count'], 206)
self.assertEqual(m['source']['sha256'],
'89573d7d2e60675bed26e986532ef6f17b02b9893a06b35f05a4a0e1a8e0512c')
self.assertEqual([r['physical_lines'] for r in m['records']],
[[1,1],[2,3],[4,4],[5,5],[6,6],[7,7],[8,8],[9,9],[10,10]])
self.assertEqual([r['field_count'] for r in m['records']], [3] * 9)
self.assertEqual([r['fields'] for r in m['records']], [
['item_id', 'label', 'note'],
['0007', 'NA', 'first line\r\nsecond line'],
['0042', 'NULL', 'plain'], ['0000', '', 'unquoted-empty'],
['0001', '', 'quoted-empty'], ['0002', ' ', 'one-space'],
['0003', ' ', 'quoted-one-space'], ['0004', ' NA ', 'padded-token'],
['0005', '0', 'literal-zero']])
def test_absent_and_empty_field_differ(self):
a = inspect_bytes(CASES['short'])['records'][1]
b = inspect_bytes(CASES['present_empty_note'])['records'][1]
self.assertEqual(a['fields'], ['0007', 'NA'])
self.assertEqual(b['fields'], ['0007', 'NA', ''])
def test_blank_keeps_location(self):
records = inspect_bytes(CASES['blank'])['records']
self.assertEqual(records[2], dict(record_id='R0002', physical_lines=[3,3],
field_count=0, fields=[]))
self.assertEqual(records[3]['physical_lines'], [4,4])
def test_partial_parse_is_marked(self):
m = inspect_bytes(CASES['unclosed'])
self.assertFalse(m['parse_complete'])
self.assertEqual(m['errors'], [{'code': 'csv_syntax', 'physical_lines': [2,2]}])
def test_downstream_gate(self):
rows = accepted_rows(CASES['ready'])
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]['values'],
{'item_id': '0007', 'label': 'NA', 'note': 'plain'})
self.assertEqual(rows[0]['record_id'], 'R0001')
self.assertEqual(rows[0]['physical_lines'], [2, 2])
self.assertEqual(rows[0]['source_sha256'],
hashlib.sha256(CASES['ready']).hexdigest())
self.assertEqual(accepted_rows(CASES['header_only']), [])
for name in ['main', 'short', 'long_first', 'duplicate_header', 'unclosed']:
with self.subTest(name=name), self.assertRaises(ValueError):
accepted_rows(CASES[name])
def test_quote_spelling_requires_original(self):
a = b'item_id,label,note\n1,,x\n'
b = b'item_id,label,note\n1,"",x\n'
ma, mb = inspect_bytes(a), inspect_bytes(b)
self.assertEqual(ma['records'], mb['records'])
self.assertNotEqual(ma['source']['sha256'], mb['source']['sha256'])
def test_cli_keeps_bytes_and_refuses_existing_output(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source, output = root / 'input.csv', root / 'bundle'
source.write_bytes(MAIN)
cmd = [sys.executable, str(ROOT / 'intake.py'), str(source), str(output)]
run = subprocess.run(cmd, capture_output=True, text=True)
self.assertEqual(run.returncode, 2)
self.assertEqual(run.stdout.strip(), 'semantics_pending')
saved = (output / 'source.csv').read_bytes()
self.assertEqual(saved, MAIN)
m = json.loads((output / 'manifest.json').read_text(encoding='utf-8'))
self.assertEqual(hashlib.sha256(saved).hexdigest(), m['source']['sha256'])
again = subprocess.run(cmd, capture_output=True, text=True)
self.assertNotEqual(again.returncode, 0)
self.assertEqual((output / 'source.csv').read_bytes(), MAIN)
if __name__ == '__main__':
unittest.main(verbosity=2)
11. probe_pandas.py 完整文件
"""Version-scoped negative controls; requires existing pandas 2.2.3."""
import csv
import io
import json
import sys
import pandas as pd
from fixtures import MAIN
def read(raw=MAIN, **options):
return pd.read_csv(io.BytesIO(raw), **options)
def main():
if pd.__version__ != '2.2.3':
raise SystemExit('This reproduction targets pandas 2.2.3; no installation is performed.')
default = read()
dtype_only = read(dtype=str)
no_na_only = read(keep_default_na=False)
both = read(dtype=str, keep_default_na=False)
assert default.item_id.tolist() == [7, 42, 0, 1, 2, 3, 4, 5]
assert dtype_only.item_id.tolist() == ['0007','0042','0000','0001','0002','0003','0004','0005']
assert dtype_only.label.isna().tolist() == [True,True,True,True,False,False,False,False]
assert no_na_only.item_id.tolist() == [7, 42, 0, 1, 2, 3, 4, 5]
assert no_na_only.label.tolist() == ['NA','NULL','','',' ',' ',' NA ','0']
assert both.label.tolist() == ['NA','NULL','','',' ',' ',' NA ','0']
assert default.label.iloc[4:7].tolist() == [' ', ' ', ' NA ']
policy = read(dtype=str, keep_default_na=False, na_values={'label': ['']})
assert policy.label.isna().tolist() == [False,False,True,True,False,False,False,False]
short = read(b'a,b,c\n1,2\n', dtype=str, keep_default_na=False)
assert short.iloc[0].tolist() == ['1','2','']
wide = read(b'a,b\n1,2,3\n', dtype=str, keep_default_na=False)
assert wide.index.tolist() == ['1'] and wide.iloc[0].tolist() == ['2','3']
duplicate = read(b'id,id\n0007,0042\n', dtype=str, keep_default_na=False)
assert duplicate.columns.tolist() == ['id', 'id.1']
assert list(csv.DictReader(io.StringIO('id,id\n0007,0042\n'))) == [{'id': '0042'}]
assert list(csv.reader(io.StringIO('a,b\n1,2,3\n'), strict=True))[1] == ['1','2','3']
assert len(read(b'a,b\n1,2\n\n3,4\n')) == 2
try:
read(b'a,b\n1,2\n3,4,5\n', dtype=str, keep_default_na=False)
except pd.errors.ParserError:
pass
else:
raise AssertionError('Expected ParserError for this long-later fixture')
print(json.dumps({'python': sys.version.split()[0], 'pandas': pd.__version__,
'assertions': 14, 'expected_exception_checks': 1,
'result': 'PASS',
'both_labels': both.label.tolist(),
'short_row': short.iloc[0].tolist(),
'wide_index': wide.index.tolist(),
'duplicate_columns': duplicate.columns.tolist()},
ensure_ascii=False, indent=2))
if __name__ == '__main__':
main()
12. 怎样读这次运行结果
主体测试通过了 8 个 unittest 方法。其中固定输入测试逐一比较 14 份手写预期，检查状态、错误代码、返回记录数、解析是否完整和待确认记录编号；其余方法检查完整字段内容、固定 SHA-256、全部物理行范围、少字段与空字段的区别、空行定位、部分解析、下游阻断、引号拼写证据和 CLI 保存的字节。这里没有把一次测试中的多条断言算成多个独立样本。
主样本共 206 字节，SHA-256 是 89573d7d2e60675bed26e986532ef6f17b02b9893a06b35f05a4a0e1a8e0512c。manifest 返回 9 条记录，包含表头；errors 为空，pending 指向 R0003 与 R0004，状态为 semantics_pending。物理行区间依次为 [1,1], [2,3], [4,4], [5,5], [6,6], [7,7], [8,8], [9,9], [10,10]。固定预期直接写在 expected.json 和测试源码里，测试时不会调用验收器来生成标准答案。
另外的负控脚本通过 14 条断言（12 条 pandas 断言、2 条标准库 csv 断言）和 1 个 pandas 预期异常检查，输出 Python 3.12.14、pandas 2.2.3 及 PASS。它复现类型推断、缺失值推断、补字段、隐式索引、重复表头、空行跳过和一份会报错的长行样本，结果只适用于文中输入及这次默认引擎选择。未做 Python、C、pyarrow 引擎矩阵，也未测试其他 pandas 版本。
下游调用 accepted_rows(raw) 时，会从字节重新生成 manifest 并核对状态，不能通过手改 manifest 的 status 绕过这个函数。成功返回的每条对象都带有 source_sha256、record_id、physical_lines 和 values。实际接入时，可让 Agent 的工具适配层只暴露这个通过验收的结果，并保存相应源文件；不要另外开放一条不经过验收的直接 read_csv 路径。示例函数本身无法约束一个另有文件访问权限的程序。
13. 准备接到真实文件前
这个教学入口一次读取整个文件，解析后的字段和 manifest 也都留在内存中。它没有文件大小、记录数或总输出字节预算，没有恶意输入压力测试、模糊测试、并发测试、权限隔离或安全沙箱。标准库解析器可能因字段过长抛出 csv.Error，这不等于已经实现资源耗尽防护。本例把所有 csv.Error 统一记为 csv_syntax，因此仅看这个代码不能断定原文引号或分隔符有误；字段上限也可能触发同一出口，文末有可复跑的区分实验。真实不可信文件应先经过另外设计并测试的容量限制和访问控制。
固定契约没有为其他编码、BOM、区域分隔符、注释行或跳过行提供兼容分支；也没有覆盖 NUL、所有畸形引号组合、公式注入、Excel 执行行为和业务空值词典。严格解析可以报告部分语法问题，不能据此宣称完整验证了所有 CSV 规范。本文只增加了一份非法 UTF-8 固定夹具来检查解码阻断，并未穷举编码错误。
文件保存发生在读入之后，未保证正在被其他进程修改的输入能够获得一致磁盘快照。输出也不是防篡改档案，完整性校验依赖可信的预期哈希和保留策略。需要生产级审计时，还应绑定数据来源、获取时间、访问权限以及不可变存储策略；这些能力未包含在示例中。
实用的接入点是先把“文件能否按既定结构解释”变成可复跑测试。字段含义仍交给明确的业务约定，模型效果需要另行评估。本次实验没有调用模型，无法据此推断 Agent 的理解准确率。三张图是围绕固定案例绘制的原创结构图，本文做了 HTML 静态结构与源码一致性检查，没有浏览器像素验收。
14. 资料与运行版本
官方资料核对日期为 2026 年 10 月 7 日。pandas 2.2.3 read_csv用于核对 dtype、NA、索引和坏行处理选项。Python 3.12 csv用于核对方言、newline、字段读取和 line_num；该文档页面当前标注 3.12.15，本次离线运行的解释器是 3.12.14。文档说明与固定案例观察分别记录，避免把某次读入结果当成任意文件的通用保证。
15. 补充：表头看起来正确，先检查开头的三个字节
如果一份 CSV 在表格里显示的列名是 item_id、label、note，交给前文入口却得到 header_mismatch，可以先看源文件开头的字节，以及解析后表头的 repr。编辑器里的视觉结果可能看不出 U+FEFF；只截取表格截图，也无法判断它是否在原始文件里。
下面另建一组固定合成输入。plain 包含两行 CRLF 文本，共 35 字节；one_bom 只在它前面增加 EF BB BF，共 38 字节。原来的 intake.py 对 plain 返回 ready，对 one_bom 返回 blocked。后者通过 utf-8 解码后，第一个字段实际为 '\ufeffitem_id'，与合同要求的 'item_id' 不同。
plain = b'item_id,label,note\r\n0007,NA,plain\r\n'
one_bom = b'\xef\xbb\xbf' + plain
本次在 pandas 2.2.3 中使用 encoding='utf-8'、dtype=str、keep_default_na=False 读取这两个样本，得到相同列名和数据字符串。这个观察解释了两种读取路径为何看似矛盾，也说明不能拿其中一个工具的显示结果替另一个入口验收。这里没有比较其他 pandas 版本或显式指定的解析引擎，不能把结果推广成所有版本、所有 CSV 都相同。
这份补充于 2026 年 10 月 13 日独立验证。前文的严格 UTF-8 入口、14 份固定输入、五个源码文件以及 BOM 不在兼容范围内的说明均保留。新增程序只生成两个解码视图，不把带签名的文件自动改成 ready。
16. utf-8-sig 只处理起始签名，位置和顺序都重要
utf-8-sig 的比较放在 CSV 解析之前：先严格解码，再把文本交给 csv.reader。本实验中，没有签名时两种解码得到同样的字段；有一份起始签名时，utf-8-sig 不把它交给 CSV 解析器。连续放两份签名，则仍留下一个 U+FEFF，表头比较继续失败。它没有为字段里的不可见字符提供通用清理规则。
另外两个输入把相同字节放在文件内部。embedded_header 将 U+FEFF 放在第二个列名 label 前，两种解码都保留它，原入口因完整表头不匹配而阻断。embedded_value 将它放在 note 的 left 与 right 之间，两种解码都保留 left\ufeffright。原入口允许这个 note，因此该样本仍为 ready；这只反映已有有限合同，不表示其中每个字符都符合某个业务词典。
bom_before_quoted_header 在签名后紧接带引号的第一列表头。普通 utf-8 保留前面的 U+FEFF，双引号不再位于字段开头，csv.reader 返回的第一个字段为 '\ufeff"item_id"'。此后再对已解析单元格去掉 U+FEFF，剩下的仍是带两个字面引号的 '"item_id"'。utf-8-sig 在解析前处理起始签名，才会让这对引号承担 CSV 引用作用。因此，“解析后修列名”并不等价于选择另一种解码方式。
最后两份输入用于确认失败出口：一份有完整签名，但后面的数据含非法 UTF-8 字节 FF；另一份只有 EF BB 两个字节。两种严格解码都返回 UnicodeDecodeError。可选签名解码不提供坏字节修复，也不会因为开头识别成功就使后续文本有效。
17. 保留原始哈希，再看那些会掩盖问题的修复
one_bom 经 utf-8-sig 解码后的字段，与 plain 完全相同，但二者的原始 SHA-256 不同。补充报告的 byte_count、sha256 和 hex 始终来自最初的 bytes；输出目录同时保存八份未改写的 CSV 字节副本。原始哈希只用于核对字节内容，仍不能证明来源可信。
plain SHA-256: adb2e885d7d15fa7f63e108c7df017b78205981dd56735bda6aaa4ad2a702134
one_bom SHA-256: e3d7c7ae36be23cda31b8e36bdb2c48a4496cd7283ed8494cba2f27497338d8b
测试还故意制作五种错误结果，逐一确认完整预期会拒绝它们。第一种用 lstrip('\ufeff') 清掉双前缀，误把本应保留一个 U+FEFF 的表头变为匹配。第二种在全文 replace('\ufeff', '')，把 note 改成 leftright。第三种同样全局替换，掩盖第二列名里的隐藏字符，造成错误通过。第四种对解码后重新编码的文本计算“源哈希”，使 38 字节输入获得 35 字节输入的哈希。第五种把诊断视图中的表头匹配直接升级为原入口 ready，绕过仍然有效的合同。
这些负控只在测试内构造错误候选，不写回样本，也不改变 intake.py。完整预期会比较所有字段、错误类型、原入口状态和源字节证据，因此测试不会只凭“程序没报错”判定改法正确。
排查真实数据时，应先核对发送方与入口约定。如果仍要求无签名 UTF-8，就保留阻断结果并请来源端按约定重新导出。如果确实要允许一份起始签名，应另行修改和测试版本化合同，同时继续绑定原始文件。本文的 header_matches 只是解码视图中一次表头比较，没有实现一个新的完整导入合同。
18. 在独立目录重跑这组比较
下载 ZIP 后从解压目录执行下面命令。也可以把前文五个文件与以下六个补充文件都放在 src 子目录，再运行相同命令。核心诊断和测试只用标准库，不需要 pandas；程序不接受真实业务文件路径，只建立这里列出的固定输入。
python -m unittest discover -s src -p "test_*.py" -v
python src/bom_diagnostic.py --output run-bom
第一条命令共运行 21 个 unittest 方法，其中 8 个来自未改动的原入口，13 个来自本补充。第二条命令成功返回 0，表示诊断文件已写出；它不是八个输入都可被入口接收的信号。run-bom/report.json 包含完整视图，标准输出只概括第一字段。embedded_header 的第一字段正常，却因第二字段而失败，需要看完整 JSON 才能找到原因。
run-bom 必须是一个尚不存在的目录。输出包括八个原始字节副本和 report.json；已有目录会报错，代码不保证磁盘失败后的原子回滚。若写入中断，应把该目录视为不完整，换一个新目录重跑。程序不安装依赖、不联网、不读取真实用户文件，也不调用模型。
本次运行使用 Linux、Python 3.12.14，另用已安装的 pandas 2.2.3 完成可选观察。源文件、从正文提取的副本和 ZIP 解压副本都通过同一组测试，生成相同的完整诊断 JSON、固定标准输出与原始字节副本。原创 SVG 以 1120 × 810 自然比例渲染并检查；HTML 做了静态结构和源码一致性检查，没有浏览器、macOS、Windows、真实 WorkBuddy 或生产文件测试。
19. 补充输入：bom_fixtures.py
字节字面量明确写出 CRLF 和 EF BB BF，不依赖从网页复制文本时编辑器如何保存文件。
"""Eight explicit synthetic byte inputs; no real files are accepted."""
BOM = b'\xef\xbb\xbf'
BASE = b'item_id,label,note\r\n0007,NA,plain\r\n'
CASES = {
'plain': BASE,
'one_bom': BOM + BASE,
'double_bom': BOM + BOM + BASE,
'embedded_header': b'item_id,' + BOM + b'label,note\r\n0007,NA,plain\r\n',
'embedded_value': b'item_id,label,note\r\n0007,NA,left' + BOM + b'right\r\n',
'bom_before_quoted_header': BOM + b'"item_id",label,note\r\n0007,NA,plain\r\n',
'bom_invalid_utf8': BOM + b'item_id,label,note\r\n0007,NA,\xff\r\n',
'partial_signature': b'\xef\xbb',
}
20. 补充程序：bom_diagnostic.py
它直接调用未改动的 inspect_bytes 保存原入口判断，再独立生成 utf-8 与 utf-8-sig 视图。两种视图均采用相同的 CSV 方言与 newline 设置。
"""Compare decoders without changing the published intake contract."""
import argparse
import csv
import hashlib
import io
import json
from pathlib import Path
from bom_fixtures import CASES
from intake import DIALECT, HEADER, inspect_bytes
def decode_rows(raw, codec):
"""Return a diagnostic view, not an ingestion approval."""
try:
text = raw.decode(codec, errors='strict')
except UnicodeDecodeError:
return {'error': 'UnicodeDecodeError'}
try:
rows = list(csv.reader(io.StringIO(text, newline=''), **DIALECT))
except csv.Error:
return {'error': 'csv.Error'}
return {'rows': rows, 'header_matches': bool(rows) and rows[0] == HEADER}
def diagnose(raw):
original = inspect_bytes(raw)
return {
'source': {'byte_count': len(raw),
'sha256': hashlib.sha256(raw).hexdigest(),
'hex': raw.hex(' ')},
'published': {'status': original['status'],
'errors': [e['code'] for e in original['errors']]},
'utf-8': decode_rows(raw, 'utf-8'),
'utf-8-sig': decode_rows(raw, 'utf-8-sig'),
}
def make_report():
return {name: diagnose(raw) for name, raw in CASES.items()}
def summary(report):
lines = []
for name, result in report.items():
fields = []
for codec in ['utf-8', 'utf-8-sig']:
view = result[codec]
fields.append(view.get('error') or repr(view['rows'][0][0]))
lines.append(f"{name}: {result['source']['byte_count']} bytes; "
f"published={result['published']['status']}; "
f"utf-8={fields[0]}; utf-8-sig={fields[1]}")
return '\n'.join(lines) + '\n'
def save_bundle(output, report):
"""New directory only. A partial I/O failure is not an atomic bundle."""
output.mkdir(parents=True, exist_ok=False)
for name, raw in CASES.items():
(output / f'{name}.csv').write_bytes(raw)
(output / 'report.json').write_text(
json.dumps(report, ensure_ascii=True, indent=2) + '\n', encoding='utf-8')
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--output', required=True, type=Path,
help='new output directory for synthetic byte copies and report')
args = parser.parse_args()
report = make_report()
save_bundle(args.output, report)
print(summary(report), end='')
if __name__ == '__main__':
main()
21. 补充完整预期：bom_expected.json
以下预期包含每个输入的全部十六进制字节、源哈希、完整字段列表和解码错误。字段列表由案例要求手写，不从诊断程序的运行结果反推。JSON 中的 \ufeff 是 Unicode 转义写法，读取后表示那个字符本身。
{
"plain": {
"source": {
"byte_count": 35,
"sha256": "adb2e885d7d15fa7f63e108c7df017b78205981dd56735bda6aaa4ad2a702134",
"hex": "69 74 65 6d 5f 69 64 2c 6c 61 62 65 6c 2c 6e 6f 74 65 0d 0a 30 30 30 37 2c 4e 41 2c 70 6c 61 69 6e 0d 0a"
},
"published": {
"status": "ready",
"errors": []
},
"utf-8": {
"rows": [
[
"item_id",
"label",
"note"
],
[
"0007",
"NA",
"plain"
]
],
"header_matches": true
},
"utf-8-sig": {
"rows": [
[
"item_id",
"label",
"note"
],
[
"0007",
"NA",
"plain"
]
],
"header_matches": true
}
},
"one_bom": {
"source": {
"byte_count": 38,
"sha256": "e3d7c7ae36be23cda31b8e36bdb2c48a4496cd7283ed8494cba2f27497338d8b",
"hex": "ef bb bf 69 74 65 6d 5f 69 64 2c 6c 61 62 65 6c 2c 6e 6f 74 65 0d 0a 30 30 30 37 2c 4e 41 2c 70 6c 61 69 6e 0d 0a"
},
"published": {
"status": "blocked",
"errors": [
"header_mismatch"
]
},
"utf-8": {
"rows": [
[
"\ufeffitem_id",
"label",
"note"
],
[
"0007",
"NA",
"plain"
]
],
"header_matches": false
},
"utf-8-sig": {
"rows": [
[
"item_id",
"label",
"note"
],
[
"0007",
"NA",
"plain"
]
],
"header_matches": true
}
},
"double_bom": {
"source": {
"byte_count": 41,
"sha256": "d0a80e7450090d807c06b7ec47222328ab4150be1d60eda16e517dc2aad806f4",
"hex": "ef bb bf ef bb bf 69 74 65 6d 5f 69 64 2c 6c 61 62 65 6c 2c 6e 6f 74 65 0d 0a 30 30 30 37 2c 4e 41 2c 70 6c 61 69 6e 0d 0a"
},
"published": {
"status": "blocked",
"errors": [
"header_mismatch"
]
},
"utf-8": {
"rows": [
[
"\ufeff\ufeffitem_id",
"label",
"note"
],
[
"0007",
"NA",
"plain"
]
],
"header_matches": false
},
"utf-8-sig": {
"rows": [
[
"\ufeffitem_id",
"label",
"note"
],
[
"0007",
"NA",
"plain"
]
],
"header_matches": false
}
},
"embedded_header": {
"source": {
"byte_count": 38,
"sha256": "e6a1402c0b1ebd006825426e27c858f85fcae3ffe88e131cf2841d11b19ccd92",
"hex": "69 74 65 6d 5f 69 64 2c ef bb bf 6c 61 62 65 6c 2c 6e 6f 74 65 0d 0a 30 30 30 37 2c 4e 41 2c 70 6c 61 69 6e 0d 0a"
},
"published": {
"status": "blocked",
"errors": [
"header_mismatch"
]
},
"utf-8": {
"rows": [
[
"item_id",
"\ufefflabel",
"note"
],
[
"0007",
"NA",
"plain"
]
],
"header_matches": false
},
"utf-8-sig": {
"rows": [
[
"item_id",
"\ufefflabel",
"note"
],
[
"0007",
"NA",
"plain"
]
],
"header_matches": false
}
},
"embedded_value": {
"source": {
"byte_count": 42,
"sha256": "f5bf8da7781f3a42e5ecdeeaf8a7cf00f4c55fec944133aeb64ea322fece60cf",
"hex": "69 74 65 6d 5f 69 64 2c 6c 61 62 65 6c 2c 6e 6f 74 65 0d 0a 30 30 30 37 2c 4e 41 2c 6c 65 66 74 ef bb bf 72 69 67 68 74 0d 0a"
},
"published": {
"status": "ready",
"errors": []
},
"utf-8": {
"rows": [
[
"item_id",
"label",
"note"
],
[
"0007",
"NA",
"left\ufeffright"
]
],
"header_matches": true
},
"utf-8-sig": {
"rows": [
[
"item_id",
"label",
"note"
],
[
"0007",
"NA",
"left\ufeffright"
]
],
"header_matches": true
}
},
"bom_before_quoted_header": {
"source": {
"byte_count": 40,
"sha256": "b42234a3266beffe8a1df74c36498cdfb6f8ee187538840566bd124e049bb981",
"hex": "ef bb bf 22 69 74 65 6d 5f 69 64 22 2c 6c 61 62 65 6c 2c 6e 6f 74 65 0d 0a 30 30 30 37 2c 4e 41 2c 70 6c 61 69 6e 0d 0a"
},
"published": {
"status": "blocked",
"errors": [
"header_mismatch"
]
},
"utf-8": {
"rows": [
[
"\ufeff\"item_id\"",
"label",
"note"
],
[
"0007",
"NA",
"plain"
]
],
"header_matches": false
},
"utf-8-sig": {
"rows": [
[
"item_id",
"label",
"note"
],
[
"0007",
"NA",
"plain"
]
],
"header_matches": true
}
},
"bom_invalid_utf8": {
"source": {
"byte_count": 34,
"sha256": "8f5124e1fd06bce0fad315da65559e6fe422b269f1755f15adfc7ba8d1db4f68",
"hex": "ef bb bf 69 74 65 6d 5f 69 64 2c 6c 61 62 65 6c 2c 6e 6f 74 65 0d 0a 30 30 30 37 2c 4e 41 2c ff 0d 0a"
},
"published": {
"status": "blocked",
"errors": [
"invalid_utf8"
]
},
"utf-8": {
"error": "UnicodeDecodeError"
},
"utf-8-sig": {
"error": "UnicodeDecodeError"
}
},
"partial_signature": {
"source": {
"byte_count": 2,
"sha256": "f257f0501a5e137710e26f1c35ddd32ce2b5752f20274a648cb6d0347849a5a9",
"hex": "ef bb"
},
"published": {
"status": "blocked",
"errors": [
"invalid_utf8"
]
},
"utf-8": {
"error": "UnicodeDecodeError"
},
"utf-8-sig": {
"error": "UnicodeDecodeError"
}
}
}
22. 补充测试：test_bom.py
测试既检查正例，也让五种错误结果实际触发断言失败。另有带引号表头的时序检查，以及输出文件逐字节一致、已有目录不可覆盖的检查。
import copy
import csv
import hashlib
import io
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from bom_fixtures import CASES
from bom_diagnostic import decode_rows, diagnose, make_report, summary
from intake import DIALECT, HEADER, accepted_rows, inspect_bytes
ROOT = Path(__file__).resolve().parent
EXPECTED = json.loads((ROOT / 'bom_expected.json').read_text(encoding='utf-8'))
def parse_text(text):
return list(csv.reader(io.StringIO(text, newline=''), **DIALECT))
class BomTests(unittest.TestCase):
def test_all_full_rows_and_errors(self):
self.assertEqual(set(CASES), set(EXPECTED))
for name, raw in CASES.items():
with self.subTest(case=name):
self.assertEqual(diagnose(raw), EXPECTED[name])
def test_raw_hashes_and_reversible_hex(self):
for name, raw in CASES.items():
with self.subTest(case=name):
source = diagnose(raw)['source']
self.assertEqual(bytes.fromhex(source['hex']), raw)
self.assertEqual(source['sha256'], hashlib.sha256(raw).hexdigest())
self.assertEqual(source['byte_count'], len(raw))
def test_optional_signature_does_not_grant_acceptance(self):
self.assertEqual(decode_rows(CASES['one_bom'], 'utf-8-sig')['rows'],
[['item_id', 'label', 'note'], ['0007', 'NA', 'plain']])
self.assertEqual(inspect_bytes(CASES['one_bom'])['status'], 'blocked')
with self.assertRaises(ValueError):
accepted_rows(CASES['one_bom'])
self.assertEqual(accepted_rows(CASES['plain'])[0]['values']['item_id'], '0007')
def test_one_prefix_only(self):
self.assertEqual(decode_rows(CASES['double_bom'], 'utf-8-sig')['rows'][0],
['\ufeffitem_id', 'label', 'note'])
self.assertFalse(decode_rows(CASES['double_bom'], 'utf-8-sig')['header_matches'])
def test_invalid_bytes_are_not_repaired(self):
for name in ['bom_invalid_utf8', 'partial_signature']:
for codec in ['utf-8', 'utf-8-sig']:
with self.subTest(case=name, codec=codec):
self.assertEqual(decode_rows(CASES[name], codec),
{'error': 'UnicodeDecodeError'})
def test_kill_lstrip_all_prefixes(self):
wrong = diagnose(CASES['double_bom'])
rows = parse_text(CASES['double_bom'].decode('utf-8').lstrip('\ufeff'))
wrong['utf-8-sig'] = {'rows': rows, 'header_matches': rows[0] == HEADER}
self.assertTrue(wrong['utf-8-sig']['header_matches'])
with self.assertRaises(AssertionError):
self.assertEqual(wrong, EXPECTED['double_bom'])
def test_kill_global_replace_value(self):
wrong = diagnose(CASES['embedded_value'])
rows = parse_text(CASES['embedded_value'].decode('utf-8').replace('\ufeff', ''))
wrong['utf-8-sig'] = {'rows': rows, 'header_matches': rows[0] == HEADER}
self.assertEqual(rows[1][2], 'leftright')
with self.assertRaises(AssertionError):
self.assertEqual(wrong, EXPECTED['embedded_value'])
def test_kill_global_replace_hidden_header(self):
wrong = diagnose(CASES['embedded_header'])
rows = parse_text(CASES['embedded_header'].decode('utf-8').replace('\ufeff', ''))
wrong['utf-8-sig'] = {'rows': rows, 'header_matches': rows[0] == HEADER}
self.assertTrue(wrong['utf-8-sig']['header_matches'])
with self.assertRaises(AssertionError):
self.assertEqual(wrong, EXPECTED['embedded_header'])
def test_kill_normalized_hash_identity(self):
wrong = diagnose(CASES['one_bom'])
normalized = CASES['one_bom'].decode('utf-8-sig').encode('utf-8')
wrong['source']['sha256'] = hashlib.sha256(normalized).hexdigest()
self.assertEqual(wrong['source']['sha256'], EXPECTED['plain']['source']['sha256'])
with self.assertRaises(AssertionError):
self.assertEqual(wrong, EXPECTED['one_bom'])
def test_kill_diagnostic_promoted_to_ready(self):
wrong = copy.deepcopy(EXPECTED['one_bom'])
wrong['published'] = {'status': 'ready', 'errors': []}
with self.assertRaises(AssertionError):
self.assertEqual(diagnose(CASES['one_bom']), wrong)
def test_stripping_parsed_cell_is_too_late_for_quotes(self):
rows = decode_rows(CASES['bom_before_quoted_header'], 'utf-8')['rows']
self.assertEqual(rows[0][0].lstrip('\ufeff'), '"item_id"')
self.assertNotEqual(rows[0][0].lstrip('\ufeff'), 'item_id')
self.assertEqual(decode_rows(CASES['bom_before_quoted_header'], 'utf-8-sig')['rows'][0][0],
'item_id')
def test_summary_matches_fixed_output(self):
expected = (ROOT / 'bom_expected_output.txt').read_text(encoding='utf-8')
self.assertEqual(summary(make_report()), expected)
def test_cli_bundle_identity_and_no_overwrite(self):
with tempfile.TemporaryDirectory() as tmp:
output = Path(tmp) / 'bundle'
cmd = [sys.executable, str(ROOT / 'bom_diagnostic.py'), '--output', str(output)]
run = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
self.assertEqual(run.returncode, 0, run.stderr)
self.assertEqual(run.stdout, (ROOT / 'bom_expected_output.txt').read_text())
self.assertEqual(json.loads((output / 'report.json').read_text()), EXPECTED)
for name, raw in CASES.items():
self.assertEqual((output / f'{name}.csv').read_bytes(), raw)
before = {f.name: f.read_bytes() for f in output.iterdir()}
again = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
self.assertNotEqual(again.returncode, 0)
self.assertEqual({f.name: f.read_bytes() for f in output.iterdir()}, before)
if __name__ == '__main__':
unittest.main(verbosity=2)
23. 补充固定输出：bom_expected_output.txt
这份摘要只显示第一个表头字段，不能替代前面的完整预期。每个例子的原入口状态都来自原始 bytes。
plain: 35 bytes; published=ready; utf-8='item_id'; utf-8-sig='item_id'
one_bom: 38 bytes; published=blocked; utf-8='\ufeffitem_id'; utf-8-sig='item_id'
double_bom: 41 bytes; published=blocked; utf-8='\ufeff\ufeffitem_id'; utf-8-sig='\ufeffitem_id'
embedded_header: 38 bytes; published=blocked; utf-8='item_id'; utf-8-sig='item_id'
embedded_value: 42 bytes; published=ready; utf-8='item_id'; utf-8-sig='item_id'
bom_before_quoted_header: 40 bytes; published=blocked; utf-8='\ufeff"item_id"'; utf-8-sig='item_id'
bom_invalid_utf8: 34 bytes; published=blocked; utf-8=UnicodeDecodeError; utf-8-sig=UnicodeDecodeError
partial_signature: 2 bytes; published=blocked; utf-8=UnicodeDecodeError; utf-8-sig=UnicodeDecodeError
24. 可选比较：probe_bom_pandas.py
只有本机已经装有 pandas 2.2.3 时才运行 python src/probe_bom_pandas.py。版本不符或未安装会退出；脚本不安装任何依赖。它比较六份可解码样本的完整列名和数据字符串，预期均写在文件内。
这份版本锁定观察不用于声明当前 pandas 的行为，也不建立多引擎兼容矩阵。外部库的显示结果适合帮助解释现象，本文入口是否继续接收仍由原来的确定性合同决定。
"""Optional, pinned observation. Does not define the intake contract."""
import io
import json
import sys
from bom_fixtures import CASES
def main():
try:
import pandas as pd
except ImportError:
raise SystemExit('Optional probe requires an already installed pandas 2.2.3')
if pd.__version__ != '2.2.3':
raise SystemExit(f'Expected pandas 2.2.3; found {pd.__version__}')
observations = {}
for name in ['plain', 'one_bom', 'double_bom', 'embedded_header',
'embedded_value', 'bom_before_quoted_header']:
frame = pd.read_csv(io.BytesIO(CASES[name]), encoding='utf-8',
dtype=str, keep_default_na=False)
observations[name] = {'columns': frame.columns.tolist(),
'rows': frame.values.tolist()}
expected = {
'plain': {'columns': ['item_id', 'label', 'note'], 'rows': [['0007', 'NA', 'plain']]},
'one_bom': {'columns': ['item_id', 'label', 'note'], 'rows': [['0007', 'NA', 'plain']]},
'double_bom': {'columns': ['\ufeffitem_id', 'label', 'note'], 'rows': [['0007', 'NA', 'plain']]},
'embedded_header': {'columns': ['item_id', '\ufefflabel', 'note'], 'rows': [['0007', 'NA', 'plain']]},
'embedded_value': {'columns': ['item_id', 'label', 'note'], 'rows': [['0007', 'NA', 'left\ufeffright']]},
'bom_before_quoted_header': {'columns': ['item_id', 'label', 'note'], 'rows': [['0007', 'NA', 'plain']]},
}
if observations != expected:
raise RuntimeError('Pinned observation differs from handwritten expectations')
print(json.dumps({'python': sys.version.split()[0], 'pandas': pd.__version__,
'observations': observations}, ensure_ascii=True, indent=2))
if __name__ == '__main__':
main()
25. 补充：相同源哈希还不够，重放时要核对字段上限
2026 年 10 月 20 日复查时，原有 21 个测试和两组 pandas 2.2.3 观察均可重跑，但发现一个未记录的运行条件：csv.field_size_limit()。它不在前文 manifest 的 parse 字段中，也不是 DIALECT 的一部分。Python 文档说明，这个函数读取当前字段上限，传入参数则改变该上限。
使用前文同一份 206 字节 MAIN，在独立子进程把上限设为 16，结果是 blocked，只保留表头，parse_complete=false，错误为 csv_syntax、区间 [2,3]。设为 32 后，恢复原来的 semantics_pending、9 条记录和 R0003/R0004 两个待确认项。两个结果中的 source 与 parse 完全相同。这不是源文件被修改，而是当前清单没有记录足以重放结果的全部运行条件。
另用纯 ASCII 和汉字各造 16、17 字符的 note。上限为 16 时，两种 16 字符字段都能通过，17 字符都阻断；16 个汉字的 UTF-8 编码占 48 字节，并未因为超过 16 字节而失败。本次 CPython 3.12.14 观察的是解码后字段字符数，不是原文件字节预算。对应版本的 _csv.c在 parse_add_char 中比较字段长度和模块状态中的上限，可与固定实验对照。
这个设置在同一解释器内由 csv 模块共享，不是 reader 创建时保存的一份独立配置。补充测试先设 32 并创建 reader，再改成 16，随后读取 17 字符字段仍然失败。因而，在共享进程中临时调高、解析后再恢复，并不能据此宣称其他并发解析不受影响。下面每个观察都新建子进程，并在子进程的 finally 中恢复先前上限，恢复检查失败就不报告成功；父进程的字段上限从未被改写。这只是隔离本实验，不是生产沙箱或并发正确性证明。
排查时应把实际字段上限与解释器版本一并记录，并区分“解析器拒绝”与“已经证明 CSV 语法有错”。若要让入口可稳定重放，需要在受控、不会被其他解析任务修改设置的执行环境中固定并记录上限，再测试边界；不要为让文件通过而直接移除限制。字段上限也不能替代文件字节数、记录数、总输出和内存预算。前文入口源码、错误代码与接收合同在本补充中均未改变。
本补充通过 4 个 unittest 方法，覆盖相同输入不同上限、两种字符的边界、reader 创建后的设置变化，以及子进程不修改父进程设置。普通模式与 -O 均通过，子进程也采用对应模式；原有 21 个方法再次通过。只使用合成输入和 Python 标准库，没有真实业务文件、模型、WorkBuddy、浏览器像素验收、其他 Python 版本或平台的验证。
26. 字段上限复现：test_field_limit.py
前面的 ZIP 仍是原入口与 BOM 两组实验，未混入这份新补充。将以下文件另存为 src/test_field_limit.py，与 ZIP 中的 intake.py 和 fixtures.py 放在同一目录，然后在解压目录运行：
python -m unittest discover -s src -p "test_field_limit.py" -v
python -O -m unittest discover -s src -p "test_field_limit.py" -v每条命令应显示 Ran 4 tests 与 OK。加入本文件后，使用前面的 test_*.py 通配命令会运行共 25 个方法。测试对失败输入仍然期待阻断，没有把诊断成功变成导入成功。
"""Synthetic reproduction of ambient CSV field limits; no intake changes."""
import csv
import json
from pathlib import Path
import subprocess
import sys
import unittest
ROOT = Path(__file__).resolve().parent
WORKER = r'''
import csv
import io
import json
import sys
from fixtures import MAIN
from intake import DIALECT, inspect_bytes
case, limit = sys.argv[1], int(sys.argv[2])
original_limit = csv.field_size_limit()
try:
if case == "late":
csv.field_size_limit(32)
reader = csv.reader(io.StringIO("x" * 17 + "\n", newline=""), **DIALECT)
csv.field_size_limit(limit)
try:
result = {"rows": list(reader)}
except csv.Error:
result = {"error": "csv.Error"}
else:
csv.field_size_limit(limit)
if case == "main":
raw = MAIN
else:
letter, count = case.split(":")
note = ("x" if letter == "ascii" else "\u6c49") * int(count)
raw = ("item_id,label,note\n0007,NA," + note + "\n").encode("utf-8")
m = inspect_bytes(raw)
result = {
"source": m["source"], "parse": m["parse"], "status": m["status"],
"errors": m["errors"], "records": len(m["records"]),
"complete": m["parse_complete"], "pending": m["pending"],
}
if case != "main":
result["note_characters"] = len(note)
result["note_utf8_bytes"] = len(note.encode("utf-8"))
finally:
csv.field_size_limit(original_limit)
if csv.field_size_limit() != original_limit:
raise RuntimeError("CSV limit was not restored")
print(json.dumps(result))
'''
def observe(case, limit):
flags = ["-O"] if sys.flags.optimize else []
run = subprocess.run(
[sys.executable, *flags, "-c", WORKER, case, str(limit)],
cwd=ROOT, capture_output=True, text=True, timeout=10, check=True)
return json.loads(run.stdout)
class FieldLimitTests(unittest.TestCase):
def test_same_source_and_parse_metadata_can_have_different_outcomes(self):
small, large = observe("main", 16), observe("main", 32)
self.assertEqual(small["source"], large["source"])
self.assertEqual(small["source"]["byte_count"], 206)
self.assertEqual(small["source"]["sha256"],
"89573d7d2e60675bed26e986532ef6f17b02b9893a06b35f05a4a0e1a8e0512c")
self.assertEqual(small["parse"], large["parse"])
self.assertNotIn("field_size_limit", small["parse"])
self.assertEqual((small["status"], small["records"], small["complete"]),
("blocked", 1, False))
self.assertEqual(small["errors"],
[{"code": "csv_syntax", "physical_lines": [2, 3]}])
self.assertEqual((large["status"], large["records"], large["complete"]),
("semantics_pending", 9, True))
self.assertEqual(large["errors"], [])
self.assertEqual([p["record_id"] for p in large["pending"]],
["R0003", "R0004"])
def test_boundary_counts_decoded_characters_not_utf8_bytes(self):
for alphabet, width in [("ascii", 1), ("han", 3)]:
for count in [16, 17]:
with self.subTest(alphabet=alphabet, count=count):
result = observe(f"{alphabet}:{count}", 16)
self.assertEqual(result["note_characters"], count)
self.assertEqual(result["note_utf8_bytes"], width * count)
self.assertEqual(result["status"],
"ready" if count == 16 else "blocked")
self.assertEqual(result["records"], 2 if count == 16 else 1)
self.assertEqual(result["complete"], count == 16)
self.assertEqual(result["errors"], [] if count == 16 else
[{"code": "csv_syntax", "physical_lines": [2, 2]}])
def test_existing_reader_uses_later_module_limit(self):
self.assertEqual(observe("late", 16), {"error": "csv.Error"})
self.assertEqual(observe("late", 32), {"rows": [["x" * 17]]})
def test_children_do_not_change_host_setting(self):
before = csv.field_size_limit()
observe("main", 16)
self.assertEqual(csv.field_size_limit(), before)
if __name__ == "__main__":
unittest.main(verbosity=2)