BJ_ZTEITS-Developers / parking-cloud

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
从**两份判定表**生成合库迁移用的六个产物：

    # ---- 全量迁移（"零丢失"，747 GB）----
    1. sql/migration/exclude_tables.txt      每个源库里"不搬"的表
    2. sql/migration/20_union_lost_rows.sql  只存在于副本、必须补进保留侧的行
    3. sql/migration/99_verify.sql           合库后校验（快）
    4. sql/migration/98_count_reconcile.sql  逐表行数对账（"零丢失"的正面证据）

    # ---- 抽样开发库（每张表几百行，给开发用；见 10_dump_legacy.sh 的 SAMPLE=N）----
    5. sql/migration/struct_only_tables.txt  **只搬结构、不搬数据**的表（含密钥/口令）
    6. sql/migration/97_verify_sample.sql    抽样镜像的校验

⚠️ 第 3/4 与第 6 是**两套判据**，别混用：3/4 断言"目标行数 == 源库行数"，
   在抽样镜像上**必然全红**（抽样镜像的行数本来就该远小于源库）。
   抽样镜像跑 6；只有做全量迁移时才跑 3/4。

判定表（都是 import 现算，没有第二份人工清单）：

    · `check_subset.CHECKS`               —— 重名表保留哪一份（merge-conflict-plan.md）
    · `audit_junk_tables.DO_NOT_MIGRATE`  —— 废弃/备份表哪些不搬（legacy-db-actual.md §10.1）
      ⚠️ 它分两层：`空表`（0 行）与 `备份`（**有数据**，约 133 GB，业务方拍板）。
         改它必须先跑 `audit_junk_tables.py`（exit 0 才继续）再跑到这里。

## 为什么是"生成"而不是手写

`CHECKS` 是"哪张表保留哪一份"的**唯一事实来源**，`merge-conflict-plan.md` 的
判定也是基于它。手抄一份排除清单 = 第二个事实来源，两边迟早会不一致
（`AGENTS.md` §1.1 记的正是这个坑：审计脚本的判据必须跟着配置一起改）。
所以这里直接 import `CHECKS`，改名/加表只改一处。

## 列清单也现算

并集 INSERT 的列清单**从 information_schema 现算**（取两侧交集、剔除自增主键 `id`）：

- 剔除 `id` 是刻意的：两侧 `id` 空间不同（`zt_order.id=4` 的
  `wx1e9001e5940605b1` 在 `zt_order_n` 是 `id=1`），带着 `id` 插必然主键冲突。
  `id` 都是 `auto_increment`，省略即由 MySQL 分配新值；
  而**业务键（`cust_id` / `appid`）保持原值**——`TdCCarSMapper` 就是
  `left join td_c_cust_person b on a.cust_id=b.cust_id`，不依赖 `id`
- 取交集是因为保留侧列更多（`td_c_cust_person` 40 vs 36），多出来的列（如
  `etc_open_state`）在副本里没有，插入时留 NULL

## ⚠️ 绝不让密钥落盘

`td_b_wx_config` 含 `app_secret` / `encoding_aes_key` / `pay_api_secret`。
本脚本**只生成 `INSERT ... SELECT`**（列名，不含任何值）——密钥值在数据库之间直接流动，
**不经过文件、不经过终端、不经过日志**。生成物里只有列名，没有一行数据。

用法：
    python3 -I sql/tools/gen_migration_sql.py --env 私密/legacy-ro.env
