#!/usr/bin/env python3 # -*- coding: utf-8 -*- """M1b 内容级自检工具（QC 取证用）。 用途：把 QC 退回意见 #3/#4/#6 要求的「内容级核查」变成可重复执行、可留证据的 机械检查。运行： cd modules/pbl_blueprint && python3 tools/m1b_selfcheck.py 退出码 0 = 全部通过；非 0 = 有 FAIL 项（打印明细）。 检查项： A. DDL 合规（sql/m1b_ddl.sql） A1 恰好 4 条 CREATE TABLE IF NOT EXISTS，且为 4 张目标表 A2 零 FOREIGN KEY A3 零 PostgreSQL 方言（BIGSERIAL / nextval / SERIAL） A4 零对既有基表的 DDL/DML 动词（ALTER/UPDATE/DELETE/DROP/TRUNCATE/INSERT） A5 每个 UNIQUE KEY 首列为 tenant_key（规避 NULL 不去重） A6 mariadb 方言要素齐备（InnoDB + utf8mb4 + AUTO_INCREMENT） B. 表模型合规（models/m1b/*.json） B1 4 个文件均可 json.load B2 四段式：summary(数组,含 primary) / fields(数组) / indexes(数组) / codes(数组) B3 models 与 DDL 表名一一对应（双向） B4 models 字段集合与 DDL 列集合一一对应（双向，逐表） B5 models 索引集合与 DDL 索引集合一一对应（逐表，含 unique 标记） C. CRUD 定义合规（json/m1b/*.json） C1 4 个文件均可 json.load C2 含 tblname 且指向本模块 4 表之一 C3 含 params.editable（QC#3 指出的格式偏差） C4 new/update/delete_data_url 在 params 顶层（非仅嵌套在 editable 内） C5 json/ 下不含表定义特征键（summary/fields/indexes）——即 json/ 只放 CRUD D. Python 源码内容级核查（pbl_blueprint/m1b_*.py） D1 零硬编码库名（无 DBNAME = 'xxx' / dbname = 'xxx' 字面量赋值） D2 取库名走 get_module_dbname D3 只用 sqlor 标准 API（sor.C/U/D/R/I/sqlExe/sqlPaging），无编造 API D4 全部 py_compile 通过 D5 每个 .py 有效语句数 > 0（非空壳） E. 三处同步注册核验（QC#4） E1 m1b_api.py 定义 register_m1b_routes E2 __init__.py 导出 register_m1b_routes（import 行） E3 init.py 在 load_pbl_blueprint 内引用 register_m1b_routes（env 注册/调用） E4 M1b 每个对外函数在三处齐备（实现 / __init__ 导出 / init.py 注册） F. M1a 既有代码零越权改动（QC#6） F1 git diff 中 M1a 文件（非 m1b_* / 非 models|json|sql|tests|docs/tools 的 既有 .py）只允许「追加式」改动：新增行不得删除既有 M1a 函数定义行 """ import ast import json import os import re import subprocess import sys REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) M1B_TABLES = [ "pbl_blueprint_template", "pbl_blueprint_ref", "pbl_subobject_ext", "pbl_ext_field_def", ] DDL_REL = os.path.join("sql", "m1b_ddl.sql") MODELS_REL = os.path.join("pbl_blueprint", "models", "m1b") JSON_REL = os.path.join("pbl_blueprint", "json", "m1b") # 编造的 sqlor API（module-development-spec 明确：只有 C/U/D/R/I/sqlExe） FAKE_SQLOR = [ "sqlor.save", "sor.save", "sqlor.list", "sor.list", "sqlor.one", "sor.one", "sqlor.delete", "sor.delete", "sqlor.insert", "sor.insert", "sqlor.query", "sor.query", "sqlor.update", "sor.update", "sqlor.create", "sor.create", "sqlor.select", "sor.select", "sqlor.find", "sor.find", "sqlor.get", "sor.get_one", "sor.fetch", "sqlor.fetch", ] # 允许的 sqlor 调用（白名单，用于 D3 正向取证） OK_SQLOR = re.compile(r"\bsor\.(C|U|D|R|I|sqlExe|sqlPaging|sqlorContext)\s*\(") results = [] def record(check_id, ok, detail): results.append((check_id, bool(ok), detail)) flag = "PASS" if ok else "FAIL" print("[%s] %s :: %s" % (flag, check_id, detail)) def p(*parts): return os.path.join(REPO, *parts) def read(rel): with open(p(rel), "r", encoding="utf-8") as f: return f.read() def strip_sql_comments(sql): """去掉 -- 行注释与 /* */ 块注释，避免注释文本干扰动词扫描。""" sql = re.sub(r"/\*.*?\*/", " ", sql, flags=re.S) out = [] for line in sql.splitlines(): idx = line.find("--") if idx >= 0: line = line[:idx] out.append(line) return "\n".join(out) # ---------------------------------------------------------------- A. DDL def check_ddl(): if not os.path.exists(p(DDL_REL)): record("A0", False, "sql/m1b_ddl.sql 不存在") return None raw = read(DDL_REL) body = strip_sql_comments(raw) creates = re.findall( r"CREATE\s+TABLE\s+IF\s+NOT\s+EXISTS\s+`?(\w+)`?", body, flags=re.I) record("A1", len(creates) == 4 and sorted(creates) == sorted(M1B_TABLES), "CREATE TABLE IF NOT EXISTS 共 %d 条，表=%s" % (len(creates), creates)) fk = len(re.findall(r"FOREIGN\s+KEY", body, flags=re.I)) record("A2", fk == 0, "FOREIGN KEY 出现 %d 次（应为 0）" % fk) pg = re.findall(r"\b(BIGSERIAL|SERIAL|nextval)\b", body, flags=re.I) record("A3", not pg, "PostgreSQL 方言命中 %s（应为空）" % (pg or "无")) # A4：任何针对既有基表的 DDL/DML 动词；同时全文（去注释）不得出现这些动词 verbs = re.findall( r"\b(ALTER|UPDATE|DELETE|DROP|TRUNCATE|INSERT|REPLACE)\b", body, flags=re.I) base_hits = re.findall( r"\b(CREATE|ALTER|UPDATE|DELETE|DROP|TRUNCATE|INSERT)\b[^;]{0,120}?" r"\b(world|scene|entity|script|world_snapshot|scense)\b", body, flags=re.I) record("A4", not verbs and not base_hits, "去注释后 DDL/DML 动词命中 %s；基表语句命中 %s（均应为空）" % (sorted(set(v.upper() for v in verbs)) or "无", base_hits or "无")) # A5：每个 UNIQUE KEY 首列必须是 tenant_key bad_uk = [] uks = re.findall(r"UNIQUE\s+KEY\s+`?(\w+)`?\s*\(([^)]*)\)", body, flags=re.I) for name, cols in uks: first = cols.split(",")[0].strip().strip("`").strip() if first != "tenant_key": bad_uk.append("%s(首列=%s)" % (name, first)) record("A5", uks and not bad_uk, "UNIQUE KEY 共 %d 个，首列非 tenant_key 的：%s" % (len(uks), bad_uk or "无")) need = ["ENGINE=InnoDB", "utf8mb4", "AUTO_INCREMENT"] miss = [n for n in need if n.lower() not in body.lower()] record("A6", not miss, "mariadb 方言要素缺失 %s（应为空）" % (miss or "无")) return body # ------------------------------------------------------- B. models 四段式 def parse_ddl_tables(body): """从 DDL 正文解析 {table: {"cols": [...], "indexes": {name: (cols, unique)}}}""" tables = {} for m in re.finditer( r"CREATE\s+TABLE\s+IF\s+NOT\s+EXISTS\s+`?(\w+)`?\s*\((.*?)\n\)\s*ENGINE", body, flags=re.I | re.S): tname, inner = m.group(1), m.group(2) cols, idx = [], {} for line in inner.splitlines(): s = line.strip().rstrip(",").strip() if not s: continue up = s.upper() if up.startswith("PRIMARY KEY"): continue if up.startswith("UNIQUE KEY") or up.startswith("KEY ") or up.startswith("INDEX "): nm = re.match(r"(?:UNIQUE\s+)?(?:KEY|INDEX)\s+`?(\w+)`?\s*\(([^)]*)\)", s, flags=re.I) if nm: cs = [c.strip().strip("`") for c in nm.group(2).split(",")] idx[nm.group(1)] = (cs, up.startswith("UNIQUE")) continue cm = re.match(r"`(\w+)`", s) if cm: cols.append(cm.group(1)) tables[tname] = {"cols": cols, "indexes": idx} return tables def check_models(ddl_tables): mdir = p(MODELS_REL) if not os.path.isdir(mdir): record("B1", False, "models/m1b/ 目录不存在") return files = sorted(f for f in os.listdir(mdir) if f.endswith(".json")) record("B1", len(files) == 4, "models/m1b/ JSON 文件 %d 个：%s" % (len(files), files)) model_tables = {} four_ok, four_bad = True, [] for fn in files: try: with open(os.path.join(mdir, fn), "r", encoding="utf-8") as f: d = json.load(f) except Exception as e: four_ok = False four_bad.append("%s: json.load 失败 %s" % (fn, e)) continue for seg in ("summary", "fields", "indexes", "codes"): if not isinstance(d.get(seg), list): four_ok = False four_bad.append("%s: 缺 %s 数组" % (fn, seg)) sm = d.get("summary") or [] if not (sm and isinstance(sm[0], dict) and sm[0].get("primary")): four_ok = False four_bad.append("%s: summary[0].primary 缺失" % fn) if sm: model_tables[sm[0].get("table")] = d record("B2", four_ok, "四段式(summary/fields/indexes/codes) 违规：%s" % (four_bad or "无")) only_model = sorted(set(model_tables) - set(ddl_tables)) only_ddl = sorted(set(ddl_tables) - set(model_tables)) record("B3", not only_model and not only_ddl, "models↔DDL 表名对应：仅 models 有 %s，仅 DDL 有 %s（均应为空）" % (only_model or "无", only_ddl or "无")) col_bad, idx_bad = [], [] for t, d in model_tables.items(): if t not in ddl_tables: continue mcols = [f.get("name") for f in d.get("fields", [])] dcols = ddl_tables[t]["cols"] if sorted(mcols) != sorted(dcols): col_bad.append("%s: 仅models=%s 仅DDL=%s" % (t, sorted(set(mcols) - set(dcols)), sorted(set(dcols) - set(mcols)))) midx = {} for ix in d.get("indexes", []): midx[ix.get("name")] = (ix.get("fields"), bool(ix.get("unique"))) didx = ddl_tables[t]["indexes"] if sorted(midx) != sorted(didx): idx_bad.append("%s: 索引名不一致 仅models=%s 仅DDL=%s" % (t, sorted(set(midx) - set(didx)), sorted(set(didx) - set(midx)))) else: for k in midx: if list(midx[k][0]) != list(didx[k][0]) or midx[k][1] != didx[k][1]: idx_bad.append("%s.%s: models=%s DDL=%s" % (t, k, midx[k], didx[k])) record("B4", not col_bad, "models↔DDL 列一一对应违规：%s" % (col_bad or "无")) record("B5", not idx_bad, "models↔DDL 索引一一对应违规：%s" % (idx_bad or "无")) # ---------------------------------------------------------- C. json CRUD def check_crud(): jdir = p(JSON_REL) if not os.path.isdir(jdir): record("C1", False, "json/m1b/ 目录不存在") return files = sorted(f for f in os.listdir(jdir) if f.endswith(".json")) record("C1", len(files) == 4, "json/m1b/ JSON 文件 %d 个：%s" % (len(files), files)) bad_tbl, bad_edit, bad_url, bad_model = [], [], [], [] for fn in files: try: with open(os.path.join(jdir, fn), "r", encoding="utf-8") as f: d = json.load(f) except Exception as e: bad_tbl.append("%s: json.load 失败 %s" % (fn, e)) continue if d.get("tblname") not in M1B_TABLES: bad_tbl.append("%s: tblname=%r" % (fn, d.get("tblname"))) prm = d.get("params") or {} if not isinstance(prm.get("editable"), dict): bad_edit.append("%s: 缺 params.editable" % fn) for k in ("new_data_url", "update_data_url", "delete_data_url"): if not prm.get(k): bad_url.append("%s: params 顶层缺 %s" % (fn, k)) for seg in ("summary", "fields", "indexes"): if seg in d: bad_model.append("%s: 含表定义段 %s（json/ 只放 CRUD）" % (fn, seg)) record("C2", not bad_tbl, "tblname 违规：%s" % (bad_tbl or "无")) record("C3", not bad_edit, "params.editable 违规：%s" % (bad_edit or "无")) record("C4", not bad_url, "params 顶层 *_data_url 违规：%s" % (bad_url or "无")) record("C5", not bad_model, "json/ 混入表定义：%s" % (bad_model or "无")) # ------------------------------------------------------- D. Python 源码 def m1b_py_files(): pkg = p("pbl_blueprint") return sorted(f for f in os.listdir(pkg) if f.startswith("m1b_") and f.endswith(".py")) def check_python(): files = m1b_py_files() if not files: record("D1", False, "未找到 pbl_blueprint/m1b_*.py") return {} hard, no_dbname, fake, compile_bad, empty = [], [], [], [], [] funcs = {} hard_re = re.compile( r"""^\s*(?:DBNAME|dbname|db_name|DATABASE|database)\s*=\s*['"][^'"]+['"]""", re.M) for fn in files: src = read(os.path.join("pbl_blueprint", fn)) if hard_re.search(src): hard.append(fn) if "get_module_dbname" not in src and "dbname" in src.lower(): no_dbname.append(fn) for api in FAKE_SQLOR: if api in src: fake.append("%s: %s" % (fn, api)) try: compile(src, fn, "exec") except SyntaxError as e: compile_bad.append("%s: %s" % (fn, e)) try: tree = ast.parse(src) except SyntaxError: continue stmts = [n for n in ast.walk(tree) if isinstance(n, ast.stmt) and not isinstance(n, (ast.Pass,))] if len(stmts) < 3: empty.append("%s: 有效语句 %d" % (fn, len(stmts))) for n in tree.body: if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)): funcs.setdefault(n.name, []).append(fn) record("D1", not hard, "硬编码库名文件：%s（应为空）" % (hard or "无")) record("D2", not no_dbname, "未走 get_module_dbname 的文件：%s" % (no_dbname or "无")) record("D3", not fake, "编造 sqlor API：%s（应为空；白名单 sor.C/U/D/R/I/sqlExe/sqlPaging）" % (fake or "无")) record("D4", not compile_bad, "py_compile 失败：%s" % (compile_bad or "无")) record("D5", not empty, "疑似空壳文件：%s" % (empty or "无")) return funcs # --------------------------------------------------- E. 三处同步注册 def check_registration(funcs): api = read(os.path.join("pbl_blueprint", "m1b_api.py")) if os.path.exists( p("pbl_blueprint", "m1b_api.py")) else "" initpkg = read(os.path.join("pbl_blueprint", "__init__.py")) initpy = read(os.path.join("pbl_blueprint", "init.py")) record("E1", "def register_m1b_routes" in api, "m1b_api.py 定义 register_m1b_routes：%s" % ("是" if "def register_m1b_routes" in api else "否")) record("E2", "register_m1b_routes" in initpkg, "__init__.py 导出 register_m1b_routes：%s" % ("是" if "register_m1b_routes" in initpkg else "否")) record("E3", "register_m1b_routes" in initpy, "init.py 引用/注册 register_m1b_routes：%s" % ("是" if "register_m1b_routes" in initpy else "否")) # E4：M1b 对外函数（m1b_*.py 顶层定义）三处齐备 # ① 实现：m1b_*.py 中 def/async def # ② 导出：__init__.py 出现该名（from .m1b_x import ... 或 __all__） # ③ 注册：init.py 出现该名（import + env.xxx= 赋值 或 直接调用） PUBLIC_PREFIX = ("m1b_", "init_m1b", "register_m1b", "ensure_m1b", "load_m1b", "seed_platform", "ensure_platform") public = sorted(n for n in funcs if n.startswith(PUBLIC_PREFIX)) miss_exp = [n for n in public if not re.search(r"\b%s\b" % re.escape(n), initpkg)] miss_reg = [n for n in public if not re.search(r"\b%s\b" % re.escape(n), initpy)] # 注册形态取证：env.xxx = 赋值 或 函数调用 或 import 行 reg_form = {} for n in public: forms = [] if re.search(r"env\.%s\s*=" % re.escape(n), initpy): forms.append("env赋值") if re.search(r'["\']%s["\']' % re.escape(n), initpy) and \ re.search(r"M1B_ENV_EXPORTS|setattr\(env", initpy): forms.append("M1B_ENV_EXPORTS声明式env注册") if re.search(r"\b%s\s*\(" % re.escape(n), initpy): forms.append("调用") if re.search(r"^\s*from\s+\.m1b_\w*\s+import[^\n]*\b%s\b" % re.escape(n), initpy, re.M) or re.search(r"^\s*from\s+\.\s+import[^\n]*\b%s\b" % re.escape(n), initpy, re.M): forms.append("import") reg_form[n] = forms or ["未注册"] record("E4", not miss_exp and not miss_reg, "M1b 对外函数 %d 个；__init__ 未导出 %s；init.py 未注册 %s" % (len(public), miss_exp or "无", miss_reg or "无")) for n in public: record("E4.%s" % n, reg_form[n] != ["未注册"], "init.py 注册形态=%s" % "/".join(reg_form[n])) def check_m1a_untouched(): """F1/F2：M1b 改动不得越权破坏 M1a 既有代码。 界定： * M1a 业务文件 = pbl_blueprint/ 下非 m1b_* 的 .py，且排除共享入口 __init__.py / init.py（这两个文件是 module-development-spec 规定的 「三处同步」注册点，M1b 必须在此追加导出/注册，属合规追加而非越权）。 * 判定：M1a 业务文件不得被修改/删除（git diff 无 M/D）； 共享入口只允许「追加式」改动——既有 def 行零删除。 """ SHARED = ("__init__.py", "init.py") try: out = subprocess.run(["git", "diff", "--name-status", "HEAD", "--", "pbl_blueprint/"], cwd=REPO, capture_output=True, text=True, timeout=30) lines = [l for l in out.stdout.splitlines() if l.strip()] except Exception as e: lines = [] print(" (git diff 不可用：%s)" % e) m1a_biz_bad, shared_del = [], [] for l in lines: parts = l.split("\t") if len(parts) < 2: continue st, path = parts[0], parts[-1] base = os.path.basename(path) if base.startswith("m1b_") or "/m1b/" in path or not base.endswith(".py"): continue if base in SHARED: # 共享入口：只查「既有 def 行是否被删除」 try: d = subprocess.run(["git", "diff", "-U0", "HEAD", "--", path], cwd=REPO, capture_output=True, text=True, timeout=30) removed_defs = [x for x in d.stdout.splitlines() if x.startswith("-") and not x.startswith("---") and re.match(r"-\s*(async\s+)?def\s+\w+", x)] if removed_defs: shared_del.append("%s 删除既有函数 %d 个: %s" % (base, len(removed_defs), removed_defs[:3])) except Exception: pass continue if st.startswith("M") or st.startswith("D"): m1a_biz_bad.append("%s %s" % (st, path)) # F2 静态核验：M1a 关键入口与 11 表定义仍在 initpy = read(os.path.join("pbl_blueprint", "init.py")) initpkg = read(os.path.join("pbl_blueprint", "__init__.py")) m1a_ok = ("load_pbl_blueprint" in initpy and "load_pbl_blueprint" in initpkg and "TABLE_NAMES" in initpy) record("F1", not m1a_biz_bad and not shared_del and m1a_ok, "M1a 业务文件被改/删：%s；共享入口删除既有函数：%s；" "load_pbl_blueprint+TABLE_NAMES 完好：%s" % (m1a_biz_bad or "无", shared_del or "无", m1a_ok)) def main(): print("=" * 72) print("M1b 内容级自检 repo=%s" % REPO) print("=" * 72) body = check_ddl() ddl_tables = parse_ddl_tables(body) if body else {} check_models(ddl_tables) check_crud() funcs = check_python() check_registration(funcs) check_m1a_untouched() total = len(results) passed = sum(1 for _, ok, _ in results if ok) print("-" * 72) print("结果：%d/%d PASS" % (passed, total)) for cid, ok, detail in results: if not ok: print(" FAIL %s :: %s" % (cid, detail)) print("-" * 72) return 0 if passed == total else 1 if __name__ == "__main__": sys.exit(main())