#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
build.sh 硬门禁 6.4 的 DDL 扫描器 —— 先剥「SQL 注释 + 字符串字面量(含 COMMENT 串)」，再扫表。

背景（deploy_test 第 3 轮实测 Q4）
--------------------------------
``scripts/ddl/pbls_tables.sql`` 文末有一段「契约自检」注释块，其中为说明"零命中项"，
逐字写出了 ``FOREIGN KEY / REFERENCES / ENUM( / SERIAL / BIGSERIAL / nextval / TIMESTAMP``。
旧版 build.sh 门禁 6.4 直接对 DDL **原文** grep，于是：

* 方言关键字扫描被这段**注释说明文字**触发 → 门禁假阳性 ``die``（注释被当成真实 DDL）；
* ``CREATE TABLE`` 计数同理会把注释行 / ``COMMENT '...'`` 字符串里的字样算进表数 → 表数不可信；
* ``BIGINT ... AUTO_INCREMENT`` 计数亦被注释里的示例文字污染。

因此本扫描器统一先做「降噪」，降噪后的文本才是"会被数据库真正执行的 DDL 正文"，
所有断言只在它上面做：

1. 剥块注释 ``/* ... */``；
2. 剥行注释 ``-- ...``（SQL 标准要求 ``--`` 后跟空白或行尾）与 ``# ...``；
3. 剥字符串字面量 ``'...'`` / ``"..."``（含 ``''``/``""`` 双写转义与反斜杠转义）——
   ``COMMENT '...'`` / ``COMMENT "..."`` 属于此类，整体抹为占位 ``''``（幂等），
   ``DEFAULT 'pending'`` 同样抹为占位（不影响本门禁的任何断言）。

单遍字符扫描实现（非逐行正则），因此**不会**把字符串里的 ``--`` / ``#`` 误判成注释，
也不会把注释里的引号误判成字符串起点。

用法
----
::

    # 门禁核验（build.sh 调用形式）
    python3 scripts/ddl_gate_check.py --ddl scripts/ddl/pbls_tables.sql \
            --expect-tables 37 --expect-autoinc 37

    # 内置最小单测（无需 pytest，验证剥注释/剥串逻辑本身正确）
    python3 scripts/ddl_gate_check.py --selftest

    # 只打印降噪后的 DDL 正文（排障用）
    python3 scripts/ddl_gate_check.py --ddl <file> --dump-body

退出码：0 = 通过；1 = 断言未通过；2 = 参数/文件错误。
"""

from __future__ import annotations

import argparse
import os
import re
import sys

# --------------------------------------------------------------------------
# 期望值单一事实源：与 scripts/ddl/pbls_tables.sql 实际 DDL 对齐
#   37 表 = pbl_governance 3 + pbl_blueprint 11 + pbl_template 2 + pbl_validation 3
#           + pbl_compiler 5 + pbl_agent_runtime 4 + pbl_evidence 4 + pbl_assessment 3
#           + pbl_domain_ext 1 + pbl_analytics 1
#   （Q4 修复：36 → 37，新增 pbl_compile_task）
# --------------------------------------------------------------------------
DEFAULT_EXPECT_TABLES = 37
DEFAULT_EXPECT_AUTOINC = 37

# 非 mariadb 方言禁用关键字（大小写不敏感，与旧 grep -Eiq 行为一致）
FORBIDDEN_PATTERNS = [
    ("BIGSERIAL", r"BIGSERIAL"),
    ("SERIAL", r"(?<!BIG)SERIAL"),
    ("nextval", r"NEXTVAL\s*\("),
    ("FOREIGN KEY", r"FOREIGN\s+KEY"),
    ("REFERENCES", r"\bREFERENCES\b"),
    ("ENUM(", r"\bENUM\s*\("),
    ("TIMESTAMP", r"\bTIMESTAMP\b"),
]

RE_CREATE_TABLE = re.compile(
    r"CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?[`\"\']?([A-Za-z_][A-Za-z0-9_]*)",
    re.IGNORECASE,
)
RE_AUTOINC_PK = re.compile(
    r"\bBIGINT(?:\s*\(\s*\d+\s*\))?(?:\s+UNSIGNED)?\s+NOT\s+NULL\s+AUTO_INCREMENT",
    re.IGNORECASE,
)
RE_TABLE_OPTION_AUTOINC = re.compile(r"\bAUTO_INCREMENT\s*=\s*\d+", re.IGNORECASE)


# --------------------------------------------------------------------------
# 降噪：剥注释 + 剥字符串字面量（单遍扫描）
# --------------------------------------------------------------------------
def strip_sql_noise(sql: str) -> str:
    """剥掉 SQL 注释与字符串字面量，返回"纯 DDL 正文"。

    - ``/* ... */`` 块注释 → 单个空格
    - ``-- ...``（后跟空白/行尾）与 ``# ...`` 行注释 → 删到行尾（保留换行，行号可追）
    - ``'...'`` / ``"..."`` 字符串（含 ``COMMENT '...'``）→ 占位 ``''``
    - 反引号标识符 ``\\`col\\``` 原样保留（其中不会出现注释/串语义）
    """
    out = []
    i, n = 0, len(sql)
    while i < n:
        c = sql[i]
        nxt = sql[i + 1] if i + 1 < n else ""

        # 行注释 --（SQL 标准：-- 后须为空白或行尾，避免误伤 a--b 之类）
        if c == "-" and nxt == "-" and (i + 2 >= n or sql[i + 2] in " \t\r\n"):
            j = sql.find("\n", i)
            i = n if j < 0 else j
            continue

        # 行注释 #
        if c == "#":
            j = sql.find("\n", i)
            i = n if j < 0 else j
            continue

        # 块注释 /* ... */
        if c == "/" and nxt == "*":
            j = sql.find("*/", i + 2)
            if j < 0:
                i = n
            else:
                i = j + 2
            out.append(" ")
            continue

        # 字符串字面量 '...' / "..."（'' "" 双写转义 + 反斜杠转义）
        if c == "'" or c == '"':
            j = i + 1
            while j < n:
                if sql[j] == "\\":
                    j += 2
                    continue
                if sql[j] == c:
                    if j + 1 < n and sql[j + 1] == c:  # 双写转义
                        j += 2
                        continue
                    break
                j += 1
            # 占位符用 '' （不带前后空格）——保证降噪**幂等**：对已降噪文本再降噪结果不变。
            # 仅当紧随其后的是单词字符/反引号时补一个空格，避免 ''NOT NULL 这类粘连
            # （补空格的条件在二次降噪时已不成立，故仍幂等）。
            i = min(j + 1, n)
            out.append("''")
            if i < n and (sql[i].isalnum() or sql[i] in "_`"):
                out.append(" ")
            continue

        out.append(c)
        i += 1

    return "".join(out)


def _count_noise(sql: str) -> dict:
    """统计降噪剥离量，作为门禁回显证据。"""
    body = strip_sql_noise(sql)
    return {
        "raw_lines": sql.count("\n") + 1,
        "body_lines": body.count("\n") + 1,
        "raw_chars": len(sql),
        "body_chars": len(body),
        "stripped_chars": len(sql) - len(body),
        "block_comments": len(re.findall(r"/\*", sql)),
        "line_comments": len(re.findall(r"(?:^|[^:\w])--(?=\s|$)|(^|\s)#", sql, re.MULTILINE)),
        "string_literals": body.count("''"),
    }


# --------------------------------------------------------------------------
# 断言项
# --------------------------------------------------------------------------
def scan_tables(body: str):
    """返回 (表名列表, 重复表名列表)。"""
    names = [m.group(1) for m in RE_CREATE_TABLE.finditer(body)]
    seen, dup = set(), []
    for nm in names:
        low = nm.lower()
        if low in seen and low not in dup:
            dup.append(low)
        seen.add(low)
    return names, dup


def scan_autoinc(body: str) -> int:
    """主键自增列数（``BIGINT [UNSIGNED] NOT NULL AUTO_INCREMENT``）。

    表选项 ``) ENGINE=InnoDB AUTO_INCREMENT=1;`` 不计入。
    """
    cleaned = RE_TABLE_OPTION_AUTOINC.sub(" ", body)
    return len(RE_AUTOINC_PK.findall(cleaned))


def scan_forbidden(body: str):
    """返回 [(关键字, 行号, 该行文本)]，在降噪后的正文上扫描。"""
    hits = []
    for lineno, line in enumerate(body.split("\n"), 1):
        for label, pat in FORBIDDEN_PATTERNS:
            if re.search(pat, line, re.IGNORECASE):
                hits.append((label, lineno, line.strip()[:160]))
    return hits


# --------------------------------------------------------------------------
# 门禁主流程
# --------------------------------------------------------------------------
def run_gate(ddl_path: str, expect_tables: int, expect_autoinc: int,
             dump_body: bool = False) -> int:
    if not os.path.isfile(ddl_path):
        print("[ddl-gate][FATAL] DDL 文件不存在: %s" % ddl_path)
        return 2

    with open(ddl_path, encoding="utf-8") as fh:
        raw = fh.read()

    body = strip_sql_noise(raw)
    noise = _count_noise(raw)

    print("[ddl-gate] DDL: %s" % ddl_path)
    print("[ddl-gate] 降噪: 原始 %d 行/%d 字符 → 纯 DDL 正文 %d 行/%d 字符"
          "（剥离 %d 字符；块注释 %d 处、行注释 %d 处、字符串字面量 %d 处）"
          % (noise["raw_lines"], noise["raw_chars"], noise["body_lines"],
             noise["body_chars"], noise["stripped_chars"], noise["block_comments"],
             noise["line_comments"], noise["string_literals"]))

    raw_grep = len(re.findall(r"CREATE\s+TABLE\s+IF\s+NOT\s+EXISTS", raw, re.IGNORECASE))
    body_grep = len(re.findall(r"CREATE\s+TABLE\s+IF\s+NOT\s+EXISTS", body, re.IGNORECASE))
    print("[ddl-gate] 对照: 原文 'CREATE TABLE IF NOT EXISTS' = %d，降噪后 = %d"
          % (raw_grep, body_grep))

    if dump_body:
        print("---- 降噪后 DDL 正文 BEGIN ----")
        print(body)
        print("---- 降噪后 DDL 正文 END ----")

    ok = True

    # (1) 非 mariadb 方言关键字：降噪后必须零命中
    forbidden = scan_forbidden(body)
    if forbidden:
        ok = False
        print("[ddl-gate][FAIL] 非 mariadb 方言关键字命中 %d 处（降噪后正文）:" % len(forbidden))
        for label, lineno, text in forbidden[:20]:
            print("           L%-5d %-14s | %s" % (lineno, label, text))
    else:
        print("[ddl-gate][PASS] 非 mariadb 方言关键字零命中"
              "（BIGSERIAL/SERIAL/nextval/FOREIGN KEY/REFERENCES/ENUM(/TIMESTAMP，"
              "注释与 COMMENT 串已剥离，不再假阳性）")

    # (2) 表数断言
    names, dup = scan_tables(body)
    if len(names) != expect_tables:
        ok = False
        print("[ddl-gate][FAIL] CREATE TABLE 数 = %d，期望 %d" % (len(names), expect_tables))
    else:
        print("[ddl-gate][PASS] CREATE TABLE 数 = %d（期望 %d）" % (len(names), expect_tables))
    if dup:
        ok = False
        print("[ddl-gate][FAIL] 重复建表: %s" % ", ".join(dup))
    non_pbl = [n for n in names if not n.lower().startswith("pbl_")]
    if non_pbl:
        print("[ddl-gate][WARN] 非 pbl_ 前缀表 %d 张: %s" % (len(non_pbl), ", ".join(non_pbl)))

    # (3) BIGINT AUTO_INCREMENT 主键数断言
    autoinc = scan_autoinc(body)
    if autoinc != expect_autoinc:
        ok = False
        print("[ddl-gate][FAIL] BIGINT [UNSIGNED] NOT NULL AUTO_INCREMENT = %d，期望 %d"
              % (autoinc, expect_autoinc))
    else:
        print("[ddl-gate][PASS] BIGINT [UNSIGNED] NOT NULL AUTO_INCREMENT = %d（期望 %d）"
              % (autoinc, expect_autoinc))

    print("[ddl-gate] 表清单(%d): %s" % (len(names), ", ".join(names)))
    print("[ddl-gate] RESULT: %s" % ("PASS" if ok else "FAIL"))
    return 0 if ok else 1


# --------------------------------------------------------------------------
# 最小单测（无需 pytest；build.sh 在跑门禁前先跑它，证明扫描器本身可信）
# --------------------------------------------------------------------------
SELFTEST_DDL = """\
-- 本行注释里故意写 CREATE TABLE IF NOT EXISTS pbl_fake_comment (不应被计数)
# 井号注释里也写 CREATE TABLE pbl_fake_hash (不应被计数)
/* 块注释里写 FOREIGN KEY REFERENCES ENUM( SERIAL BIGSERIAL nextval TIMESTAMP
   以及 CREATE TABLE pbl_fake_block (均不应被命中/计数) */