"""

import argparse
import os
import re
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from legacy_db import load_env, db_names, connect  # noqa: E402
from check_subset import CHECKS  # noqa: E402
from audit_junk_tables import DO_NOT_MIGRATE, DO_NOT_MIGRATE_TIERS  # noqa: E402


def junk_excluded():
    """`{(库, 表)}` —— `legacy-db-actual.md` §10.1 拍板"不搬"的废弃/备份表。

    **事实来源是 `audit_junk_tables.DO_NOT_MIGRATE`**，与 `check_subset.CHECKS`
    同一个套路：绝不手抄第二份。那些条目分两层（详见该脚本文件头）：

      · `空表`（4 张）—— 0 行，不搬 = 新库少一张没人认领的表，**一行都不丢**
      · `备份`（3 张）—— **有数据**（含约 133 GB 的历史统计快照）。
        业务方 2026-10-10 拍板"备份表，无需理会"。不搬 = 那批行只留在老库
        （只读保留、未销毁，需要时可单独导出），但**新库里确实没有**。
        要改这个决定，先改 `DO_NOT_MIGRATE`，别无第二处。
    """
    return {tuple(x.split('.', 1)) for x in DO_NOT_MIGRATE}

HERE = os.path.dirname(os.path.abspath(__file__))
OUT_DIR = os.path.normpath(os.path.join(HERE, "..", "migration"))

# 目标库名（六库合一后的唯一库，见 sql/README.md §1）
TARGET_DB = "parking_cloud"

# 需要"取并集"的表：CHECKS 的自动判定只能给"有独有行"这个结论，
# 而"补进去还是只导出留痕"是人工决定（见 merge-conflict-plan.md §5）。
# 每项 = (表, 保留侧模块, 丢弃侧模块, 判重用的业务键)
UNION_TABLES = [
    # 29 行客户只有 zt_order_n 有；order 侧无 SQL 读它，真正读它的是 pay，
    # 而 pay 的 join 键是 cust_id —— 补进去对 pay 是修复而非破坏
    ("td_c_cust_person", "PAY", "ORDER", "cust_id"),
    # 3 行 2017 年的微信支付配置；WxPayApiConfigKit 按 appid 找配置，
    # 找不到直接抛 IllegalStateException
    ("td_b_wx_config", "ORDER", "PAY", "appid"),
]


def magnitude(n):
    """把 information_schema.TABLE_ROWS 的**估算**值折算成量级标签。

    为什么不把这个数原样印进产物（两个理由都实测过，不是洁癖）：

    1. **它是估算，不是实测**。MySQL 文档明说 InnoDB 的 TABLE_ROWS 对大表可差
       数十个百分点。产物里原先写的是「保留侧实测约 96,759,945 行」——把估算叫
       "实测"、印到个位数，是假精度，操作者会拿它当准数去对账。
    2. **它每次生成都不一样**。源库在线写入，估算值跟着漂移：同一张 11.9 亿行的表
       连续两次生成相差 77 行。于是「重跑生成器、看有没有 diff」这个同步检查
       **永远不可能干净**——一个每次都报 diff 的检查等于没有检查
       （与 AGENTS.md 里"一个每次都喊狼来了的守卫"同类），而且产物真被人手工改过时，
       那一处改动会淹没在这堆噪声里看不出来。

    量级标签让四个产物成为 CHECKS 的**确定性函数**，只在真的跨越量级边界时才变
    （那时 diff 出来的一行说明的是"表长大了"，不是噪声）。
    """
    if not n:
        return "（未知/空）"
    for limit, label in ((10 ** 9, "十亿级"), (10 ** 8, "亿级"), (10 ** 7, "千万级"),
                         (10 ** 6, "百万级"), (10 ** 5, "十万级"), (10 ** 4, "万级")):
        if n >= limit:
            return label
    return "万行以下"


def columns(cur, db, table):
    """[(列名, EXTRA, IS_NULLABLE), ...]，按表内顺序。"""
    cur.execute(
        "SELECT COLUMN_NAME, EXTRA, IS_NULLABLE FROM information_schema.COLUMNS "
        "WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s ORDER BY ORDINAL_POSITION",
        (db, table))
    return cur.fetchall()


def emit_excludes(dbs):
    """不搬的表，两类合起来：

    1. `check_subset.CHECKS` 里所有**重名表的副本**（保留侧见 merge-conflict-plan.md）
    2. `audit_junk_tables.DO_NOT_MIGRATE` 里**已拍板不搬**的废弃/备份表（§10.1）
       —— 又分两小类：`空表`（0 行，不搬零代价）与 `备份`（**有数据**，见文件头）

    两类都从源头 import 现算，产物里不存第二份事实。
    """
    dupes = {"%s.%s" % (dbs[drop_mod], table)
             for table, _km, drop_mod, _k, _n in CHECKS}
    junk = set(DO_NOT_MIGRATE)
    # ⚠️ 元组是 (层级, 理由)，别把变量名写反——写反了这里会静默算出 0（实测踩过）
    n_bak = sum(1 for tier, _why in DO_NOT_MIGRATE.values() if tier == '备份')
    uniq = sorted(dupes | junk)
    path = os.path.join(OUT_DIR, "exclude_tables.txt")
    with open(path, "w", encoding="utf-8") as fh:
        fh.write("# 合库时不搬的表\n")
        fh.write("# 由 sql/tools/gen_migration_sql.py 生成，**不要手改**\n")
        fh.write("#   来源一 = check_subset.py 的 CHECKS              —— 重名表的所有副本（%d 张）\n"
                 % len(dupes))
        fh.write("#   来源二 = audit_junk_tables.DO_NOT_MIGRATE       —— 已拍板不搬（%d 张：空表 %d + 有数据的备份 %d）\n"
                 % (len(junk), len(junk) - n_bak, n_bak))
        fh.write("# 用途：mysqldump --ignore-table=<此行> ；保留侧见 merge-conflict-plan.md\n")
        fh.write("# 行数=%d\n" % len(uniq))
        if n_bak:
            fh.write("#\n"
                     "# ⚠️ 来源二里的 `备份` 那 %d 张**是有数据的**（含约 133 GB 的历史统计快照）——\n"
                     "#    不是「空表所以无所谓」，是业务方 2026-10-10 拍板「备份表，无需理会」。\n"
                     "#    数据仍在六个老库里（只读保留、未销毁）；哪天要的话从**老库**单独导出，\n"
                     "#    别往新库里补。判断依据与理由串见 ../legacy-db-actual.md §10.1。\n"
                     % n_bak)
        fh.write("\n")
        for ln in sorted(dupes):
            fh.write(ln + "\n")
        if junk:
            fh.write("\n# ---- 以下为 §10.1 拍板不搬的废弃/备份表（非重名表副本，空表 %d + 备份 %d）----\n"
                     % (len(junk) - n_bak, n_bak))
            for ln in sorted(junk):
                fh.write(ln + "\n")
    return path, uniq


def emit_verify(cur, dbs):
    """生成 99_verify.sql。

    校验项同样**从 CHECKS 现算**，理由和 exclude 清单一样：任何手抄的"应保留
    X 表"清单都会和 CHECKS 漂移。这里生成三类断言：

      §3 每张重名表的**保留份**行数 == 目标库该表行数（保留份搬全了、且没多搬）
      §4 每张并集表：丢弃侧"还没进目标库"的行数 == 0（并集补全了）
      §1/§2/§5 覆盖率与字符集是静态 SQL（不依赖 CHECKS），直接写死

    注意：§3 必须**跳过并集表**——并集表的目标行数 = 保留侧 + 补进的行，
    拿它跟"保留侧"比必然不等，那是设计如此，不是错误。并集表由 §4 负责。

    ## §3 的判据是**不对称**的（彩排后改的）

    原先写的是"必须完全相等"，这在**在线**源库上必然报红：verify 跑在 dump 之后
    （T2 > T1），源库只要还在写，保留侧就会比目标多。彩排实测就是这样——
    zt_sys 38 张表里 `sys_op_log` 差了 2 行（源 448435 / 目标 448433）。
    一个每次都在高写入表上喊红的校验，会训练操作者忽略红色，比没有校验更坏。

    所以判据分开：
      · 目标 > 保留侧 → **🔴 一定有问题**（新增只会出现在源侧，目标不可能凭空多）
      · 目标 < 保留侧 → **⚠️ 需要解释**（是 dump 之后的在线新增？还是真漏搬？）
      · 相等 → OK
    "少"怎么解释见 98_count_reconcile.sql 的水位线法；维护窗口内跑则应全 OK。
    """
    union_names = {t for t, _k, _d, _key in UNION_TABLES}

    # (表, 保留库)，去重：parking_lot 在 CHECKS 里出现 3 次但保留侧只有 zt_park
    keep_pairs = sorted({(t, dbs[km]) for t, km, _dm, _k, _n in CHECKS})
    keep_pairs = [(t, db) for t, db in keep_pairs if t not in union_names]

    rows = []
    sizes = []
    for t, db in keep_pairs:
        # 现读体量：COUNT(*) 在 InnoDB 上是全索引扫描。小表无所谓，
        # 但 park_in_out_record 实测 ~9700 万行 / 32 GB（两条 COUNT(*) 要数分钟）——
        # 把量级写进产物头部，操作者才知道"这一步慢是正常的"，
        # 而不是以为卡死了。别在这里凭感觉写"体量不大"（我写过，是错的）。
        # 注意写的是 magnitude(n) 那个**量级标签**，不是原样数字，理由见 magnitude()。
        cur.execute("SELECT TABLE_ROWS FROM information_schema.TABLES "
                    "WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s", (db, t))
        r = cur.fetchone()
        n = r[0] if r and r[0] else 0
        sizes.append((t, db, n))
        note = ""
        if n > 5_000_000:
            note = ("   -- 保留侧{m}（TABLE_ROWS 估算），这里两条 COUNT(*) 需数分钟"
                    .format(m=magnitude(n)))
        rows.append(
            "SELECT '{t}' AS `表`, '{db}' AS `保留侧`, k.`保留侧行数`, t.`目标行数`,\n"
            "       IF(t.`目标行数` = k.`保留侧行数`, 'OK',\n"
            "          IF(t.`目标行数` > k.`保留侧行数`,\n"
            "             CONCAT('🔴 多 ', t.`目标行数` - k.`保留侧行数`, ' 行——目标不该比源多，只可能是重复导入/多搬了副本'),\n"
            "             CONCAT('⚠️ 少 ', k.`保留侧行数` - t.`目标行数`,\n"
            "                    ' 行——需解释：dump 之后的在线新增，还是真漏搬？（见 98 的水位线法）'))) AS `结论`\n"
            "  FROM (SELECT COUNT(*) AS `目标行数` FROM `{tgt}`.`{t}`) t\n"
            "  JOIN (SELECT COUNT(*) AS `保留侧行数` FROM `{db}`.`{t}`) k{n}".format(
                t=t, db=db, tgt=TARGET_DB, n=note))

    # 每行自带 "-- " 前缀，模板里就不要再加 -- 了（否则首行比其余行多一层缩进）
    # 按**表名**排序，不按体量排序：体量每次生成都在漂移，拿它当排序键会让两张体量
    # 相近的表在两次生成之间互换行序，diff 看起来像"内容改了"，其实只是估算值动了。
    size_note = "\n".join(
        "--   %-34s %-12s %s" % (t, db, magnitude(n))
        for t, db, n in sorted(sizes, key=lambda x: (x[0], x[1])))


    drops = []
    for t, keep_mod, drop_mod, key in UNION_TABLES:
        drop_db = dbs[drop_mod]
        drops.append(
            "SELECT '{t}' AS `表`, '{dd}' AS `丢弃侧来源`,\n"
            "       (SELECT COUNT(*) FROM `{dd}`.`{t}` d\n"
            "         WHERE NOT EXISTS (SELECT 1 FROM `{tgt}`.`{t}` k\n"
            "                            WHERE k.`{key}` = d.`{key}`)) AS `仍未进目标库的行数`".format(
                t=t, dd=drop_db, tgt=TARGET_DB, key=key))

    static = """-- ---------------------------------------------------------------------------
