2026.11.04 AI学习

AI编程（五十八）：数据迁移 — CSV和数据库互相搬家

作者：climbing

CSV 是你最早的老家当（第25到36篇全是在那练的），数据库是刚搬进的新房（第49到57篇）。可两边现在是”鸡犬之声相闻，老死不相往来”：Excel 里攒的客户名单进不了数据库，数据库里的价格记录导不回 CSV 给同事看。今天学数据迁移：把数据从一个”容器”搬进另一个容器。听着玄乎，其实就是”读出来→转格式→写进去”三步，但搬家搬不好，数据会丢、会重、会串行——今天把这门手艺练顺。

一、真实场景：Excel里攒的家当，进不了数据库的门

接着讲你的价格监控工具。数据库用得越来越顺，某天同事发来一份 客户名单.csv：”这些客户挨个盯一下价格呗。”你愣住——名单在 CSV 里，程序却只会读 shop.db。反过来也一样：老板说”把这半年的价格记录发我个 Excel 看看”，你翻遍代码，只会往数据库里”存”，不会往外”拿”成 CSV。

这就是数据迁移的日常：数据住在哪儿，从来不是你说了算。Excel 同事发的是 CSV，老系统存的是文本文件，新系统用的是数据库，网上下的还是 JSON。你的程序要是只会守着一个家，就永远得手工倒腾——复制粘贴、格式错乱、少一行多一行，全是事故现场。

所以正经程序员都得会”搬家”：让数据想住哪儿住哪儿，程序负责打包和拆包。而且好消息是——”读 CSV”（第25篇）、”写 CSV”（第25篇）、”读写数据库”（第50、55篇）你全学过了！今天不是学新轮子，是把两组合已知零件拼成一座桥。今天唯一的新概念，是搬家的”三步流程”本身。

二、新概念：数据迁移 = 抽取 → 转换 → 装载

所有数据搬家，不管从哪儿搬到哪儿，都是同一套三步：

  1. 抽取（Extract）：从源头把数据读出来——读 CSV、查数据库、抓网页，都是”抽取”；
  2. 转换（Transform）：把数据整理成目的地要的样子——列名对不对得上？日期格式认不认？空值怎么办？
  3. 装载（Load）：把整理好的数据写进目的地——写进数据库、写成 CSV、发给 Excel。

行话叫 ETL（Extract-Transform-Load），听起来高大上，其实你第36篇的”数据分析流水线”干的就是这个。今天的区别只是：流水线是”读进来分析完输出报表”，迁移是”读进来转格式存进另一个家”——终点不同，三步骨架一模一样。

用搬家打比方就更直白了：从旧家搬衣服到新家（抽取：从衣柜拿出来；转换：叠好、按季节分袋；装载：放进新衣柜）。搬家最容易出事的从来不是”拿”和”放”，而是中间的”整理”——衣服塞错袋、贵重物品混进杂物、搬一半漏了两箱。数据迁移的事故也全在”转换”这步：列名没对上、类型变了形、重复导入翻了倍。今天重点练的就是这步。

三、方向一：CSV → 数据库（把老家当搬进新房）

先干最急的活：把同事给的 客户名单.csv 搬进数据库。假 CSV 长这样：

姓名,城市,月预算
张三,沈阳,3000
李四,北京,5000
王五,上海,4500

三步走，代码一共 30 行：

# import_csv.py — 方向一：把 CSV 搬进数据库（抽取→转换→装载）
import csv
import sqlite3

# 第1步 抽取：把 CSV 读出来
rows = []
with open("客户名单.csv", encoding="utf-8-sig") as f:   # utf-8-sig 防 BOM 坑，见坑2
    reader = csv.DictReader(f)                          # 每行是 {"姓名": ..., "城市": ...}
    for row in reader:
        rows.append(row)
print(f"从 CSV 抽到 {len(rows)} 条")

# 第2步 转换：整理成数据库要的样子
clean = []
for row in rows:
    name = row["姓名"].strip()
    city = row["城市"].strip()
    budget = int(row["月预算"])          # CSV 里全是字符串！要转成数字，见坑1
    clean.append((name, city, budget))

