#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""M11b-2b 自测：表定义 JSON 符合 database-table-definition-spec（四段式内联口径）。

口径变更（M11b-2b-B1）：world_sync/models/pbl_runtime_event.json、pbl_entity_state.json
已由「key=schema 引用式」改为**四段式内联**（summary/fields/indexes/codes），列集与权威真源
modules/pbl_runtime_ext/models/*.json 内容级对齐，由 m11b2b_column_diff.py 做双向列集 diff 留证，
不再靠 authority_path 运行时引用。本脚本断言口径随之更新。

覆盖全部 4 个落点（world_sync 真源 2 + 应用打包镜像 2），任一不符即 RESULT: FAIL。

断言集（判定项，计入 RESULT）：
  A1 JSON 可解析且根为对象；
  A2 根键 == {"summary","fields","indexes","codes"}（四段式，不再带 key/schema 引用外壳）；
  A3 fields 段非空（四段式产物必须自带列结构体，列集内联落地；旧「不得自带结构体」检查废除）；
  A4 fields[]/indexes[]/codes[] 逐条键集合 ⊆ spec 白名单
     （SPEC_FIELD_KEYS / SPEC_INDEX_KEYS / SPEC_CODE_KEYS；禁止把私有元信息藏进标准结构体条目里）；
  A5 summary[0].name 非空且 == 文件名去 .json（表名与文件名一致；旧 authority_module/
     authority_path/table 必填检查随引用式口径一并废除）；
  A7 真源与镜像 sha256 两两相同（禁止内容级分叉）。

U 段（上游体检项，不计入本任务 RESULT，但逐条打印 DEFECT 供冒泡留证）：
  A6 权威真源 modules/pbl_runtime_ext/models/*.json 自身对 spec 四段式白名单/抽象类型的符合性。
     —— 该侧为 M11b-1 已批准产物，本任务按 PM 裁决「不得反向修改」，故只体检登记不判定。

信息性打印（不计入判定）：summary 段条目上超出 SPEC_SUMMARY_KEYS 的表级元信息键
（owner/milestone/decision_ref/decision_basis/structure_note/consumers/mirror_note 等）
只打印 info 提示——表级元信息允许登记在 summary 记录上，不作为失败判定。
"""

import hashlib
import json
import os
import sys

SOURCE_TABLES = [
    "pbl_runtime_event.json",
    "pbl_entity_state.json",
]

# 落点 1：world_sync 真源（modules 侧，唯一手改入口）
MODULE_SRC_DIR = "modules/world_sync/models"
# 落点 2：应用打包镜像（只由 sync_models_mirror.py 生成，禁止手工双写）
APP_MIRROR_DIR = os.path.join("apps", "scense", "pkgs", "world_sync", "models")
# 权威真源（pbl_runtime_ext，M11b-1 已批准产物，本任务只读体检）
AUTHORITY_DIR = "modules/pbl_runtime_ext/models"

# database-table-definition-spec 四段式与各段键白名单
SPEC_ROOT_KEYS = {"summary", "fields", "indexes", "codes"}
SPEC_SUMMARY_KEYS = {"name", "title", "primary", "catelog"}
SPEC_FIELD_KEYS = {"name", "title", "type", "length", "dec", "nullable", "default"}
SPEC_INDEX_KEYS = {"name", "idxtype", "idxfields"}
SPEC_CODE_KEYS = {"field", "table", "valuefield", "textfield", "cond"}

ABSTRACT_TYPES = {
    "str", "char", "short", "int", "long", "float", "double", "ddouble",
    "decimal", "date", "time", "datetime", "timestamp", "text", "bin",
}
NEED_LENGTH = {"str", "char", "float", "double", "ddouble", "decimal"}
NEED_DEC = {"float", "double", "ddouble", "decimal"}

_failures = []          # 判定项失败
_upstream_defects = []  # 上游体检项缺陷（登记不判定）
_assert_count = 0
_upstream_count = 0


def check(cond, label):
    global _assert_count
    _assert_count += 1
    if not cond:
        _failures.append(label)
        print("  FAIL  %s" % label)
    else:
        print("  ok    %s" % label)
    return bool(cond)


def check_upstream(cond, label):
    global _upstream_count
    _upstream_count += 1
    if not cond:
        _upstream_defects.append(label)
        print("  DEFECT  %s" % label)
    else:
        print("  ok      %s" % label)
    return bool(cond)


def find_root():
    here = os.path.dirname(os.path.abspath(__file__))
    cur = here
    for _ in range(10):
        if os.path.isdir(os.path.join(cur, "modules")) and os.path.isdir(os.path.join(cur, "apps")):
            return cur
        parent = os.path.dirname(cur)
        if parent == cur:
            break
        cur = parent
    return os.path.abspath(os.path.join(here, os.pardir, os.pardir, os.pardir))


def sha256_of(path):
    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_json(path):
    with open(path, "r", encoding="utf-8") as fh:
        return json.load(fh)


def as_list(val):
    return val if isinstance(val, list) else []


def stem_of(path):
    base = os.path.basename(path)
    return base[:-5] if base.endswith(".json") else base


def validate_spec(path):
    """A1~A5：world_sync 侧四段式表定义（真源与镜像执行同一套断言）。"""
    print("== spec(inline four-section): %s" % path)
    try:
        data = load_json(path)
    except Exception as exc:  # noqa: BLE001
        check(False, "A1 json-parseable (%s)" % exc)
        return None
    if not check(isinstance(data, dict), "A1 json-is-object"):
        return None

    # --- A2 根键必须恰好是四段式（引用式外壳 key/schema 视为不合规）---
    check(set(data.keys()) == SPEC_ROOT_KEYS,
          "A2 root-keys == {summary,fields,indexes,codes} (got %s)" % sorted(data.keys()))

    summary = as_list(data.get("summary"))
    fields = as_list(data.get("fields"))
    indexes = as_list(data.get("indexes"))
    codes = as_list(data.get("codes"))

    check(len(summary) >= 1, "A2 summary non-empty array (got %d)" % len(summary))
    check(isinstance(data.get("fields"), list), "A2 fields is array")
    check(isinstance(data.get("indexes"), list), "A2 indexes is array")
    check(isinstance(data.get("codes"), list), "A2 codes is array")

    # --- A3 列结构体内联落地：fields 必须非空（旧口径「引用体内不含结构体」检查已废除）---
    if not check(len(fields) > 0, "A3 fields non-empty (inline column set, got %d)" % len(fields)):
        return data

    # --- A4 各段条目键集合 ⊆ spec 白名单 ---
    for s_i, s_rec in enumerate(summary):
        if not isinstance(s_rec, dict):
            check(False, "A4 summary[%d] is object" % s_i)
            continue
        extra = set(s_rec.keys()) - SPEC_SUMMARY_KEYS
        if extra:
            print("  info  summary[%d] table-level meta keys (not judged): %s" % (s_i, sorted(extra)))

    field_names = []
    for f in fields:
        if not isinstance(f, dict):
            check(False, "A4 field entry is object (%r)" % f)
            continue
        fname = str(f.get("name", "?"))
        field_names.append(fname)
        extra = set(f.keys()) - SPEC_FIELD_KEYS
        check(not extra, "A4[%s] field keys subset whitelist (extra=%s)" % (fname, sorted(extra)))
        check(fname != "?", "A4[%s] field.name present" % fname)

    for idx in indexes:
        if not isinstance(idx, dict):
            check(False, "A4 index entry is object (%r)" % idx)
            continue
        iname = str(idx.get("name", "?"))
        extra = set(idx.keys()) - SPEC_INDEX_KEYS
        check(not extra, "A4[%s] index keys subset whitelist (extra=%s)" % (iname, sorted(extra)))

    for code in codes:
        if not isinstance(code, dict):
            check(False, "A4 code entry is object (%r)" % code)
            continue
        extra = set(code.keys()) - SPEC_CODE_KEYS
        check(not extra, "A4[code:%s] keys subset whitelist (extra=%s)" % (code.get("field", "?"), sorted(extra)))

    # 列集自洽性（附加判定：内联后本文件自身必须可独立建表）
    for idx in indexes:
        if not isinstance(idx, dict):
            continue
        iname = str(idx.get("name", "?"))
        for ifld in as_list(idx.get("idxfields")):
            check(ifld in field_names, "A4[%s] idxfield present in fields: %s" % (iname, ifld))
        check(idx.get("idxtype") in ("unique", "index"), "A4[%s] idxtype unique|index (%r)" % (iname, idx.get("idxtype")))

    for f in fields:
        if not isinstance(f, dict):
            continue
        fname = str(f.get("name", "?"))
        t = f.get("type")
        check(t in ABSTRACT_TYPES, "A3[%s] abstract type (%r)" % (fname, t))
        if t in NEED_LENGTH:
            check(isinstance(f.get("length"), int) and f.get("length") > 0, "A3[%s] positive int length" % fname)
        if t in NEED_DEC:
            check(isinstance(f.get("dec"), int) and f.get("dec") > 0, "A3[%s] positive int dec" % fname)
        check(str(f.get("nullable", "no")) in ("yes", "no"), "A3[%s] nullable in yes|no" % fname)

    # --- A5 表名一致性：summary[0].name 非空且 == 文件名去 .json（旧引用字段检查已废除）---
    expected_name = stem_of(path)
    if not check(len(summary) >= 1 and isinstance(summary[0], dict), "A5 summary[0] is object"):
        return data
    s0_name = str(summary[0].get("name", "")).strip()
    check(bool(s0_name), "A5 summary[0].name non-empty")
    check(s0_name == expected_name,
          "A5 summary[0].name == filename stem (%r vs %r)" % (s0_name, expected_name))
    return data


def validate_authority(path):
    """A6（上游体检）：pbl_runtime_ext 权威真源对 spec 的符合性，只登记不判定。"""
    print("== authority(upstream audit): %s" % path)
    if not os.path.isfile(path):
        check_upstream(False, "A6 authority file present: %s" % path)
        return
    try:
        data = load_json(path)
    except Exception as exc:  # noqa: BLE001
        check_upstream(False, "A6 json-parseable (%s)" % exc)
        return
    check_upstream(isinstance(data, dict), "A6 root-is-object")
    check_upstream(set(data.keys()) <= SPEC_ROOT_KEYS,
                   "A6 root-keys subset of four-section (got %s)" % sorted(data.keys()))

    summary = data.get("summary") or []
    check_upstream(len(summary) == 1, "A6 summary has exactly one record (got %d)" % len(summary))
    if summary and isinstance(summary[0], dict):
        s0 = summary[0]
        check_upstream(set(s0.keys()) <= SPEC_SUMMARY_KEYS,
                       "A6 summary keys subset whitelist (got %s)" % sorted(s0.keys()))
        check_upstream(bool(s0.get("name")), "A6 summary.name non-empty")
        check_upstream(isinstance(s0.get("primary"), list) and len(s0.get("primary") or []) > 0,
                       "A6 summary.primary is non-empty array")

    fields = data.get("fields") or []
    check_upstream(len(fields) > 0, "A6 fields non-empty")
    names = []
    for f in fields:
        if not isinstance(f, dict):
            check_upstream(False, "A6 field-is-object")
            continue
        fname = f.get("name", "?")
        names.append(fname)
        check_upstream(set(f.keys()) <= SPEC_FIELD_KEYS,
                       "A6[%s] field keys subset whitelist (got %s)" % (fname, sorted(f.keys())))
        t = f.get("type")
        check_upstream(t in ABSTRACT_TYPES, "A6[%s] abstract type (%r)" % (fname, t))
        if t in NEED_LENGTH:
            check_upstream(isinstance(f.get("length"), int) and f.get("length") > 0,
                           "A6[%s] positive int length" % fname)
        if t in NEED_DEC:
            check_upstream(isinstance(f.get("dec"), int) and f.get("dec") > 0, "A6[%s] positive int dec" % fname)
        check_upstream(str(f.get("nullable", "no")) in ("yes", "no"), "A6[%s] nullable in yes|no" % fname)

    for idx in (data.get("indexes") or []):
        if not isinstance(idx, dict):
            check_upstream(False, "A6 index-is-object")
            continue
        iname = idx.get("name", "?")
        check_upstream(set(idx.keys()) <= SPEC_INDEX_KEYS,
                       "A6[%s] index keys subset whitelist (got %s)" % (iname, sorted(idx.keys())))
        check_upstream(isinstance(idx.get("idxfields"), list) and len(idx.get("idxfields") or []) > 0,
                       "A6[%s] idxfields non-empty array" % iname)
        check_upstream(idx.get("idxtype") in ("unique", "index"), "A6[%s] idxtype unique|index" % iname)
        for ifld in (idx.get("idxfields") or []):
            check_upstream(ifld in names, "A6[%s] idxfield exists in fields: %s" % (iname, ifld))

    for code in (data.get("codes") or []):
        if not isinstance(code, dict):
            check_upstream(False, "A6 code-is-object")
            continue
        check_upstream(set(code.keys()) <= SPEC_CODE_KEYS,
                       "A6 code keys subset whitelist (got %s)" % sorted(code.keys()))
        if code.get("table") == "appcodes_kv":
            check_upstream("parentid=" in str(code.get("cond", "")), "A6 appcodes_kv cond uses parentid=")
        check_upstream("." not in str(code.get("table", "")), "A6 codes.table has no dot notation")


def main():
    root = find_root()
    print("workspace root: %s" % root)
    paths = []
    for name in SOURCE_TABLES:
        paths.append(os.path.join(root, MODULE_SRC_DIR, name))
        paths.append(os.path.join(root, APP_MIRROR_DIR, name))

    print("\n--- A1~A5: 4 个落点全部执行同一套四段式 spec 断言 ---")
    for p in paths:
        if not os.path.isfile(p):
            check(False, "file exists: %s" % p)
            continue
        validate_spec(p)

    print("\n--- A7: 真源与镜像 sha256 两两相同 ---")
    for name in SOURCE_TABLES:
        src = os.path.join(root, MODULE_SRC_DIR, name)
        dst = os.path.join(root, APP_MIRROR_DIR, name)
        if os.path.isfile(src) and os.path.isfile(dst):
            hs, hd = sha256_of(src), sha256_of(dst)
            print("  %s\n    src=%s\n    dst=%s" % (name, hs, hd))
            check(hs == hd, "A7 sha256 equal: %s" % name)
        else:
            check(False, "A7 both files present for sha256: %s" % name)

    print("\n--- A6: 权威真源（pbl_runtime_ext）上游体检，只登记不判定 ---")
    for name in SOURCE_TABLES:
        validate_authority(os.path.join(root, AUTHORITY_DIR, name))

    print("\njudged assertions executed: %d" % _assert_count)
    print("upstream audit checks executed: %d" % _upstream_count)
    if _upstream_defects:
        print("UPSTREAM DEFECTS (%d) -> 已登记，需冒泡 agent.pm / pbl_runtime_ext owner：" % len(_upstream_defects))
        for d in _upstream_defects:
            print("  * %s" % d)
    if _failures:
        print("FAILURES (%d):" % len(_failures))
        for f in _failures:
            print("  - %s" % f)
        print("RESULT: FAIL")
        return 1
    print("RESULT: PASS")
    return 0


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