1. 从 xlrd 到 pymysqlExcel 批量导入数据库的真实痛点如果你手头有一份几万行的 Excel用 Python 把它塞进 MySQL最直觉的写法是 for 循环里一条条cursor.execute()。我最早也是这么干的5000 行跑了将近两分钟中途网络抖一下还得从头再来。后来换成executemany一次性提交同样的数据量压到几秒差别非常明显。这篇要解决的就是这条完整链路Python 读取 Excel → 字段映射与类型转换 →pymysql的executemany批量写入 → 分批提交与异常回滚 → 最后用行数比对验证一致性。适合已经会写基础 Python、但被大批量导入卡住的同学也适合需要把 Excel 当临时数据源做一次性迁移的场景。核心检索词先摆出来python excel 批量导入数据库、pymysql executemany 用法、executemany 分批提交。这三个词基本覆盖了你会搜到的所有问题。为什么不用xlrd了因为xlrd从 2.0 开始只支持.xls读.xlsx会直接报错。现在更推荐openpyxl或pandas。本文用openpyxl读.xlsx用pymysql写库两者都是纯 Python 实现装起来干净。还有一个容易被忽略的点调用凭据的管理。如果你后续要接大模型做字段映射建议、或者用 AI 辅助生成 SQL散落各处的 Key 很容易失控。TaoToken 提供统一 Key 管理把模型调用凭据收敛到一处配合本文的数据库写入链路整个数据管道会更清爽。官网入口在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_end 后面第三节会给具体配置。先把环境装好pip install openpyxl pymysql数据库这边假设你已经有一张目标表。为了演示我建一张简单的用户表CREATE TABLE user_import ( id INT PRIMARY KEY AUTO_INCREMENT, user_name VARCHAR(64), age INT, city VARCHAR(64), signup_date DATE ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;Excel 结构对应五列user_name、age、city、signup_date。注意 Excel 里的日期读出来可能是datetime对象也可能是字符串这个差异后面会专门处理。2. TaoToken 统一 Key 的前置准备与凭据收敛在写导入脚本之前先把凭据这件事理清楚。很多人的导入脚本里硬编码了数据库密码甚至把模型 API Key 也写死在代码里一旦要换环境就得全局搜索替换。TaoToken 的思路是把模型调用凭据统一管理你只需要维护一个 Key不同模型、不同用途都走同一个入口。这一步不是必须的但如果你打算在导入流程里加入 AI 能力——比如让模型帮你把中文列名映射成英文字段、或者自动生成建表语句——那统一 Key 会省很多事。我试过在字段映射环节接一个模型对话接口把 Excel 表头丢过去让它返回推荐的字段名和类型准确率还不错。TaoToken 的接入文档在 https://taotoken.net/doc API 基址是 https://taotoken.net/api 。注意 API 地址不带 UTM 参数保持干净。获取 Key 的入口在 https://taotoken.net/api-keys 登录后在控制台 https://taotoken.net/console 可以看到用量。具体怎么用假设你要在导入脚本里加一个「表头智能映射」的小功能可以这样组织配置。先建一个config.py# config.py import os # 数据库配置 DB_CONFIG { host: 127.0.0.1, port: 3306, user: root, password: os.getenv(DB_PASSWORD, your_password), database: demo, charset: utf8mb4, } # TaoToken 统一 Key 配置 TAOTOKEN_BASE_URL https://taotoken.net/api TAOTOKEN_API_KEY os.getenv(TAOTOKEN_API_KEY, ) TAOTOKEN_MODEL claude-sonnet-4-5把 Key 放在环境变量里而不是写死在代码中。Linux/macOS 下export TAOTOKEN_API_KEYsk-你的key export DB_PASSWORD你的数据库密码Windows PowerShell$env:TAOTOKEN_API_KEYsk-你的key $env:DB_PASSWORD你的数据库密码这样做的直接好处是脚本可以进 GitKey 不会泄露换机器只需要重新导出环境变量。如果你用 Claude Code 做辅助开发可以在 https://taotoken.net/claude-code 看到接入方式把 Base URL 指向 TaoToken 的 API 地址即可。需要强调的是TaoToken 在这里的角色是凭据统一管理不是数据库代理。数据库连接还是走你自己的pymysql两者互不干扰。模型调用只在需要 AI 辅助的环节出现比如字段映射建议、异常数据解释。如果你暂时不需要 AI 能力这一节可以跳过直接看第三节的导入脚本。但有一点值得提前做把数据库配置和模型配置都收敛到config.py后续无论加什么功能入口都是统一的。这是我在多个项目里踩过坑之后的习惯——配置散落是维护成本的最大来源。3. 可复制的 pymysql executemany 配置与 Excel 解析脚本这一节是核心给出完整可运行的脚本。先看整体结构再逐段解释。# import_excel.py import pymysql from openpyxl import load_workbook from datetime import datetime, date from config import DB_CONFIG # 字段映射Excel 列索引 - 数据库字段 COLUMN_MAPPING { 0: user_name, 1: age, 2: city, 3: signup_date, } INSERT_SQL INSERT INTO user_import (user_name, age, city, signup_date) VALUES (%s, %s, %s, %s) BATCH_SIZE 5000 def get_connection(): return pymysql.connect(**DB_CONFIG) def parse_date(value): 把 Excel 里的日期统一转成 date 对象 if value is None: return None if isinstance(value, datetime): return value.date() if isinstance(value, date): return value if isinstance(value, str): for fmt in (%Y-%m-%d, %Y/%m/%d, %Y.%m.%d): try: return datetime.strptime(value.strip(), fmt).date() except ValueError: continue raise ValueError(f无法解析日期: {value!r}) def parse_int(value): if value is None or value : return None return int(float(value)) def read_excel(path, sheet_nameNone): 读取 Excel返回生成器逐行产出元组 wb load_workbook(path, read_onlyTrue, data_onlyTrue) ws wb[sheet_name] if sheet_name else wb.active rows ws.iter_rows(min_row2, values_onlyTrue) # 跳过表头 for row in rows: if row is None or all(cell is None for cell in row): continue yield ( str(row[0]).strip() if row[0] is not None else None, parse_int(row[1]), str(row[2]).strip() if row[2] is not None else None, parse_date(row[3]), ) wb.close() def batch_insert(excel_path, sheet_nameNone): conn get_connection() cursor conn.cursor() total 0 batch [] try: for record in read_excel(excel_path, sheet_name): batch.append(record) if len(batch) BATCH_SIZE: cursor.executemany(INSERT_SQL, batch) conn.commit() total len(batch) print(f已提交 {total} 行) batch.clear() if batch: cursor.executemany(INSERT_SQL, batch) conn.commit() total len(batch) print(f已提交 {total} 行) except Exception as e: conn.rollback() print(f导入失败已回滚当前批次: {e}) raise finally: cursor.close() conn.close() return total if __name__ __main__: count batch_insert(users.xlsx) print(f导入完成共 {count} 行)几个关键点展开说。字段映射用字典而不是硬编码索引。上面COLUMN_MAPPING定义了 Excel 列索引到数据库字段的对应关系。实际项目里 Excel 列顺序经常变把映射抽出来改一处就行。如果你用 TaoToken 接模型做智能映射可以让模型读表头返回这个字典。类型转换单独成函数。parse_date处理了datetime、date、字符串三种情况字符串又尝试了三种常见格式。parse_int先转float再转int因为 Excel 里的数字读出来可能是1.0这种浮点。这两个函数是踩坑重灾区建议直接抄。read_onlyTrue和data_onlyTrue。read_only让openpyxl用流式读取几万行也不会吃爆内存data_only让公式单元格返回计算结果而不是公式本身。这两个参数在大文件场景下是必须的。生成器逐行产出。read_excel用yield而不是一次性返回列表配合batch列表做分批内存占用稳定在BATCH_SIZE条左右。这就是executemany分批提交的核心思路。executemany的参数模板。SQL 里用%s占位batch是元组列表每个元组的元素顺序必须和 SQL 里的字段顺序一致。这是最容易出错的地方——顺序错了不会报错但数据会串位。分批提交与回滚。每BATCH_SIZE条commit一次异常时rollback当前未提交的批次。注意rollback只能回滚未提交的部分已经commit的批次不会撤销。如果你需要全量原子性把BATCH_SIZE设大或者去掉中间commit但那样内存和事务日志压力会上升。折中方案是分批提交 记录已提交行数失败后从断点续传。BATCH_SIZE设多少合适5000 到 10000 是比较稳的区间。太小提交频繁网络往返多太大单次事务日志膨胀回滚慢。我实测 5000 行一批10 万行数据大概 20 秒左右取决于磁盘和网络。4. 验证请求与成功结果行数比对与重复导入检查脚本跑完不算完得验证数据真的进去了而且没重复、没串位。这一步很多人跳过结果线上发现数据对不上再回头查成本高得多。第一步比对行数。Excel 有效行数减去表头应该等于数据库里的行数from openpyxl import load_workbook import pymysql from config import DB_CONFIG def count_excel_rows(path): wb load_workbook(path, read_onlyTrue) ws wb.active count 0 for row in ws.iter_rows(min_row2, values_onlyTrue): if row and any(cell is not None for cell in row): count 1 wb.close() return count def count_db_rows(): conn pymysql.connect(**DB_CONFIG) cursor conn.cursor() cursor.execute(SELECT COUNT(*) FROM user_import) n cursor.fetchone()[0] cursor.close() conn.close() return n excel_n count_excel_rows(users.xlsx) db_n count_db_rows() print(fExcel: {excel_n}, DB: {db_n}, 一致: {excel_n db_n})如果数字对不上先查是不是有空行被跳过、或者有行因为类型转换失败被吞掉。上面的read_excel里对空行做了continue这是有意为之但你要确认业务上这些空行确实该跳过。第二步抽样比对内容。行数对了不代表内容对。随机抽几行把 Excel 原值和数据库查出来的值放一起看def sample_check(path, sample_ids): wb load_workbook(path, read_onlyTrue, data_onlyTrue) ws wb.active excel_rows {} for idx, row in enumerate(ws.iter_rows(min_row2, values_onlyTrue), start1): if idx in sample_ids: excel_rows[idx] row wb.close() conn pymysql.connect(**DB_CONFIG) cursor conn.cursor(pymysql.cursors.DictCursor) for idx in sample_ids: cursor.execute(SELECT * FROM user_import WHERE id %s, (idx,)) db_row cursor.fetchone() print(f行 {idx} Excel: {excel_rows.get(idx)}) print(f行 {idx} DB: {db_row}) cursor.close() conn.close()第三步重复导入检查。如果你不小心跑了两次脚本数据会翻倍。加一个唯一索引或者导入前清空表能避免但更稳的是导入前检查def is_table_empty(): conn pymysql.connect(**DB_CONFIG) cursor conn.cursor() cursor.execute(SELECT COUNT(*) FROM user_import) n cursor.fetchone()[0] cursor.close() conn.close() return n 0 if not is_table_empty(): raise RuntimeError(目标表非空请先清空或改用增量导入)或者用INSERT IGNORE/ON DUPLICATE KEY UPDATE做幂等写入但那需要表上有唯一键。对于一次性导入导入前清空是最简单的。成功结果长这样已提交 5000 行 已提交 10000 行 已提交 12345 行 导入完成共 12345 行 Excel: 12345, DB: 12345, 一致: True看到行数一致、抽样内容对得上基本就稳了。如果这一步发现日期字段差了一天多半是时区问题检查parse_date有没有做时区转换。5. 本篇常见报错排查401、local proxy failed、reading choices这一节列几个真实会撞上的报错以及排查路径。报错一pymysql.err.OperationalError: (1045, Access denied for user)这是数据库认证失败不是 TaoToken 的 401。检查DB_CONFIG里的 user 和 password确认环境变量DB_PASSWORD有没有导出成功。在 Python 里打印一下from config import DB_CONFIG print(DB_CONFIG[user], DB_CONFIG[password][:2] ***)如果密码是空字符串说明环境变量没读到。注意os.getenv的默认值只在变量不存在时生效变量存在但为空字符串时不会用默认值。报错二pymysql.err.ProgrammingError: (1064, You have an error in your SQL syntax)多半是 SQL 里的字段名和表结构对不上或者executemany的占位符数量与元组元素数量不一致。数一下INSERT_SQL里的%s有几个batch里每个元组有几个元素必须相等。我踩过的坑是复制粘贴时多了一个%s报错信息不会直接告诉你数量不匹配得自己数。报错三openpyxl.utils.exceptions.InvalidFileException: openpyxl does not support the old .xls file format你拿.xls文件喂给了openpyxl。两个选择用 Excel 另存为.xlsx或者装xlrd1.2.0专门读.xls。后者版本锁死不推荐。统一转.xlsx最省事。报错四local proxy failed或连接超时如果你在调用 TaoToken 的模型接口时看到这个先确认TAOTOKEN_BASE_URL是https://taotoken.net/api不要带多余路径。然后检查网络能不能通curl -I https://taotoken.net/api如果返回 401说明网络通但 Key 不对检查TAOTOKEN_API_KEY环境变量。如果返回连接超时检查本机网络配置。注意不要使用任何非正规的网络工具企业环境通常有合规的出口方案走正规渠道即可。报错五reading choices相关错误这个通常出现在模型返回的 JSON 解析环节。如果你用模型做字段映射返回内容里可能带了 markdown 代码块标记直接json.loads会失败。处理方式import json import re def extract_json(text): match re.search(r\{.*\}, text, re.DOTALL) if not match: raise ValueError(未找到 JSON 内容) return json.loads(match.group())报错六pymysql.err.DataError: (1406, Data too long for column)Excel 里某个单元格内容超过了数据库字段长度。排查方式是先扫一遍 Excel 各列最大长度def max_lengths(path): wb load_workbook(path, read_onlyTrue, data_onlyTrue) ws wb.active lengths {} for row in ws.iter_rows(min_row2, values_onlyTrue): for i, cell in enumerate(row): if cell is not None: lengths[i] max(lengths.get(i, 0), len(str(cell))) wb.close() return lengths对照表结构把字段长度调大或者在导入前截断。关于 Claude Code / Cline MCP / Codex auth.json 的三件套如果你用这些工具辅助开发配置里必须写全三样Base URL、Key、Model ID。以 Claude Code 为例Base URL 填https://taotoken.net/apiKey 填你的 TaoToken KeyModel ID 填你用的模型名。缺任何一个都会报认证或路由错误。Cline 的 MCP 配置同理在 settings 里把这三项对齐。Codex 的auth.json里对应的是api_base、api_key、model三个字段。6. 长期编码与 Agent 场景下的凭据管理建议导入脚本写完之后如果你还要持续维护数据管道或者用 Agent 做自动化凭据管理会变成日常问题。我的建议是把 TaoToken 的 Key 和数据库密码都走环境变量脚本里只留读取逻辑。这样无论是本地跑、CI 跑还是容器里跑配置方式一致。对于长期编码场景Coding Plan 提供了更稳定的调用额度入口在 https://taotoken.net/coding-plan 。如果你只是偶尔验证模型输出用模型对话页面就够了https://taotoken.net/chat 。需要看用量和 Key 状态就去控制台 https://taotoken.net/console Key 管理在 https://taotoken.net/api-keys 。回到导入本身最后给一个实用技巧把BATCH_SIZE和INSERT_SQL做成命令行参数这样同一份脚本能复用到不同表和不同数据量import argparse parser argparse.ArgumentParser() parser.add_argument(--excel, requiredTrue) parser.add_argument(--table, requiredTrue) parser.add_argument(--batch-size, typeint, default5000) args parser.parse_args()然后INSERT_SQL用 f-string 拼表名表名来自命令行注意做白名单校验防止注入BATCH_SIZE用args.batch_size。这样一份脚本能覆盖大部分导入需求不用每次改代码。数据导入这件事核心就三点类型转换要稳、分批提交要控内存、导入后要验证。把这三件事做扎实几万行还是几十万行差别只是多等几秒。