# 第3步 装载：写进数据库
conn = sqlite3.connect("shop.db")
conn.execute("""CREATE TABLE IF NOT EXISTS customers (
    name TEXT, city TEXT, budget INTEGER
)""")
conn.executemany("INSERT INTO customers VALUES (?, ?, ?)", clean)   # 批量插入，见坑3
conn.commit()
conn.close()
print(f"已装载 {len(clean)} 条进 customers 表")

三个新面孔（其实都是老朋友换了衣服）：

  • csv.DictReader：第25篇你用过按位置读的 csv.reader，DictReader 是升级版——每行直接变成字典，用列名取值 row["姓名"]，比 row[0] 不容易错位。列换了顺序它都不慌。
  • 转换那一步别省：CSV 里”3000″是字符串，不 int() 转一下，存进数据库就是文字”3000″，回头想算”平均预算”就抓瞎。CSV 无类型，数据库有类型——转换就是给数据”办落户手续”。
  • executemany：一次性插入一批，比循环里一条条 execute 快得多，代码也短。它接收一个”元组列表”，就是上面 clean 那个形状。

跑完，SELECT * FROM customers; 一看，CSV 的家当整整齐齐躺进数据库了。

跟AI说需求：把开头那段注释和 CSV 样例发给AI：”这是我的 CSV，这几列要进 SQLite 的 customers 表，月预算存成整数，帮我写导入脚本”——它连 BOM、strip 这些细节都会顺手处理，你负责读懂每一行为啥这么写。

四、方向二：数据库 → CSV（把新房的货发回老家）

老板要半年价格记录的 Excel？反着来，还是三步，这次倒着走：

# export_csv.py — 方向二：把数据库查询结果导出成 CSV
import csv
import sqlite3

# 第1步 抽取：从数据库查出来
conn = sqlite3.connect("shop.db")
conn.row_factory = sqlite3.Row        # 让查询结果能按列名取值
rows = conn.execute(
    "SELECT name, price, date FROM prices ORDER BY date"
).fetchall()
conn.close()
print(f"从数据库抽到 {len(rows)} 条")

# 第2步 转换：整理成 CSV 要的样子（这次几乎不用转，列名当表头）
header = ["商品名", "价格", "日期"]    # 对外发的文件，表头用中文更礼貌

# 第3步 装载：写成 CSV
with open("价格记录导出.csv", "w", encoding="utf-8-sig", newline="") as f:
    writer = csv.writer(f)
    writer.writerow(header)                       # 先写表头
    for row in rows:
        writer.writerow([row["name"], row["price"], row["date"]])
print("导出完成：价格记录导出.csv")

三个细节打个招呼：

  • conn.row_factory = sqlite3.Row：一行设置，查询结果就能 row["name"] 按列名取值，不用记第几列——跟 DictReader 一个思路，取值靠名字不靠位置。
  • newline=""：csv 模块的官方要求。不写它，Windows 上导出的 CSV 每行之间会多一个空行（坑4细说）。
  • 表头写中文：存进数据库用英文列名（程序好处理），发给人看的 CSV 用中文表头（老板好阅读）——这就是”转换”步的价值：同一个数据，出门换身衣服。

Excel 双击打开 价格记录导出.csv，齐活。注意 Excel 双击打开 CSV 可能把长数字变科学计数法、把日期自动改格式——给同事发之前自己先双击看一眼（坑5）。

五、双向小工具：一个程序，两个方向

两个脚本合成一个”搬家工具”，命令行指个方向就干活：

# migrate.py — 双向数据迁移工具
# 用法：python migrate.py csv2db 客户名单.csv customers
#       python migrate.py db2csv prices 价格记录导出.csv
import csv
import sqlite3
import sys


