#!/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 """ 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", } results = [] # (ok: bool, line: str) 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) 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}") # 交叉核对：代码里的列名全部在册 sql_path = os.path.join(WS_ROOT, SQL_MODULE) if os.path.isfile(sql_path) and tname in COLS_IN_CODE: src = open(sql_path, "r", encoding="utf-8").read() m = re.search(COLS_IN_CODE[tname] + r"\s*=\s*\((.*?)\)", src, re.S) if m: code_cols = re.findall(r'"([a-zA-Z_][a-zA-Z0-9_]*)"', m.group(1)) missing = [c for c in code_cols if c not in names] check(not missing, f"[cross-check-sql] {rel} 覆盖 {COLS_IN_CODE[tname]} {len(code_cols)} 列" + ("" if not missing else f" 缺列={missing}")) else: check(False, f"[cross-check-sql] {rel} 未能解析 {COLS_IN_CODE[tname]}") # --- 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() 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())