BJ_ZTEITS-Developers / parking-cloud

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
「看着像废弃/备份表」的逐张确认 —— 六库合一前，给名字可疑的表逐个出证据。

## 为什么需要它

[`legacy-db-actual.md`](../legacy-db-actual.md) §8 数出这批残留：
`del<日期>` 前缀 2 张、`_<日期>` 后缀 12 张、`_copyN` 8 张、`_bak` 14 张、
`temp` 词 13 张、`test` 词 1 张。同一节的警告是：

> ⚠️ 这批**不能仅凭名字删**。`month_dchz`、`table_15_free` 这类"看着像临时表"的，
> 实际是被代码正常引用的业务表。**逐张确认后再决定**。

⚠️⚠️ **反方向的风险更大**，这是本脚本真正的存在理由：
**也 不能仅凭名字搬**。把 `sys_org_copy` 这类历史副本原样搬进新库，
它就会永久留在新库里——而 `00` 建库脚本与后来的建表语句都不知道它存在，
等于往新库注入一批**无来源、无归属、以后没人敢删**的表。

## 判据：名字只是"嫌疑"，证据是三条

1. **精确行数**（`COUNT(*)`，不是 `TABLE_ROWS` 估算）。
   估算对大表可差数十个百分点，而"这张表到底是不是空的"恰恰是这里的关键判断——
   用估算值判空等于没判。
2. **代码引用**：老工程搬进 `parking-module-*` 之后的 mapper XML 与 Java 里逐字搜。
   命中分**表位置**（硬证据）与**疑似**（大概率是 SQL 别名）两级，见下面坑 4。
3. **建表时间**：`_bak` 不等于"陈旧"——实测有 `sys_op_advice_bak` 是 **2026-02-25**
   建的（109 万行 / 200 MB）。近期建的备份表很可能是**运维正在用**的，不是历史残渣。
   （这一条 2026-10-10 已由业务方拍板「备份表，无需理会」，见 `DO_NOT_MIGRATE`。）

**代码引用数是硬约束**：只要落在表位置就不许按名字丢（见 `verdict()`）；
只有疑似命中的降为 🟠 并贴出原文，仍不自动放行。

## 「不搬」分两层，守卫能查的东西不一样（2026-10-10 加）

`DO_NOT_MIGRATE` 是**唯一**的"哪些表不搬进新库"的事实来源，值带层级：

| 层级 | 判据来源 | 守卫能查什么 | 不搬的代价 |
|---|---|---|---|
| `空表` | **数据自证**：0 行 + 无表位置引用 | 逐条复核仍存在 / 仍 0 行 / 仍无引用，任一不成立 exit 2 | **零**（本来就 0 行） |
| `备份` | **外部拍板**：业务方说"这是备份，无需理会" | 只能守**前提**——仍存在、仍无表位置引用。**证明不了决定本身对不对** | 那批历史行只留在老库。老库只读保留（回退路径）、**没有被销毁**，但新库里确实没有它们 |

⚠️ 这是 `AGENTS.md` §1.1 推论 4 的另一面：手工维护的清单必须有一个**不消费它**的
独立信号去反查。`空表` 那层有——行数与引用都是独立读出来的；`备份` 那层**没有**
——"这是不是你需要的备份"是个业务判断，任何查询都答不了。所以这层只能做两件事：

1. 把**谁、什么时候、凭什么**写进理由串。这是**唯一**的留痕处（报告是从这份清单生成的，
   别处没有第二份记录）；
2. 死盯"代码里还是没人读它"这个前提。前提一破就 exit 2，必须重新问业务方，
   **不许**因为"清单上已经写了不搬"就放行。

## ⚠️ 本工具踩过的五个坑（都是它自己犯的，记下来免得后人重犯）

1. **`temp` 用朴素子串匹配** → 把 `charger_ticket_template`、`td_c_msg_template`
   两张**真业务表**报成"名含 temp"。
   **本工具的整个存在理由就是"别仅凭名字下结论"，而它第一版自己就犯了。**
   现在 `temp` / `test` 按**独立词**匹配（前后不能是字母）。
2. **代码引用用朴素子串匹配** → 名为 `temp` 的表报出「924 处引用 / 924 个文件」，
   因为 `temp` 是 `template`、`tempFile`、Java 局部变量 `temp` 的子串。
   现在：XML 里按**词边界**匹配；Java 里**只认带引号的字面量**
   （表名在 Java 里只以字符串字面量出现，裸标识符一定是变量）。
3. **自检报了假警报** → 第一版的"分表族漏排"检查用族**前缀**判断，
   于是把 `pay_order_success_1202` 这类**有意放行**的变体表报成 🔴。
   误报的守卫等于没有守卫（`AGENTS.md` §1.1）。现在的自检换成了
   **两个独立实现互证**：MySQL 的 `LIKE` 与 Python 正则必须对"数字后缀的族成员"
   给出一致答案——它们才可能真的不一致。
4. **词边界仍然挡不住 SQL 别名** → 修完坑 2，`zt_park.temp`（7 行的真垃圾表）
   还是报出 6 个文件，因为 `duty_roster_base_temp temp` 里的 `temp` 是**表别名**，
   前后都是空格，词边界判据完全无法区分。
   现在命中分两级：`表位置`（`FROM/JOIN/…` 之后 + Java 带引号字面量）算硬证据，
   其余算**疑似**（🟠，不自动放行），并把**命中行原文**贴进报告——
   看的人一眼能分辨 `FROM xxx temp`（别名）与 `FROM temp`（真引用）。