def csv2db(csv_file, table):
    """CSV → 数据库"""
    with open(csv_file, encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        rows = list(reader)
    if not rows:
        print("CSV 是空的，没啥可搬"); return
    columns = list(rows[0].keys())                # 表头就是列名
    clean = [tuple(r[c].strip() for c in columns) for r in rows]

    conn = sqlite3.connect("shop.db")
    col_defs = ", ".join(f"{c} TEXT" for c in columns)
    conn.execute(f"CREATE TABLE IF NOT EXISTS {table} ({col_defs})")
    placeholders = ", ".join("?" for _ in columns)
    conn.executemany(
        f"INSERT INTO {table} VALUES ({placeholders})", clean
    )
    conn.commit()
    conn.close()
    print(f"{csv_file} → {table}：搬了 {len(clean)} 条")


def db2csv(table, csv_file):
    """数据库 → CSV"""
    conn = sqlite3.connect("shop.db")
    conn.row_factory = sqlite3.Row
    rows = conn.execute(f"SELECT * FROM {table}").fetchall()
    conn.close()
    if not rows:
        print("表里没数据，没啥可搬"); return

    columns = rows[0].keys()
    with open(csv_file, "w", encoding="utf-8-sig", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(columns)
        for row in rows:
            writer.writerow([row[c] for c in columns])
    print(f"{table} → {csv_file}：搬了 {len(rows)} 条")


if __name__ == "__main__":
    direction, a, b = sys.argv[1], sys.argv[2], sys.argv[3]
    if direction == "csv2db":
        csv2db(a, b)
    elif direction == "db2csv":
        db2csv(a, b)
    else:
        print("方向只认 csv2db 或 db2csv")

注意这个版本的 csv2db 是”通用搬法”：列名跟着 CSV 表头走、类型全存 TEXT，适合”先搬进来再说”。上一节的 import_csv.py 是”定制造搬法”：列和类型都提前定好，适合正经业务。临时倒腾用通用版，正式入库用定制版——通用版灵活但类型全变文字，定制版麻烦但数据规规矩矩。

用 pandas 的同学还有条捷径：pandas.read_csv(...).to_sql("customers", conn) 两行搞定导入，pandas.read_sql("SELECT ...", conn).to_csv(...) 两行搞定导出。偷懒可以，但三步流程要心里有数——pandas 只是把三步打包了，转换步出了问题（类型不对、列名不配）它可不替你兜底。

六、增量迁移：只搬新来的，别翻倍导入

搬家工具能跑了，新坑马上来：明天同事又发来一份 客户名单-新增.csv，你顺手再跑一遍 csv2db——得，昨天那 3 条又插了一遍，表里 6 条，查个”客户总数”直接翻倍。重复导入是数据迁移的头号事故，疫苗有三针：

疫苗一：导入前先清场（全量重来）。 适合”这份 CSV 就是全量真相”的场景——导入前把表删了重建，或 DELETE FROM customers 清空再灌。简单粗暴，但只适合全量文件，别拿它处理”新增名单”。

疫苗二：导入时查重（有就跳过）。 适合增量文件——插之前先问一句”这人有了吗”：

# 增量导入：已有就跳过，只搬新来的
existing = {r[0] for r in conn.execute("SELECT name FROM customers")}
new_rows = [r for r in clean if r[0] not in existing]
print(f"名单里 {len(clean)} 条，其中新增 {len(new_rows)} 条")
conn.executemany("INSERT INTO customers VALUES (?, ?, ?)", new_rows)

用集合（set）查重是行家手法：existing 收集所有已有姓名，not in 一个集合是瞬间出结果，几千条也不卡。

疫苗三：给表装个唯一键。 治本的方法是在数据库层面上”禁止同名”：CREATE TABLE customers (name TEXT UNIQUE, ...)，重复插入时数据库直接拒绝。这是数据库的正经玩法（约束），第52篇之后的进阶话题，今天先知道有这回事。

七、五个坑点疫苗（先打再上路）

  1. CSV 里的数字全是字符串。 row["月预算"] 拿到的是”3000″不是 3000，不转 int() 就入库，回头求和、比大小全是错。转换步的核心工作就是这个——CSV 无类型，搬进有类型的家必须办落户。顺手防一手脏数据：int(row["月预算"] or 0) 或 try/except，别让一个空格让整个程序崩掉。
  1. BOM 和编码专咬中文。 Excel 另存的 UTF-8 CSV 开头藏了个看不见的 BOM 字符，第一列的列头会变成 "ufeff姓名"，程序拿 "姓名" 去取值直接 KeyError。解法就是代码里的 encoding="utf-8-sig"——读写 CSV 涉及中文，认准这个，它读时自动去 BOM、写时按规矩来。要是遇到 GBK 编码的老文件（报错 'gbk' codec can't decode... 或满屏乱码），换成 encoding="gbk" 读。
  1. executemany 要的是”元组的列表”，不是”字典的列表”。 [(名字, 城市, 预算), ...] 这个形状才对，直接把 DictReader 的字典列表塞进去会报错。还有：表名列名永远不能用 f-string 拼用户输入（SQL 注入坑，第51篇打过疫苗），占位符 ? 只管值不管列名。
  1. Windows 导出 CSV 每行多个空行。 open(..., "w") 不加 newline=""，csv 模块的换行和 Windows 的换行叠罗汉，导出的文件行距加倍，Excel 里看着像空了一行。官方文档钦定解法：newline=""，写导出代码时无脑带上。
  1. Excel 双击打开会”自作聪明”。 长订单号变科学计数法（1.23E+15）、以 0 开头的编号被吃掉前导零、日期被改成 1/5 还是 5/1 全看系统心情——Excel 打开 CSV 会自动”转换格式”，数据本身没坏，是显示被改了。给人发 CSV 前自己先双击验一眼；对方要用 Excel 正经处理，宁可导出成 .xlsx（pandas 一行 to_excel），别让 Excel 糟蹋了你的数据。

八、跟AI协作的正确姿势（迁移特供版）

  1. 把 CSV 头几行和目标表结构一起发给AI。 只说”帮我把 CSV 导入数据库”，AI 只能猜你的列；贴上 head -5 客户名单.csv 的输出和 PRAGMA table_info(customers); 的结果，它连列名映射和类型转换都写得一步到位。
  1. 让AI把转换规则写成人话注释。 迁移代码三个月后没人记得”为什么这里 strip 那里 int”。让AI在每步转换旁边写清”这一行在干嘛”，比如”# 预算转整数，空值按0处理，防止脏数据炸掉整批导入”——迁移代码的注释就是你的”搬家清单”。
  1. 搬完让AI写验货脚本。 “检查 customers 表：行数要等于 CSV 行数减表头、月预算平均值要在合理范围、名字不能有空的”——搬家搬没搬丢，得对单验收，别”跑完没报错”就当成了。这跟第57篇”恢复必须亲手破坏一次”一个道理：迁移必须对数验收。

九、今日小练习

练习1：做个搬家对账单。 写一个 check_migrate.py：输入 CSV 文件和表名，分别数出 CSV 的行数和表里的行数，一致打印”搬齐了”，不一致打印差几条、差在哪个姓名上（提示：把两边的姓名都收进集合，做差集就知道谁没搬过去；想想 existing - new 和 new - existing 各是啥意思）。

练习2：温度单位大迁移。 有个 温度记录.csv（列：日期,城市,温度,单位，单位列的值是”摄氏”或”华氏”），搬进数据库 temps 表，要求温度统一存成摄氏度存进 REAL 列（提示：华氏转摄氏的公式 AI 一秒告诉你；这题的转换步比前面都”厚”——同一列要按单位列分情况处理，这正是 ETL 里 T 的日常）。

练习3：踩一次重复导入的坑。 故意把同一份 CSV 对同一张表跑两遍 csv2db，亲眼看看数据翻倍的现场，然后用”疫苗二”的查重法改好再跑一遍验证（提示：SELECT COUNT(*) 前后对比；自己制造一次事故再修复，比读十遍坑点疫苗都长记性）。

下篇预告

下一篇（第59篇）：性能优化入门 — 索引。 数据库用着用着，几千条数据查询就开始”想一想”了，几万条能等到你去接杯水。问题出在数据库每次查询都在”逐行翻账本”。下一篇给它装上”目录”——索引：一个让查询从”翻全书”变”查目录”的魔法。概念只此一个，效果立竿见影，是数据库从”能用”到”好用”的第一道分水岭。