让 AI 写出一个可以运行的 CSV 去重脚本并不难,真正容易被忽略的是结果是否可靠。简单样例往往覆盖不了列顺序变化、带引号字段、单元格换行和非法数据等情况。接下来将从明确去重规则开始,用八个测试逐步发现问题、修正实现,并梳理 AI 生成代码交付前应完成的验收工作。
让 AI 写一个 CSV 去重脚本,要求按 id 去重,保留第一条记录。
拿几行数据一试,重复项确实没了。但如果把 id 挪到第二列,或者在单元格里加一个换行,脚本还能给出正确结果吗?
下面就拿这样一个小脚本做实验。先看它在哪些数据上出错,再补上解析和校验,最后用 8 个测试检查修改结果。代码可以直接复制运行。
实验说明:下面是为教学构造的演示版和修正版,代码由 AI 辅助整理,已在 Python 3.9.6 环境实际执行。演示版故意保留常见简化写法;本文不把它归因于某个模型,也不据此评价模型能力。测试数据均为自造数据。
写代码之前,得先确定“去重”具体指什么。本例按下面这些规则处理:
这里有些选择取决于业务。比如,id 两侧的空格要不要去掉?本例保留原样。如果你的系统要求去掉,就把规则和测试一起改掉,别让脚本替你做决定。
def demo_dedupe(text):
lines = text.splitlines()
result, seen = [lines[0]], set()
for line in lines[1:]:
key = line.split(',')[0]
if key not in seen:
seen.add(key)
result.append(line)
return 'n'.join(result) + 'n'
先给它一份最简单的数据。运行后,2,B 和 1,A 都在,后面的 2,C 被删除,符合预期:
id,name
2,B
1,A
2,C
问题在于,这份数据刚好绕开了代码里的几个假设。
代码拿第一列当 id。如果 id 实际在第二列,两条 name 相同、id 不同的记录就会被误删。
直接按逗号切开也有问题:带引号的 a,1 和 a,2 本来是两个完整的 id,切开后却只剩下相同的前半截。
字段里有换行时,问题更不容易发现。一个单元格里的多行文字会被拆成多条记录,其中重复的文字可能被删掉。程序没有报错,数据却变了。
逗号、引号和字段内的换行,交给 Python 的 csv 模块处理就好。它默认把字段读成字符串,也能保留 001 这样的值。文件读写时如何设置 newline,可以查 Python 官方 csv 文档。
剩下的是业务校验:有没有 id 列,列名是否重复,每条记录的列数对不对,id 能不能为空。这些都需要自己写。
下面的函数接收已经解码的文本,校验成功后,返回去重后的 CSV 文本:
import csv
import io
def dedupe_csv(text):
reader = csv.reader(io.StringIO(text, newline=''), strict=True)
output = io.StringIO(newline='')
writer = csv.writer(output, lineterminator='n')
try:
header = next(reader, None)
if not header or any(not name.strip() for name in header):
raise ValueError('表头不能为空')
if len(set(header)) != len(header) or 'id' not in header:
raise ValueError('表头必须唯一且包含 id')
key_index = header.index('id')
writer.writerow(header)
seen = set()
for row in reader:
if len(row) != len(header):
raise ValueError(f'截至物理行 {reader.line_num}:列数不匹配')
key = row[key_index]
if not key.strip():
raise ValueError(f'截至物理行 {reader.line_num}:id 不能为空')
if key not in seen:
seen.add(key)
writer.writerow(row)
except csv.Error as exc:
raise ValueError(f'CSV 解析失败:{exc}') from exc
return output.getvalue()
这里有两个细节。
错误提示写的是“截至物理行”。一条 CSV 记录可能跨好几行,所以解析器报出的行号,不一定是第几条数据。
函数只在整段数据校验成功后返回文本,不直接改写源文件。真正写入文件时,可以先保存到另一个文件,核对结果后再替换原数据。
把前面的两个函数、import 语句和下面的测试放进同一个 Python 文件,就能运行。
测试会把输出重新读成 CSV,比较字段值和记录顺序。没有直接比较整段文本,是因为重新写出后,换行符或引号的写法可能变化;我们要确认的是数据有没有被改错。
CASES = [
('重复 id 保留首条和原顺序',
'id,namen2,Bn1,An2,Cn',
[['id', 'name'], ['2', 'B'], ['1', 'A']]),
('id 不在第一列',
'name,idnA,1nA,2nB,1n',
[['name', 'id'], ['A', '1'], ['A', '2']]),
('带引号的 id 内含逗号',
'id,noten"a,1",Xn"a,2",Yn',
[['id', 'note'], ['a,1', 'X'], ['a,2', 'Y']]),
('单元格内含多行文字',
'id,noten1,"startnsamensamenend"n',
[['id', 'note'], ['1', 'startnsamensamenend']]),
('前导零必须保留',
'id,namen001,An1,Bn',
[['id', 'name'], ['001', 'A'], ['1', 'B']]),
('空 id 应拒绝', 'id,namen,An', ValueError),
('缺少 id 表头应拒绝', 'code,namen1,An', ValueError),
('列数不匹配应拒绝', 'id,namen1,A,extran', ValueError),
]
def verify(fn, source, expected):
try:
result = fn(source)
except Exception as exc:
return expected is ValueError and isinstance(exc, ValueError)
if expected is ValueError:
return False
actual = list(csv.reader(io.StringIO(result, newline='')))
return actual == expected
for fn in (demo_dedupe, dedupe_csv):
passed = 0
print('n' + fn.__name__)
for label, source, expected in CASES:
ok = verify(fn, source, expected)
passed += ok
print(('PASS' if ok else 'FAIL') + ' | ' + label)
print(f'{passed}/{len(CASES)} passed')
在 Python 3.9.6 中运行这 8 个用例,结果如下:
| 验收项 | 演示版 | 修正版 |
|---|---|---|
| 重复 id 保留首条和原顺序 | 通过 | 通过 |
| id 不在第一列 | 失败 | 通过 |
| 带引号的 id 内含逗号 | 失败 | 通过 |
| 单元格内含多行文字 | 失败 | 通过 |
| 前导零必须保留 | 通过 | 通过 |
| 空 id 应拒绝 | 失败 | 通过 |
| 缺少 id 表头应拒绝 | 失败 | 通过 |
| 列数不匹配应拒绝 | 失败 | 通过 |
演示版通过 2 项,修正版通过 8 项。
这能说明修正版处理了这里列出的情况,但不能说明它放进生产环境就不会出错。用例是针对已知问题设计的,并非随机抽样;没有测到的地方,还需要继续检查。
8 项都通过了,这个函数还只能处理约定范围内的数据。用之前,下面几件事也得考虑:
处理几百行临时文本,这个实现可能够用。每天处理几 GB 的客户数据,就要另外考虑流式读写、去重索引,以及处理到一半失败后怎么办。
先不要写代码。请把任务拆成:
1. 输入格式与有效范围;
2. 输出必须保持的规则;
3. 重复值、空值和非法输入的处理方式;
4. 至少一个正常用例、边界用例和失败用例。
请先列出仍需我确定的业务规则,不要自行补全。
规则确定后再实现,并逐项运行测试。
最后分别说明:已验证的行为、失败项、尚未覆盖的场景。
这段提示词能帮你把要求说清楚,写出来的代码仍然要跑测试。把预期结果和用例一起留下,下次换模型或修改实现时,就能重新检查,而不用再凭“看着没问题”来判断。
你遇到过脚本没报错、数据却被处理错的情况吗?可以在评论里说说是哪类数据、出了什么问题,不用贴真实业务数据。