Python 读取 CSV 对比 MySQL：存在更新、不存在保留的配置骨架与验证

1. 为什么 CSV 对比 MySQL 的同步逻辑总写乱做数据补丁、配置导入、商品资料合并时最常见的需求就是CSV 里有一批数据MySQL 表里也有一批数据现在要按某个业务主键ISBN、商品编码、配置 key做比对——两边都有的用 CSV 的值更新 MySQLMySQL 有但 CSV 没有的原样保留不能删也不能清空。听起来简单但真写起来很多人第一版代码都会踩三个坑一是把「不存在则保留」写成了「不存在则插入空值」二是逐行SELECT再逐行UPDATE几千行就跑了几分钟三是没有 dry-run直接改库改错了只能翻备份。我试过最稳的做法是把这件事拆成三步先读 CSV 建内存索引再一次性把 MySQL 里命中的行捞出来做差异比对最后只对真正有变化的行执行 UPDATE。整个过程用一份config.toml管住数据库连接和字段映射跑之前先 dry-run 打印差异确认无误再真正写库。下面这套骨架可以直接复制去改适合数据补丁、配置导入、主数据合并这类场景Python 基础一般也能跟下来。核心检索词先明确Python 读取 CSV 对比 MySQL存在更新、不存在保留本质是一个「按主键做 upsert但只 upsert 命中的部分」的同步逻辑。它不负责删除也不负责插入新行只负责把 CSV 里已有的记录同步到库里。想清楚这个边界代码就不会写歪。2. 前置准备TaoToken 与依赖环境这套脚本本身不依赖任何在线服务纯本地跑。但如果你后续想把这套同步逻辑接到模型侧做字段清洗、异常值判断或者用 coding agent 帮你生成字段映射可以先把 TaoToken 的接入准备好。它的 API 地址是https://taotoken.net/api官网在https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content模型对话、Coding Plan、控制台、API Keys、接入文档都有对应入口按需取用即可。本地依赖只需要三个pip install pandas pymysql tomlitomli是 Python 3.11 以下读 TOML 用的3.11 可以直接用标准库tomllib。数据库这边MySQL 5.7 和 8.0 都能跑注意字符集统一用utf8mb4否则中文书名、配置描述容易出乱码。CSV 文件建议用utf-8-sig编码保存Excel 导出的 CSV 默认带 BOM用utf-8读第一列会多出\ufeff主键比对直接失败这是最高频的坑。3. 可复制配置config.toml 骨架把连接信息和字段映射全部外置到config.toml代码里不写死任何库名表名。这样换一个同步任务只改配置不改代码。# config.toml [database] host 127.0.0.1 port 3306 user sync_user password your_password db book_center charset utf8mb4 [sync] # CSV 路径与编码 csv_path ./wys.csv csv_encoding utf-8-sig # MySQL 目标表 table book_master # 业务主键CSV 列名 - MySQL 列名 key_csv ISBN key_db isbn # 主键清洗规则去掉横线再比对 key_strip - # dry-run 为 true 时只打印差异不写库 dry_run true # 批量提交行数 batch_size 500 # 字段映射CSV 列名 MySQL 列名 # 只写需要同步的字段没写的字段一律不动 [field_map] 书名 book_name 定价 price 出版者 publisher 出版时间 publish_date 中图分类 clc_code字段映射这块是重点只映射你要更新的列。MySQL 表里那些 CSV 没有的列比如创建时间、内部备注、状态位因为不在field_map里UPDATE 语句根本不会碰它们天然满足「不存在保留」。这比先查全表再逐字段判断要干净得多。读取配置的代码import tomli def load_config(path./config.toml): with open(path, rb) as f: return tomli.load(f) cfg load_config()如果你用的是 Python 3.11把import tomli换成import tomllib as tomli即可其余不变。4. 核心实现读 CSV、比对、生成差异先写主键清洗和 CSV 读取。CSV 里 ISBN 可能带横线MySQL 里存的是纯数字所以两边都要按同一规则清洗后再比对否则永远匹配不上。import pandas as pd import pymysql def clean_key(val, strip_chars): if val is None: return s str(val).strip() for ch in strip_chars: s s.replace(ch, ) return s def read_csv_index(cfg): df pd.read_csv( cfg[sync][csv_path], dtypestr, encodingcfg[sync][csv_encoding], keep_default_naFalse, # 关键空值读成 不读成 nan ) key_csv cfg[sync][key_csv] strip_chars cfg[sync].get(key_strip, ) df[_key] df[key_csv].map(lambda v: clean_key(v, strip_chars)) # 主键去重后出现的覆盖先出现的 df df.drop_duplicates(subset_key, keeplast) return df.set_index(_key)keep_default_naFalse这一行必须加。默认情况下 pandas 会把空单元格读成nan写回 CSV 或拼 SQL 时就变成字符串nan这正是原方案里提到的那个坑。设成False后空值就是空字符串行为可控。接着从 MySQL 捞出命中的行。不要逐行查用IN批量查def fetch_db_rows(cfg, keys): db cfg[database] conn pymysql.connect( hostdb[host], portdb[port], userdb[user], passworddb[password], databasedb[db], charsetdb[charset], cursorclasspymysql.cursors.DictCursor, ) key_db cfg[sync][key_db] table cfg[sync][table] rows {} keys list(keys) with conn.cursor() as cur: for i in range(0, len(keys), 1000): chunk keys[i:i1000] placeholders ,.join([%s] * len(chunk)) sql fSELECT * FROM {table} WHERE {key_db} IN ({placeholders}) cur.execute(sql, chunk) for r in cur.fetchall(): rows[clean_key(r[key_db], cfg[sync].get(key_strip, ))] r conn.close() return rows然后做差异比对生成待更新列表。只有 CSV 和 DB 都有、且至少一个映射字段值不同的行才进入更新队列def build_diff(cfg, csv_index, db_rows): field_map cfg[field_map] to_update [] for key, csv_row in csv_index.iterrows(): if key not in db_rows: continue # 库里没有 - 保留不插入 db_row db_rows[key] changed {} for csv_col, db_col in field_map.items(): new_val str(csv_row.get(csv_col, )).strip() old_val if db_row.get(db_col) is None else str(db_row[db_col]).strip() if new_val ! old_val: changed[db_col] new_val if changed: to_update.append((key, changed)) return to_update到这里to_update就是全部差异。dry-run 模式下直接打印它不碰数据库。5. 验证请求先 dry-run 再执行 upsertdry-run 打印差异格式清晰一点方便肉眼核对def dry_run_report(to_update, limit20): print(f[DRY-RUN] 待更新行数: {len(to_update)}) for key, changed in to_update[:limit]: print(f key{key}) for col, val in changed.items(): print(f {col} - {val}) if len(to_update) limit: print(f ... 其余 {len(to_update) - limit} 行省略)真正执行时按主键逐行 UPDATE用参数化查询防注入按batch_size提交def apply_updates(cfg, to_update): db cfg[database] table cfg[sync][table] key_db cfg[sync][key_db] batch cfg[sync][batch_size] conn pymysql.connect( hostdb[host], portdb[port], userdb[user], passworddb[password], databasedb[db], charsetdb[charset], autocommitFalse, ) affected 0 try: with conn.cursor() as cur: for i, (key, changed) in enumerate(to_update, 1): cols , .join(f{c}%s for c in changed) sql fUPDATE {table} SET {cols} WHERE {key_db}%s cur.execute(sql, list(changed.values()) [key]) affected cur.rowcount if i % batch 0: conn.commit() conn.commit() except Exception as e: conn.rollback() raise e finally: conn.close() return affected主流程串起来if __name__ __main__: cfg load_config() csv_index read_csv_index(cfg) db_rows fetch_db_rows(cfg, csv_index.index) diff build_diff(cfg, csv_index, db_rows) if cfg[sync][dry_run]: dry_run_report(diff) else: n apply_updates(cfg, diff) print(f[DONE] 实际更新行数: {n})成功结果长这样dry-run 阶段打印出待更新行数和每行字段变化把dry_run改成false再跑输出[DONE] 实际更新行数: N。核对时用一条 SQL 验证SELECT COUNT(*) FROM book_master WHERE isbn IN (9787111111111,9787111111112);再抽查几行的字段值确认 CSV 的新值已经写进去而 CSV 里没有的 ISBN 对应的行完全没动。行数对得上、变更记录对得上这次同步就算验证通过。6. 本篇常见错排查主键匹配不上差异永远是 0。九成是编码或清洗规则不一致。先确认 CSV 用utf-8-sig读再确认两边key_strip规则相同。可以在read_csv_index后打印csv_index.index[:5]在fetch_db_rows后打印list(db_rows.keys())[:5]肉眼比一下。空值被写成字符串 nan。检查pd.read_csv是否带了keep_default_naFalse。如果 CSV 里本来就有字面量nan那是数据问题需要在清洗阶段单独处理。UPDATE 把不该动的列清空了。说明field_map里映射了 CSV 中不存在的列或者csv_row.get(csv_col, )取到了空串还照样更新。可以在build_diff里加一条规则new_val为空且old_val非空时跳过避免用空值覆盖库里的有效数据。批量 IN 查询报 placeholder 数量超限。MySQL 对IN列表长度有上限代码里已经按 1000 分块如果还报错就把块调小到 500。事务没提交跑完看库没变化。确认apply_updates里conn.commit()执行到了异常分支走了rollback。dry-run 为 true 时本来就不写库别把这两种情况搞混。字段类型不匹配导致写入失败。CSV 全是字符串MySQL 里price是 DECIMAL、publish_date是 DATE 时直接写字符串可能被拒。稳妥做法是在field_map旁边加一个类型转换表写入前按目标类型转换转换失败的行单独记日志不阻塞整批。7. 把同步逻辑接到模型侧做字段清洗字段映射和空值处理稳定之后下一步通常是把脏数据交给模型做标准化比如把「出版时间」里各种格式统一成YYYY-MM-DD或者把中图分类的简写补全。这时候可以用 TaoToken 的模型对话能力先做小批量试跑确认清洗规则可靠再全量跑。API 地址是https://taotoken.net/api接入文档里有请求格式和鉴权说明API Keys 在控制台生成。如果这套同步脚本要长期跑、还要配合 coding agent 改字段映射可以看下 Coding Plan把生成映射、写校验 SQL 这类重复劳动交给 agent 处理人只负责 dry-run 核对和最终提交。整套骨架的价值在于配置外置、差异先看后写、只更新命中字段。把这三点守住CSV 对比 MySQL 的同步就不会再出「该保留的被清空、该更新的没更新」这类问题。