5. **分表族清单只登记了 1/2，而自检发现不了** → `FAMILY` 只列了
   `tp_p_eqp_inout_record`，漏了 `tp_p_eqp_log`（113 张）。漏网的族表恰好因为
   "数字后缀不带下划线"没被 `_<日期> 后缀` 判据命中，**纯属侥幸**
   （改天族名里带个下划线就中招）。
   比错报更糟的是：自检拿 MySQL `LIKE` 去核对 Python 正则，而**两者共用同一份
   不完整的清单**——这就是 `AGENTS.md` §1.1 说的「一个能通过一切输出的守卫等于没有守卫」。
   现在加了实现 C `sharded_in_code()`：**不看清单**，只问"代码里有没有 `"前缀"` 这个
   字面量"，再**双向**反查清单（漏登记的、登记了却没在拼的，都 exit 2）。
   （另有一版实现 B `structural_families()` 只看名字形状，它**会**把人工快照
   误报成族，所以只当候选生成器、不下结论——「一个每次都喊狼来了的守卫，
   一样等于没有守卫」。）

6. **自检告警写进了报告，却打在生成流程的前半段** → `bad` 的最后一批
   （`DO_NOT_MIGRATE` 复核）是在后面几条循环里才追加的，而告警块原先放在
   「自检」小节。实测注入 4 条假条目：**退出码是 2，报告里却一个字都没提**。
   现在告警统一收在文末「## 自检告警」一节，报告上面留了指路。
   教训：「守卫真的会拦」不能靠读代码相信——**注入坏数据看它响不响**。

> 坑 1→2→4 是同一个病的三次发作：**"命中文本"不等于"引用表"**。
> 每次修都是在给判据补上下文（整词 → 词边界+引号 → 表位置）。
> 别再想着用一条正则把这件事做对了。
>
> 坑 3→5 是另一个病：**守卫与被守卫对象共享同一份错误假设**，于是守卫必然通过。
> 判据（清单、枚举、白名单）一旦是手工维护的，就必须有一个**不消费它**的
> 独立信号去反查它。
>
> 坑 6 是第三个病：**守卫没哑，只是报告没把它的声音放出来**。
> 守卫和报告是两件事，两边都要验。

## ⚠️ 必须排除分表族，否则会把活表当垃圾

老库有 **2 个族、共 228 张**运行期拼后缀的活表（实测，与
`legacy-db-actual.md` §6 一致）：

| 前缀 | 后缀 | 张数 | 范围 |
|---|---|---:|---|
| `tp_p_eqp_inout_record` | `yyyyMM` | 115 | 201709 ~ 202703 |
| `tp_p_eqp_log` | `yyyyMM` | 113 | 201711 ~ 202703 |

它们**长得和"`_<日期>` 后缀的备份表"一模一样**，但每一张都是活表。
注意数字后缀是**直接拼接**的（`tp_p_eqp_log201711`），没有下划线，
所以连 `_<6位年月> 后缀` 判据都碰不到它们——**漏登记的族不会报错，只会静默多出 113 张"嫌疑表"**。

> ⚠️ 这张表踩过两次，都记在坑 5：第一版只登记了 `tp_p_eqp_inout_record`
> （漏掉 `tp_p_eqp_log` 113 张）；`parking_running_statistic` 曾被当成第三个族，
> **那是错的**——代码从没拼过它，那几张是人工快照。现在清单由 `sharded_in_code()`
> **双向**反查（见坑 5）。

## ⚠️ 只读

只发 `SELECT COUNT(*)` 与读 `information_schema`。**不 SELECT 任何列值**，
所以即便嫌疑表里有敏感数据（如 `td_b_wx_config` 那类），输出里也不可能带出来。

用法：
    python3 -I sql/tools/audit_junk_tables.py                    # 报告打屏
    python3 -I sql/tools/audit_junk_tables.py --emit docs/xxx.md
    python3 -I sql/tools/audit_junk_tables.py --exact-limit 0    # 不限量，全取精确行数

退出码：0 = 报告已生成（**不代表"已处置"**，处置结论要人拍板后写回文档）。
        2 = 自检失败（分表族判定两套实现不一致），此时报告内容不可信。
