Python自动化实战：用TaoToken统一Key批量修改数据库字符集配置

1. 多实例 MySQL 字符集统一为什么手动改容易翻车多库环境里做字符集统一麻烦的从来不是ALTER TABLE这条语句本身而是它前面那一堆准备工作十几个实例的 host、port、账号密码散落在不同脚本、不同.env、不同同事的备忘录里有人用utf8有人用utf8mb4还有人 collation 写的是utf8mb4_general_ci改完发现外键关联的两张表排序规则不一致直接报错。更别提改到一半连接断了你根本不知道哪些表已经转换、哪些还没动。我这次要做的场景很具体把一批 MySQL 实例里的业务库从旧的utf8/utf8mb3统一迁到utf8mb4utf8mb4_bin用 Python 脚本批量跑同时把「连接配置」和「模型调用凭证」这两类敏感信息收敛到一处管理。前者用config.toml做多实例配置后者用 TaoToken 的统一 Key 来接管——这样脚本里不再硬编码任何密钥换实例、加实例只改配置文件。这篇文章交付三样东西一份可直接复制的config.toml骨架、TaoToken 统一 Key 的接入步骤、以及批量执行字符集变更后的验证 SQL 和回滚检查动作。适合手里有多个 MySQL 实例、正在做字符集治理、又不想把密码写死在代码里的后端或运维同学。下面按「先讲清楚问题 → 再搭配置 → 再跑脚本 → 再验证回滚」的顺序走每一步都能直接跟做。2. 用 TaoToken 统一 Key 接管脚本里的凭证管理先说清楚为什么要引入 TaoToken。批量改字符集的脚本本身不复杂但一旦你想在脚本里加一点「智能」能力——比如让模型帮你判断某张表的 collation 是否真的需要转换、或者生成变更报告摘要——就会遇到凭证问题每个实例一套数据库密码已经够乱了再叠加大模型 API Key管理成本直接翻倍。TaoToken 在这里的角色是「统一 Key 入口」你只需要在平台申请一个 Key脚本通过它来调用模型能力不用在每台机器、每个脚本里分别配置不同的模型凭证。数据库连接信息仍然走config.toml模型调用凭证走 TaoToken两类敏感信息各归其位。接入步骤不复杂按下面走第一步打开模型对话页面确认你要用的模型可用https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite第二步进入控制台创建 API Keyhttps://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite第三步在 API Keys 页面复制你的 Keyhttps://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite第四步如果你后续要做长期编码或 Agent 任务可以了解 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite接口地址统一用https://taotoken.net/api不要带 UTM 参数。Key 拿到后不要写进代码放进环境变量或config.toml的独立字段里脚本运行时读取。这样即使脚本被分享出去凭证也不会泄露。注意TaoToken 是模型调用的统一入口不是数据库代理也不替代你的 MySQL 客户端。数据库连接仍然由pymysql直连TaoToken 只负责模型侧凭证收敛。3. 可复制的 config.toml 骨架与 Python 批量脚本配置文件的思路是一个[taotoken]段放模型凭证一个[[databases]]数组放多个 MySQL 实例。每个实例有自己的别名、连接参数和目标字符集。这样加实例就是复制一段配置不用改代码。# config.toml [taotoken] api_key sk-your-taotoken-key base_url https://taotoken.net/api model gpt-4o-mini [[databases]] alias order_db host 10.0.0.11 port 3306 user charset_admin password your_password_1 database order_center target_charset utf8mb4 target_collate utf8mb4_bin [[databases]] alias user_db host 10.0.0.12 port 3306 user charset_admin password your_password_2 database user_center target_charset utf8mb4 target_collate utf8mb4_bin [[databases]] alias log_db host 10.0.0.13 port 3306 user charset_admin password your_password_3 database log_center target_charset utf8mb4 target_collate utf8mb4_binPython 脚本分三块读配置、连库、逐表转换。关键点是先查当前字符集再决定是否转换避免对已经是utf8mb4的表重复执行ALTER大表上这是纯浪费。# charset_migrate.py import tomllib import pymysql from pymysql.cursors import DictCursor def load_config(pathconfig.toml): with open(path, rb) as f: return tomllib.load(f) def get_tables(cursor, db_name): cursor.execute( SELECT TABLE_NAME, TABLE_COLLATION FROM information_schema.TABLES WHERE TABLE_SCHEMA %s AND TABLE_TYPE BASE TABLE, (db_name,) ) return cursor.fetchall() def convert_table(cursor, table_name, charset, collate): sql ( fALTER TABLE {table_name} fCONVERT TO CHARACTER SET {charset} COLLATE {collate} ) cursor.execute(sql) def migrate_instance(cfg): conn pymysql.connect( hostcfg[host], portcfg[port], usercfg[user], passwordcfg[password], databasecfg[database], charsetutf8mb4, cursorclassDictCursor, autocommitFalse, ) changed, skipped [], [] try: with conn.cursor() as cursor: tables get_tables(cursor, cfg[database]) for t in tables: name t[TABLE_NAME] current (t[TABLE_COLLATION] or ).lower() if current.startswith(cfg[target_charset]): skipped.append(name) continue convert_table(cursor, name, cfg[target_charset], cfg[target_collate]) changed.append(name) conn.commit() except Exception as e: conn.rollback() print(f[{cfg[alias]}] 失败并回滚: {e}) raise finally: conn.close() print(f[{cfg[alias]}] 已转换 {len(changed)} 张表, 跳过 {len(skipped)} 张) return changed, skipped if __name__ __main__: config load_config() for db in config[databases]: migrate_instance(db)几个容易忽略的细节autocommitFalse是故意的让每张表的ALTER在同一个事务里提交出错能整体回滚charsetutf8mb4是连接层字符集和表字符集是两回事别混表名用反引号包起来防止关键字冲突。4. 验证请求与成功结果跑完脚本后查什么脚本跑完不代表改对了。字符集变更最容易出问题的地方是列级 collation 没跟着变或者外键关联的两张表 collation 不一致。所以验证要分三层库级、表级、列级。先看库级默认字符集SELECT SCHEMA_NAME, DEFAULT_CHARACTER_SET_NAME, DEFAULT_COLLATION_NAME FROM information_schema.SCHEMATA WHERE SCHEMA_NAME IN (order_center, user_center, log_center);再看表级确认没有漏网的表SELECT TABLE_SCHEMA, TABLE_NAME, TABLE_COLLATION FROM information_schema.TABLES WHERE TABLE_SCHEMA order_center AND TABLE_COLLATION NOT LIKE utf8mb4%;这条 SQL 如果返回空说明该库所有表都已经是utf8mb4系。最后查列级这是最容易被忽略的一层SELECT TABLE_NAME, COLUMN_NAME, CHARACTER_SET_NAME, COLLATION_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA order_center AND CHARACTER_SET_NAME IS NOT NULL AND COLLATION_NAME NOT LIKE utf8mb4%;同样返回空才算干净。如果列级还有残留说明CONVERT TO CHARACTER SET没覆盖到某些列通常是TEXT/VARCHAR之外的二进制列或者视图依赖需要单独处理。成功结果长这样脚本输出[order_db] 已转换 42 张表, 跳过 3 张三条验证 SQL 全部返回空集。这时候你可以在脚本里加一段模型调用让 TaoToken 帮你把变更结果整理成一句话报告import requests def summarize(changed_map, api_key, base_url, model): prompt 用一句话总结以下数据库字符集变更结果 str(changed_map) resp requests.post( f{base_url}/v1/chat/completions, headers{Authorization: fBearer {api_key}}, json{model: model, messages: [{role: user, content: prompt}]}, timeout30, ) return resp.json()[choices][0][message][content]这段调用走的就是 TaoToken 的统一 Keybase_url用https://taotoken.net/api模型名按你在模型对话页面确认的填。凭证从config.toml的[taotoken]段读脚本里不出现明文 Key。5. 本篇常见错排查字符集变更踩过的坑报错一Illegal mix of collations。这是外键或 JOIN 时两张表 collation 不一致导致的。排查方法找出关联的两张表分别查TABLE_COLLATION如果一张是utf8mb4_bin、另一张是utf8mb4_general_ci统一成同一个。注意utf8mb4_bin是区分大小写的如果你的业务依赖不区分大小写的查询要慎重选 collation。报错二ALTER TABLE卡住不动。大表上CONVERT TO CHARACTER SET会重建整张表锁表时间长。排查先看SHOW PROCESSLIST确认是不是在等锁如果是考虑用pt-online-schema-change或gh-ost做在线变更。脚本里可以加一个表大小判断超过阈值就跳过并记录人工处理。报错三连接层字符集没设对中文变问号。pymysql.connect里的charset参数控制的是客户端和服务器之间的传输编码和表字符集是两码事。如果连接层用了latin1即使表是utf8mb4写入的中文也会乱码。统一设成utf8mb4。报错四TaoToken 调用返回 401。检查config.toml里的api_key是否完整复制、有没有多余空格确认base_url是https://taotoken.net/api而不是带 UTM 的官网地址。如果还是不行去 API Keys 页面重新生成一个 Key 试试。报错五脚本跑一半断了不知道改到哪。这就是为什么建议每张表单独提交、并记录日志。可以在convert_table成功后往一张migration_log表写一条记录断点续跑时先查这张表跳过已完成的。提示字符集变更前一定要有备份。CONVERT TO CHARACTER SET虽然理论上可逆但 collation 从_bin改回_general_ci时如果数据里有大小写不同的重复值回滚会失败。备份是唯一的保险。6. 长期跑批与 Agent 场景的凭证收敛建议如果你只是偶尔跑一次字符集变更上面的配置够用了。但如果你在做的是「多实例 多任务」的长期自动化——比如每天定时巡检字符集、自动生成变更报告、甚至让 Agent 根据报错自动决定下一步——那凭证管理就要再收敛一层。我的做法是把 TaoToken 的 Key 放在环境变量里config.toml只存数据库连接信息模型凭证通过os.environ读取。这样配置文件可以进版本库数据库密码用单独的 secrets 文件或密钥管理服务而模型 Key 永远不落盘。长期编码或 Agent 任务可以走 Coding Plan把模型调用额度集中管理https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite接入文档在这里遇到参数问题可以对照查https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite最后说一个我自己的习惯每次批量变更前先拿一个测试实例跑一遍完整流程包括验证 SQL 和回滚动作。确认没问题了再对生产实例执行。字符集这种事宁可多花十分钟验证也别在生产上赌运气。