ARTICLE DETAIL

资讯详情

深耕网站视觉设计与运营推广的一线实战洞察。

用Codex辅助开发CSV合并脚本:需求拆解、实现与验收

用Codex辅助开发CSV合并脚本:需求拆解、实现与验收 上周接到一个挺无聊但又不能出错的需求把三份 CSV 合并成一份。说无聊是因为这本质上就是数据搬运工干的活把 A 表、B 表、C 表倒腾成一张总表说不能出错是因为其中一份是用户基础信息一份是近三个月订单明细一份是支付流水任何一行错位、一个字段对不上后面所有统计报表都会变成废纸。要是以前我大概率会打开 Excel 硬扛或者花半小时写个一次性脚本。但这次我决定换一种方式用 Codex 来写这个合并脚本我来盯需求和验收。整条链路走下来从需求拆解、提示词设计到脚本落地、验收测试有不少心得值得沉淀。这篇就来完整复盘这次实操内容包括三份 CSV 的合并逻辑怎么定、Codex 辅助开发的提示词怎么给、完整可用的 Python 脚本长什么样、以及验收测试到底要验哪些东西。适合正在用或准备用 Codex 写数据处理脚本的朋友也适合被 CSV 合并这种杂活反复折腾的运营和开发。1. 需求拆解三份 CSV 到底要怎么合并1.1 先看清三份数据各自的“脾气”拿到三份 CSV第一件事不是急着写代码而是先把文件打开看一遍。这一步特别重要很多人的合并脚本翻车就是因为没摸清原始数据的情况。我这次手头三份文件具体是这样的文件关键字段行数编码user_info.csv用户ID、姓名、城市、注册时间约 1.2 万UTF-8order_info.csv订单号、用户ID、商品、金额、下单时间约 6.8 万GBKpayment_info.csv支付单号、订单号、支付渠道、支付状态约 6.6 万UTF-8三份文件的共同点是都有“用户ID”或“订单号”这类关联字段但问题也不少。最典型的问题是字段名不统一用户表里叫user_id订单表里叫uid支付表里甚至直接叫用户ID中文。如果不做映射直接用 pandas 的 concat 或 merge要么报 KeyError要么生成一堆重复列。其次是编码不统一。订单表是 GBK直接用 Python 的默认编码去读会直接抛UnicodeDecodeError就算勉强读出来打印到控制台也是一堆乱码。后来我把所有文件统一转成 UTF-8 再处理世界清净了。第三类是隐藏的脏数据。比如某份 CSV 里出现了空行、表头前有看不见的空格、某个字段值前后带\ufeff之类的不可见字符。这些坑不打开原始文件逐行看一遍根本不会被发现。所以我的第一个建议不管用不用 Codex先写几行命令把三份文件的行数、表头、前两行数据打印出来心里有个底再设计合并逻辑。1.2 合并逻辑不是只有一种“合并 CSV”这个词其实很模糊很多人一上来就默认是纵向拼接实际上完全取决于业务诉求。我在这次需求里梳理出三种常见合并逻辑每种都有对应的使用场景。纵向合并行堆叠把多份结构相同或相似的表按行拼成一张长表。适合“上个月订单表 这个月订单表 下个月订单表”这种同构数据的汇总也可以用在一份大 CSV 被拆成多个分片之后重新拼回去。横向合并按主键关联以某个公共字段为基准把多份表的字段拼到同一行里。适合“用户表 订单聚合表 支付表”这种宽表加工场景业务上其实就是 SQL 里的 JOIN。还有一种更复杂的混合场景先把订单和支付按订单号横向关联再把多个月的关联结果纵向堆叠起来。这种最费劲因为对字段一致性的要求更高。理解合并逻辑最好的类比是积木。纵向合并相当于把几摞同规格的板件倒进同一个箱子里只要尺寸一样就行横向合并则像拼插模型每一块的编号必须对得上位置错了整个结构就塌了。1.3 把需求翻译成 Codex 能听懂的话Codex 这类工具最大的特点是你给它什么样的输入它就给你什么样的输出。你要是只丢一句“帮我合并 CSV”它大概率给你一个用 pandas 纵向堆叠的通用脚本完全不管你的字段映射和业务规则。我这次在让 Codex 写代码之前先把需求整理成了一份“提示词需求单”包含这几个要素输入文件路径、格式、编码方式每份文件的字段清单字段映射关系user_iduid用户ID合并逻辑横向还是纵向缺失值怎么处理填空字符串还是填 0输出字段顺序输出编码必须utf-8-sig异常处理要求遇到重复主键怎么办把这些信息按结构给到 Codex它生成的代码基本不会跑偏。如果信息给得太模糊生成出来的脚本看起来像模像样一跑就报错来回拉扯反而更耗时。2. Codex 辅助开发的正确姿势2.1 安装与 PATH 配置Codex 是一款命令行工具当前主流安装方式是通过 npm 全局安装npm install -g openai/codex安装完成之后运行codex --version确认是否成功。这一步看着简单实际上有不少人在这一步就卡住了因为 Windows 的 PowerShell 会报“无法将 codex 项识别为 cmdlet、函数、脚本文件或可运行程序的名称。”这个报错十有八九是你本机的 npm 全局包目录没有加入 PATH 环境变量。解决方法是找到 npm 的全局安装根目录Windows 下一般是npm config get prefix把输出路径比如C:\Users\你的用户名\AppData\Roaming\npm加到系统 PATH 里然后重新打开终端。如果用的是 PowerShell也可以直接用下面这条命令临时追加$env:Path ;C:\Users\你的用户名\AppData\Roaming\npmmacOS 或 Linux 上如果遇到类似问题多半是 Node 从包管理器安装时npm 全局目录没有被 shell 加载检查一下.zshrc或.bashrc里的路径配置就行。这些都是基于常见实践的做法不同系统细节略有差异但排查思路是完全通用的。2.2 提示词怎么写才不翻车跟 Codex 沟通本质上跟带新人是一个道理。你只说“把数据整理一下”新人无从下手你要把期望结果、约束条件、文件位置全部讲清楚他才能高效产出。我这次给 Codex 的提示词大致长这样我有三份 CSV 文件 1. /data/user_info.csv - user_id, name, city, register_time - 编码 UTF-8 2. /data/order_info.csv - order_id, uid, product, amount, order_time - 编码 GBK 3. /data/payment_info.csv - payment_id, order_id, pay_channel, pay_status - 编码 UTF-8 合并规则 - 以 order_id 关联 order_info 和 payment_info以 user_id 关联 user 表 - user_id 字段别名user_info.user_id order_info.uid - 输出字段顺序user_id, name, city, order_id, product, amount, pay_channel, pay_status - 缺失字段填空字符串 - 输出到 /data/final_report.csv编码 utf-8-sig - 重复主键保留第一条打印警告日志这份提示词包含了路径、字段、映射、规则、输出要求五类关键信息。Codex 生成脚本后我只需要微调一小部分逻辑就能直接跑通。另外一个很实用的技巧在提示词里附上每份文件的表头和前两行数据而不是贴整份文件。这样 Codex 既能理解数据结构又不会因为输入过长浪费上下文窗口。2.3 别让 Codex 一次干太多活有一种错误我见过很多次试图让 Codex 一步到位生成一个能处理所有边界情况的完美脚本结果它写出来的代码臃肿不说还可能遗漏最核心的逻辑。合理的做法是把任务拆成小步骤让 Codex 逐步完成第一步写一个读取三份 CSV 并输出前五行的探测脚本第二步在第一步基础上实现字段映射和主键关联第三步增加编码处理、缺失值填充、日志输出第四步加一个简单的行数统计和 MD5 校验功能每完成一步先人工跑一遍确认结果再进入下一步。这个流程表面上看起来多了几次交互实际上每一步都能快速验证比等 Codex 一口气输出一个 300 行的大脚本然后满是 bug 要高效得多。因为 Codex 的上下文窗口是有限的塞入太多需求要么导致输出截断要么生成到一半开始丢细节拆开反而是最优解。3. 完整脚本实现与逐段拆解3.1 脚本一纵向合并 CSV 文件先说最常见的纵向合并场景。假设你手上有好几份结构一致的分片 CSV要合成一份总文件。用 Python 标准库就能搞定不必依赖任何第三方库。import csv import sys from collections import OrderedDict INPUT_FILES [ part_1.csv, part_2.csv, part_3.csv, ] OUTPUT_FILE merged_stack.csv def read_csv_rows(filepath): with open(filepath, r, encodingutf-8-sig, newline) as f: reader csv.DictReader(f) rows list(reader) return rows def merge_rows(rows_list): merged [] for rows in rows_list: merged.extend(rows) return merged def write_csv_rows(rows, output_path): if not rows: return fieldnames list(OrderedDict.fromkeys(key for row in rows for key in row.keys())) with open(output_path, w, encodingutf-8-sig, newline) as f: writer csv.DictWriter(f, fieldnamesfieldnames, extrasactionignore) writer.writeheader() writer.writerows(rows) if __name__ __main__: all_rows [] for path in INPUT_FILES: try: rows read_csv_rows(path) print(f{path}: {len(rows)} rows) all_rows.extend(rows) except FileNotFoundError: print(f[WARN] file not found: {path}) write_csv_rows(all_rows, OUTPUT_FILE) print(fmerged total: {len(all_rows)} rows - {OUTPUT_FILE})这里有几个关键细节其一读取时用encodingutf-8-sig而不是utf-8。因为很多从 Excel 另存为 CSV 的文件会带 BOM 头用utf-8读取时表头第一列会多出一个\ufeff字符输出后整个字段名都对不上。用utf-8-sig能自动帮你剥掉这个前缀这也是解决“手机打开正常、电脑打开不正常”最核心的一招。其二写文件时同样用newline防止 CSV 里出现多余空行。Windows 环境下用记事本或 Excel 打开生成的 CSV如果每两行之间有个空行多半就是没加这个参数。其三用OrderedDict.fromkeys去重并保留字段顺序保证最终输出的列顺序跟原始文件出现顺序一致。有些文件表头顺序不一样直接写死字段列表反而容易出错。3.2 脚本二按主键横向合并三份表横向合并才是这次需求的主角。三份文件要按user_id和order_id关联成一张宽表核心逻辑类似 SQL 里的 LEFT JOIN。import csv def load_table(filepath, key_field, encodingutf-8-sig): table {} with open(filepath, r, encodingencoding, newline) as f: reader csv.DictReader(f) for row_num, row in enumerate(reader, start2): key row.get(key_field, ).strip() if key in table: print(f[WARN] {filepath} duplicate key {key} at row {row_num}) continue table[key] row return table def merge_tables(user_path, order_path, payment_path, output_path): users load_table(user_path, user_id) orders load_table(order_path, uid, encodinggbk) payments load_table(payment_path, order_id) output_rows [] for order_id, order in orders.items(): uid order.get(uid, ).strip() user users.get(uid, {}) payment payments.get(order_id, {}) output_rows.append({ user_id: uid, name: user.get(name, ), city: user.get(city, ), order_id: order_id, product: order.get(product, ), amount: order.get(amount, ), pay_channel: payment.get(pay_channel, ), pay_status: payment.get(pay_status, ), }) fieldnames [user_id, name, city, order_id, product, amount, pay_channel, pay_status] with open(output_path, w, encodingutf-8-sig, newline) as f: writer csv.DictWriter(f, fieldnamesfieldnames) writer.writeheader() writer.writerows(output_rows) print(forders: {len(orders)}, output: {len(output_rows)}) if __name__ __main__: merge_tables( user_info.csv, order_info.csv, payment_info.csv, final_report.csv, )这个脚本的可取之处在于用字典来模拟 LEFT JOIN。以订单表作为主表遍历每一条订单通过用户 ID 查用户信息通过订单号查支付信息查不到就用空字符串兜底。这样写比直接用 pandas 的 merge 更可控的地方在于你能清楚看到每一份表在关联时的细节问题。比如订单表里出现了一个user_id在用户表中不存在脚本会静默填上空值同时你也可以根据需求把它记入日志。对于一份一次性数据处理任务来说这种透明度和可控性远比几行代码的简洁重要。3.3 为什么优先用标准库而不是 pandas我知道不少人看到 CSV 合并第一反应就是上 pandas。但我的观点是能用标准库解决的事就不要轻易引入第三方依赖。原因有三一是环境兼容性。很多数据分析场景下目标机器可能没有预装 pandas你又没有权限安装标准库在任何 Python 环境里都能直接跑不会出现“跑起来才报 ModuleNotFoundError”的尴尬。二是字段映射的精确性。pandas 的merge确实方便但对字段别名的处理反而容易写得很绕。用csv.DictReader读出来的是字典字段映射就是一次dict.get(key, )的事逻辑一目了然。三是性能。CSV 文件本身是文本格式几万行的数据量级Python 标准库处理耗时也就零点几秒到几秒根本不值得为此引入一个几百 MB 依赖的库。当然如果你的数据量已经到百万行以上或者后续还要做复杂分组统计那上 pandas 完全合理工具选型本来就该跟业务体量匹配。3.4 日志、警告和可观测性脚本里加日志很多人容易忽略但实际跑数据时特别重要。我在两个脚本里都留了警告输出文件找不到时打印 WARN重复主键时打印 WARN。这些日志不是为了好看而是在出问题时让你能快速定位。比如横向合并后发现输出行数比预期少了 2000 行如果脚本里没有任何日志你得一行行去对有了日志一条重复键的警告就能告诉你问题出在哪张表、哪一行。对一次性脚本来说把日志写在终端上就够用了。更规范的做法是输出到log文件不过那是工程化范畴的事这里不展开。4. 验收测试怎么证明脚本是真的“对”4.1 用三份迷你的样例数据做验证脚本写完不等于任务完成必须做验收。我习惯的做法是先造三份能手工算清楚结果的迷你 CSV跑完脚本后拿实际输出跟手工结果对比。比如这次我构造了这样的数据用户表user_id,name,city u001,张三,北京 u002,李四,上海 u003,王五,广州订单表注意编码是 GBKorder_id,uid,product,amount 1001,u001,键盘,299 1002,u002,鼠标,99 1003,u003,显示器,1299支付表payment_id,order_id,pay_channel,pay_status p001,1001,支付宝,成功 p002,1002,微信,成功 p003,1003,银行卡,失败这三份表一行不多一行不少期望结果我用脑子都能算出来第三行支付状态是“失败”其余字段全部匹配。脚本跑完肉眼比对三行数据一致这算第一层验收。4.2 自动化断言让代码去验证代码数据量小的时候肉眼比对没问题但几十万行的真实数据靠肉眼显然不现实。我建议在脚本里直接写几组断言来做自动化验证关键校验点包括输出行数是否等于订单表行数横向合并且订单为主表时主键字段是否有空值主键是否唯一关键金额字段能否转成数字避免文本串入用户表里的每个user_id是否都能在订单表里找到对应关系按需求决定是否允许空下面是一段简单的验证脚本import csv with open(final_report.csv, encodingutf-8-sig) as f: rows list(csv.DictReader(f)) total len(rows) assert total 3, fexpected 3 rows, got {total} for row in rows: for field in [user_id, order_id, product]: assert row.get(field, ).strip(), f{field} is empty: {row} amount row.get(amount, 0) float(amount) order_ids [row[order_id] for row in rows] assert len(order_ids) len(set(order_ids)), duplicate order_id found print(all assertions passed)这种断言式的验证方式把验收标准从“人脑记忆”变成了“代码强制”。即使换一个人来接手这个脚本只要跑一遍测试文件就能知道结果对不对。4.3 MD5 校验让文件指纹替你盯梢热词里有一条“csv 文件怎么进行 md5 校验”这正好是我这次实践的重头戏。MD5 的作用可以理解为给文件生成一个唯一指纹。只要文件内容有任何变动哪怕只是改了一个空格MD5 值都会变。利用这个特性可以在脚本反复修改之后快速判断输出文件是否产生了意外变化。Linux 或 macOS 上直接算md5sum final_report.csvWindows 的 PowerShell 用Get-FileHash -Algorithm MD5 final_report.csvPython 里也能算import hashlib def md5_of_file(path): h hashlib.md5() with open(path, rb) as f: for chunk in iter(lambda: f.read(8192), b): h.update(chunk) return h.hexdigest() print(md5_of_file(final_report.csv))我这次实际经历特别能说明 MD5 的价值。第一次脚本生成后Codex 输出了正确的表头顺序结果也挺好。但等我让它优化代码结构、增加日志之后再跑一遍输出文件的表头顺序悄悄变了。肉眼扫一遍根本看不出来但 MD5 一对比两个哈希值完全不同立刻警觉再一查果然某列顺序被调了个位置。从此以后凡是涉及脚本修改我都会先把上一次的输出文件保留一份算好 MD5改完代码再跑一次对比 MD5。一致说明输出没受干扰不一致就要人工确认差异是不是预期内。这个习惯帮我在后面好几次任务里都提前发现问题。4.4 边界测试清单真实世界的数据永远不会像样例那么乖巧所以验收阶段要刻意构造一些边界情况来测试脚本的健壮性。我列了几项自己常用的检查空文件只有表头没有任何数据脚本不应崩单行文件表头加一行数据关联不上另一张表时缺失值应被填为空字符串重复主键同一条订单出现两次脚本应按预设规则取第一条并打印警告字段包含逗号和换行符CSV 字段本身可以用引号包裹特殊字符正确处理才能保证不串列编码混用一张表 UTF-8一张表 GBK读取时逐表指定编码这些边界情况短时间内不一定会全遇到但提前测一遍心里就有底。尤其是字段里带逗号和换行符的情况很多人以为 CSV 格式很简单实际上如果某字段内容是“备注今天, 天气不错”不带引号包起来就会把一列拆成两列。用csv.DictReader能自动处理这种转义这也是我选择标准库而不是自己写 split 来解析 CSV 的原因。5. 常见问题与排错实录5.1 命令找不到codex / npm / git 无法识别这个问题在 Windows 用户里出现的频率非常高尤其新电脑刚配好环境的时候。报错信息一般是“无法将 codex 项识别为 cmdlet、函数、脚本文件或可运行程序的名称。”绝大多数原因是 PATH 环境变量没配置好。npm 全局安装的包没有放到系统能搜索到的目录里终端自然找不到。解决办法前面说过了把 npm 全局目录加进 PATH或者干脆在安装 Node.js 时勾选“Add to PATH”选项。我个人还建议装完任何命令行工具之后做的第一件事就是重启一个新的终端窗口然后工具名 --version确认一下。因为很多工具刚安装完旧终端窗口不会自动刷新环境变量你继续输入命令大概率报错然后误以为是安装失败了。5.2 Codex 上下文不足或模型不支持的应对用 Codex 写大需求时有时会碰到类似 “error running remote compact task: codex ran out of room in the models context” 的报错翻译过来就是上下文窗口被塞满了。这时候最有效的办法是把需求拆成更小的子任务一次只让 Codex 处理一步。比如别让它一口气读完整份 CSV 的全部内容只给它表头和前五行为样例别让它同时实现合并、去重、统计、校验四个功能先做合并再说。还有一种情况是模型配置错误比如提示说 “the gpt-5.6-sol model is not supported when using codex with a chatgpt account”。这通常是因为配置里指定了一个当前环境不支持的模型名或者接入的是第三方模型接口而模型名跟服务商的实际模型不匹配。解决方法是检查配置文件里的模型字段把它改成官方支持的模型名称。如果你用的是第三方兼容接口那更要仔细对齐模型名两边差一个字符都会直接报 not supported。5.3 CSV 编码与 Excel 打开乱码“手机打开正常电脑打开不正常”这个热词在 CSV 场景里太经典了。手机上很多解析器会自动检测编码即使文件是 GBK 或 UTF-8 都能正确显示而 Windows 的 Excel 默认按本地编码打开如果你输出的 CSV 是标准 UTF-8 且没有 BOMExcel 就会把中文显示成乱码。解决方案不是改成 GBK 编码而是把文件写成 UTF-8 with BOM也就是 Python 里的utf-8-sig。这样 Excel 打开时能通过 BOM 识别出文件是 UTF-8中文就能正常显示。这是我在编码问题上踩了无数次坑之后总结出来的稳定方案。5.4 脚本生成了但数据对不上怎么办Codex 生成的脚本跑出错误结果先别急着责怪工具先做排查。我的经验是分三步走第一步缩小数据范围。把输入替换成四五行的手工样例看输出跟手工计算是否一致。如果连样例都对不上那说明合并逻辑本身理解错了回到需求定义检查字段映射和主键选择。第二步对比新旧版本输出。如果之前有一个可用的旧脚本把新旧脚本的输出文件做一次 diff定位每一处差异判断是预期变化还是回归 bug。第三步用 MD5 做最终判定。把新输出与上次验证过的输出做哈希对比一致就没问题不一致就重点排查表头顺序、行数、填入的空值是否合理。这套排查流程其实不复杂但它把“看起来对”变成了“比对过、一定对”这两种状态在数据任务里的差距就是灾难与正常的差距。最后再分享一个小技巧如果你之后也会经常用 Codex 处理 CSV 相关任务建议把这次用到的提示词模板、脚本框架、验证代码都保存到一个固定的目录里。下次遇到类似需求不需要从头开始把文件名和字段映射改一改就能直接用。我自己的体会是让 Codex 合并 CSV 这件事真正有价值的不是省下的那半小时而是它倒逼我把需求和验收标准先写清楚。以前写一次性脚本我经常是“跑通就行”的心态现在养成了“先定验收标准再写代码”的习惯质量明显上来了。哪怕你已经很熟悉 CSV 处理从这个场景入手感受一下 AI 辅助开发的节奏也会发现新的效率空间。
返回列表