最近用 Codex 把一个重复了三次的 CSV 合并需求做完了,从需求确认、脚本生成到验收测试,整个流程比想象中顺。三份 CSV 分别是不同月份导出的销售明细,字段结构一致,但因为来源系统不同,存在表头重复、日期格式不统一、个别空行和重复主键的问题。如果靠手工打开 Excel 复制粘贴,不仅慢,而且极容易漏数据;我选择直接在终端里让 Codex 参与,用自然语言把需求说清楚,让它生成脚本,我再人工检查逻辑、补上边界处理,最后写验收测试确认结果没问题。
这篇就完整记录一下当时的做法:需求怎么看、Codex 怎么装、脚本怎么生成、验收测试怎么做,以及踩过的几个环境坑。如果你也是第一次接触 Codex CLI,或者正在纠结怎么让 AI 帮你处理 CSV,这篇文章正好能给你一条可以照搬的路径,脚本和命令我都会贴出来。
1. 需求拆解:三份CSV合并到底要解决什么问题
1.1 数据背景:三份月度明细的痛点
先说实际场景。我手上拿到jan.csv、feb.csv、mar.csv三个文件,每份大概几百到上千行,结构都是相同的列:order_id、date、customer_id、product_id、amount、status。需求方最终要一份merged.csv,里面包含这三个月的全部有效订单,并且要求:
- 去掉完全重复的行;
order_id必须唯一,不能出现同一个订单在合并后出现两次;- 日期格式统一为
YYYY-MM-DD; - 金额保留两位小数;
- 排除掉
status为cancelled的订单,但要在统计里报告被过滤的数量; - 最后按日期升序排列。
这种需求听起来很简单,但真正处理的时候会发现一堆细节。比如第一份文件是带 BOM 的 UTF-8,第二份是 GBK 编码,第三份虽然是无 BOM 的 UTF-8,但里面有 Windows 换行符。如果直接用 Excel 或者普通文本编辑器去合并,很容易出现中文乱码、表头被当成数据、空行混入等问题。
我之所以强调“需求拆解”,是因为这一步直接决定脚本怎么写。如果连“哪些列参与去重”“空字符串算不算有效值”都没想清楚,AI 生成出来的脚本大概率只能跑通表面流程,经不起验收。
1.2 合并方案的边界:追加、关联还是清洗
CSV 合并其实有三种常见形态,很多人一开始会混在一起:
- 追加合并:多份文件结构相同,直接上下拼接。这次的月度明细就属于这种。
- 横向关联:不同文件有不同的列,需要按某个主键把它们接成一张宽表,比如订单表关联用户表。
- 清洗合并:在追加或关联基础上,还要做去重、格式规范化、过滤脏数据。这类需求最容易被低估。
这次三个文件其实是第 1 种加第 3 种。也就是说,核心动作是“追加”,但真正花时间的是“清洗”。如果直接cat jan.csv feb.csv mar.csv > merged.csv,那只能得到一个文件,完全没法解决表头重复和垃圾数据的问题。所以在跟 Codex 描述需求时,我把重点放在了字段规则、去重逻辑和输出格式上,而不是简单说“帮我合并三个 CSV”。
1.3 为什么用Codex而不是手写脚本
有人可能会问:这种脚本自己写也不难,为什么要用 Codex?我的理由很实际:这类脚本本身不难,但很容易在边界处理上遗漏,比如文件编码、空值、重复主键、路径分隔符等。让 Codex 生成初版脚本,我再补充测试用例和异常分支,效率比自己从零写要高。
Codex 在这里的角色不是“自动代替我想”,而是“帮我搭骨架”。我会用自然语言告诉它文件有哪些、列名是什么、输出规则是什么,它会给出一个能运行的 Python 脚本。然后我逐行检查关键部分,把不够严谨的地方改掉。这个流程其实和带一个初级开发干活很像:AI 先出 draft,我做 code review。
2. 环境准备:把 Codex CLI 跑起来
2.1 安装前置:Node.js 和 npm 的那些坑
Codex CLI 本质是一个命令行工具,最常见的安装方式是通过 npm 全局安装。所以第一步先确认机器上有 Node.js 和 npm。
如果你在 Windows 的 PowerShell 里敲npm -v提示:
npm : 无法将“npm”项识别为 cmdlet、函数、脚本文件或可运行程序的名称。那就是环境变量没配好,或者 Node.js 根本没装上。我一般会先检查:
node -v能不能输出版本号;npm -v能不能输出版本号;- 如果不行,去 Node.js 官网下载 LTS 版本安装,安装时记得勾选“Add to PATH”。
顺带一提,codex命令本身也可能遇到同样的问题:
codex : 无法将“codex”项识别为 cmdlet、函数、脚本文件或可运行程序的名称。这种情况要么是 npm 全局安装的目录不在 PATH 里,要么是安装失败。可以试试重新执行全局安装命令,或者直接用npx codex的方式运行,能规避一部分 PATH 问题。
2.2 安装 Codex CLI 并完成基础配置
我当时是在 macOS 上操作的,安装命令很简单:
npm install -g @openai/codex安装完之后,运行codex --version确认版本。如果是第一次使用,还需要做基本配置,通常是设置 API Key 或登录账号。这个过程网上信息很杂,但核心就是让 Codex CLI 知道用什么身份去调用模型接口。
配置完成后,我习惯先在交互模式里跑一句最简单的测试:
codex exec "把下面这句话翻译成英语:你好"如果它能正常返回结果,说明整个链路已经通了。此时再进入正式的 CSV 合并需求,避免后面真正使用时突然冒出来一堆环境问题。
2.3 用自然语言向 Codex 描述需求
Codex 不是搜索引擎,它的输出质量非常依赖你给的上下文。我把这次需求写成了下面这样,几乎就是一段简单的任务说明:
我有三个CSV文件:jan.csv、feb.csv、mar.csv。 它们的列名都是:order_id, date, customer_id, product_id, amount, status。 请写一个Python脚本,实现: 1. 读取这三个文件,自动识别UTF-8和GBK编码; 2. 按 order_id 去重,保留第一次出现的记录; 3. 过滤掉 status 为 cancelled 的订单; 4. 把 date 统一成 YYYY-MM-DD 格式; 5. amount 保留两位小数; 6. 按 date 升序排序; 7. 输出 merged.csv,同时在控制台打印每个文件的原始行数、过滤行数、最终行数。这里有个很重要的技巧:把验收标准直接写进需求里。这样 Codex 生成的脚本会天然带统计信息,后面做验收测试时能直接用这些输出来对比。如果只说“帮我合并文件”,它大概率只会给你一个简单拼接,不会考虑去重、编码这些细节。
3. 完整脚本:Codex 生成的合并脚本与设计
3.1 脚本整体设计:输入输出、容错、可追溯
Codex 给的第一版脚本用了 Python 的标准库csv,没有依赖 pandas,这样在绝大多数机器上都能直接跑。整体结构是:
- 定义输入文件列表和输出文件路径;
- 写一个读取函数,尝试多种编码;
- 用一个字典保存去重后的记录,
order_id作为 key; - 边读边统计每份文件的原始行数和被过滤的数量;
- 最后统一排序并写入输出文件。
这个设计的好处是内存占用可控,对于几千行的 CSV 完全够用。如果数据量达到几十万行,可能需要用流式处理,但这次不需要。更重要的是,脚本在运行时会打印统计信息,这为后面的验收测试提供了依据。
当然,Codex 生成的脚本并不完美。第一版里它直接用order_id作为主键去重,但没有考虑文件中可能出现完全重复的行且order_id相同但其他字段不同。当时我人工补了一个逻辑:如果多个order_id相同,保留最新日期的那一行。这个细节在需求里没有明确,但实际业务中很常见。
3.2 核心代码逐段解析
最终我保留了 Codex 生成的骨架,并手工调整了它的一些处理逻辑。下面这段是完整可运行的版本,你可以直接替换文件路径使用。
#!/usr/bin/env python3 # -*- coding: utf-8 -*- import csv import sys from pathlib import Path INPUT_FILES = ["jan.csv", "feb.csv", "mar.csv"] OUTPUT_FILE = "merged.csv" KEY_COLUMN = "order_id" DATE_COLUMN = "date" ENCODINGS = ["utf-8-sig", "utf-8", "gbk"] def read_csv_with_fallback(path): """尝试多编码读取,返回 DictReader 的行列表。""" for enc in ENCODINGS: try: with open(path, "r", encoding=enc, newline="") as f: return list(csv.DictReader(f)) except UnicodeDecodeError: continue raise ValueError(f"无法识别文件编码: {path}") def normalize_date(value): """把常见日期格式统一成 YYYY-MM-DD。""" value = value.strip() for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%d/%m/%Y", "%m/%d/%Y"): try: from datetime import datetime return datetime.strptime(value, fmt).strftime("%Y-%m-%d") except ValueError: continue return value def normalize_amount(value): """保留两位小数,非数字返回 0。""" try: return f"{float(value):.2f}" except ValueError: return "0.00" def main(): all_rows = {} total_raw = 0 total_cancelled = 0 total_duplicate = 0 for path in INPUT_FILES: if not Path(path).exists(): print(f"警告: 文件不存在 {path}") continue rows = read_csv_with_fallback(path) print(f"读取 {path}: {len(rows)} 行") total_raw += len(rows) for row in rows: # 跳过空行 if not row or not row.get(KEY_COLUMN): continue # 过滤取消订单 if row.get("status", "").strip().lower() == "cancelled": total_cancelled += 1 continue # 规范化字段 row["date"] = normalize_date(row.get("date", "")) row["amount"] = normalize_amount(row.get("amount", "")) key = row.get(KEY_COLUMN).strip() if key in all_rows: total_duplicate += 1 # 保留更晚的日期 if row.get("date") > all_rows[key].get("date"): all_rows[key] = row else: all_rows[key] = row # 按日期排序 sorted_rows = sorted(all_rows.values(), key=lambda r: r.get(DATE_COLUMN, "")) # 写输出 if not sorted_rows: print("没有有效数据,不生成输出文件") sys.exit(1) fieldnames = ["order_id", "date", "customer_id", "product_id", "amount", "status"] with open(OUTPUT_FILE, "w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(sorted_rows) print(f"\n统计结果:") print(f"原始行数: {total_raw}") print(f"过滤取消订单: {total_cancelled}") print(f"重复订单: {total_duplicate}") print(f"最终行数: {len(sorted_rows)}") if __name__ == "__main__": main()代码里最关键的两点:
- 多编码尝试:先试着用带 BOM 的 UTF-8 读,再试纯 UTF-8,最后试 GBK。这解决了很多乱码问题。
- 去重策略:用
order_id作为主键,如果出现重复,保留日期更大的那条记录。这个逻辑比简单“保留第一次出现”更贴近业务。
3.3 机器生成的脚本也要人工检查
很多人以为让 Codex 生成脚本就能直接跑,其实不行。它生成的代码通常能运行,但业务逻辑是否准确必须人工把关。我在拿到初版后做了三件事:
- 一遍逐行读代码,重点看列名是否和实际 CSV 的表头一致;
- 用一份只有 5 行的小样本文件去跑,观察输出是否符合预期;
- 让 Codex 解释它为什么这么设计,尤其去重和排序逻辑。
人工检查并不代表不信任 Codex,而是因为 AI 没有见过真实数据的“隐藏规则”。比如我当时发现第一份 CSV 的表头前面有一个不可见字符,如果不处理,order_id这个列名是匹配不上的。这种问题只靠看代码发现不了,必须结合实际数据。
4. 验收测试:怎么证明合并结果是对的
4.1 验收标准:从需求反推检查项
脚本跑完只是第一步,重点是要能回答“合并结果对不对”。我把验收标准分成了四个层面:
- 文件级:
merged.csv存在,且不是空文件; - 数量级:最终行数 + 取消订单数 + 重复订单数 = 三份文件原始行数之和,前提是没有额外空行干扰;
- 主键唯一性:
order_id没有重复; - 数据一致性:每条记录的必填字段非空,日期格式正确,金额是两位小数。
这个思路也可以用到其他 CSV 合并场景里:不要只看“能不能合并”,还要看“合并完之后数字对不对得上”。我在验收脚本里直接把这几条变成了自动化检查。
4.2 写一个验收脚本:行数、主键、MD5
除了人工抽查,我写了一个独立验收脚本。它的作用不是生成 merged.csv,而是验证 merged.csv 是否满足所有条件。
#!/usr/bin/env python3 # -*- coding: utf-8 -*- import csv import hashlib import sys from pathlib import Path OUTPUT_FILE = "merged.csv" KEY_COLUMN = "order_id" def file_md5(path): """分块计算文件的 MD5,避免大文件一次性读入内存。""" h = hashlib.md5() with open(path, "rb") as f: chunk = f.read(8192) while chunk: h.update(chunk) chunk = f.read(8192) return h.hexdigest() def main(): if not Path(OUTPUT_FILE).exists(): print("失败: 输出文件不存在") sys.exit(1) rows = [] with open(OUTPUT_FILE, "r", encoding="utf-8-sig", newline="") as f: reader = csv.DictReader(f) fieldnames = reader.fieldnames rows = list(reader) errors = [] # 1. 文件非空 if len(rows) == 0: errors.append("输出文件为空") # 2. 字段完整性 required_fields = ["order_id", "date", "customer_id", "product_id", "amount", "status"] for field in required_fields: if field not in fieldnames: errors.append(f"缺少字段: {field}") # 3. 主键唯一性 seen = set() for i, row in enumerate(rows): key = row.get(KEY_COLUMN, "").strip() if not key: errors.append(f"第 {i+1} 行 order_id 为空") elif key in seen: errors.append(f"第 {i+1} 行 order_id 重复: {key}") seen.add(key) # 4. 日期格式 date_patterns = set() for i, row in enumerate(rows): date_val = row.get("date", "").strip() if len(date_val) != 10: errors.append(f"第 {i+1} 行日期格式异常: {date_val}") date_patterns.add(date_val) # 5. 金额格式 for i, row in enumerate(rows): amount_val = row.get("amount", "").strip() parts = amount_val.split(".") if len(parts) != 2 or len(parts[1]) != 2: errors.append(f"第 {i+1} 行金额格式异常: {amount_val}") # 6. 输出统计信息 total_cancelled = 0 for row in rows: if row.get("status", "").strip().lower() == "cancelled": total_cancelled += 1 if total_cancelled > 0: errors.append(f"输出文件中仍包含已取消订单: {total_cancelled} 行") print(f"输出行数: {len(rows)}") print(f"不同日期数量: {len(date_patterns)}") print(f"文件 MD5: {file_md5(OUTPUT_FILE)}") if errors: print("\n验收失败:") for e in errors: print(f"- {e}") sys.exit(1) else: print("\n验收通过") print(f"输出文件: {OUTPUT_FILE} ({Path(OUTPUT_FILE).stat().st_size} bytes)") if __name__ == "__main__": main()关于 MD5 校验,很多人会问“CSV 文件怎么进行 MD5 校验”。其实它不复杂:MD5 是针对文件二进制内容的哈希值。同一个文件复制到另一台电脑上,MD5 应该完全一致,所以它常被用来验证“文件是否被完整传输”或“文件是否被改动过”。
命令行下可以直接用系统工具:
# macOS / Linux md5sum merged.csv # Windows PowerShell Get-FileHash merged.csv -Algorithm MD5如果两边的哈希值一样,就说明文件内容完全一致。在验收脚本里,计算 MD5 的主要目的是留一个固定快照,方便以后对比“这份 merged.csv 有没有被二次修改”。不过要注意:MD5 只是完整性校验,不是数据正确性校验。它不能告诉你order_id是否唯一,也不能告诉你日期格式是否规范。所以要搭配前面的业务规则检查一起用。
4.3 手动抽查与边界用例
自动化脚本跑通后,我还会手动抽查几行。比如从原文件里随机抽一个order_id,到 merged.csv 里找它,确认金额、日期、状态没有被改错。这个动作看起来原始,但非常有效,能发现自动化脚本没覆盖到的业务错误。
边界用例我也专门测了几种:
- 如果某一行
order_id是空字符串,脚本应该跳过而不是报错; - 如果三份文件里同一个订单出现两次,但一份状态是 completed,一份是 cancelled,到底算不算重复?我在脚本里的逻辑是先过滤 cancelled 再去重,所以最终保留的是 completed;
- 如果某一行
amount是abc,会被转成0.00,验收时我需要确认是否合理。
这些边界情况不一定每个项目都一样,但如果你正在做类似需求,一定要在验收阶段把它们列出来。宁可多测试几次,也不要等下游同事发现数据有问题再回来排查。
5. 常见问题与排查实录
5.1 环境类问题速查表
在准备环境阶段,最容易卡住的就是各种命令无法识别。我把这次遇到的问题和解决方案整理成一个速查表。
| 症状 | 可能原因 | 处理方式 |
|---|---|---|
npm无法识别 | Node.js 未安装或未加入 PATH | 安装适合系统的 Node.js LTS 版本,勾选自动配置 PATH |
codex无法识别 | npm 全局安装目录不在 PATH | 检查 npm 全局 bin 目录,临时用npx codex代替 |
| Codex 提示模型不支持 | 当前账号或配置的模型版本不支持该命令 | 检查 Codex 配置中的模型名,换成合适的模型 |
| Codex 上下文溢出 | 对话太多,模型上下文窗口占满 | 把需求拆小,给 Codex 提供精简后的文件摘要,而不是粘贴整个 CSV |
| Python 读取 CSV 中文乱码 | 文件编码不匹配 | 脚本里增加utf-8-sig、utf-8、gbk多编码尝试 |
这里面我特别想强调的是“Codex 上下文溢出”的问题。我们很容易在一段对话里塞大量内容,最后模型提示运行空间不够。解决方案不是换更大的模型,而是学会精简输入:把 CSV 文件内容先做描述性统计,或者只让 Codex 看几行样例,让它写脚本,而不是让它直接处理全部数据。
5.2 数据处理类问题
除了环境问题,CSV 本身也有一些很隐蔽的坑。最常见的是:
- 带 BOM 的表头:用普通文本编辑器打开没问题,但在 Python 里字段名会变成
\ufefforder_id。解决办法是读取时用utf-8-sig编码; - 混合换行符:有些文件是
\r\n,有些是\n,直接读可能会导致多出的空行。csv模块在读取时声明newline=""能避免大部分问题; - 日期格式不统一:有
2024/01/05,有05/01/2024,脚本里必须先明确规则,否则排序会乱。
这些坑都不是 Codex 能凭“聪明”跳过的,它需要你明确告诉它数据长什么样、规则是什么。所以我建议在跑正式处理前,先用命令看一下文件的前几行和编码,比如在 Linux/macOS 上有file jan.csv,Windows 上可以用编辑器查看。
5.3 Codex 使用中的几个坑
最后说几个 Codex 本身的实操体会。
第一,Codex 生成的代码风格不一定统一,有时会用 pandas,有时会用标准库。如果你像我一样不想额外装依赖,就在需求里明确写“使用 Python 标准库 csv 模块,不要使用 pandas”。它通常会照做。
第二,发现 Codex 生成的脚本有问题时,可以直接在对话里追问“这里为什么这样写”,也可以让它重新生成某个函数,不需要推翻重来。它的上下文记忆能力挺强,但一旦对话太长,可能会忘记前面的规定。所以我习惯把关键约束写在第一条消息里,中间如果发现它省略了某个约束,就重新强调一遍。
第三,千万别让 Codex 在没有任何数据样例的情况下硬写解析逻辑。我踩过一次坑:它默认了所有整数列都是字符串,结果后面做数据统计时才发现amount被当成了字符串,排序完全不对。后来我改成先让它读前 5 行数据,再生成脚本,效果好很多。
最后再说个小技巧:合并之前先备份原始 CSV。哪怕脚本跑得很顺,也要保留原始文件,一方面方便验收对比,另一方面万一后续发现处理规则有偏差,还能重新跑一遍。我一般会把原始文件放在data/raw/目录下,输出放在data/processed/,这样结构清晰,也好写脚本。
这次整个流程走下来,我的体会是:Codex 这类工具最擅长的不是“替你思考”,而是“快速把你想清楚的需求转成可运行的代码”。如果你自己都没想明白合并规则、验收标准,直接丢一句“帮我合并CSV”给 AI,大概率只能得到一个表面能跑、实际经不起推敲的脚本。反过来,当你把需求拆得越细,验收标准定得越硬,Codex 能发挥的作用就越大。至少在处理三份 CSV 合并这件事上,我从开始到验收只用了不到半天,这在以前是不可想象的。