#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
重名表「丢弃份是不是保留份的子集」精确校验 —— 六库合一的最后一道数据闸门。

## 为什么需要它

[`merge-conflict-plan.md`](../merge-conflict-plan.md) §0 列的**必须逐一确认的例外**有两条，
其中第 2 条是：

> 两侧行数都非 0 且差异大时，不能假定"主份已包含副本全部行"——
> 要抽样比对，确认没有只存在于副本的历史数据。

抽样比对只能给出"抽样命中"这种**概率性**结论（§2.1 里 `park_in_out_record`
那次是抽 300 个回查、命中 300/300）。本工具把这件事做成**精确**的：

    所有六个库都在**同一台实例**上（见 [`legacy-db-actual.md`](../legacy-db-actual.md) §1），
    所以可以直接发跨库 `LEFT JOIN`，一次问清"丢弃份里有多少行的主键在保留份里**找不到**"。

那个数就是**合库后会真正丢掉的行数**。0 ⇒ 这份可以安全丢弃；非 0 ⇒ 必须人工处置
（取并集 / 查清调用方 / 单独导出）。

## ⚠️ 只读且只看键列

本脚本**只发 SELECT**，且只 SELECT **主键列**（外加显式配置的键列）——
绝不 `SELECT *`。这一条是刻意的：清单里有些表（如 `td_b_wx_config`）
装着微信支付配置，通配查询会把密钥打进终端和日志。只投影键列，
就**不可能**有任何敏感列值出现在输出里。

## ⚠️ 键选错 = 假警报（本脚本踩过的坑）

`--key` 用来把判定从「主键」换成「业务键」，**副本换过 id 空间时，只有业务键才说真话**。
但反过来也成立：**业务键选错会造出成片假独有**。已修掉的两例：

- `t_vip_card_cust_info`：按主键 `id` 比 → 27 行"独有"；按 `card_no` 比 → **0**。
  同一批卡换了 id，不是丢了数据
- `td_b_wx_config`：按 `org_id` 比 → 18 行"独有"；按 `appid` 比 → **3**。
  老库的 `org_id` **大量为 NULL**，而 `NULL = NULL` 在 SQL 里不成立，
  于是 join 必然落空 —— 那 18 行是**连接失败**，不是数据缺失。
  这张表的真实查询键是 `appid`：`WxPayApiConfigKit` 就是按 appId 找配置的
  （找不到直接抛 `IllegalStateException`）

> 规律：**键列里只要可能有 NULL，用它做 join 就一定会误报**。
> 选业务键前先问一句「这个列在老数据里是否普遍为空」。

用法：
    python3 -I sql/tools/check_subset.py --env 私密/legacy-ro.env
    python3 -I sql/tools/check_subset.py --env 私密/legacy-ro.env --only td_c_cust_person
    python3 -I sql/tools/check_subset.py --env 私密/legacy-ro.env --sample 20