-- §1 覆盖率：源库里每张表都必须在目标库里出现
--
-- 注意：被排除的**副本**与保留份**同名**，所以它们在这里也会显示"已存在"
--       （存在的是保留份那份）——本查询查不出"该排除的没排除"，
--       那个由 §3 的行数相等来保证。本查询专抓"表整个没了"（dump 漏表）。
--
-- ⚠️ 但 §10.1 拍板**不搬**的那几张（{junk_list}）
--    是**有意**不在目标库里的——其中三张是**有数据的备份**（约 133 GB），不是空表。
--    不在这里排掉，本查询就会永远报红——
--    一个每次都喊红的校验会训练操作者忽略红色，比没有校验更坏（§3 的注释同此理）。
-- 期望：0 行
-- ---------------------------------------------------------------------------
SELECT s.TABLE_SCHEMA AS `缺失的源库`, s.TABLE_NAME AS `目标库缺少的表`,
       s.TABLE_ROWS AS `源库约行数`
  FROM information_schema.TABLES s
  LEFT JOIN information_schema.TABLES t
         ON t.TABLE_SCHEMA = @tgt AND t.TABLE_NAME = s.TABLE_NAME
 WHERE FIND_IN_SET(s.TABLE_SCHEMA, @src_dbs)
   AND s.TABLE_TYPE = 'BASE TABLE'
   AND CONCAT(s.TABLE_SCHEMA, '.', s.TABLE_NAME) NOT IN ({junk_in})
   AND t.TABLE_NAME IS NULL
 ORDER BY s.TABLE_SCHEMA, s.TABLE_NAME;

-- ---------------------------------------------------------------------------
-- §2 字符集残留：合库后不应再有任何非 utf8mb4 的表或列
--    （改写在 10_dump_legacy.sh 的 rewrite_charset，自测见 tests/）
-- 期望：两条查询各 0 行
-- ---------------------------------------------------------------------------
SELECT TABLE_NAME AS `非 utf8mb4 的表`, TABLE_COLLATION AS `排序规则`
  FROM information_schema.TABLES
 WHERE TABLE_SCHEMA = @tgt AND TABLE_TYPE = 'BASE TABLE'
   AND (TABLE_COLLATION IS NULL OR TABLE_COLLATION NOT LIKE 'utf8mb4%')
 ORDER BY TABLE_NAME;

SELECT TABLE_NAME AS `含非 utf8mb4 列的表`, COLUMN_NAME AS `列`,
       CHARACTER_SET_NAME AS `列字符集`, COLLATION_NAME AS `列排序规则`
  FROM information_schema.COLUMNS
 WHERE TABLE_SCHEMA = @tgt
   AND CHARACTER_SET_NAME IS NOT NULL
   AND CHARACTER_SET_NAME NOT LIKE 'utf8mb4%'
 ORDER BY TABLE_NAME, ORDINAL_POSITION;"""

    # `static` 是当作**值**传给外层 .format() 的，里面再写 {placeholder} 不会被替换，
    # 所以这里显式替换。（不用 .format() 是刻意的：SQL 里 % 和反引号很多，
    # 万一以后谁加个花括号，.format 会直接抛，replace 不会。）
    junk = sorted(DO_NOT_MIGRATE)
    static = static.replace(
        '{junk_in}',
        ",".join("'%s'" % x for x in junk) or "''"
    ).replace(
        '{junk_list}',
        "、".join("`%s`" % x for x in junk) or "（无）"
    )

    sql = """-- 合库后校验（在**目标库所在实例**上执行；要求六个源库仍同实例在侧）
--
-- 由 sql/tools/gen_migration_sql.py 从 check_subset.py 的 CHECKS 生成 —— 不要手改
-- 判定依据见 ../merge-conflict-plan.md
--
-- 🚫 **只对"全量迁移"的库有意义，别在抽样镜像（SAMPLE=N）上跑。**
--    §3/§4 断言的是"目标行数 == 源库行数"；抽样镜像的行数本来就远小于源库，
--    跑上去会**必然全红**——而一个每次都报红的校验，和一个永远不报红的一样没用
--    （AGENTS.md §1.1 推论 5）。抽样镜像请跑 97_verify_sample.sql。
--
-- 跑法：
--     mysql -h$TARGET_HOST -u$TARGET_USER -p < 99_verify.sql
--
-- 期望：§1 零行、§2 两个查询零行、§3 全部 OK、§4 全部 0。
-- 任何一条不满足 → **先别删源库**（源库是回滚路径）。
--
-- ⚠️ §3/§4 会 COUNT(*) 十余张表。它们的**实测体量**（生成时现读）：
--
{size_note}
--
-- 小表瞬间返回；`park_in_out_record`（约 9700 万行）那两条 COUNT(*) 要数分钟——
-- 慢是正常的，别以为卡死。真正的大表（zt_park 的 16 亿行那些）不在这里。

SET @tgt := '{tgt}';
SET @src_dbs := '{src_dbs}';

{static}

-- ---------------------------------------------------------------------------
-- §3 重名表：目标库行数必须**等于**保留侧行数
--    多于 → 把该丢的副本也搬进来了；少于 → 保留份没搬全
--    并集表不在此列（它们的目标行数天生大于保留侧，由 §4 负责）
-- 期望：全部 OK
-- ---------------------------------------------------------------------------
{rows};

-- ---------------------------------------------------------------------------
-- §4 并集：丢弃侧还有多少行没进目标库
--    NOT EXISTS 的键与 20_union_lost_rows.sql 用的是同一个业务键
-- 期望：全部 0
-- ---------------------------------------------------------------------------
{drops};

-- ---------------------------------------------------------------------------
-- §5 目标库默认字符集
-- 期望：utf8mb4 / utf8mb4_general_ci
-- ---------------------------------------------------------------------------
SELECT DEFAULT_CHARACTER_SET_NAME AS `默认字符集`, DEFAULT_COLLATION_NAME AS `默认排序规则`
  FROM information_schema.SCHEMATA WHERE SCHEMA_NAME = @tgt;
