1. 为什么要把 SQL 文件转成 CSV真实场景与踩坑点你手里可能有一份从同事那拷来的.sql文件里面是CREATE TABLE加一堆INSERT INTO想拿去做数据分析但 Excel 打开乱码、Navicat 又懒得装。这时候用 Python pandas 把 SQL 文件导出成 CSV是最省事的路径。核心检索词就是python pandas 将 sql 文件变为 csv 文件它解决的是「本地 SQL 文本 → 结构化表格 → 通用 CSV」这条链路。先说清楚它适合谁做数据清洗的分析师、写课程作业的学生、需要把测试库数据导出来做报表的后端。不适合谁数据量上千万行、需要增量同步的生产场景那种情况应该走数据库原生导出工具。我试过直接拿pd.read_sql去读一个.sql文件结果报错因为 pandas 不认识 SQL 文本文件它只认数据库连接。所以正确姿势是两步先把 SQL 文件灌进一个本地 SQLite 数据库再用 pandas 从数据库里查出来。SQLite 的好处是零配置、单文件、Python 标准库自带sqlite3不用装 MySQL 服务。这里有个常见误区很多人以为pymysql连本地就能读.sql文件其实pymysql连的是运行中的 MySQL 服务而.sql文件只是文本。如果你本地没跑 MySQLconnect(hostlocalhost)会直接抛连接拒绝。所以本文用 SQLite 做中转全程不依赖任何外部服务。还有一个坑是编码。SQL 文件如果是utf-8带 BOM或者gbk直接读会报UnicodeDecodeError。导出 CSV 时如果不在to_csv里指定encodingutf-8-sigExcel 打开中文就是乱码。这两个点后面都会给具体参数。整个流程拆成四步建 SQLite 库并执行 SQL 文件、用 pandas 查询、导出 CSV、验证行数和字段。下面按顺序给可复制的代码。2. 前置准备TaoToken 配置与本地环境搭建这一节解决「工具从哪来」。如果你只是本地跑 pandas其实不需要任何在线服务。但如果你想让 AI 帮你生成建库脚本、排查read_sql报错或者用 Claude Code 这类编码助手来写转换逻辑就需要一个稳定的模型接入点。TaoToken 在这里的角色是提供 API 入口让你在编辑器或脚本里调用模型。先明确三件套任何接入场景都绕不开Base URL、API Key、Model ID。Base URL 用https://taotoken.net/api注意这个地址不带任何查询参数。API Key 在控制台生成路径是 console。Model ID 根据你用的模型填比如claude-sonnet-4-20250514这类。如果你用 Claude Code 做编码辅助配置方式是在项目里设置环境变量或配置文件。核心是让工具知道往哪发请求。文档在 doc里面有各客户端的接入说明。API Keys 管理页在 api-keys。本地环境这边你需要 Python 3.8 以上然后装 pandaspip install pandasSQLite 不用装sqlite3是标准库。验证一下import sqlite3 import pandas as pd print(pd.__version__) print(sqlite3.sqlite_version)输出类似2.2.2和3.45.1就说明环境 OK。如果你要用pymysql连真实 MySQL再pip install pymysql但本文主流程不需要。关于模型调用如果你想在脚本里让 AI 帮你把 SQL 文件里的字段映射成中文列名可以用 模型对话 先试提示词确认输出格式后再写进代码。长期做数据管道的话Coding Plan 更适合因为转换脚本会反复迭代。这里给一个最小可用的配置片段假设你在用某个支持 OpenAI 兼容接口的客户端配置文件写成 JSON{ base_url: https://taotoken.net/api, api_key: sk-你的key, model: claude-sonnet-4-20250514 }注意base_url后面不要加/v1之外的路径具体以文档为准。Key 不要提交到 Git放环境变量里export TAOTOKEN_API_KEYsk-你的key这样脚本里用os.environ.get(TAOTOKEN_API_KEY)读取避免硬编码。环境准备好后进入下一步建库。3. 可复制配置建 SQLite 库并执行 SQL 文件这一步是整个流程的核心。假设你有一个data.sql文件内容长这样CREATE TABLE consumer ( id INTEGER PRIMARY KEY, name TEXT, city TEXT, amount REAL ); INSERT INTO consumer (id, name, city, amount) VALUES (1, 张三, 北京, 120.5); INSERT INTO consumer (id, name, city, amount) VALUES (2, 李四, 上海, 88.0); INSERT INTO consumer (id, name, city, amount) VALUES (3, 王五, 广州, 200.0);现在写一个sql_to_csv.py先建库再执行。关键点是executescript可以一次执行多条语句比execute逐条跑方便import sqlite3 import pandas as pd import os SQL_FILE data.sql DB_FILE local.db CSV_FILE output.csv # 1. 读取 SQL 文件处理编码 with open(SQL_FILE, r, encodingutf-8) as f: sql_text f.read() # 2. 连接 SQLite不存在会自动创建 conn sqlite3.connect(DB_FILE) cursor conn.cursor() # 3. 执行整个 SQL 脚本 cursor.executescript(sql_text) conn.commit() # 4. 用 pandas 查询 df pd.read_sql(SELECT * FROM consumer, conn) print(行数:, len(df)) print(df.head()) # 5. 导出 CSV df.to_csv(CSV_FILE, indexFalse, encodingutf-8-sig, sep,) conn.close() print(导出完成:, CSV_FILE)逐段解释。open那里如果文件是 GBK把encoding改成gbk否则会报UnicodeDecodeError: utf-8 codec cant decode byte。executescript会自动处理分号分隔但注意它不会自动提交所以后面必须conn.commit()否则数据没落盘。pd.read_sql的第一个参数是 SQL 查询语句第二个是连接对象。这里查的是consumer表你可以换成SELECT name, city FROM consumer WHERE amount 100做过滤。返回的df就是标准 DataFrame后续随便处理。to_csv的参数是重点。indexFalse去掉 pandas 自动加的行号列不然 CSV 会多一列Unnamed: 0。encodingutf-8-sig带 BOMExcel 打开中文不乱码。sep,是默认分隔符如果你要导给某些只认分号的系统改成sep;。如果你坚持用 MySQL 而不是 SQLite配置片段是这样import pymysql conn pymysql.connect( hostlocalhost, port3306, userroot, password你的密码, databaseflowers, charsetutf8mb4 ) df pd.read_sql(SELECT * FROM consumer, conn)但前提是你本地 MySQL 服务在跑且consumer表已存在。用 SQLite 的好处就是不用管这些。还有一个细节如果 SQL 文件里有CREATE DATABASE或USE语句SQLite 不支持会报near USE: syntax error。处理办法是手动删掉这些行或者用正则过滤import re sql_text re.sub(r(?i)^\s*(CREATE DATABASE|USE)\s.*?;, , sql_text, flagsre.MULTILINE)这样就能兼容从 MySQL 导出的 SQL 文件。配置写好后跑一遍看结果。4. 验证请求与成功结果行数比对和字段校验导出完不能直接交差得验证数据没丢没变形。最直接的方法是行数比对SQL 文件里有多少条INSERTCSV 就应该有多少行不含表头。先数 SQL 文件里的 INSERTwith open(SQL_FILE, r, encodingutf-8) as f: content f.read() insert_count content.lower().count(insert into) print(SQL 中 INSERT 条数:, insert_count)再数 CSV 行数df_check pd.read_csv(CSV_FILE, encodingutf-8-sig) print(CSV 数据行数:, len(df_check)) assert insert_count len(df_check), 行数不一致检查是否漏执行如果两个数相等说明没漏。如果 CSV 行数少可能是executescript中途报错但没抛异常或者有重复主键导致插入失败。SQLite 默认对主键冲突会报IntegrityError所以一般不会静默丢数据。字段校验看列名和数据类型print(列名:, df_check.columns.tolist()) print(数据类型:) print(df_check.dtypes) print(空值统计:) print(df_check.isnull().sum())正常输出应该是[id, name, city, amount]id是 int64name/city是 objectamount是 float64。如果amount变成 object说明有非数字字符混进去了得回去查 SQL 文件。再抽样比对具体值print(df_check[df_check[name] 张三])应该输出id1, city北京, amount120.5。如果中文变成乱码检查to_csv的encoding参数和read_csv的是否一致。成功结果长这样行数: 3 id name city amount 0 1 张三 北京 120.5 1 2 李四 上海 88.0 2 3 王五 广州 200.0 导出完成: output.csv SQL 中 INSERT 条数: 3 CSV 数据行数: 3 列名: [id, name, city, amount]看到行数一致、列名正确、中文正常就算验证通过。如果这一步你用了模型辅助生成校验代码可以在 模型对话 里贴报错让它帮你改。5. 本篇常见错排查401、编码、read_sql 报错这一节列真实会撞上的报错对照改。报错一UnicodeDecodeError: utf-8 codec cant decode byte 0xd5 in position 0原因SQL 文件是 GBK 编码。解决open(SQL_FILE, r, encodinggbk)。不确定编码就用chardet探测import chardet with open(SQL_FILE, rb) as f: print(chardet.detect(f.read()))报错二pandas.errors.DatabaseError: Execution failed on sql SELECT * FROM consumer: no such table: consumer原因executescript没执行成功或者表名拼错。先确认 SQL 文件里确实有CREATE TABLE consumer再检查conn.commit()有没有漏。可以在executescript后加一句查询验证cursor.execute(SELECT name FROM sqlite_master WHERE typetable) print(cursor.fetchall())输出[(consumer,)]说明表建好了。报错三401 Unauthorized或local proxy failed这个出现在你调用模型 API 时。401 是 Key 无效或没带。检查请求头里Authorization: Bearer sk-xxx格式对不对Key 有没有过期。local proxy failed通常是本地网络配置问题确认base_url写的是https://taotoken.net/api没有多余斜杠或路径。如果用了客户端工具去 api-keys 重新生成一个 Key 再试。报错四KeyError: reading choices或返回结构里没有choices原因模型返回格式和你解析的字段不匹配。OpenAI 兼容接口返回{choices: [{message: {content: ...}}]}如果你按别的结构取就会 KeyError。打印完整响应体看结构import json print(json.dumps(resp, ensure_asciiFalse, indent2))报错五OAuth token expired用 Claude Code 或类似工具时OAuth 凭证过期。重新走一遍授权流程或者改用 API Key 方式接入。文档 doc 里有各方式的说明。报错六CSV 打开中文乱码不是代码报错是 Excel 的锅。to_csv用encodingutf-8-sigread_csv也用encodingutf-8-sig。如果还乱用记事本打开 CSV 另存为 ANSI。报错七to_csv后多了一列Unnamed: 0忘了indexFalse。加上就行。排查顺序建议先看 SQL 文件编码再看建表是否成功最后看导出参数。大部分问题出在前两步。6. 语义一致 CTA把流程固化成脚本到这里python pandas 将 sql 文件变为 csv 文件的完整链路已经跑通读 SQL、建 SQLite、read_sql、to_csv、验证行数。你可以把上面的代码合并成一个函数传文件路径就能跑def sql_to_csv(sql_file, db_file, csv_file, table, encodingutf-8): with open(sql_file, r, encodingencoding) as f: sql_text f.read() conn sqlite3.connect(db_file) conn.cursor().executescript(sql_text) conn.commit() df pd.read_sql(fSELECT * FROM {table}, conn) df.to_csv(csv_file, indexFalse, encodingutf-8-sig) conn.close() return len(df) rows sql_to_csv(data.sql, local.db, output.csv, consumer) print(f导出 {rows} 行)如果转换逻辑要反复调、字段映射经常变用 Coding Plan 配合编码助手会更顺因为你可以把报错直接贴进去让它改。临时验证某个模型对 SQL 的理解能力用 模型对话 就够。Key 管理和文档分别在 api-keys 和 doc。最后留一个实用技巧如果 SQL 文件特别大executescript会一次性加载进内存可能卡住。改成逐条执行并分批提交for stmt in sql_text.split(;): stmt stmt.strip() if stmt: cursor.execute(stmt) conn.commit()这样内存占用低也方便在报错时定位是哪条语句出问题。