#!/usr/bin/env python3 """migrate_collation.py — 统一跨库字符排序规则（collation），使跨库 UNION/JOIN 可用 背景（P0-6 的后半部分） ---------------------- 日期**格式**已统一为 `YYYYMMDD`（见 migrate_date_format.py），但跨库 `UNION` 仍报： ERROR 1271 (HY000): Illegal mix of collations for operation 'UNION' 原因是列的 **collation** 不同： tushare_data 63 张表 / 316 列 utf8mb4_unicode_ci akshare_data 18 张表 utf8mb4_unicode_ci（库内自身不一致） fin_data_clean 全部 utf8mb4_0900_ai_ci 统一目标选 `utf8mb4_0900_ai_ci`：MySQL 8+ 默认，且 3 个库里 2 个已经是它。 **重要前提**：collation 变更 MySQL **只能走 COPY 算法（整表重建）**， 期间该表禁止写入。因此必须在有备份、且无调度任务的窗口执行。 用法 ---- python3 Scripts/bin/migrate_collation.py --report python3 Scripts/bin/migrate_collation.py --execute [--only 库.表] [--limit N] python3 Scripts/bin/migrate_collation.py --verify `--execute` 按表体积**升序**执行：小表先完成，便于尽早发现机制问题。 每张表执行前后都记录 collation 与行数，任一异常即停止后续（fail-fast）。 """ from __future__ import annotations import argparse import os import sys import time from datetime import datetime import pymysql TARGET_COLLATION = "utf8mb4_0900_ai_ci" TARGET_CHARSET = "utf8mb4" SCHEMAS = ("tushare_data", "akshare_data", "fin_data_clean") def connect(): pwd = os.environ.get("MYSQL_PASSWORD") if not pwd: print("✗ 需要 MYSQL_PASSWORD（可先 `set -a; . ~/.config/findatahub/env; set +a`）") sys.exit(2) return pymysql.connect( host="127.0.0.1", port=3306, user=os.environ.get("MYSQL_USER", "root"), password=pwd, charset="utf8mb4", connect_timeout=10, read_timeout=3600, autocommit=True, ) def pending_tables(cur) -> list[tuple[str, str, float, int]]: """返回待转换的表：(库, 表, MB, 估算行数)，按体积升序。""" cur.execute( "SELECT table_schema, table_name, " " ROUND((data_length+index_length)/1024/1024, 1), table_rows " "FROM information_schema.tables " "WHERE table_schema IN (%s, %s) AND table_type='BASE TABLE' " " AND table_collation <> %s " "ORDER BY (data_length+index_length) ASC", (SCHEMAS[0], SCHEMAS[1], TARGET_COLLATION), ) return [(r[0], r[1], float(r[2] or 0), int(r[3] or 0)) for r in cur.fetchall()] def table_collation(cur, sch: str, tbl: str) -> str: cur.execute( "SELECT table_collation FROM information_schema.tables " "WHERE table_schema=%s AND table_name=%s", (sch, tbl), ) r = cur.fetchone() return r[0] if r else "?" def row_count(cur, sch: str, tbl: str) -> int: cur.execute(f"SELECT COUNT(*) FROM `{sch}`.`{tbl}`") return int(cur.fetchone()[0]) def leftover_columns(cur, sch: str, tbl: str) -> list[tuple[str, str]]: """该表内 collation 不等于目标的字符列（CONVERT 后应为空）。""" cur.execute( "SELECT column_name, collation_name FROM information_schema.columns " "WHERE table_schema=%s AND table_name=%s AND collation_name IS NOT NULL " " AND collation_name <> %s", (sch, tbl, TARGET_COLLATION), ) return list(cur.fetchall()) def cmd_report(cur) -> None: todo = pending_tables(cur) total_mb = sum(t[2] for t in todo) print(f"待转换表：{len(todo)} 张，合计 {total_mb:,.0f} MB") print(f"目标：{TARGET_CHARSET} / {TARGET_COLLATION}（MySQL 8 默认，且 2/3 的库已是它）") print() print(f"{'库.表':<46}{'MB':>10}{'估算行数':>14}") print("-" * 72) for sch, tbl, mb, rows in todo: print(f"{sch+'.'+tbl:<46}{mb:>10,.1f}{rows:>14,}") print("-" * 72) cur.execute( "SELECT table_schema, COUNT(*) FROM information_schema.tables " "WHERE table_schema IN (%s, %s, %s) AND table_type='BASE TABLE' " " AND table_collation=%s GROUP BY table_schema", SCHEMAS + (TARGET_COLLATION,), ) print("已符合目标的表数：", ", ".join(f"{s}={c}" for s, c in cur.fetchall()) or "无") def cmd_execute(cur, only: str | None, limit: int | None) -> None: todo = pending_tables(cur) if only: todo = [t for t in todo if f"{t[0]}.{t[1]}" == only] if not todo: print(f"✗ --only {only!r} 未匹配待转换表（可能已转换）") return if limit: todo = todo[:limit] if not todo: print("✓ 没有待转换的表") return print(f"开始转换 {len(todo)} 张表（按体积升序，fail-fast）") print("⚠ collation 变更只能走 COPY（整表重建），期间该表禁止写入\n") t_all = time.time() for i, (sch, tbl, mb, est) in enumerate(todo, 1): before_col = table_collation(cur, sch, tbl) before_rows = row_count(cur, sch, tbl) t0 = time.time() print(f"[{i}/{len(todo)}] {sch}.{tbl} ({mb:,.1f} MB, {before_rows:,} 行) …", flush=True) try: cur.execute( f"ALTER TABLE `{sch}`.`{tbl}` " f"CONVERT TO CHARACTER SET {TARGET_CHARSET} COLLATE {TARGET_COLLATION}" ) except Exception as e: # noqa: BLE001 print(f" ✗ ALTER 失败，已停止后续：{str(e)[:200]}") return el = time.time() - t0 after_col = table_collation(cur, sch, tbl) after_rows = row_count(cur, sch, tbl) left = leftover_columns(cur, sch, tbl) ok = (after_col == TARGET_COLLATION and after_rows == before_rows and not left) print(f" {before_col} → {after_col} | 行数 {before_rows:,} → {after_rows:,} | " f"残留异类列 {len(left)} | 耗时 {el:.1f}s {'✓' if ok else '✗'}", flush=True) if not ok: print(" ✗ 校验未通过，已停止后续（请人工检查）") return print(f"\n全部完成，总耗时 {time.time()-t_all:.1f}s") def cmd_verify(cur) -> int: ok = True print(f"{'库':<18}{'表数':>6}{'符合目标':>10}{'不符合':>8} 结论") print("-" * 60) for sch in SCHEMAS: cur.execute( "SELECT COUNT(*), SUM(table_collation=%s) FROM information_schema.tables " "WHERE table_schema=%s AND table_type='BASE TABLE'", (TARGET_COLLATION, sch), ) total, good = cur.fetchone() good = int(good or 0) total = int(total or 0) good_flag = good == total ok = ok and good_flag print(f"{sch:<18}{total:>6}{good:>10}{total-good:>8} {'✓' if good_flag else '✗'}") print("-" * 60) # 端到端：跨库 UNION 与 JOIN 必须可用（这才是转换的实际收益） try: cur.execute( "(SELECT trade_date FROM akshare_data.stock_daily WHERE trade_date='20260902' LIMIT 1) " "UNION ALL " "(SELECT trade_date FROM tushare_data.stock_daily WHERE trade_date='20260916' LIMIT 1)" ) rows = cur.fetchall() print(f"跨库 UNION : ✓ 返回 {len(rows)} 行（无需显式 COLLATE）") except Exception as e: # noqa: BLE001 ok = False print(f"跨库 UNION : ✗ {str(e)[:140]}") try: cur.execute( "SELECT COUNT(*) FROM akshare_data.stock_daily a " "JOIN tushare_data.stock_daily b " " ON a.ts_code = b.ts_code AND a.trade_date = b.trade_date " "WHERE a.trade_date = '20260902'" ) print(f"跨库 JOIN 同标的对账 : ✓ 命中 {cur.fetchone()[0]:,} 行") except Exception as e: # noqa: BLE001 ok = False print(f"跨库 JOIN 同标的对账 : ✗ {str(e)[:140]}") print() print("校验" + ("通过 ✓" if ok else "未通过 ✗")) return 0 if ok else 1 def main() -> int: ap = argparse.ArgumentParser() g = ap.add_mutually_exclusive_group(required=True) g.add_argument("--report", action="store_true") g.add_argument("--execute", action="store_true") g.add_argument("--verify", action="store_true") ap.add_argument("--only", default=None, help="只处理 库.表") ap.add_argument("--limit", type=int, default=None, help="最多处理前 N 张（按体积升序）") args = ap.parse_args() print(f"=== migrate_collation {datetime.now():%Y-%m-%d %H:%M:%S} ===", flush=True) conn = connect() try: cur = conn.cursor() if args.report: cmd_report(cur) return 0 if args.execute: cmd_execute(cur, args.only, args.limit) return 0 return cmd_verify(cur) finally: conn.close() if __name__ == "__main__": sys.exit(main())