""".format(tgt=TARGET_DB,
           src_dbs=",".join(dbs[m] for m in ("SYS", "PARK", "PAY", "ORDER", "BILL", "CHARGE")
                            if m in dbs),
           size_note=size_note,
           static=static,
           rows="\nUNION ALL\n".join(rows),
           drops="\nUNION ALL\n".join(drops))

    path = os.path.join(OUT_DIR, "99_verify.sql")
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(sql)
    return path


def emit_reconcile(cur, dbs):
    """生成 98_count_reconcile.sql：**逐表**行数对账——"零丢失"的正面证据。

    与 99_verify.sql 的分工（为什么不并在一个文件里）：

      · `99` §3 只数 17 张**重名表**的保留侧，秒级，适合反复跑
      · 本文件数**所有**被搬的表（709 张减去不搬的副本）。非重名表（如 zt_sys 的
        38 张）**只有这里才被数到行数**——`99` §1 只查"表在不在"，不查行数。
        彩排正是靠逐表 COUNT(*) 才看出 `sys_op_log` 差 2 行
      · 代价：709 张表各两条 COUNT(*)，zt_park 那 16 亿行要让服务端做全索引扫描，
        总时长为**小时级**。所以拆成独立文件、并按库分段（见文件头），
        别把它塞进要反复跑的 99

    ## 水位线法：把"少 N 行"从"可能是 churn"变成"已核对"

    对**有 `auto_increment` 主键 `id`** 的追加型表，"少"的部分是不是 dump 之后的
    在线新增，可以精确判定：

        差值 == 源库中 id > 目标库 MAX(id) 的行数   ⇒ 少掉的全是水位线之后的新增，没丢

    彩排的 `sys_op_log` 就是这么判的（目标 id∈[187152,635584]、源 id∈[187152,635586]，
    `id > 635584` 恰好 2 行 == 差值 2 ⇒ 未丢）。**没有** auto_increment `id` 的表
    判不了，只能落到 ⚠️ 让人看——不假装它 OK。
    """
    # 不搬的表：与 exclude_tables.txt 同一判据（CHECKS 的副本 + §10.1 拍板不搬的
    # 废弃/备份表），别手抄。漏掉后一类，本文件会为那 7 张表各报一行"目标库没有它"的假缺口。
    # 顺带一个实打实的好处：那两张统计快照（3.86 亿 + 2.41 亿 = 6.27 亿行）不数了，
    # 而 zt_park 段总共才约 16 亿行——本文件最慢的一段因此少掉约四成开销。
    excluded = {(dbs[drop_mod], t) for t, _km, drop_mod, _k, _n in CHECKS}
    excluded |= junk_excluded()
    # 并集表跳过：目标行数 = 保留侧 + 补进的行，与"源侧"比必然不等（由 99 §4 负责）
    union_names = {t for t, _k, _d, _key in UNION_TABLES}

    sections = []
    src_dbs = [dbs[m] for m in ("SYS", "PARK", "PAY", "ORDER", "BILL", "CHARGE") if m in dbs]
    for db in src_dbs:
        cur.execute(
            "SELECT TABLE_NAME, TABLE_ROWS FROM information_schema.TABLES "
            "WHERE TABLE_SCHEMA=%s AND TABLE_TYPE='BASE TABLE' ORDER BY TABLE_NAME", (db,))
        tables = [(n, r or 0) for n, r in cur.fetchall()]

        # "这张表有没有 auto_increment 的 `id`" —— **按库一次问完**，别在下面的循环里
        # 逐表查（679 张 × 30ms/往返 ≈ 20s，改一次判定表就要重跑一次；
        # 与红线第 10 条"禁止在循环中执行数据库查询"同一个道理）。
        cur.execute(
            "SELECT TABLE_NAME FROM information_schema.COLUMNS "
            "WHERE TABLE_SCHEMA=%s AND COLUMN_NAME='id' "
            "AND EXTRA LIKE '%%auto_increment%%'", (db,))
        has_id_set = {n for (n,) in cur.fetchall()}

        rows, total = [], 0
        for t, approx in tables:
            if (db, t) in excluded or t in union_names:
                continue
            total += approx
            has_id = t in has_id_set

            if has_id:
                wm_sub = ("(SELECT COUNT(*) FROM `{db}`.`{t}` x "
                          "WHERE x.`id` > t.`目标最大id`)".format(db=db, t=t))
                wm_col = wm_sub
                ok_wm = "k.`源侧行数` - t.`目标行数` = " + wm_sub
                wm_note = "`id` 是 auto_increment，可判"
                target_sub = ("(SELECT COUNT(*) AS `目标行数`, MAX(`id`) AS `目标最大id`"
                              " FROM `{tgt}`.`{t}`)".format(tgt=TARGET_DB, t=t))
            else:
                # 没有 auto_increment `id` 就**别引用 `id`**——否则这里就是一句 SQL 报错，
                # 把整段对账打断。给个 NULL 列，让它落到 ⚠️ 由人判。
                wm_col = "NULL"
                ok_wm = "0 = 1"
                wm_note = "无 auto_increment `id`，判不了"
                target_sub = ("(SELECT COUNT(*) AS `目标行数`, NULL AS `目标最大id`"
                              " FROM `{tgt}`.`{t}`)".format(tgt=TARGET_DB, t=t))

            rows.append(
                "SELECT '{db}' AS `源库`, '{t}' AS `表`, k.`源侧行数`, t.`目标行数`,\n"
                "       t.`目标最大id`, {wm_col} AS `水位线之后新增`,\n"
                "       IF(t.`目标行数` = k.`源侧行数`, 'OK',\n"
                "          IF(t.`目标行数` > k.`源侧行数`,\n"
                "             CONCAT('🔴 多 ', t.`目标行数` - k.`源侧行数`, ' 行——只可能是重复导入'),\n"
                "          IF({ok_wm},\n"
                "             'OK（少掉的全是 dump 之后源库新增的行，水位线已核对）',\n"
                "             CONCAT('⚠️ 少 ', k.`源侧行数` - t.`目标行数`, ' 行，需人工核对（{wm_note}）')))) AS `结论`\n"
                "  FROM (SELECT COUNT(*) AS `源侧行数` FROM `{db}`.`{t}`) k\n"
                "  JOIN {target_sub} t\n".format(db=db, t=t, wm_col=wm_col, ok_wm=ok_wm,
                                                wm_note=wm_note, target_sub=target_sub))

        if not rows:
            continue
        heavy = ""
        if total > 50_000_000:
            heavy = ("--   ⚠️ 本段源侧总量 {m}（TABLE_ROWS 估算，且大型 InnoDB 表这个估算值\n"
                     "--      偏低）。这一段是**小时级**的，别以为卡死。\n").format(m=magnitude(total))
        sections.append(
            "-- @@DB {db}\n"
            "-- {db}：{n} 张表参与对账，源侧总量 {m}\n"
            "{heavy}"
            "{body};\n"
            "-- @@END {db}\n".format(db=db, n=len(rows), m=magnitude(total), heavy=heavy,
                                     body="\nUNION ALL\n".join(rows)))

    header = """-- 逐表行数对账（全表，重，但这是"零丢失"的正面证据）
--
-- 由 sql/tools/gen_migration_sql.py 从 check_subset.py 的 CHECKS + 源库元数据生成 —— 不要手改
--
-- 🚫 **只对"全量迁移"的库有意义，别在抽样镜像（SAMPLE=N）上跑。** 判据是
--    "目标行数 == 源库行数"，抽样镜像必然每一行都报差异 → 一份永远全红的对账表，
--    下一个人只会把它加进忽略列表（AGENTS.md §1.1 推论 5）。抽样镜像跑 97_verify_sample.sql。
--
-- 位置：合库的**最后一步**，在 10_dump_legacy.sh load 与 20_union_lost_rows.sql 之后。
--       在**目标库所在实例**上执行（要能同时看到源库与 parking_cloud）。
--
-- 跑法（可只跑一个库，别一上来就整跑 709 张表）：
--     # 先看有哪些段
--     grep '^-- @@DB' 98_count_reconcile.sql
--     # 只跑 zt_sys 那一段
--     sed -n '/^-- @@DB zt_sys$/,/^-- @@END zt_sys$/p' 98_count_reconcile.sql \\
--       | mysql -h$TARGET_HOST -u$TARGET_USER -p
--
-- 读法（**判据是不对称的**，理由见 99_verify.sql §3 的说明与 AGENTS.md §1.1）：
--   · OK                                  —— 相等
--   · OK（水位线已核对）                  —— 目标少了，但少掉的正好是 dump 之后源库新增的
--                                            行，逐行对得上水位线 ⇒ 没丢
--   · 🔴 多 N 行                          —— 目标比源还多，只可能是重复导入，一定要查
--   · ⚠️ 少 N 行，需人工核对              —— 没有 auto_increment `id` 判不了水位线；
--                                            源库在维护窗口内静默时，这一项应该消失
--
-- ⚠️ 若某张表整体没搬进来，本文件会在这张表上以 `1146 Table ... doesn't exist` 报错并停下
--    （mysql 客户端默认遇错即停——**别加 --force**，那会把错误藏起来）。
--    出现 1146 说明 `99_verify.sql` §1 本该先报；先去查 §1。
--
-- ⚠️ 每一段都自带 `-- @@DB <库>` / `-- @@END <库>` 标记，便于分段执行；别删标记。

"""
    path = os.path.join(OUT_DIR, "98_count_reconcile.sql")
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(header)
        fh.write("\n".join(sections))
    return path, sections


def emit_unions(cur, dbs):
    chunks = []
    for table, keep_mod, drop_mod, key in UNION_TABLES:
        keep_db, drop_db = dbs[keep_mod], dbs[drop_mod]
        kcols = {}   # 列名 -> (extra, nullable)
        for name, extra, nullable in columns(cur, keep_db, table):
            kcols[name] = (extra, nullable)
        common = [n for n, _e, _n2 in columns(cur, drop_db, table) if n in kcols]
        # 剔除两侧任一侧是 auto_increment 的列（就是 id）——让目标库重新分配
        ins = [c for c in common
               if "auto_increment" not in kcols[c][0].lower()]
        if key not in ins:
            raise SystemExit("%s：业务键 %s 不在可插入列里" % (table, key))

        collist = ", ".join("`%s`" % c for c in ins)
        select_cols = ", ".join("d.`%s`" % c for c in ins)
        chunks.append("""
