1. 从 xlrd 到 pymysql：Excel 批量导入数据库的真实痛点

如果你手头有一份几万行的 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_source=taotoken_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 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

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_KEY="sk-你的key" export DB_PASSWORD="你的数据库密码"

Windows PowerShell：

$env:TAOTOKEN_API_KEY="sk-你的key" $env:DB_PASSWORD="你的数据库密码"

这样做的直接好处是：脚本可以进 Git，Key 不会泄露；换机器只需要重新导出环境变量。如果你用 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_name=None): """读取 Excel，返回生成器，逐行产出元组""" wb = load_workbook(path, read_only=True, data_only=True) ws = wb[sheet_name] if sheet_name else wb.active rows = ws.iter_rows(min_row=2, values_only=True) # 跳过表头 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_name=None): 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_only=True和data_only=True。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_only=True) ws = wb.active count = 0 for row in ws.iter_rows(min_row=2, values_only=True): 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(f"Excel: {excel_n}, DB: {db_n}, 一致: {excel_n == db_n}")

如果数字对不上，先查是不是有空行被跳过、或者有行因为类型转换失败被吞掉。上面的read_excel里对空行做了continue，这是有意为之，但你要确认业务上这些空行确实该跳过。

第二步，抽样比对内容。行数对了不代表内容对。随机抽几行，把 Excel 原值和数据库查出来的值放一起看：

def sample_check(path, sample_ids): wb = load_workbook(path, read_only=True, data_only=True) ws = wb.active excel_rows = {} for idx, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=1): 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，或者装xlrd==1.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_only=True, data_only=True) ws = wb.active lengths = {} for row in ws.iter_rows(min_row=2, values_only=True): 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/api，Key 填你的 TaoToken Key，Model 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", required=True) parser.add_argument("--table", required=True) parser.add_argument("--batch-size", type=int, default=5000) args = parser.parse_args()

然后INSERT_SQL用 f-string 拼表名（表名来自命令行，注意做白名单校验防止注入），BATCH_SIZE用args.batch_size。这样一份脚本能覆盖大部分导入需求，不用每次改代码。

数据导入这件事，核心就三点：类型转换要稳、分批提交要控内存、导入后要验证。把这三件事做扎实，几万行还是几十万行，差别只是多等几秒。