CREATE TABLE IF NOT EXISTS `pbl_alpha` (
  `tenant_id` VARCHAR(64) NOT NULL COMMENT '租户ID(强制打头)',
  `id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键',
  `note` VARCHAR(200) NULL COMMENT "串里写 CREATE TABLE pbl_fake_str / TIMESTAMP / FOREIGN KEY 都不应命中",
  `status` VARCHAR(32) NOT NULL DEFAULT 'pending' COMMENT 'appcodes:pbl_x(a/b)',
  `dash` VARCHAR(32) NULL DEFAULT 'a--b # c' COMMENT '串内 -- 与 # 不是注释起点',
  PRIMARY KEY (`tenant_id`, `id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='表注释里也有 CREATE TABLE pbl_fake_tbl';

CREATE TABLE IF NOT EXISTS pbl_beta (
  tenant_id VARCHAR(64) NOT NULL COMMENT '租户ID',
  id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '主键',
  PRIMARY KEY (tenant_id, id)
) ENGINE=InnoDB AUTO_INCREMENT=1;
"""

SELFTEST_BAD_DDL = """\
CREATE TABLE IF NOT EXISTS pbl_bad (
  id BIGSERIAL PRIMARY KEY,
  ts TIMESTAMP NOT NULL,
  kind VARCHAR(8) NOT NULL DEFAULT 'x' COMMENT '枚举',
  CONSTRAINT fk_x FOREIGN KEY (id) REFERENCES pbl_alpha (id)
);
"""


def selftest() -> int:
    fails = []

    def check(name, got, want):
        if got != want:
            fails.append("%s: got=%r want=%r" % (name, got, want))
            print("[selftest][FAIL] %-46s got=%r want=%r" % (name, got, want))
        else:
            print("[selftest][PASS] %-46s = %r" % (name, got))

    body = strip_sql_noise(SELFTEST_DDL)

    # 1) 注释/COMMENT 串里的 CREATE TABLE 不被计数；真实 2 张表被计数
    names, dup = scan_tables(body)
    check("注释+COMMENT串干扰下的表数", len(names), 2)
    check("表名集合", sorted(n.lower() for n in names), ["pbl_alpha", "pbl_beta"])
    check("重复表", dup, [])

    # 2) 注释里逐字写出的方言关键字不触发假阳性
    check("注释内方言关键字假阳性", scan_forbidden(body), [])

    # 3) AUTO_INCREMENT 主键计数：2 处（表选项 AUTO_INCREMENT=1 不计）
    check("AUTO_INCREMENT 主键数", scan_autoinc(body), 2)

    # 4) 字符串里的 -- / # 不被当成注释起点（否则 dash 列后半段会被吃掉）
    check("串内 -- 未截断(列仍存在)", "dash" in body, True)
    check("串内 # 未截断(PRIMARY KEY 仍在)", "PRIMARY KEY" in body, True)

    # 5) 真实违规必须被抓到（不能因为降噪而漏报）
    bad_body = strip_sql_noise(SELFTEST_BAD_DDL)
    bad_labels = sorted({h[0] for h in scan_forbidden(bad_body)})
    check("真实违规关键字被抓到", bad_labels,
          ["BIGSERIAL", "FOREIGN KEY", "REFERENCES", "TIMESTAMP"])

    # 6) 降噪幂等（对已降噪文本再降噪不应改变结果）
    check("降噪幂等", strip_sql_noise(body), body)

    print("[selftest] RESULT: %s" % ("PASS" if not fails else "FAIL (%d)" % len(fails)))
    return 0 if not fails else 1


# --------------------------------------------------------------------------
def main(argv=None) -> int:
    ap = argparse.ArgumentParser(
        description="build.sh 门禁 6.4：剥注释/COMMENT 串后再扫 DDL（方言 + 表数 + 自增主键）")
    ap.add_argument("--ddl", help="DDL 文件路径")
    ap.add_argument("--expect-tables", type=int, default=DEFAULT_EXPECT_TABLES,
                    help="期望 CREATE TABLE 数（默认 %d）" % DEFAULT_EXPECT_TABLES)
    ap.add_argument("--expect-autoinc", type=int, default=DEFAULT_EXPECT_AUTOINC,
                    help="期望 BIGINT AUTO_INCREMENT 主键数（默认 %d）" % DEFAULT_EXPECT_AUTOINC)
    ap.add_argument("--dump-body", action="store_true", help="打印降噪后的 DDL 正文（排障）")
    ap.add_argument("--selftest", action="store_true", help="只跑内置最小单测")
    args = ap.parse_args(argv)

    if args.selftest:
        return selftest()
    if not args.ddl:
        ap.error("需要 --ddl <path>（或使用 --selftest）")
    return run_gate(args.ddl, args.expect_tables, args.expect_autoinc, args.dump_body)


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