-- ---------------------------------------------------------------------------
-- %s：把只存在于 %s.%s 的行补进 %s.%s
--   判重键 %s（两侧 id 空间不同，不能按 id 判重，也不能带 id 插入）
--   可插入列 %d 个（= 两侧交集 − auto_increment 的 id）
-- ---------------------------------------------------------------------------
INSERT INTO `%s`.`%s` (%s)
SELECT %s
FROM `%s`.`%s` d
WHERE NOT EXISTS (
        SELECT 1 FROM `%s`.`%s` k WHERE k.`%s` = d.`%s`
      );""" % (table, drop_db, table, TARGET_DB, table, key, len(ins),
               TARGET_DB, table, collist, select_cols, drop_db, table,
               TARGET_DB, table, key, key))

    path = os.path.join(OUT_DIR, "20_union_lost_rows.sql")
    with open(path, "w", encoding="utf-8") as fh:
        fh.write("""-- 补进"只存在于副本"的行（六库合一的唯一数据搬运动作）
--
-- 由 sql/tools/gen_migration_sql.py 生成 —— 列清单现算，不要手改本文件
-- 判定依据见 ../merge-conflict-plan.md §2.2 / §2.3 / §5
--
-- 🚫 **抽样镜像（SAMPLE=N）不要跑本文件。** 它的前提是"保留侧已被完整灌入"——
--    抽样镜像里保留侧只抽了几百行，此时往丢弃侧取"目标库还没有的行"，取到的会是
--    几乎所有行（几亿行里的绝大部分），既没有意义也可能把开发库撑爆。
--
-- 前提：本文件在**目标库所在实例**上执行，且六个源库还在同一实例上
--       （六库同实例已实测，见 ../legacy-db-actual.md §1）
-- 顺序：先按 10_dump_legacy.sh 把保留侧灌进 parking_cloud，再执行本文件
--
-- ⚠️ **执行时机（别当成一个可以随便穿插的步骤）**
--
--   本文件的判据是"**丢弃侧**里目标库还没有的行"（NOT EXISTS）。也就是说：
--   本文件跑完之后，谁再往丢弃侧写一行，那行就永远不会进目标库——
--   而这正是它要搬的那 29 行客户数据同一类的漏。
--
--   所以正确顺序是：
--       停写/切流 → ③ load → ④ 本文件 → ⑤ 99 校验（+ 98 全表对账）→ 放流量到新库
--   不是"搬完随手跑一下"。绝不能在写流量还指向老库副本时先跑一遍了事。
--
-- ⚠️ td_b_wx_config 含微信支付密钥。本文件只有列名、没有任何值——
--    密钥值由 INSERT ... SELECT 在库间直接流动，不落盘。
--    请勿"为了方便"改成带值的 INSERT。
""")
        fh.write("\n".join(chunks))
        fh.write("\n")
    return path


# ---------------------------------------------------------------------------
# 抽样开发库（`10_dump_legacy.sh` 的 SAMPLE=N）
#
# 全量迁移是 747 GB / 2~3 天的活；开发不需要那个精度，只需要"每张表有几百行、
# 结构一模一样"。抽样镜像的**正确性主张变了**，所以校验的判据也必须跟着变：
#
#   全量迁移主张：源库有多少行，目标库就有多少行      → 98/99 §3§4
#   抽样镜像主张：每张表都在、列结构一致、每张表都有样例行 → 97
#
# 把 98/99 套在抽样镜像上跑，会得到一个**永远全红**的校验——而永远报红的校验和
# 永远不报红的一样没用（AGENTS.md §1.1 推论 5），它只会训练人忽略红色。
# ---------------------------------------------------------------------------

# 含密钥/口令的**列名**判据。这个正则控制的是"哪些表只搬结构不搬数据"，
# 所以两个方向的代价**极不对称**，取判据时一律**从严**：
#   · 多判进来一张 → 开发库里少一张表的样例行。代价低，而且看得见。
#   · 漏判一张     → 生产的密钥/口令被写进开发机的 dump 文件。代价高，而且看不见。
# 判据**只读 information_schema 的列名**，一行数据都不读——所以本脚本永远
# 不会把密钥读进内存，更不会写进任何产物（同 §"绝不让密钥落盘"）。
SECRET_COLUMN_RE = (
    r"secret|passwd|password|_aes|aes_|private_key|pulic_key|pub_key"
    r"|apikey|api_key|sdk_key|sign_key|access_key|accesskey"
    r"|id_card|idcard|bank_card|bankcard|credential|^token$|_token$")


# ---------------------------------------------------------------------------
# 列类型归一化：**MySQL 5.7 与 8 对整型的 COLUMN_TYPE 写法不一样**
#
#   5.7（源库）：`int(11)` / `bigint(32)` —— 带"显示宽度"
#   8.0（目标库）：`int` / `bigint` —— 8.0.19 起显示宽度被废弃，也不再写进 COLUMN_TYPE
#
# 显示宽度**不影响存储、不影响取值**（官方文档：deprecated, no effect），
# 但它会出现在 COLUMN_TYPE 字符串里。不归一化的话，§2 会把**每一个整型列**都报成
# "类型变了" —— 第一次在本地跑就报了上万行。一份每次都报差异的守卫等于没有守卫
# （AGENTS.md §1.1 推论 5），所以 §2 比的是**归一化后**的类型。
#
# `tinyint(1)` 在 8.0 里被特意保留（BOOLEAN 的别名），但两边**都**归一化成
# `tinyint`，所以不必为它开例外——少一个特例就少一处能漂的地方。
#
# 规则只写一份（INT_TYPES），Python 侧与生成到 SQL 里的表达式都从它来，
# 避免"两边各写一遍正则、改了一边"这种漂法（推论 7：按字符串比对的判据要防手滑）。
# ---------------------------------------------------------------------------
INT_TYPES = ("tinyint", "smallint", "mediumint", "int", "integer", "bigint")

COLTYPE_NORM_RE = r"^(%s)\([0-9]+\)" % "|".join(INT_TYPES)

# SQL 里的同一个正则：MySQL 字符串字面量中 `\\` 才是一个反斜杠，
# 即正则引擎最终收到的是 `\(`——**不能**只写一个反斜杠，那样 `(` 会被当成分组。
COLTYPE_NORM_SQL = r"^(%s)\\([0-9]+\\)" % "|".join(INT_TYPES)


def norm_coltype(ctype):
    """Python 侧归一化。与 SQL 侧的 `sql_norm_coltype` 必须同一条规则——
    两者一旦漂开，§2 要么静默放行、要么永远报红，两种都是坏的。"""
    return re.sub(COLTYPE_NORM_RE, r"\1", ctype)


def sql_norm_coltype(expr):
    """SQL 侧归一化表达式（`REGEXP_REPLACE` 是 MySQL 8 才有的；目标库就是 MySQL 8）。

        bigint(32) unsigned → bigint unsigned
        int(11)            → int
        varchar(64)        → varchar(64)   （不动）
        decimal(10,2)      → decimal(10,2) （不动）
        enum('0','1')      → enum('0','1') （不动）
    """
    return "REGEXP_REPLACE(%s, '%s', '$1')" % (expr, COLTYPE_NORM_SQL)


def kept_tables(cur, dbs):
    """`[(源库, 表名)]` —— 六库合一后**应该出现在目标库**的物理表。

    判据与 `exclude_tables.txt` 同源（六库全部物理表 − CHECKS 丢弃的副本
    − DO_NOT_MIGRATE 拍板不搬的），所以不会出现"排除清单排了、这里还留着"。
    """
    excluded = {(dbs[drop_mod], t) for t, _km, drop_mod, _k, _n in CHECKS}
    excluded |= {tuple(x.split('.', 1)) for x in DO_NOT_MIGRATE}
    out = []
    for db in sorted(set(dbs.values())):
        cur.execute(
            "SELECT TABLE_NAME FROM information_schema.TABLES "
            "WHERE TABLE_SCHEMA=%s AND TABLE_TYPE='BASE TABLE'", (db,))
        for (t,) in cur.fetchall():
            if (db, t) not in excluded:
                out.append((db, t))
    return sorted(out)


def detect_struct_only(cur, dbs):
    """`{(源库, 表名)}` —— 含密钥/口令，抽样时**只搬结构不搬数据**。

    `LOWER(COLUMN_NAME) REGEXP <SECRET_COLUMN_RE>`，写在 SQL 里而不是 Python 里，
    是为了让判据本身可以被 `SHOW` 出来复现——手抄一份"敏感表清单"就是第二个事实来源。
    """
    dbl = sorted(set(dbs.values()))
    cur.execute(
        "SELECT DISTINCT TABLE_SCHEMA, TABLE_NAME FROM information_schema.COLUMNS "
        "WHERE TABLE_SCHEMA IN (%s) AND LOWER(COLUMN_NAME) REGEXP %%s"
        % ",".join(["%s"] * len(dbl)),
        tuple(dbl) + (SECRET_COLUMN_RE,))
    return {(s, t) for s, t in cur.fetchall()}


def source_empty(cur, dbs, tables):
    """`{(源库, 表名)}` —— 在**源库里就一行都没有**的表（精确判定）。

    为什么不能只看 `information_schema.TABLE_ROWS`：那是**估算**（AGENTS.md §1.1
    推论 3 明确禁过"把估算当实测"）。所以每张表都要发一条**真查询**。

    ⚠️ 但**别**写成"每张表一条 `SELECT ... LIMIT 1`"：源库在张北，一次往返约 30ms，
    679 张就是 20 秒，而这个生成器改一次判定表就要重跑一次
    （与红线第 10 条"禁止在循环中执行数据库查询"同一个道理）。
    改成**一条** `UNION ALL` 把 679 个 `EXISTS(...)` 问完，实测 21s → 不到 1s。

    顺带把"估算当候选筛子"那一步**整个去掉**了：既然现在一次问完所有表，
    就不必先用 `TABLE_ROWS` 缩范围。那一步本身还带个隐患——估算会滞后
    （刚被清空的表可能仍报非 0），漏掉一张真空表就会被当成 kind=1，
    进而在 §4 报一条**假红**。现在没有这个筛子，也就没有这个隐患。

    这个集合的作用：抽样镜像里"空表"是允许的，但只允许**源库本来就空**的那些。
    不把它们挑出来，§4 就会为它们**永远报红**（推论 5）。
    """
    if not tables:
        return set()
    exprs = " UNION ALL ".join(
        "SELECT %s AS db, %s AS tbl, EXISTS(SELECT 1 FROM `%s`.`%s`) AS has_row"
        % (sql_lit(db), sql_lit(t), db, t) for db, t in sorted(tables))
    cur.execute("SELECT db, tbl FROM (%s) x WHERE x.has_row = 0" % exprs)
    return {(db, t) for db, t in cur.fetchall()}


def emit_struct_only(struct_only):
    """产物 5：`struct_only_tables.txt`（格式同 `exclude_tables.txt`，喂 `--ignore-table`）。

    样例只会列出**表名**，不会出现列名之外的任何东西——产物里不含一行数据。
    """
    path = os.path.join(OUT_DIR, "struct_only_tables.txt")
    with open(path, "w", encoding="utf-8") as fh:
        fh.write("# 抽样镜像里**只搬结构、不搬数据**的表\n")
        fh.write("# 由 sql/tools/gen_migration_sql.py 生成，**不要手改**\n")
        fh.write("#   判据 = 表里存在列名匹配 %s\n" % SECRET_COLUMN_RE)
        fh.write("#   为什么：这些表装的是支付/短信/云账号的密钥与口令，\n")
        fh.write("#   抽样会把生产的真实凭据写进**开发机的 dump 文件**。列一行都不搬。\n")
        fh.write("#   判据从严：多判进来只是少几张表的样例行，漏判是密钥落盘。\n")
        fh.write("# 用途：10_dump_legacy.sh 的 SAMPLE= 模式下，数据那一遍 --ignore-table\n")
        fh.write("# 行数=%d\n" % len(struct_only))
        for db, t in sorted(struct_only):
            fh.write("%s.%s\n" % (db, t))
    return path


def source_columns(cur, tables):
    """`[(源库, 表名, 列名, 列序, 列类型)]` —— 生成时把源库的列结构**冻结**下来。

    ⚠️ 为什么要在生成时冻结、而不是让校验 SQL 当场去 information_schema 读源库：
       抽样镜像跑在**开发机**上，那台机器上**没有** `zt_park`/`zt_bill` 这些源库。
       若校验 SQL 直接 join 源 schema，每一列都会报"目标库没有这一列"——
       又是一份**永远全红**的校验（AGENTS.md §1.1 推论 5）。
       冻成 `_cols` 之后，97 是一份**自包含**的校验：目标库上有 mysql 就能跑。
       代价：源库改结构后要重跑本生成器，否则快照过期——那是**要报出来的信号**，
       不是假阳性（AGENTS.md 推论 3：生成物要么确定，要么标明它为什么会漂）。

    `COLUMN_TYPE` 不含字符集（`varchar(64)` 就是 `varchar(64)`），正合用：
    字符集是生成时**故意**改的，比它会把每列都报出来。
    **存进去的是归一化后的类型**（见 `norm_coltype`）——5.7 的 `int(11)` 与
    8.0 的 `int` 是同一个东西，不是同一串字符。

    性能：**按库一次查完**（`TABLE_NAME IN (...)`），别写成"每张表一条 SELECT"。
    源库在张北、生成器在本地，一次往返约 30ms——679 条就是 20 秒，
    而这个生成器是要被反复重跑的（改了判定表就要重跑）。
    这和项目红线第 10 条"禁止在循环中执行数据库查询"是同一个道理，
    只是那条管的是业务代码、这里管的是工具代码。
    """
    by_db = {}
    for db, t in tables:
        by_db.setdefault(db, []).append(t)

    out = []
    for db in sorted(by_db):
        names = sorted(by_db[db])
        qmarks = ",".join(["%s"] * len(names))
        cur.execute(
            "SELECT TABLE_NAME, COLUMN_NAME, ORDINAL_POSITION, COLUMN_TYPE "
            "FROM information_schema.COLUMNS "
            "WHERE TABLE_SCHEMA=%s AND TABLE_NAME IN (%s) "
            "ORDER BY TABLE_NAME, ORDINAL_POSITION" % (("%s", qmarks)),
            tuple([db] + names))
        for t, name, pos, ctype in cur.fetchall():
            out.append((db, t, name, pos, norm_coltype(ctype)))
    # ⚠️ 调用方按 (tbl, col) 建主键、生成的文件也按这个顺序——
    #    查询结果已经是 (db,tbl,pos) 有序，这里再显式排一次，别依赖隐式顺序。
    out.sort(key=lambda r: (r[0], r[1], r[3]))
    return out


def sql_lit(s):
    """MySQL 字符串字面量。**必须转义单引号**：`COLUMN_TYPE` 里就有，
    例如 `enum('0','1')` / `set('a','b')`——不转义会直接把 INSERT 语句撕成两半。"""
    return "'" + str(s).replace("'", "''") + "'"


def emit_verify_sample(cur, dbs, struct_only):
    """产物 6：`97_verify_sample.sql` —— 抽样镜像的校验（判据见 §抽样开发库）。

    生成的是 `_kept` + `_cols` 两张临时表 + 五段断言，期望**全部 0 行**。
    `_kept` 里每张表带一个 kind：0=只搬结构 / 1=抽样 / 2=源库本来就空。
    `_cols` 是生成时冻结的源库列结构（见 `source_columns`）。
    """
    path = os.path.join(OUT_DIR, "97_verify_sample.sql")
    db_list = sorted(set(dbs.values()))
    kept = kept_tables(cur, dbs)
    so = struct_only & set(kept)
    sampled = [kt for kt in kept if kt not in so]
    empty = source_empty(cur, dbs, sampled)
    cols = source_columns(cur, kept)

    def kind(kt):
        if kt in so:
            return 0
        return 2 if kt in empty else 1

    with open(path, "w", encoding="utf-8") as fh:
        fh.write("""-- 抽样开发库校验（**不是**全量迁移校验，别跟 98 / 99 混着看）