退出码：0 = 所有待丢弃份都已被证明是子集（或已被人工判定）；1 = 仍有需人工处置的。
"""

import argparse
import os
import sys

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

# ---------------------------------------------------------------------------
# 待校验清单：唯一的事实来源
#
# 每项 = (表, 保留侧模块, 丢弃侧模块, 显式键列 or None 表示自动取主键, 备注)
# 模块名对应 env 文件里的 LEGACY_DB_<模块>；注意历史填写里
#   PAY → zt_order（pay 域活库 + 2018 订单归档）
#   ORDER → zt_order_n（order 域活库）
# —— 库名与业务域是错位的，所以这里必须写「模块」而不是直接写库名，
#    由 db_names() 解析，避免把错位的名字抄错第二次。
#
# ⚠️ 第 4 列**不要留 None 就完事**：副本换过 id 空间时主键会满屏假独有。
#    已固化三处业务键（见文件头「键选错 = 假警报」）：
#      t_vip_card_cust_info → card_no      sys_org → org_code
#      td_b_wx_config       → appid        （org_id 为 NULL 会造出 18 行假独有）
# ---------------------------------------------------------------------------
CHECKS = [
    # ---- §2.1 「主份是超集且活跃」5 组：本工具要给每份副本一个精确的"可丢"证明 ----
    ("parking_lot", "PARK", "BILL", "pl_no", "zt_bill 1603 行（无更新）"),
    ("parking_lot", "PARK", "PAY", "pl_no", "zt_order 1465 行（无更新）"),
    ("parking_lot", "PARK", "ORDER", "pl_no", "zt_order_n 1979 行（无更新）"),
    ("park_in_out_record", "PARK", "PAY", "order_id",
     "§2.1 已抽 300/300 命中，此处改做精确全量"),
    ("t_vip_card_cust_info", "BILL", "PAY", "card_no",
     "zt_order 1892 行陈旧；按 id 会假报 27 行，按 card_no 为 0"),
    ("tp_p_berths", "PARK", "ORDER", None,
     "zt_order_n 20554 行；按 id 有 31 行独有，但 1681 行同 id 已被改写"),
    ("tp_e_employee_charge", "PARK", "ORDER", None, "zt_order_n 940 行"),

    # ---- §2.2 🔴「动手前必须查调用方」4 组 ----
    ("sys_org", "SYS", "PARK", "org_code", "zt_park 20 行（残缺副本）"),
    ("sys_org", "SYS", "ORDER", "org_code", "zt_order_n 20 行（残缺副本）"),
    ("sys_org", "SYS", "PAY", "org_code", "zt_order 0 行"),
    ("td_c_cust_person", "PAY", "ORDER", "cust_id", "方向相反：活的在 zt_order（超集）"),
    ("td_c_coupon_person", "PAY", "ORDER", None, "zt_order_n 5720 行待定性"),
    ("td_b_settlement_record", "ORDER", "PAY", None, "§2.2：zt_order 90 行建议并集"),

    # ---- §1 「结构完全一致」7 组：结构同，但数据来源仍要定 ----
    ("income_temp", "ORDER", "PARK", None, "353/353 行，两侧同数"),
    ("month_dchz", "ORDER", "PARK", "time,pl_no", "199/199 行，两侧同数；两侧都无主键"),
    ("table_15_free", "ORDER", "PARK", "dateStr,pl_no", "344/342 行，差 2；两侧都无主键"),
    ("td_b_settlement", "ORDER", "PAY", None, "两侧皆空"),
    ("td_b_settlement_org", "ORDER", "PAY", None, "zt_order 0 行"),
    ("td_b_settlement_serial", "ORDER", "PAY", None, "zt_order 0 行"),
    ("td_b_settlement_share_record", "ORDER", "PAY", None, "20/33 行，两侧都有数据"),
    ("td_p_parking_share_order_temp", "ORDER", "PAY", None, "临时表"),

    # ---- §2.3 🟡 无活跃方 3 组 ----
    ("tb_b_park_income_day", "ORDER", "PAY", None, "zt_order 1796 行待定性"),
    ("td_b_wx_config", "ORDER", "PAY", "appid",
     "两侧 27/22 行且都无更新时间；⚠️ 本表含支付配置，仅比对 appid，不比对内容"),
]


def pk_columns(cur, db, table):
    """主键列（按索引顺序）。没有主键返回 []。"""
    cur.execute(
        "SELECT COLUMN_NAME FROM information_schema.STATISTICS "
        "WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s AND INDEX_NAME='PRIMARY' "
        "ORDER BY SEQ_IN_INDEX", (db, table))
    return [r[0] for r in cur.fetchall()]


def one(cur, db, table):
    cur.execute("SELECT COUNT(*) FROM `%s`.`%s`" % (db, table))
    return cur.fetchone()[0]


def check(cur, table, keep_db, drop_db, key, sample):
    """返回 dict 结论。只投影键列，不做 SELECT *。"""
    # 键列：显式给的优先，否则自动取「丢弃侧」的主键（要丢的那份才是被探测的对象）
    if key:
        cols = [c.strip() for c in key.split(",")]
    else:
        cols = pk_columns(cur, drop_db, table)
    res = {"table": table, "keep_db": keep_db, "drop_db": drop_db,
           "key": cols, "keep_n": None, "drop_n": None, "drop_only": None,
           "samples": [], "note": ""}

    if not cols:
        res["note"] = "两侧都没有主键，且未配置键列 → 无法自动判定"
        return res

    keep_pk = pk_columns(cur, keep_db, table)
    missing_in_keep = [c for c in cols if c not in keep_pk]
    if missing_in_keep:
        # 键不是保留侧主键时的两点说明：
        #   ① 结论强度下降：只能说「这个键值在保留侧不存在」，不能说「这一行不存在」；
        #   ② 若该键在**丢弃侧**也不唯一，行数会把「同一个业务对象」重复计数，
        #      所以下面同时给出行数与**去重后的键数**，两者不等就说明键不唯一。
        res["note"] = ("键列 %s 在保留侧不是主键 → 结论是「键值不存在」而非「行不存在」"
                       % ",".join(missing_in_keep))

    on = " AND ".join("k.`%s` = d.`%s`" % (c, c) for c in cols)
    res["keep_n"] = one(cur, keep_db, table)
    res["drop_n"] = one(cur, drop_db, table)

    distinct = ("COUNT(DISTINCT d.`%s`)" % cols[0] if len(cols) == 1 else
                "COUNT(DISTINCT CONCAT_WS('\\x1f', %s))"
                % ", ".join("d.`%s`" % c for c in cols))
    cur.execute(
        "SELECT COUNT(*), %s FROM `%s`.`%s` d LEFT JOIN `%s`.`%s` k ON %s "
        "WHERE k.`%s` IS NULL"
        % (distinct, drop_db, table, keep_db, table, on, cols[0]))
    res["drop_only"], res["drop_only_distinct"] = cur.fetchone()

    if res["drop_only"]:
        cur.execute(
            "SELECT %s FROM `%s`.`%s` d LEFT JOIN `%s`.`%s` k ON %s "
            "WHERE k.`%s` IS NULL LIMIT %d"
            % (", ".join("d.`%s`" % c for c in cols),
               drop_db, table, keep_db, table, on, cols[0], sample))
        res["samples"] = cur.fetchall()
    return res


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--env", default=os.path.join(
        os.path.dirname(os.path.abspath(__file__)), "..", "..", "私密", "legacy-ro.env"))
    ap.add_argument("--only", action="append",
                    help="只跑这几张表（可重复；默认全部）")
    ap.add_argument("--sample", type=int, default=10,
                    help="发现独有行时最多列出几条键（默认 10）")
    ap.add_argument("--timeout", type=int, default=600,
                    help="单条 SQL 的读超时秒数（默认 600；大表逐行探测需要）")
    ap.add_argument("--key", action="append", metavar="表=列1,列2",
                    help="覆盖某张表的键（按表名匹配，可重复）。"
                         "用途：按主键判定「行不存在」后，再按**业务键**判定一次 —— "
                         "副本若只是换了 id 空间，用业务键才不会误判成「丢了数据」")
    a = ap.parse_args()

    cfg = load_env(a.env)
    dbs = db_names(cfg)

    # --key 的覆盖：只按表名匹配，允许同一张表在不同保留/丢弃组合上都换键
    over = {}
    for kv in (a.key or []):
        t, _, cols = kv.partition("=")
        if not cols:
            print("--key 格式应为 表=列[,列]", file=sys.stderr)
            return 2
        over[t.strip()] = cols.strip()

    want = [c for c in CHECKS if not a.only or c[0] in a.only]
    if over:
        # 显式给了键的表，即使没在 CHECKS 的 --only 里也要跑，避免"改了键却没跑"
        seen = {c[0] for c in want}
        for t in over:
            if t not in seen:
                want += [c for c in CHECKS if c[0] == t]
    if not want:
        print("没有匹配的检查项", file=sys.stderr)
        return 2

    conn = connect(cfg, read_timeout=a.timeout)
    need_manual = []
    try:
        with conn.cursor() as cur:
            print("=" * 78)
            print("重名表「丢弃份 ⊂ 保留份」精确校验（跨库 LEFT JOIN，只看键列）")
            print("=" * 78)
            for table, keep_mod, drop_mod, key, note in want:
                kd, dd = dbs[keep_mod], dbs[drop_mod]
                r = check(cur, table, kd, dd, over.get(table, key), a.sample)
                head = "%s  ← 保留 %s(%s) / 丢弃 %s(%s)" % (table, keep_mod, kd, drop_mod, dd)
                print("\n" + head)
                print("  备注: %s" % note)
                if r["keep_n"] is None:
                    print("  ✋ %s" % r["note"])
                    need_manual.append(head)
                    continue
                print("  保留侧 %d 行 / 丢弃侧 %d 行 / 键 %s"
                      % (r["keep_n"], r["drop_n"], "+".join(r["key"])))
                if r["note"]:
                    print("  ⚠️ %s" % r["note"])
                if r["drop_only"] == 0:
                    print("  ✅ 丢弃侧**全部**落在保留侧内 → 可安全丢弃（丢 0 行）")
                else:
                    uniq = ""
                    if r["drop_only_distinct"] != r["drop_only"]:
                        uniq = ("（去重后 %d 个键 —— 键在丢弃侧不唯一，"
                                "上面是行数）" % r["drop_only_distinct"])
                    print("  🔴 丢弃侧有 **%d 行**在保留侧找不到%s → 丢掉会真的丢数据"
                          % (r["drop_only"], uniq))
                    if r["samples"]:
                        print("      独有键样本: %s"
                              % ", ".join(str(s[0]) if len(s) == 1 else str(s)
                                          for s in r["samples"]))
                    need_manual.append(head)
    finally:
        conn.close()

    print("\n" + "=" * 78)
    if need_manual:
        print("需人工处置 %d 组：" % len(need_manual))
        for h in need_manual:
            print("  · %s" % h)
        print("\n处置方向：独有行取并集（`INSERT ... SELECT` 补进保留份，再丢弃副本）；"
              "\n          无主键无法判定的，先给它定一个业务键再跑一次。")
    else:
        print("✅ 全部待丢弃份都已被证明是保留份的子集 —— 可安全合并。")
    print("=" * 78)
    return 1 if need_manual else 0


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