#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""M11b-2b 表定义自检（database-table-definition-spec 机械断言）。

覆盖范围（QC #4 要求）:
  1. 全部 4 个落点都跑同一套 spec 断言：
     - modules/world_sync/models/pbl_runtime_event.json          (唯一真源)
     - modules/world_sync/models/pbl_entity_state.json           (唯一真源)
     - apps/scense/pkgs/world_sync/models/pbl_runtime_event.json (打包镜像)
     - apps/scense/pkgs/world_sync/models/pbl_entity_state.json  (打包镜像)
  2. 真源与镜像 sha256 两两相同的一致性断言（消除内容级分叉）。
  3. 各段键集合 ⊆ spec 白名单的断言（summary/fields/indexes/codes 均不得出现未定义键）。
  4. 抽象类型断言（禁止 VARCHAR/BIGINT/DATETIME(3)/JSON 等方言具体类型）。
  5. 主键 id str(32) 断言 + 与 pbl_runtime_sql.py 实际列名交叉核对。

任一断言不符即 RESULT: FAIL 且退出码 1。
用法: python3 modules/world_sync/scripts/m11b2_selftest.py

────────────────────────────────────────────────────────────────────────────
[cross-check-sql] 提取器的列名解析规则（M11b-2d-B 固化，下次结构变更照此扩展）
────────────────────────────────────────────────────────────────────────────
pbl_runtime_sql.py 里 ``EVENT_COLUMNS`` / ``STATE_COLUMNS`` 有两种合法真源形态，
提取器必须**同时**兼容，否则自检会在代码演进时假阳（M11b-2c 实跑 4 FAIL 即此因）：

  形态 A（旧·两段式手抄字面量）
      EVENT_COLUMNS = (
          "id", "event_id", ...
      )
      → 正则 ``NAME\\s*=\\s*\\((.*?)\\)`` + 引号内标识符，直接取字面量列名。

  形态 B（M11b-2b 起·四段式 models/*.json 机械派生，当前真态）
      STATE_TABLE   = "pbl_entity_state"
      STATE_COLUMNS = columns_of(STATE_TABLE)
      列名不再手抄，由 :mod:`world_sync.pbl_runtime_authority` 按
      ``models/<table>.json`` 的 ``fields[].name``（spec 四段式第三段之前的
      fields 段）有序导出，fail-closed（模型缺失/非四段式/fields 空 →
      AuthorityResolutionError，不退回旧常量）。
      解析优先级：
        B1 运行时取真值：sys.path 注入 ``modules/world_sync``，import
           ``world_sync.pbl_runtime_sql``，读模块属性（= 生产代码实际使用的列契约，
           与「手抄正则」相比更强：常量表达式怎么变都能拿到最终值）。
        B2 静态回落：import 失败（依赖/环境差异）时，按形态 B 正则解析出
           ``columns_of(<TABLE_CONST>)`` → ``<TABLE_CONST> = "<table>"`` → 读唯一真源
           ``modules/world_sync/models/<table>.json`` 的 ``fields[].name``。
      A/B 都解析不出 → 记 FAIL（不静默跳过），保证门禁不失效。

覆盖强度说明（未放宽任何断言）：形态 A 原本只断「代码列 ⊆ fields 列」单向；
形态 B 下列集由 fields 派生，单向断言退化为恒真，故补两条等价替代断言把强度
恢复到「代码契约 == 表契约」双向：
  - [cross-check-sql]      代码列 ⊆ fields 列（原有，语义不变）
  - [cross-check-sql-nocov] fields 列 ⊆ 代码列（真源不得声明代码不认的幽灵列）
  - [cross-check-sql-order] 代码列顺序 == fields 声明顺序（columns_of 有序导出契约）
  - [cross-check-sql-count] 列数与代码注释宣称的列数一致（防静默丢列）
"""

from __future__ import annotations

import hashlib
import json
import os
import re
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
MODULE_DIR = os.path.abspath(os.path.join(HERE, ".."))          # modules/world_sync
WS_ROOT = os.path.abspath(os.path.join(MODULE_DIR, "..", ".."))  # 机构工作空间根

SRC_DIR = os.path.join("modules", "world_sync", "models")
MIRROR_DIR = os.path.join("apps", "scense", "pkgs", "world_sync", "models")

TABLES = ("pbl_runtime_event", "pbl_entity_state")

# ---- spec 白名单键（database-table-definition-spec）----
ROOT_KEYS = {"summary", "fields", "indexes", "codes"}
SUMMARY_KEYS = {"name", "title", "primary", "catelog", "comment"}
FIELD_KEYS = {"name", "title", "type", "length", "dec", "nullable", "default"}
INDEX_KEYS = {"name", "idxtype", "idxfields"}
CODE_KEYS = {"field", "table", "valuefield", "textfield", "cond"}

# 抽象类型（spec 表格）
ABSTRACT_TYPES = {
    "str", "char", "short", "int", "long", "float", "double", "ddouble",
    "decimal", "date", "time", "datetime", "timestamp", "text", "bin",
}
NEEDS_LENGTH = {"str", "char", "float", "double", "ddouble", "decimal"}
NEEDS_DEC = {"float", "double", "ddouble", "decimal"}

# 真源 fields 必须覆盖的代码侧列名（pbl_runtime_sql.py）
SQL_MODULE = os.path.join("modules", "world_sync", "world_sync", "pbl_runtime_sql.py")
COLS_IN_CODE = {
    "pbl_runtime_event": "EVENT_COLUMNS",
    "pbl_entity_state": "STATE_COLUMNS",
}
# 代码注释里宣称的列数（防静默丢列；改模型时同步更新此表，不一致即 FAIL）
DECLARED_COL_COUNT = {
    "pbl_runtime_event": 25,
    "pbl_entity_state": 10,
}

results = []  # (ok: bool, line: str)
_CODE_COLS_CACHE = {}  # (tname, const) -> (cols|None, how)


def check(ok: bool, line: str) -> bool:
    results.append((bool(ok), line))
    return bool(ok)


def sha256_of(path: str) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as fh:
        for chunk in iter(lambda: fh.read(65536), b""):
            h.update(chunk)
    return h.hexdigest()


def load(path: str):
    with open(path, "r", encoding="utf-8") as fh:
        return json.load(fh)


# --------------------------------------------------------------------------
# 代码侧列名提取（解析规则见文件头「[cross-check-sql] 提取器的列名解析规则」）
# --------------------------------------------------------------------------
def _cols_from_literal(src: str, const: str):
    """形态 A：NAME = ("a", "b", ...) 手抄字面量元组。"""
    m = re.search(const + r"\s*=\s*\((.*?)\)", src, re.S)
    if not m:
        return None
    cols = re.findall(r'"([a-zA-Z_][a-zA-Z0-9_]*)"', m.group(1))
    return cols or None


def _table_from_columns_of(src: str, const: str):
    """形态 B 静态解析：NAME = columns_of(<TABLE_CONST>) → <TABLE_CONST> = "<table>"。"""
    m = re.search(const + r"\s*=\s*columns_of\(\s*([A-Za-z_][A-Za-z0-9_]*)\s*\)", src)
    if not m:
        return None
    tconst = m.group(1)
    m2 = re.search(r"\b" + re.escape(tconst) + r"\s*=\s*[\"']([a-zA-Z_][a-zA-Z0-9_]*)[\"']", src)
    return m2.group(1) if m2 else None


def _cols_from_runtime(const: str):
    """形态 B1：import 生产模块，读运行时常量真值（最贴近实际 SQL 契约）。"""
    try:
        if MODULE_DIR not in sys.path:
            sys.path.insert(0, MODULE_DIR)
        mod = __import__("world_sync.pbl_runtime_sql", fromlist=["pbl_runtime_sql"])
        cols = getattr(mod, const, None)
        if isinstance(cols, (list, tuple)) and cols:
            return [str(c) for c in cols]
    except Exception:  # noqa: BLE001 - 环境不具备时回落静态解析，不因此放宽断言
        return None
    return None


def _cols_from_authority_model(table: str):
    """形态 B2：读唯一真源 models/<table>.json 的 fields[].name（有序）。"""
    p = os.path.join(WS_ROOT, SRC_DIR, table + ".json")
    if not os.path.isfile(p):
        return None
    try:
        doc = load(p)
    except Exception:  # noqa: BLE001
        return None
    fields = doc.get("fields") or []
    cols = [str(f.get("name")) for f in fields if f.get("name")]
    return cols or None


def extract_code_columns(tname: str, const: str):
    """返回 (cols|None, how)。cols=None 表示解析失败（调用方记 FAIL，不静默跳过）。"""
    key = (tname, const)
    if key in _CODE_COLS_CACHE:
        return _CODE_COLS_CACHE[key]

    out = (None, "unresolved")
    sql_path = os.path.join(WS_ROOT, SQL_MODULE)
    if os.path.isfile(sql_path):
        src = open(sql_path, "r", encoding="utf-8").read()
        # A) 旧两段式字面量
        cols = _cols_from_literal(src, const)
        if cols:
            out = (cols, "literal-tuple(形态A)")
        else:
            # B1) 运行时真值
            cols = _cols_from_runtime(const)
            if cols:
                out = (cols, "runtime-import columns_of(形态B1)")
            else:
                # B2) 静态解析表名 → 唯一真源 models/*.json
                table = _table_from_columns_of(src, const)
                if table:
                    cols = _cols_from_authority_model(table)
                    if cols:
                        out = (cols, "static columns_of(%s) -> models/%s.json fields[]"
                               % (table, table) + "(形态B2)")
    _CODE_COLS_CACHE[key] = out
    return out


def assert_spec(rel: str) -> None:
    """对单个落点文件跑全套 spec 断言。"""
    path = os.path.join(WS_ROOT, rel)
    if not check(os.path.isfile(path), f"[exists] {rel}"):
        return
    try:
        doc = load(path)
        check(True, f"[json-parse] {rel} OK")
    except Exception as exc:  # noqa: BLE001
        check(False, f"[json-parse] {rel} FAIL: {exc}")
        return

    # --- 根键四段式 ---
    root = set(doc.keys())
    check(root <= ROOT_KEYS, f"[root-keys] {rel} ⊆ summary/fields/indexes/codes -> {sorted(root)}"
          + ("" if root <= ROOT_KEYS else f"  多余={sorted(root - ROOT_KEYS)}"))

    # --- summary: 恰好一条，键白名单 ---
    summary = doc.get("summary") or []
    check(len(summary) == 1, f"[summary-count] {rel} 恰好 1 条 -> {len(summary)}")
    if not summary:
        return
    s0 = summary[0]
    skeys = set(s0.keys())
    check(skeys <= SUMMARY_KEYS, f"[summary-keys] {rel} ⊆ {sorted(SUMMARY_KEYS)}"
          + ("" if skeys <= SUMMARY_KEYS else f"  非白名单键={sorted(skeys - SUMMARY_KEYS)}"))
    tname = os.path.basename(rel)[: -len(".json")]
    check(s0.get("name") == tname, f"[summary-name] {rel} name=={tname} -> {s0.get('name')}")
    check(bool(str(s0.get("title") or "").strip()), f"[summary-title] {rel} title 非空")
    primary = s0.get("primary")
    check(isinstance(primary, list) and primary == ["id"],
          f"[summary-primary] {rel} primary==['id'] (array) -> {primary!r}")
    check(str(s0.get("catelog") or "") in {"entity", "relation", "dimession", "indication", ""},
          f"[summary-catelog] {rel} catelog 合法 -> {s0.get('catelog')}")

    # --- fields ---
    fields = doc.get("fields") or []
    check(len(fields) >= 1, f"[fields-count] {rel} >=1 -> {len(fields)}")
    names = []
    for f in fields:
        fn = str(f.get("name") or "?")
        names.append(fn)
        tag = f"{rel}#{fn}"
        fkeys = set(f.keys())
        check(fkeys <= FIELD_KEYS, f"[field-keys] {tag} ⊆ {sorted(FIELD_KEYS)}"
              + ("" if fkeys <= FIELD_KEYS else f"  非白名单键={sorted(fkeys - FIELD_KEYS)}"))
        t = f.get("type")
        check(t in ABSTRACT_TYPES, f"[abstract-type] {tag} type 抽象 -> {t!r}")
        # 方言具体类型黑名单形态（含括号/空格的大写写法）
        check(bool(re.fullmatch(r"[a-z]+", str(t or ""))), f"[type-lower-no-dialect] {tag} -> {t!r}")
        check(bool(str(f.get("title") or "").strip()), f"[field-title] {tag} title 非空(DDL COMMENT 来源)")
        if t in NEEDS_LENGTH:
            ln = f.get("length")
            check(isinstance(ln, int) and not isinstance(ln, bool) and ln > 0,
                  f"[field-length] {tag} length 为正整数 -> {ln!r}")
        if t in NEEDS_DEC:
            dc = f.get("dec")
            check(isinstance(dc, int) and not isinstance(dc, bool) and dc > 0,
                  f"[field-dec] {tag} dec 为正整数 -> {dc!r}")
        check(str(f.get("nullable") or "yes") in {"yes", "no"}, f"[field-nullable] {tag} -> {f.get('nullable')}")
    check("id" in names, f"[field-id] {rel} 含 id 主键列")
    idf = next((f for f in fields if f.get("name") == "id"), None)
    if idf:
        check(idf.get("type") == "str" and idf.get("length") == 32,
              f"[id-str32] {rel} id type=str length=32 -> {idf.get('type')}/{idf.get('length')}")
        check(idf.get("nullable") == "no", f"[id-notnull] {rel} id nullable=no")
    check(len(names) == len(set(names)), f"[field-unique] {rel} 列名无重复")

    # --- indexes ---
    indexes = doc.get("indexes") or []
    inames = []
    for ix in indexes:
        nm = str(ix.get("name") or "?")
        inames.append(nm)
        tag = f"{rel}#{nm}"
        ikeys = set(ix.keys())
        check(ikeys <= INDEX_KEYS, f"[index-keys] {tag} ⊆ {sorted(INDEX_KEYS)}"
              + ("" if ikeys <= INDEX_KEYS else f"  非白名单键={sorted(ikeys - INDEX_KEYS)}"))
        check(ix.get("idxtype") in {"unique", "index"}, f"[index-type] {tag} -> {ix.get('idxtype')}")
        ixfs = ix.get("idxfields")
        check(isinstance(ixfs, list) and len(ixfs) > 0, f"[index-idxfields-array] {tag} -> {ixfs!r}")
        if isinstance(ixfs, list):
            unknown = [c for c in ixfs if c not in names]
            check(not unknown, f"[index-cols-exist] {tag} 索引列都在 fields 中 -> {ixfs}"
                  + ("" if not unknown else f"  缺列={unknown}"))
    check(len(inames) == len(set(inames)), f"[index-unique-name] {rel} 索引名无重复")
    # 租户隔离：所有唯一索引以 tenant_id 打头
    for ix in indexes:
        if ix.get("idxtype") == "unique":
            ixfs = ix.get("idxfields") or []
            check(bool(ixfs) and ixfs[0] == "tenant_id",
                  f"[tenant-first] {rel}#{ix.get('name')} 唯一索引 tenant_id 打头 -> {ixfs}")

    # --- 交叉核对：代码里的列名全部在册（兼容形态 A/B，见文件头解析规则）---
    const = COLS_IN_CODE.get(tname)
    if const:
        sql_path = os.path.join(WS_ROOT, SQL_MODULE)
        if not os.path.isfile(sql_path):
            check(False, f"[cross-check-sql] {rel} 代码侧真源缺失 {SQL_MODULE}")
        else:
            code_cols, how = extract_code_columns(tname, const)
            if not code_cols:
                check(False, f"[cross-check-sql] {rel} 未能解析 {const}"
                      f"（形态A字面量/形态B运行时/形态B静态 三级解析均失败）")
            else:
                missing = [c for c in code_cols if c not in names]
                check(not missing, f"[cross-check-sql] {rel} 覆盖 {const} {len(code_cols)} 列"
                      f" via {how}" + ("" if not missing else f"  缺列={missing}"))
                # 等价替代断言（四段式下把单向覆盖升级为双向契约一致，强度不降反升）
                nocov = [c for c in names if c not in set(code_cols)]
                check(not nocov, f"[cross-check-sql-nocov] {rel} fields 无代码未声明列"
                      f"（fields⊆{const}）" + ("" if not nocov else f"  幽灵列={nocov}"))
                check(list(code_cols) == list(names),
                      f"[cross-check-sql-order] {rel} {const} 顺序 == fields 声明顺序"
                      f"（columns_of 有序导出契约）")
                declared = DECLARED_COL_COUNT.get(tname)
                if declared is not None:
                    check(len(code_cols) == declared and len(names) == declared,
                          f"[cross-check-sql-count] {rel} 列数==宣称 {declared}"
                          f" -> code={len(code_cols)} fields={len(names)}")

    # --- codes ---
    codes = doc.get("codes") or []
    seen_fields = set()
    for c in codes:
        ckeys = set(c.keys())
        check(ckeys <= CODE_KEYS, f"[code-keys] {rel}#{c.get('field')} ⊆ {sorted(CODE_KEYS)}"
              + ("" if ckeys <= CODE_KEYS else f"  非白名单键={sorted(ckeys - CODE_KEYS)}"))
        check(str(c.get("table") or "") and "." not in str(c.get("table")),
              f"[code-table-no-dot] {rel}#{c.get('field')} table={c.get('table')}")
        if str(c.get("table")) == "appcodes_kv":
            check(str(c.get("cond") or "").startswith("parentid="),
                  f"[code-cond-parentid] {rel}#{c.get('field')} cond={c.get('cond')}")
        cf = c.get("field")
        check(cf not in seen_fields, f"[code-no-dup] {rel} field={cf} 不重复")
        seen_fields.add(cf)
        check(cf in names, f"[code-field-exists] {rel} field={cf} 在 fields 中")

    # 全文无方言具体类型字样（type 值层面已断言，这里兜底扫 type 键）
    dialect = [f.get("name") for f in fields
               if re.search(r"(VARCHAR|BIGINT|DATETIME\(|TIMESTAMP\(|NVARCHAR|INT4|INT8)", str(f.get("type")))]
    check(not dialect, f"[no-dialect-types] {rel} 无方言具体类型" + ("" if not dialect else f" 违规={dialect}"))


def main() -> int:
    paths = []
    for t in TABLES:
        paths.append(os.path.join(SRC_DIR, t + ".json"))
    for t in TABLES:
        paths.append(os.path.join(MIRROR_DIR, t + ".json"))

    print("== 自检落点（4 个路径，真源 + 镜像同一套 spec 断言）==")
    for p in paths:
        print("  - " + p)
    print()

    for rel in paths:
        assert_spec(rel)

    print("== 真源 vs 镜像 sha256 一致性断言 ==")
    for t in TABLES:
        src_rel = os.path.join(SRC_DIR, t + ".json")
        mir_rel = os.path.join(MIRROR_DIR, t + ".json")
        sp, mp = os.path.join(WS_ROOT, src_rel), os.path.join(WS_ROOT, mir_rel)
        if os.path.isfile(sp) and os.path.isfile(mp):
            hs, hm = sha256_of(sp), sha256_of(mp)
            print(f"  {t}: src={hs}")
            print(f"  {t}: mir={hm}")
            check(hs == hm, f"[mirror-sha256-equal] {t} 真源与镜像字节级一致 -> {hs == hm}")
        else:
            check(False, f"[mirror-sha256-equal] {t} 文件缺失，无法比对")
    print()

    print("== 代码侧列名解析方式（形态 A/B 兼容诊断）==")
    for t in TABLES:
        cols, how = extract_code_columns(t, COLS_IN_CODE[t])
        print(f"  {COLS_IN_CODE[t]}: n={len(cols) if cols else 0} via {how}")
    print()

    passed = sum(1 for ok, _ in results if ok)
    failed = [(line) for ok, line in results if not ok]
    for ok, line in results:
        print(("PASS  " if ok else "FAIL  ") + line)
    print()
    print(f"断言合计 {len(results)}：PASS {passed} / FAIL {len(failed)}")
    if failed:
        print("RESULT: FAIL")
        return 1
    print("RESULT: PASS")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())