--
-- 由 sql/tools/gen_migration_sql.py 从判定表 + 源库元数据生成 —— 不要手改
--
-- 用途：`10_dump_legacy.sh` 的 **SAMPLE=N** 模式灌完之后跑。
--
-- ⚠️ 它与 98_count_reconcile.sql / 99_verify.sql §3§4 的判据**正好相反**。
--    那两份断言"目标行数 == 源库行数"，是为**零丢失的全量迁移**写的；
--    抽样镜像的行数**本来就该**远小于源库，在那两份上必然全红。
--    两个主张不一样，就别拿同一把尺子量：
--        全量迁移主张 = 源库多少行，目标库就多少行
--        抽样镜像主张 = 每张表都在、列结构一致、每张表都落了样例行
--
-- 跑法（两个开关都别省）：
--     mysql --force -N -h$TARGET_HOST -u$TARGET_USER -p < 97_verify_sample.sql
--   · `--force` —— 一段出错**不要中止**。§4 那段逐表 COUNT(*) 是硬编码表名的
--     UNION ALL，撞到一张不存在的表就是 `ERROR 1146`，**不加 --force 时 mysql 会
--     从这里直接退出，§5 永远不会执行**（实测踩到：只看到 §1 的输出加一条 ERROR，
--     还以为是 §5 通过了）。
--   · `-N` —— 不印列名。要的是"干净时**完全没有输出**"这个判决；
--     印列名的话空结果集也会留一行表头，"空=通过"就不成立了。
-- 期望：§1~§5 **全部 0 行**。任何一段有输出 → 见该段的标题说明。
--
-- 反向用例（**证明这五段真会报红，不是空转**）见 tests/97_negative_test.sh。