"""

import argparse
import datetime
import os
import pathlib
import re
import sys

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

# ---------------------------------------------------------------------------
# 嫌疑判据：唯一的事实来源。改这里，不要改报告。
#
# 顺序有意义：先报最像"人为备份"的，再报词命中的——报告里按这个顺序排。
PATTERNS = [
    ('del 前缀带日期',   re.compile(r'^del_?\d{6,8}', re.I)),
    ('_copy 副本',       re.compile(r'_copy', re.I)),
    ('_bak 备份',        re.compile(r'_bak', re.I)),
    ('_<8位日期> 后缀',  re.compile(r'_\d{8}$')),
    ('_<6位年月> 后缀',  re.compile(r'_\d{6}$')),
    ('_<4位数字> 后缀',  re.compile(r'_\d{4}$')),
    # 裸数字后缀（**无下划线**）≥4 位：这正是"运行期拼后缀"的形状
    # （`tp_p_eqp_log201711`）。活族成员在前面就被 FAMILY 拦掉了，
    # 能走到这里的说明**没有登记过的族认领它**——实测抓到
    # `parking_running_statistic2023` / `…2025` 两张人工快照。
    ('裸<数字>后缀',     re.compile(r'(?<=[a-z_])\d{4,}$', re.I)),
    # ⚠️ `temp` / `test` 必须按**独立词**匹配，不能用朴素子串——
    # 见文件头坑 1：朴素子串会把 `template` 报成临时表。
    ('名含 temp',        re.compile(r'(?<![a-z])temp(?![a-z])', re.I)),
    ('名含 test',        re.compile(r'(?<![a-z])test(?![a-z])|test$', re.I)),
]

# ⚠️ 分表族前缀的**唯一事实来源**，必须与 `legacy-db-actual.md` §6 一致。
# 数字后缀**直接拼在名字后面**（`tp_p_eqp_log201711`），没有下划线——
# 这也是它们能躲过 `_<6位年月> 后缀` 判据的原因。
#
# 漏登记一个前缀 = 那个族的**每一张表**都会被报成"疑似备份表"。
# 本清单的完整性由 `structural_families()` 反向校验（它不看这份清单）。
FAMILY_PREFIXES = (
    # 库           前缀                      后缀     张数
    'tp_p_eqp_inout_record',   # zt_park  yyyyMM   115  201709~202703
    'tp_p_eqp_log',            # zt_park  yyyyMM   113  201711~202703
)
# ⚠️ §6 一度把 `parking_running_statistic`（2 张：…2023、…2025）也当成族，
# **那是错的**（§6 已于 2026-10-10 改正）：全仓只有 `FROM parking_running_statistic`
# （裸表名，SQL 里），没有任何代码把它的表名拼出来。真正的族在 Java 里必然有
# `"前缀" + 日期` 的拼装。那两张是人工快照，已改列入嫌疑判据，
# 并按 §10.1 拍板**不搬**（见下面 `DO_NOT_MIGRATE` 的 `备份` 层）。
# 判据见 `sharded_in_code()`：**代码里有没有 `"前缀"` 这个字面量**。
FAMILY = re.compile(r'^(%s)\d+$' % '|'.join(FAMILY_PREFIXES), re.I)

# 另有一批**以族前缀开头、但不是族成员**的表：`tp_p_eqp_inout_record_temp`、
# `pay_order_success` 等。它们照常走嫌疑判据，但要在报告里显式交代一句，
# 免得看的人以为是漏排。
# `pay_order` 是历史包袱前缀（代码里并无 `pay_order{yyyy}` 分表），
# 留着只为让这批表出现在"有意放行"清单里。
FAMILY_PREFIX = re.compile(
    r'^(%s)' % '|'.join(FAMILY_PREFIXES + ('pay_order',)), re.I)

# 结构判据用：名字以 >=4 位数字结尾（可带一个下划线）即"像族成员"。
STRUCT_FAMILY = re.compile(r'^(.+?)_?(\d{4,})$')

# 建表时间在这个月数内的，单独标出来——`_bak` 不等于陈旧，见文件头判据 3。
RECENT_MONTHS = 12

CODE_EXTS = ('.java', '.xml')
DATE_FMT = '%Y-%m-%d'

# ---------------------------------------------------------------------------
# 已拍板的处置：**不搬进新库**的表。`gen_migration_sql.py` 直接 import 这份
# （与 `check_subset.CHECKS` 同一个套路：单一事实来源，不手抄第二份）。
#
# 值 = (层级, 理由)，层级**决定守卫能查什么**（详见文件头「「不搬」分两层」）：
#
#   `空表` —— 0 行 + 无表位置引用。判据是**数据自证**的，守卫能完整复核。
#   `备份` —— **有数据**的，判据是**外部拍板**。守卫只能守它的前提（仍存在、
#             仍无引用）；"这是不是你需要的备份"本工具证明不了。
#
# ⚠️ 两层的风险不对称：
#   空表不搬 —— **一行都不丢**，只是新库少一张没人认领的表。
#   备份不搬 —— 那批历史行**只留在老库**（老库只读保留，是回退路径，没有销毁，
#               需要时可以单独导出），但新库里确实没有。**理由串是唯一留痕处**。
# 有数据、又没拿到业务方明确答复的，一律按 §10 的"先全搬"走——"没引用"只是
# **当前代码**的结论，外部报表/BI/运维 SQL 有没有在读，本工具证明不了。
#
# ⚠️ 加/删条目后必须重跑本脚本：`main()` 会逐条复核"它还存不存在、还是不是空表、
#    还有没有代码引用"，任一条不成立就 exit 2。清单会漂，守卫得跟着一起动。
# 层级标签的**全集**（下面 `DO_NOT_MIGRATE` 的值的第一个元素只许取这里的）。
# 为什么要有这个常量：标签是**字符串比较**的判据，写错一个字（`'备份'` → `'备份表'`）
# 不会抛任何错——只会让"按层级统计"的数字静默变成 0，而那个数字会写进
# `exclude_tables.txt` 的头部。生成物里的假数字比报错难查得多（实测刚踩过）。
# 本脚本与 `gen_migration_sql.py` 都拿它校验，不各写一份。
DO_NOT_MIGRATE_TIERS = ('空表', '备份')

DO_NOT_MIGRATE = {
    # ---- 层级一：空表（数据自证）----
    'zt_order.td_b_pay_blncac_diff_copy':
        ('空表', '_copy 副本，0 行（zt_order）'),
    'zt_order_n.td_b_pay_blncac_result_bak':
        ('空表', '_bak 备份，0 行（zt_order_n）'),
    'zt_park.del20181116td_g_first_guide_info_config':
        ('空表', 'del<日期> 残留，0 行（zt_park）'),
    'zt_park.del20181116td_g_guide_park_berths':
        ('空表', 'del<日期> 残留，0 行（zt_park）'),
    # ---- 层级二：有数据的备份（**外部拍板**，见 legacy-db-actual.md §10.1）----
    # 这三张的共同形态：代码全仓只读裸表名（无后缀那张），带后缀的是**人工滚动快照**，
    # 建表时间与"无后缀那张"只差几分钟——是 `CREATE TABLE … SELECT` 出来存档的。
    # 前提（代码零引用）由 main() 逐条盯着，破了就 exit 2，必须重新问。
    'zt_park.parking_running_statistic2023': (
        '备份',
        '滚动快照（建表 2024-05-22，3.86 亿行 / 约 83 GB），代码零引用；'
        '**2026-10-10 业务方拍板「备份表，无需理会」**'),
    'zt_park.parking_running_statistic2025': (
        '备份',
        '滚动快照（建表 2026-04-17，2.41 亿行 / 约 50 GB），代码零引用；'
        '**2026-10-10 业务方拍板「备份表，无需理会」**'),
    'zt_sys.sys_op_advice_bak': (
        '备份',
        '滚动前的旧表副本（建表 2026-02-25，109 万行 / 200 MB），代码零引用；'
        '**2026-10-10 业务方拍板「备份表，无需理会」**'),
}


def do_not_migrate_tables():
    """合库时"不搬"的 `库.表` 列表，供 `gen_migration_sql.py` 并入排除清单。

    **单一事实来源就是上面的 `DO_NOT_MIGRATE`**，别在别处再抄一份。
    """
    return sorted(DO_NOT_MIGRATE)


def do_not_migrate_tier(db_table):
    """`'空表'` / `'备份'` / `None`。守卫按层级决定查什么。"""
    e = DO_NOT_MIGRATE.get(db_table)
    return e[0] if e else None


def module_of(path, root):
    """从路径取模块名：`src` 之前最深的那个 `parking-*`。"""
    parts = path.relative_to(root).parts
    try:
        si = parts.index('src')
    except ValueError:
        return None
    for p in reversed(parts[:si]):
        if p.startswith('parking-'):
            return p
    return parts[0] if parts else None


# ---------------------------------------------------------------------------
# 一、从库里取嫌疑表

def all_tables(conn, db):
    with conn.cursor() as cur:
        cur.execute(
            "SELECT TABLE_NAME, TABLE_ROWS, DATA_LENGTH, CREATE_TIME "
            "FROM information_schema.TABLES WHERE TABLE_SCHEMA=%s "
            "AND TABLE_TYPE='BASE TABLE' ORDER BY TABLE_NAME", (db,))
        return cur.fetchall()


def family_catalog(conn, db):
    """独立实现 A：**用 MySQL 的 LIKE** 数分表族的数字后缀成员。

    刻意不复用 Python 的 FAMILY 正则——自检要的就是"两套独立实现互证"。
    """
    cond = ' OR '.join(
        "TABLE_NAME LIKE '%s%%%%'" % p.replace('_', r'\_') for p in FAMILY_PREFIXES)
    sql = ("SELECT TABLE_NAME FROM information_schema.TABLES "
           "WHERE TABLE_SCHEMA=%%s AND TABLE_TYPE='BASE TABLE' AND (%s)" % cond)
    with conn.cursor() as cur:
        cur.execute(sql, (db,))
        names = [r[0] for r in cur.fetchall()]
    by_prefix = {p: [] for p in FAMILY_PREFIXES}
    for n in names:
        low = n.lower()
        for p in by_prefix:
            if low.startswith(p) and low[len(p):].isdigit():
                by_prefix[p].append(n)
                break
    return names, by_prefix


def structural_families(names, min_members=2):
    """独立实现 B（**不依赖 `FAMILY_PREFIXES`**）：只看名字形状。

    名字以 >=4 位数字结尾者，按**去掉数字后的前缀**归组；同前缀 >= `min_members`
    张就算"结构上像族"。**它只是候选**——人工快照（`tp_p_eqp_20180915`）也长这样，
    必须再用 `sharded_in_code()` 判一次，见自检。
    """
    groups = {}
    for n in names:
        m = STRUCT_FAMILY.match(n.lower())
        if m:
            groups.setdefault(m.group(1), []).append(n)
    return {p: sorted(v) for p, v in groups.items() if len(v) >= min_members}


SHARD_SKIP = {'target', 'node_modules', '.git', 'logs', '.dump'}


def strip_comments(txt, is_xml):
    """抹掉注释，只留下**可执行文本**（字符串字面量保留原样）。

    ⚠️ 2026-10-23 修正：`sharded_in_code()` 原先直接对**原文**跑正则，
    而本仓 Java 的 Javadoc 习惯用 Markdown 反引号写表名（`` `tp_p_eqp` ``），
    反引号恰好被当成"引号字面量"⇒ 三个**静态表**被判成"代码在拼分表族"，
    报告退出码 2、整份不可信。

    判据的本意是「运行期拼表名 ⇒ 代码里必有 `"前缀"` 字符串字面量」，
    注释里的反引号不是字面量，必须先剥掉。字符串内部的字符原样保留
    （包括 `\\"` 转义、Java 字符字面量），XML 只剥 `<!-- -->`。
    """
    out = []
    i, n = 0, len(txt)
    in_str = False       # Java 双引号字符串 / XML 单引号 SQL 字符串共用
    in_chr = False       # 仅 Java：字符字面量
    quote = '"'
    while i < n:
        c = txt[i]
        if in_str or in_chr:
            out.append(c)
            if c == '\\' and i + 1 < n:
                out.append(txt[i + 1])
                i += 2
                continue
            if c == quote:
                if in_str:
                    in_str = False
                else:
                    in_chr = False
            i += 1
            continue
        if c == '"':
            in_str, quote = True, '"'
            out.append(c)
            i += 1
            continue
        if c == "'" and not is_xml:
            in_chr, quote = True, "'"
            out.append(c)
            i += 1
            continue
        if not is_xml and txt.startswith('//', i):
            j = txt.find('\n', i)
            i = n if j < 0 else j
            continue
        if not is_xml and txt.startswith('/*', i):
            j = txt.find('*/', i + 2)
            i = n if j < 0 else j + 2
            out.append('\n')
            continue
        if is_xml and txt.startswith('<!--', i):
            j = txt.find('-->', i + 4)
            i = n if j < 0 else j + 3
            out.append('\n')
            continue
        out.append(c)
        i += 1
    return ''.join(out)


def sharded_in_code(root, prefixes):
    """独立实现 C（**权威判据**）：**可执行代码**里有没有 `"<前缀>"` 这个字面量。

    运行期分表必然要在 Java（或 XML 的动态 SQL）里把表名拼出来，
    即 `"tp_p_eqp_log" + yyyyMM` 这种写法——于是代码里一定存在
    `"tp_p_eqp_log"` 这个独立字符串。反过来，**静态表**在 SQL 里永远只以
    裸名字出现（`FROM parking_running_statistic`），Java 里不会有它的引号字面量。

    这条判据和 `scan_code()` 里 Java 分支的口径一致（都是"只认带引号的字面量"），
    但**不消费 `FAMILY_PREFIXES`**，所以能反过来证伪那份清单。
    ⚠️ 注释（含 Javadoc 里的反引号写法的表名）不是字面量，先 `strip_comments()`。
    """
    if not prefixes:
        return set()
    rx = re.compile(r'["\'`](%s)["\'`]'
                    % '|'.join(re.escape(p) for p in prefixes), re.I)
    hits = set()
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [d for d in dirnames if d not in SHARD_SKIP]
        if os.sep + 'src' + os.sep not in dirpath + os.sep:
            continue
        for fn in filenames:
            if not (fn.endswith('.java') or fn.endswith('.xml')):
                continue
            try:
                txt = open(os.path.join(dirpath, fn), encoding='utf-8',
                           errors='ignore').read()
            except OSError:
                continue
            txt = strip_comments(txt, fn.endswith('.xml'))
            for m in rx.finditer(txt):
                hits.add(m.group(1).lower())
    return hits


def candidates(conn, dbs):
    """{库: [(表名, 判据, 估算, 字节, 建表时间)]}，已排除分表族。

    另返回自检所需：families（按族排除的）、familyish（前缀属族但未按族排除的）。
    """
    out, families, familyish = {}, [], []
    for mod, db in sorted(dbs.items()):
        hits = []
        for name, est, size, created in all_tables(conn, db):
            if FAMILY.match(name):
                families.append('%s.%s' % (db, name))
                continue
            if FAMILY_PREFIX.match(name):
                familyish.append('%s.%s' % (db, name))
            for label, rx in PATTERNS:
                if rx.search(name):
                    hits.append((name, label, est, size or 0, created))
                    break
        if hits:
            out[db] = hits
    return out, families, familyish


# ---------------------------------------------------------------------------
# 二、从代码里找引用

def scan_code(root, names):
    """{表名小写: {相对路径: (模块, kind, 行号, 命中行原文)}}。

    一次遍历、对每个文件跑一条"所有名字的并集"正则——比每个表名各 grep 一遍
    少几百倍 IO。

    **命中分强弱两类**（见文件头坑 2、坑 4）：
      - `sql`  —— 出现在**表位置**：`FROM/JOIN/INTO/UPDATE/TABLE/EXISTS` 后面，
        或 Java 里的**带引号字面量**。这是强证据。
      - `other`—— 其它位置的词边界命中。**弱证据**，主要来源是 **SQL 别名**：
        `FROM duty_roster_base_temp temp` 里的 `temp` 是别名，不是表名，
        实测就靠这个把 7 行的 `zt_park.temp` 报成了「有代码引用」。

    两类都报，**不据此自动下结论**——但把命中行原文一起贴出来，
    看的人一眼就能分辨是 `FROM xxx temp`（别名）还是 `FROM temp`（真引用）。
    """
    if not names:
        return {}
    alt = '|'.join(re.escape(n) for n in sorted(names, key=len, reverse=True))
    rx_sql = re.compile(
        r'(?i)\b(?:from|join|into|update|table|exists)\s+`?(%s)(?![a-z0-9_`])' % alt)
    rx_xml = re.compile(r'(?<![a-z0-9_])(%s)(?![a-z0-9_])' % alt, re.I)
    rx_java = re.compile(r'["\'`](%s)["\'`]' % alt, re.I)

    def line_at(txt, pos):
        return txt.count('\n', 0, pos) + 1

    found = {}
    skip = {'target', 'node_modules', '.git', 'logs', '.dump'}
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [d for d in dirnames if d not in skip]
        if os.sep + 'src' + os.sep + 'main' + os.sep not in dirpath + os.sep:
            continue
        for fn in filenames:
            is_java = fn.endswith('.java')
            if not (is_java or fn.endswith('.xml')):
                continue
            fp = os.path.join(dirpath, fn)
            try:
                txt = open(fp, encoding='utf-8', errors='ignore').read()
            except OSError:
                continue
            per_file = {}
            for kind, rx in (('sql', rx_java if is_java else rx_sql),
                             ('other', None if is_java else rx_xml)):
                if rx is None:
                    continue
                for m in rx.finditer(txt):
                    n = m.group(1).lower()
                    cur = per_file.setdefault(n, ['other', 0, ''])
                    if kind == 'sql':                      # 强证据覆盖弱证据
                        cur[0] = 'sql'
                    if not cur[2]:                         # 只在首条命中时存原文
                        ln = line_at(txt, m.start())
                        cur[1] = ln
                        cur[2] = txt.splitlines()[ln - 1].strip()[:160]
            if not per_file:
                continue
            module = module_of(pathlib.Path(fp), root) or '?'
            rel = os.path.relpath(fp, root)
            for n, (kind, ln, snip) in per_file.items():
                found.setdefault(n, {})[rel] = (module, kind, ln, snip)
    return found


# ---------------------------------------------------------------------------
# 三、结论

def verdict(n_rows, n_files, n_sql, months_old):
    """处置建议。**代码引用数是硬约束**：只要有人查，就不许按名字丢。

    引用分两级（见 `scan_code`）：`sql` = 表位置命中，是硬证据；
    只有 `other` 命中（典型是 SQL 别名 `FROM xxx temp`）则降为 🟠——
    大概率是噪声，但**不自动放行**，仍需人眼确认一条。
    """
    if n_sql:
        return '🔴 有代码引用', '**不能按名字处置**——先看引用是活代码还是死代码'
    if n_files:
        return '🟠 疑似引用（非表位置）', (
            '%d 个文件命中，但**没有一处出现在表位置**（`FROM/JOIN/…` 之后或 Java 字面量），'
            '多半是 SQL 别名或变量——人工扫一眼明细即可判' % n_files)
    if months_old is not None and months_old <= RECENT_MONTHS:
        return '🟠 近期创建', ('建表距今 %d 个月，很可能是运维在用而非历史残渣，'
                              '先问运维' % months_old)
    if n_rows == 0:
        return '🟢 空表且无引用', '可丢（建议仍导出留痕）'
    if n_rows is None:
        return '🟡 无引用但行数未知', '行数超限未数，定处置前先取精确 COUNT(*)'
    return '🟡 有数据但无引用', '导出留痕后丢，或随迁（默认`先全搬`）'


def months_since(created, now):
    if not created:
        return None
    return (now.year - created.year) * 12 + (now.month - created.month)


def extract_rows(conn, db, table, limit, est):
    """精确行数。超过 limit 就先不数——报告里显式标出来，别拿估算冒充。"""
    if limit and (est or 0) > limit:
        return None
    with conn.cursor() as cur:
        cur.execute('SELECT COUNT(*) FROM `%s`.`%s`' % (db, table))
        return cur.fetchone()[0]


def human(n):
    for unit in ('B', 'KB', 'MB', 'GB', 'TB'):
        if n < 1024:
            return '%.1f %s' % (n, unit)
        n /= 1024.0
    return '%.1f PB' % n


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('--root', default=os.path.join(
        os.path.dirname(os.path.abspath(__file__)), '..', '..'),
        help='merged 仓库根（默认 parking-cloud/）')
    ap.add_argument('--emit', help='把报告写到文件（默认打到 stdout）')
    ap.add_argument('--exact-limit', type=int, default=5_000_000,
                    help='估算行数超过它就不取精确值（0=不限）')
    a = ap.parse_args()

    cfg = load_env(a.env)
    dbs = db_names(cfg)
    conn = connect(cfg, read_timeout=900)
    now = datetime.datetime.now()
    bad = []
    try:
        cand, families, familyish = candidates(conn, dbs)

        # ---- 自检 1/2：MySQL LIKE（实现 A） vs Python 正则（实现 B）----
        like_total = regex_total = 0
        for mod, db in sorted(dbs.items()):
            _, by_prefix = family_catalog(conn, db)
            like_total += sum(len(v) for v in by_prefix.values())
        regex_total = len(families)
        if like_total != regex_total:
            bad.append('分表族计数不一致：MySQL LIKE 数出 %d，Python 正则数出 %d'
                       % (like_total, regex_total))

        # ---- 自检 2/2：结构候选 × 代码判据（实现 C，不看清单）----
        # 这一条才是真守卫：A 与 B 都消费 FAMILY_PREFIXES，清单漏了族它们一起漏。
        names_all = []
        for mod, db in sorted(dbs.items()):
            names_all += [n for n, _r, _s, _c in all_tables(conn, db)]
        struct = structural_families(names_all)
        sharded = sharded_in_code(pathlib.Path(a.root),
                                  set(struct) | set(FAMILY_PREFIXES))

        # (a) 代码在拼、清单没登记 → 漏了族，该族的表会被整批错报成备份表
        for p in sorted(sharded - set(FAMILY_PREFIXES)):
            v = struct.get(p, [])
            bad.append('`%s` 在代码里被拼装（分表族），但清单未登记'
                       '（结构上 %d 张%s）——漏一个族，它的表会整批被错报成备份表'
                       % (p, len(v),
                          '：%s%s' % ('、'.join(v[:3]), '…' if len(v) > 3 else '')
                          if v else ''))
        # (b) 清单登记了、代码却没拼 → 名字像族但不是族，那几张是人工快照/残留
        for p in sorted(set(FAMILY_PREFIXES) - sharded):
            v = struct.get(p, [])
            bad.append('清单登记了 `%s`，但**代码里没有任何地方拼这个表名**——'
                       '它不是分表族，那 %d 张（%s%s）应从嫌疑清单里逐张确认，'
                       '现在是被错放行了' % (p, len(v), '、'.join(v[:3]),
                                        '…' if len(v) > 3 else ''))
        # 结构上像族、代码不拼、清单也没登记 → 人工快照，照常走嫌疑判据（不告警）
        snapshots = sorted(set(struct) - sharded - set(FAMILY_PREFIXES))

        names = {n.lower() for hits in cand.values() for n, _, _, _, _ in hits}
        code = scan_code(pathlib.Path(a.root), names)

        lines = []
        w = lines.append
        w('# 废弃/备份嫌疑表 —— 逐张确认')
        w('')
        w('> 本文件由 `sql/tools/audit_junk_tables.py` 生成，**不要手改**（重跑会覆盖）。')
        w('> 判据与结论口径见该脚本文件头。**名字只是嫌疑**，证据是精确行数 + 代码引用 + 建表时间。')
        w('')
        w('## 自检')
        w('')
        w('- 分表族（运行期拼后缀的活表）已排除 **%d 张**；'
          '独立用 MySQL `LIKE` 复查得 **%d 张**——%s'
          % (regex_total, like_total, '✅ 一致' if like_total == regex_total
             else '🔴 不一致，先修 `FAMILY` 再读结论'))
        for p in sorted(FAMILY_PREFIXES):
            v = struct.get(p, [])
            w('  - `%s` **%d 张**（%s ~ %s）'
              % (p, len(v), v[0] if v else '—', v[-1] if v else '—'))
        w('- 族清单完整性：用 `sharded_in_code()`（**代码里有没有 `"前缀"` 字面量**）'
          '双向反查——%s'
          % ('✅ 登记的都真在拼、在拼的都登记了'
             if not any('在代码里被拼装' in b or '代码里没有任何地方拼' in b for b in bad)
             else '🔴 见下方告警'))
        if snapshots:
            w('- 名字像族、**但代码没拼**（人工快照，不是族，照常走嫌疑判据）：')
            for p in snapshots:
                v = struct[p]
                w('  - `%s` %d 张：%s' % (p, len(v),
                                        '、'.join('`%s`' % x for x in v[:6])
                                        + ('…' if len(v) > 6 else '')))
        w('- 分表族前缀下**未按族排除**的成员（有意放行，照常走嫌疑判据）：')
        for t in familyish:
            w('  - `%s`' % t)
        w('- 代码引用口径：命中分两级——`表位置`（`FROM/JOIN/INTO/UPDATE/TABLE/EXISTS` 之后，'
          '或 Java 里的**带引号字面量**）记为硬证据；'
          '其余词边界命中记为**疑似**，因为 `FROM duty_roster_base_temp temp` 里的 `temp` '
          '是**别名不是表名**，实测就靠这个把 7 行的 `zt_park.temp` 报成了「有引用」。'
          '两级都会贴出命中行原文，一眼可判')
        w('- 扫描根：`%s`' % a.root)
        w('')
        w('> ⚠️ 自检项分散在生成过程的各处（代码引用要扫完仓才知道、'
          '`DO_NOT_MIGRATE` 要复核完才敢确认），所以'
          '**告警统一在文末「自检告警」一节列出**——别只看这上面几行。'
          '（踩过：告警块原先写在这一节里，`DO_NOT_MIGRATE` 那批复核是在它**之后**'
          '才追加进 `bad` 的，于是退出码 2、报告里一个字都没提。）')
        w('')

        total = sum(len(v) for v in cand.values())
        w('## 总计 %d 张嫌疑表' % total)
        w('')
        w('| 库 | 张数 |')
        w('|---|---:|')
        for db, hits in sorted(cand.items()):
            w('| `%s` | %d |' % (db, len(hits)))
        w('')

        buckets, refd, seen = {}, [], {}
        for db, hits in sorted(cand.items()):
            w('## `%s`（%d 张）' % (db, len(hits)))
            w('')
            w('| 表 | 命中判据 | 精确行数 | 大小 | 建表时间 | 代码引用 | 建议处置 |')
            w('|---|---|---:|---:|---|---|---|')
            for name, label, est, size, created in hits:
                n = extract_rows(conn, db, name, a.exact_limit, est)
                refs = code.get(name.lower(), {})       # {rel: (模块, kind, 行号, 原文)}
                n_sql = sum(1 for r in refs.values() if r[1] == 'sql')
                key = '%s.%s' % (db, name)
                seen[key] = (n, n_sql, len(refs))
                v, why = verdict(n, len(refs), n_sql, months_since(created, now))
                decided = DO_NOT_MIGRATE.get(key)
                if decided:
                    # 已拍板不搬：盖掉"待确认/建议"那类结论——那条问题**已经有人回答了**，
                    # 报告里再留着"先问运维"会让人以为还悬着。
                    # 但**不盖证据**：万一有（哪怕疑似）代码命中，原样附在理由后面。
                    v = '⬛ 已拍板不搬（%s）' % decided[0]
                    why = decided[1]
                    if refs:
                        why += ('（⚠️ 但代码里有 %d 处命中，其中表位置 %d 处'
                                '——见上面明细，不搬的理由存疑）' % (len(refs), n_sql))
                buckets[v] = buckets.get(v, 0) + 1
                if refs:
                    refd.append((db, name, refs))
                n_txt = ('{:,}'.format(n) if n is not None
                         else '超限未数（估 {:,}）'.format(est or 0))
                if not refs:
                    ref_txt = '—'
                elif n_sql:
                    ref_txt = '表位置 **%d** / 疑似 %d' % (n_sql, len(refs) - n_sql)
                else:
                    ref_txt = '表位置 **0** / 疑似 %d' % len(refs)
                w('| `%s` | %s | %s | %s | %s | %s | %s —— %s |' % (
                    name, label, n_txt, human(size),
                    created.strftime(DATE_FMT) if created else '—',
                    ref_txt, v, why))
            w('')

        if refd:
            w('## 🔴/🟠 有命中的：明细（要逐条判断是活代码还是死代码）')
            w('')
            w('> `表位置` = 硬证据；`疑似` = 命中词不在表位置，'
              '**看下面贴的原文**：`FROM tmp_x temp` 这种是别名，不算引用。')
            w('')
            for db, name, refs in refd:
                sql_hits = {k: r for k, r in refs.items() if r[1] == 'sql'}
                other_hits = {k: r for k, r in refs.items() if r[1] != 'sql'}
                w('### `%s`.`%s`' % (db, name))
                w('')
                for tag, group in (('表位置', sql_hits), ('疑似', other_hits)):
                    if not group:
                        continue
                    w('**%s**（%d）' % (tag, len(group)))
                    w('')
                    for f in sorted(group)[:12]:
                        module, _, ln, snip = group[f]
                        w('- `%s`:%d — `%s`' % (f, ln, snip))
                    if len(group) > 12:
                        w('- …另 %d 个文件' % (len(group) - 12))
                    w('')

        # ---- 复核 DO_NOT_MIGRATE：它列的前提，现在还成立吗？----
        # 两层共有的前提是"**代码里没人读它**"；`空表` 层多一条"还是 0 行"。
        unknown_tiers = sorted({t for t, _w in DO_NOT_MIGRATE.values()
                                if t not in DO_NOT_MIGRATE_TIERS})
        if unknown_tiers:
            bad.append('`DO_NOT_MIGRATE` 里有未知层级 %s——只许 %s。'
                       '标签是字符串比较的判据，写错一个字不会报错，'
                       '只会让按层级的统计（以及生成物头部的数字）静默变 0'
                       % ('、'.join(unknown_tiers), '、'.join(DO_NOT_MIGRATE_TIERS)))
        for db_table, (tier, _why) in sorted(DO_NOT_MIGRATE.items()):
            if db_table not in seen:
                bad.append('`%s` 已不在嫌疑清单里（改名/删表/判据变了）——'
                           '先查清再动 `DO_NOT_MIGRATE`' % db_table)
                continue
            n, n_sql, n_files = seen[db_table]
            if n_sql:
                bad.append('`%s` 现在有 %d 处**表位置**代码引用，"不搬"的理由不成立了'
                           '——这是两层共有的前提，比行数重要' % (db_table, n_sql))
            elif n_files:
                bad.append('`%s` 现在有 %d 处**疑似**命中（都不在表位置）——'
                           '先人眼扫一遍报告里的明细，确认是 SQL 别名还是真引用，'
                           '再决定这份清单要不要改' % (db_table, n_files))
            if tier == '空表' and n != 0:
                bad.append('`%s` 现在是%s，**已经不是空表**了——'
                           '要么改层级成 `备份`（得有业务方拍板），要么从清单里去掉'
                           % (db_table,
                              '有 %s 行' % '{:,}'.format(n) if n is not None
                              else '行数超限未数'))
        n_empty = sum(1 for t, _w in DO_NOT_MIGRATE.values() if t == '空表')
        w('## 已拍板：不搬进新库的表（%d 张 = 空表 %d + 有数据的备份 %d）'
          % (len(DO_NOT_MIGRATE), n_empty, len(DO_NOT_MIGRATE) - n_empty))
        w('')
        w('> 单一事实来源是 `audit_junk_tables.py` 的 `DO_NOT_MIGRATE`，'
          '迁移脚本（`gen_migration_sql.py`）直接 import 它并写进 '
          '`sql/migration/exclude_tables.txt`（`mysqldump --ignore-table`）、'
          '`99_verify.sql` §1 的 `NOT IN`、`98_count_reconcile.sql` 的排除集。')
        w('>')
        w('> **两层的守卫强度不一样**（见脚本文件头）：')
        w('> `空表` 靠**数据自证**——仍存在、仍 0 行、仍无表位置引用，逐条复核；')
        w('> `备份` 是**外部拍板**，本工具只能守它的前提——仍存在、仍无表位置引用。'
          '**"这是不是你需要的备份"没有任何查询能回答**。')
        w('>')
        w('> ⚠️ 所以 `备份` 那几张**是有数据的**：不搬 = 那批历史行只留在老库'
          '（老库只读保留、未被销毁，需要时可单独导出），但**新库里确实没有**。'
          '理由串里的"谁、什么时候、凭什么"是这件事**唯一**的留痕处。')
        w('')
        w('| 表 | 层级 | 理由 |')
        w('|---|---|---|')
        for db_table, (tier, why) in sorted(DO_NOT_MIGRATE.items()):
            # 把"有没有代码命中"也标在这张表里：它是读者最先看的一处，
            # 只写理由串会让人以为干净（退出码 2 的告警在**文末**，容易先被略过）。
            st = seen.get(db_table)
            flag = (' ⚠️ **有 %d 处命中（表位置 %d）**，见上文明细；自检已报红'
                    % (st[2], st[1])) if st and st[2] else ''
            w('| `%s` | %s | %s%s |' % (db_table, tier, why, flag))
        w('')

        w('## 结论分布')
        w('')
        w('| 结论 | 张数 |')
        w('|---|---:|')
        for k, c in sorted(buckets.items()):
            w('| %s | %d |' % (k, c))
        w('')
        w('> 处置决定拍板后，写回 [`../legacy-db-actual.md`](../legacy-db-actual.md) §8，'
          '并在 `check_subset.py` / 排除清单里落地——**别只改这份报告**（它是生成物）。')
        w('')
        # ⚠️ 告警必须放文末：`bad` 的最后一批（DO_NOT_MIGRATE 复核）是在
        #    上面几条循环之后才追加的。放在「自检」小节里会**看不见**——
        #    实测踩过：退出码是 2，报告里却一个字都没提。
        w('## 自检告警')
        w('')
        if bad:
            w('> 🔴 **自检未通过（退出码 2），以下结论不可信**：')
            for b in bad:
                w('> - %s' % b)
        else:
            w('✅ 全部自检通过（退出码 0）。')
            w('')
            w('即：分表族两套实现一致、族清单被代码判据双向确认、'
              '`DO_NOT_MIGRATE` 逐条的**前提**仍成立——两层共有的前提是'
              '**代码里没人读它**，`空表` 层另有"仍是 0 行"。')
            w('')
            w('⚠️ 注意这条自检的**边界**：它只能证明"不就已知前提报警"，'
              '证明不了"不搬这个决定本身是对的"——`备份` 层的理由（"这是备份，不用管"）'
              '是业务判断，报告里的任何一行都不是它的证据。')

        text = '\n'.join(lines) + '\n'
        if a.emit:
            with open(a.emit, 'w', encoding='utf-8') as fh:
                fh.write(text)
            print('已写出 %s（%d 张嫌疑表，排除分表族 %d 张%s）'
                  % (a.emit, total, regex_total, '，⚠️ 自检未通过' if bad else ''))
        else:
            print(text)
    finally:
        conn.close()
    return 2 if bad else 0


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