SET @tgt := '%s';
-- ⚠️ 与 10_dump_legacy.sh 的 SAMPLE= 保持一致（只用于标出"某张表抽多了"）
SET @sample_n := 500;

-- ⚠️ **必须选一个默认库**：下面要 `DROP` / `CREATE TEMPORARY TABLE`，而这两条语句
--    都要有默认库，否则直接 `ERROR 1046 (3D000) No database selected`。
--    （其余表名全是 `parking_cloud`.`x` 全限定写法，这个 USE 只是给临时表一个落脚点，
--      所以跑法里**不用**再传库名参数——传了也不会错，但别以为它是必需的。）
USE `%s`;

-- _kept：六库合一后**应该有**的表。kind 决定这张表该有多少行。
--   0 = 只搬结构（表里有密钥/口令列，数据一律不搬）→ 必须 0 行
--   1 = 抽样                                       → 必须 >= 1 行且 <= @sample_n
--   2 = 源库本来就空（已逐张 LIMIT 1 实查确认）     → 必须 0 行
--
-- ⚠️ 这里比 98_count_reconcile.sql 多 **2 张**（`td_c_cust_person` / `td_b_wx_config`）。
--    那两张在**全量**迁移里要被"取并集"（20_union_lost_rows.sql 往保留侧补行），
--    98 因此跳过它们（目标行数 = 保留侧 + 补进的，跟源侧比必然不等）。
--    抽样镜像不做并集，它们就是普通表 → 必须在这里，别照着 98 的清单把这两张删掉。
DROP TEMPORARY TABLE IF EXISTS `_kept`;
CREATE TEMPORARY TABLE `_kept` (
  `db`   VARCHAR(64) NOT NULL,
  `tbl`  VARCHAR(64) NOT NULL,
  `kind` TINYINT     NOT NULL,
  PRIMARY KEY (`db`, `tbl`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
INSERT INTO `_kept` VALUES
""" % (TARGET_DB, TARGET_DB))

        rows = ["('%s','%s',%d)" % (db, t, kind((db, t))) for db, t in kept]
        for i in range(0, len(rows), 200):
            fh.write(",".join(rows[i:i + 200]))
            fh.write(",\n" if i + 200 < len(rows) else ";\n")

        # _cols：**生成时冻结**的源库列结构。为什么不直接 join 源库 schema，见 §2 的说明。
        fh.write("""
-- _cols：源库列结构的**生成时快照**（不是现在去读源库，理由见 §2）。
--   列序 pos 一并带上——只比列名+类型会漏掉"列序被打乱"，而列序也是结构的一部分。
DROP TEMPORARY TABLE IF EXISTS `_cols`;
CREATE TEMPORARY TABLE `_cols` (
  `db`   VARCHAR(64)  NOT NULL,
  `tbl`  VARCHAR(64)  NOT NULL,
  `col`  VARCHAR(64)  NOT NULL,
  `pos`  SMALLINT     NOT NULL,
  `type` VARCHAR(255) NOT NULL,
  PRIMARY KEY (`tbl`, `col`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
INSERT INTO `_cols` VALUES
""")

        # ⚠️ 必须转义单引号：COLUMN_TYPE 里就有（`enum('0','1')` / `set('a','b')`），
        #    不转义会把这批 INSERT 直接撕成两半。
        crows = ["(%s,%s,%s,%d,%s)" % (sql_lit(db), sql_lit(t), sql_lit(c),
                                       p, sql_lit(ty)) for db, t, c, p, ty in cols]
        for i in range(0, len(crows), 100):
            chunk = crows[i:i + 100]
            fh.write(",".join(chunk))
            fh.write(",\n" if i + 100 < len(crows) else ";\n")

        # `%NORM%` 是目标侧的归一化列类型表达式；`%NORMDESC%` 是它用的正则（写进注释）。
        # 用 replace 而不是 %-format / .format：SQL 里 % 和反引号很多，
        # 与本文件其它地方（见 emit_verify 的 {junk_in}）保持同一种做法。
        _norm = sql_norm_coltype("t.COLUMN_TYPE")
        fh.write("""
-- ---------------------------------------------------------------------------
-- §1 表都在：`_kept` 里每张表都要在目标库出现
-- ---------------------------------------------------------------------------
SELECT k.`db` AS `源库`, k.`tbl` AS `表` FROM `_kept` k
  LEFT JOIN information_schema.TABLES t
    ON t.TABLE_SCHEMA=@tgt AND t.TABLE_NAME=k.`tbl` AND t.TABLE_TYPE='BASE TABLE'
 WHERE t.TABLE_NAME IS NULL
 ORDER BY 1, 2;

-- ---------------------------------------------------------------------------
-- §2 列结构一致：列名 + 列类型 + 列序，对的是**生成时冻结的源库快照**（`_cols`）
--    ⚠️ 这里**不读源库 schema**：抽样镜像跑在开发机上，那台机器**没有** zt_park 等源库，
--    join 过去会每列都报"缺列"——一份永远全红的守卫等于没有守卫（AGENTS.md §1.1 推论 5）。
--    所以源库的列结构在 gen_migration_sql.py 里冻结成了 `_cols`。
--    `_cols` 过期（源库改过结构但没重跑生成器）时 §2 会报差异——那是**要的信号**，不是假阳性。
--
--    ⚠️ 也**不比字符集/排序规则**——源库是 utf8/utf8_bin/latin1，目标是 utf8mb4，
--    字符集是生成时**故意**改的（见 10_dump_legacy.sh 的 rewrite_charset）。
--    比了会把每一列都报出来。COLUMN_TYPE 不含字符集（`varchar(64)` 就是 `varchar(64)`），正合用。
--
--    ⚠️ **整型的显示宽度也要先归掉**：源库 5.7 报 `int(11)`，目标 8.0 报 `int`
--    （8.0.19 起显示宽度被废弃）。不归的话**每一个整型列都会报差异**——
--    本地第一次跑就报了上万行，正是"永远全红的守卫"。两边用同一条规则：
--        %NORMDESC%
--    （规则定义只写一份，见 gen_migration_sql.py 的 INT_TYPES / sql_norm_coltype。）
-- ---------------------------------------------------------------------------
-- §2a 源库有、目标库没有（或类型/列序变了）的列
SELECT c.`db` AS `源库`, c.`tbl` AS `表`, c.`col` AS `列`, c.`pos` AS `源库列序`,
       c.`type` AS `源类型`,
       IFNULL(%NORM%, '（目标库没有这一列）') AS `目标类型（已归一化）`,
       IFNULL(t.ORDINAL_POSITION, 0) AS `目标列序`
  FROM `_cols` c
  LEFT JOIN information_schema.COLUMNS t
    ON t.TABLE_SCHEMA=@tgt AND t.TABLE_NAME=c.`tbl` AND t.COLUMN_NAME=c.`col`
 WHERE t.COLUMN_NAME IS NULL
    OR %NORM% <> c.`type`
    OR t.ORDINAL_POSITION <> c.`pos`
 ORDER BY 1, 2, 4;

-- §2b 目标库多出来的列（源库没有）
SELECT t.TABLE_NAME AS `表`, t.COLUMN_NAME AS `多出来的列`,
       t.COLUMN_TYPE AS `类型`, t.ORDINAL_POSITION AS `列序`
  FROM information_schema.COLUMNS t
  JOIN `_kept` k ON k.`tbl` = t.TABLE_NAME
  LEFT JOIN `_cols` c
    ON c.`tbl` = t.TABLE_NAME AND c.`col` = t.COLUMN_NAME
 WHERE t.TABLE_SCHEMA=@tgt AND c.`col` IS NULL
 ORDER BY 1, 4;

-- ---------------------------------------------------------------------------
-- §3/§4 行数：结构表必须 0 行、源库本空的必须 0 行、抽样表必须 >= 1 行
--    这是该文件最值钱的一段：dump **静默漏掉某张表**时，§1/§2 全绿，
--    只有这里会响。
-- ---------------------------------------------------------------------------
SELECT * FROM (
  SELECT k.`db` AS `源库`, k.`tbl` AS `表`,
         CASE k.`kind` WHEN 0 THEN '只搬结构' WHEN 2 THEN '源库本来就空' ELSE '抽样' END AS `类型`,
         x.`行数`,
         CASE WHEN k.`kind` IN (0, 2) AND x.`行数` <> 0 THEN '🔴 不该有数据'
              WHEN k.`kind` = 1 AND x.`行数` = 0       THEN '🔴 抽样没落地'
              WHEN k.`kind` = 1 AND x.`行数` > @sample_n THEN '⚠️ 超过抽样条数'
              ELSE 'OK' END AS `结论`
    FROM `_kept` k
    JOIN (
""".replace("%NORM%", _norm).replace("%NORMDESC%", COLTYPE_NORM_RE))
        # 目标库逐表 COUNT(*)：与 98_count_reconcile.sql 同一套写法，
        # 不用 PREPARE + GROUP_CONCAT —— 那条路的 group_concat_max_len 默认只有
        # 1024 字节，677 张表拼出来的 SQL 会被**静默截断**，得到一个看不出错的错。
        exprs = ["      SELECT '%s' AS `表`, COUNT(*) AS `行数` FROM `%s`.`%s`"
                 % (t, TARGET_DB, t) for _db, t in kept]
        fh.write("\n    UNION ALL\n".join(exprs))
        fh.write("""
    ) x ON x.`表` = k.`tbl`
) z WHERE z.`结论` <> 'OK'
 ORDER BY z.`源库`, z.`表`;

-- ---------------------------------------------------------------------------
-- §5 目标库字符集：全部 utf8mb4（表级 + 列级）
-- ---------------------------------------------------------------------------
SELECT TABLE_NAME AS `表`, '表级' AS `位置`, TABLE_COLLATION AS `实际`
  FROM information_schema.TABLES
 WHERE TABLE_SCHEMA=@tgt AND TABLE_TYPE='BASE TABLE' AND TABLE_COLLATION NOT LIKE 'utf8mb4%'
UNION ALL
SELECT TABLE_NAME, CONCAT('列 ', COLUMN_NAME), COLLATION_NAME
  FROM information_schema.COLUMNS
 WHERE TABLE_SCHEMA=@tgt AND COLLATION_NAME IS NOT NULL AND COLLATION_NAME NOT LIKE 'utf8mb4%'
 ORDER BY 1, 2;

DROP TEMPORARY TABLE IF EXISTS `_cols`;
DROP TEMPORARY TABLE IF EXISTS `_kept`;
""")
    return path, len(so), len(empty)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--env", default=os.path.join(
        HERE, "..", "..", "私密", "legacy-ro.env"))
    a = ap.parse_args()

    cfg = load_env(a.env)
    dbs = db_names(cfg)
    # 层级标签写错一个字不会抛错，只会让下面的分类计数静默变 0，
    # 而那个数字要写进 exclude_tables.txt 的头部——成品里的假数字最难查。
    unknown = sorted({t for t, _w in DO_NOT_MIGRATE.values()
                      if t not in DO_NOT_MIGRATE_TIERS})
    if unknown:
        print("DO_NOT_MIGRATE 里有未知层级：%s（只许 %s）"
              % ("、".join(unknown), "、".join(DO_NOT_MIGRATE_TIERS)), file=sys.stderr)
        return 2
    need = ({c[1] for c in CHECKS} | {c[2] for c in CHECKS}
            | {u[1] for u in UNION_TABLES} | {u[2] for u in UNION_TABLES})
    missing = [m for m in sorted(need) if m not in dbs]
    if missing:
        print("env 里缺模块：%s" % ", ".join(missing), file=sys.stderr)
        return 2

    if not os.path.isdir(OUT_DIR):
        os.makedirs(OUT_DIR)

    conn = connect(cfg, read_timeout=120)
    try:
        with conn.cursor() as cur:
            p1, uniq = emit_excludes(dbs)
            p2 = emit_unions(cur, dbs)
            p3 = emit_verify(cur, dbs)
            p4, sections = emit_reconcile(cur, dbs)
            # ---- 抽样开发库那一套（判据不同，见文件头）----
            struct_only = detect_struct_only(cur, dbs)
            p5 = emit_struct_only(struct_only)
            p6, n_so, n_empty = emit_verify_sample(cur, dbs, struct_only)
    finally:
        conn.close()

    print("已生成（全量迁移）：")
    print("  %s  （%d 张表不搬）" % (p1, len(uniq)))
    print("  %s  （%d 处并集）" % (p2, len(UNION_TABLES)))
    print("  %s  （合库后校验，快）" % p3)
    print("  %s  （逐表行数对账，%d 段 / 慢）" % (p4, len(sections)))
    print("已生成（抽样开发库，SAMPLE=N 用；判据与上面那份相反）：")
    print("  %s  （%d 张表只搬结构·含密钥口令）" % (p5, len(struct_only)))
    print("  %s  （抽样校验；另有 %d 张源库本来就空，已按「允许为空」入表）"
          % (p6, n_empty))
    return 0


if __name__ == "__main__":
    sys.exit(main())