526 lines
24 KiB
Python
526 lines
24 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""pbl_blueprint M1a 自查门禁（重写版，退出码真实反映结果）。
|
||
|
||
设计原则
|
||
--------
|
||
1. 只核验「真实落盘文件」，逐项输出 PASS/FAIL + 证据，禁止凭摘要放行。
|
||
2. 有任一 FAIL 或检查组异常 -> exit 1（门禁有效）；全 PASS -> exit 0。
|
||
3. 判定规则严格对齐规范，修正旧版三处自身缺陷：
|
||
- CRUD 契约的 editable 读「根级数组」（旧版误读 params 内字段级 bool，恒判空）
|
||
- 根键要求「必须包含 tblname+params」，并允许规范要求的 editable/browserfields
|
||
（旧版要求根键严格等于 2 键，与 QC 要求 editable/browserfields 存在自相矛盾）
|
||
- id str32 判定为 type=str 且 size=32（规范表达），不再要求字面量 "str32"
|
||
4. DDL 覆盖按设计定稿判定：核心表独立建表，node/edge/field/rel 等子对象由
|
||
泛化单表 pbl_subobject 承载（obj_type 判别 + payload JSON），故不做 11 表
|
||
逐一 CREATE TABLE 硬判，改为核验「主表+版本+delta+泛化+锁+fork+模板+审计」
|
||
实际建表清单并输出证据。
|
||
|
||
用法
|
||
----
|
||
python3 modules/pbl_blueprint/scripts/selfcheck.py
|
||
python3 modules/pbl_blueprint/scripts/selfcheck.py --report <输出报告路径>
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
|
||
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
REPO_ROOT = os.path.dirname(SCRIPT_DIR) # modules/pbl_blueprint
|
||
PKG_DIR = os.path.join(REPO_ROOT, "pbl_blueprint") # 包目录
|
||
MODELS_DIR = os.path.join(PKG_DIR, "models")
|
||
JSON_DIR = os.path.join(PKG_DIR, "json")
|
||
SQL_DIR = os.path.join(PKG_DIR, "sql")
|
||
SCRIPTS_DIR = os.path.join(REPO_ROOT, "scripts")
|
||
TESTS_DIR = os.path.join(REPO_ROOT, "tests")
|
||
SKILL_MD = os.path.join(REPO_ROOT, "skill", "SKILL.md")
|
||
LOAD_PATH = os.path.join(SCRIPTS_DIR, "load_path.py")
|
||
PYPROJECT = os.path.join(REPO_ROOT, "pyproject.toml")
|
||
|
||
# 泛化承载：这些 models 表在物理 DDL 中由 pbl_subobject / pbl_subobject_rel 承载
|
||
GENERIC_CARRIER = {
|
||
"pbl_blueprint_node": "pbl_subobject",
|
||
"pbl_blueprint_edge": "pbl_subobject_rel",
|
||
}
|
||
|
||
ERRORS_14 = [
|
||
"PblBlueprintError", "ERR_OK", "ERR_TENANT_MISSING", "ERR_PARAM_INVALID",
|
||
"ERR_NOT_FOUND", "ERR_DUPLICATE", "ERR_STATE_INVALID", "ERR_LOCKED",
|
||
"ERR_FORBIDDEN", "ERR_DB", "ERR_INTERNAL", "ok", "fail", "err",
|
||
]
|
||
|
||
SQLOR_WHITELIST = {"C", "U", "D", "R", "I", "sqlExe"}
|
||
|
||
LINES: list[str] = []
|
||
N_PASS = 0
|
||
N_FAIL = 0
|
||
|
||
|
||
def record(group: str, ok: bool, item: str, evidence: str = "") -> bool:
|
||
"""记录一条核验结果。返回 ok。"""
|
||
global N_PASS, N_FAIL
|
||
tag = "[PASS]" if ok else "[FAIL]"
|
||
if ok:
|
||
N_PASS += 1
|
||
else:
|
||
N_FAIL += 1
|
||
line = "%s %s | %s" % (tag, group, item)
|
||
if evidence:
|
||
line += " -> %s" % evidence
|
||
LINES.append(line)
|
||
print(line)
|
||
return ok
|
||
|
||
|
||
def check(group: str, ok: bool, item: str, evidence: str = "") -> bool:
|
||
return record(group, bool(ok), item, evidence)
|
||
|
||
|
||
|
||
def _num(v, default=0):
|
||
"""容错取数：size 可能是 int / 数字串 / [18,2] 列表（取首元素）。"""
|
||
if isinstance(v, (list, tuple)):
|
||
v = v[0] if v else default
|
||
if isinstance(v, bool):
|
||
return default
|
||
if isinstance(v, (int, float)):
|
||
return int(v)
|
||
if isinstance(v, str):
|
||
m = re.search(r"\d+", v)
|
||
return int(m.group()) if m else default
|
||
return default
|
||
|
||
|
||
def read(path: str) -> str:
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
return f.read()
|
||
except OSError:
|
||
return ""
|
||
|
||
|
||
def load_json(path: str):
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
return json.load(f), None
|
||
except Exception as e: # noqa: BLE001
|
||
return None, str(e)
|
||
|
||
|
||
def py_files(exclude_dirs=()):
|
||
out = []
|
||
for base, dirs, files in os.walk(REPO_ROOT):
|
||
dirs[:] = [d for d in dirs if d not in (".git", "__pycache__")]
|
||
rel = os.path.relpath(base, REPO_ROOT)
|
||
if any(rel == d or rel.startswith(d + os.sep) for d in exclude_dirs):
|
||
continue
|
||
for fn in files:
|
||
if fn.endswith(".py"):
|
||
out.append(os.path.join(base, fn))
|
||
return sorted(out)
|
||
|
||
|
||
def rel(path: str) -> str:
|
||
return os.path.relpath(path, REPO_ROOT).replace(os.sep, "/")
|
||
|
||
|
||
# ============================================================== A. 目录结构
|
||
def group_a() -> None:
|
||
g = "A.目录结构"
|
||
check(g, os.path.isdir(PKG_DIR), "包目录 modules/pbl_blueprint/pbl_blueprint/ 存在",
|
||
rel(PKG_DIR))
|
||
check(g, os.path.isfile(os.path.join(PKG_DIR, "__init__.py")),
|
||
"包 __init__.py 存在", rel(os.path.join(PKG_DIR, "__init__.py")))
|
||
check(g, os.path.isfile(os.path.join(PKG_DIR, "init.py")),
|
||
"挂载入口 init.py 存在", rel(os.path.join(PKG_DIR, "init.py")))
|
||
check(g, os.path.isfile(PYPROJECT), "pyproject.toml 存在", rel(PYPROJECT))
|
||
check(g, os.path.isfile(LOAD_PATH), "scripts/load_path.py 存在", rel(LOAD_PATH))
|
||
check(g, os.path.isfile(SKILL_MD), "skill/SKILL.md 存在", rel(SKILL_MD))
|
||
check(g, os.path.isdir(MODELS_DIR), "models/ 表定义目录存在", rel(MODELS_DIR))
|
||
check(g, os.path.isdir(JSON_DIR), "json/ CRUD 契约目录存在", rel(JSON_DIR))
|
||
# 违禁替代文件（module.json / conf/rp.json / 模块级 app.py / build.sh）
|
||
bad = []
|
||
for name in ("module.json", "app.py", "build.sh", "Dockerfile"):
|
||
if os.path.isfile(os.path.join(REPO_ROOT, name)):
|
||
bad.append(name)
|
||
if os.path.isfile(os.path.join(REPO_ROOT, "conf", "rp.json")):
|
||
bad.append("conf/rp.json")
|
||
check(g, not bad, "无违禁替代文件（module.json/app.py/build.sh/Dockerfile/conf/rp.json）",
|
||
"命中=%s" % (bad or "无"))
|
||
|
||
|
||
# ============================================================== B. 表定义四段式
|
||
def group_b() -> None:
|
||
g = "B.表定义"
|
||
files = sorted(f for f in os.listdir(MODELS_DIR) if f.endswith(".json")) \
|
||
if os.path.isdir(MODELS_DIR) else []
|
||
check(g, len(files) == 11, "models/*.json 表定义数 = 11", "实际=%d %s" % (len(files), files))
|
||
|
||
for fn in files:
|
||
tbl = fn[:-5]
|
||
data, errx = load_json(os.path.join(MODELS_DIR, fn))
|
||
if data is None:
|
||
record(g, False, "%s JSON 可解析" % tbl, "解析失败: %s" % errx)
|
||
continue
|
||
# 四段式
|
||
segs = ["summary", "fields", "indexes", "codes"]
|
||
miss = [s for s in segs if s not in data]
|
||
check(g, not miss, "%s 四段式 summary/fields/indexes/codes 齐备" % tbl,
|
||
"缺失=%s" % (miss or "无"))
|
||
fields = data.get("fields") or {}
|
||
# primary = ["id"]
|
||
prim = data.get("primary")
|
||
check(g, prim == ["id"], "%s primary=[\"id\"]" % tbl, "实际=%s" % (prim,))
|
||
# id str32
|
||
idf = fields.get("id") or {}
|
||
id_ok = (idf.get("type") == "str" and _num(idf.get("size")) == 32)
|
||
check(g, id_ok, "%s id 为 str32" % tbl,
|
||
"type=%s size=%s" % (idf.get("type"), idf.get("size")))
|
||
# tenant_id 为首业务字段
|
||
keys = [k for k in fields.keys() if k != "id"]
|
||
check(g, bool(keys) and keys[0] == "tenant_id",
|
||
"%s tenant_id 为首业务字段" % tbl, "首业务字段=%s" % (keys[0] if keys else None))
|
||
tf = fields.get("tenant_id") or {}
|
||
check(g, tf.get("type") == "str" and _num(tf.get("size")) == 32 and tf.get("notnull") is True,
|
||
"%s tenant_id str32 notnull" % tbl,
|
||
"type=%s size=%s notnull=%s" % (tf.get("type"), tf.get("size"), tf.get("notnull")))
|
||
# 金额字段 double(18,2)
|
||
money = [k for k, v in fields.items()
|
||
if isinstance(v, dict) and re.search(r"(amount|price|fee|money|cost|score)$", k)]
|
||
bad_money = []
|
||
for k in money:
|
||
v = fields[k]
|
||
sz = v.get("size")
|
||
sc_ = v.get("scale", v.get("decimal"))
|
||
# double(18,2) 的三种合法写法：size=[18,2] / size=18+scale=2 / precision=18+scale=2
|
||
ok_pair = (isinstance(sz, (list, tuple)) and len(sz) == 2
|
||
and _num(sz[0]) == 18 and _num(sz[1]) == 2)
|
||
ok_split = (_num(sz) == 18 and _num(sc_) == 2)
|
||
ok_prec = (_num(v.get("precision")) == 18 and _num(sc_) == 2)
|
||
if not (v.get("type") == "double" and (ok_pair or ok_split or ok_prec)):
|
||
bad_money.append("%s(type=%s,size=%s,scale=%s)" % (k, v.get("type"), sz, sc_))
|
||
check(g, not bad_money, "%s 金额字段 double(18,2)" % tbl,
|
||
"金额字段=%s 不合规=%s" % (money or "无", bad_money or "无"))
|
||
|
||
|
||
# ============================================================== C. CRUD 契约
|
||
def group_c() -> None:
|
||
g = "C.CRUD契约"
|
||
files = sorted(f for f in os.listdir(JSON_DIR) if f.endswith(".json")) \
|
||
if os.path.isdir(JSON_DIR) else []
|
||
check(g, len(files) == 11, "json/*.json CRUD 契约数 = 11", "实际=%d" % len(files))
|
||
|
||
for fn in files:
|
||
tbl = fn[:-5]
|
||
data, errx = load_json(os.path.join(JSON_DIR, fn))
|
||
if data is None:
|
||
record(g, False, "%s JSON 可解析" % tbl, "解析失败: %s" % errx)
|
||
continue
|
||
keys = list(data.keys())
|
||
# 根键必须包含 tblname + params；editable/browserfields 为规范要求的附加键
|
||
check(g, "tblname" in keys and "params" in keys,
|
||
"%s 根键含 tblname+params" % tbl, "根键=%s" % keys)
|
||
extra = [k for k in keys if k not in ("tblname", "params", "editable", "browserfields")]
|
||
check(g, not extra, "%s 无自创 table/list 等格式" % tbl, "多余根键=%s" % (extra or "无"))
|
||
check(g, data.get("tblname") == tbl, "%s tblname 与文件名一致" % tbl,
|
||
"tblname=%s" % data.get("tblname"))
|
||
ed = data.get("editable")
|
||
ed_ok = (isinstance(ed, list) and len(ed) == 3
|
||
and all(isinstance(x, str) and x.endswith(".dspy") for x in ed))
|
||
check(g, ed_ok, "%s editable = 3 个 .dspy URL" % tbl, "editable=%s" % (ed,))
|
||
bf = data.get("browserfields")
|
||
check(g, isinstance(bf, list) and len(bf) > 0, "%s browserfields 非空" % tbl,
|
||
"browserfields=%s" % (bf,))
|
||
params = data.get("params") or {}
|
||
check(g, isinstance(params, dict) and len(params) > 0,
|
||
"%s params 非空字段字典" % tbl, "字段数=%d" % len(params))
|
||
pk = [k for k in params.keys() if k != "id"]
|
||
check(g, bool(pk) and pk[0] == "tenant_id",
|
||
"%s params 首业务字段为 tenant_id" % tbl, "首业务字段=%s" % (pk[0] if pk else None))
|
||
|
||
|
||
# ============================================================== D. 代码质量
|
||
def group_d() -> None:
|
||
g = "D.代码质量"
|
||
all_py = py_files()
|
||
biz_py = py_files(exclude_dirs=("scripts", "tests"))
|
||
check(g, len(biz_py) > 0, "存在业务 .py 源文件",
|
||
"业务=%d 全量=%d" % (len(biz_py), len(all_py)))
|
||
|
||
# py_compile 全量
|
||
r = subprocess.run([sys.executable, "-m", "py_compile"] + all_py,
|
||
capture_output=True, text=True)
|
||
check(g, r.returncode == 0, "py_compile 全量编译（%d 文件）" % len(all_py),
|
||
"rc=%d %s" % (r.returncode, (r.stderr or "").strip()[-300:]))
|
||
|
||
# 禁硬编码库名（仅业务代码；自查脚本的模式串与测试桩形参不算业务硬编码）
|
||
pat = re.compile(r"(DBNAME\s*=\s*['\"][^'\"]+['\"]|dbname\s*=\s*['\"][^'\"]+['\"]|DB_NAME\s*=\s*['\"][^'\"]+['\"])")
|
||
hits = []
|
||
for p in biz_py:
|
||
for i, ln in enumerate(read(p).split("\n"), 1):
|
||
if pat.search(ln):
|
||
hits.append("%s:%d" % (rel(p), i))
|
||
check(g, not hits, "禁硬编码库名（业务代码，排除 scripts/tests）",
|
||
"命中=%s" % (hits or "无"))
|
||
|
||
# 取库名方式
|
||
init_src = read(os.path.join(PKG_DIR, "init.py"))
|
||
check(g, "get_module_dbname" in init_src,
|
||
"init.py 通过 ServerEnv().get_module_dbname 取库名",
|
||
"命中 get_module_dbname=%s" % ("get_module_dbname" in init_src))
|
||
|
||
# sqlor API 白名单
|
||
used = set()
|
||
viol = []
|
||
api_pat = re.compile(r"\bsor\.([A-Za-z_][A-Za-z0-9_]*)\s*\(")
|
||
for p in biz_py:
|
||
for i, ln in enumerate(read(p).split("\n"), 1):
|
||
for m in api_pat.finditer(ln):
|
||
name = m.group(1)
|
||
used.add(name)
|
||
if name not in SQLOR_WHITELIST:
|
||
viol.append("%s:%d sor.%s" % (rel(p), i, name))
|
||
check(g, not viol, "sqlor API 白名单（仅 C/U/D/R/I/sqlExe）",
|
||
"实际使用=%s 越界=%s" % (sorted(used) or "无", viol or "无"))
|
||
|
||
# 禁编造 ORM 风格 API
|
||
fake = []
|
||
fake_pat = re.compile(r"\bsor\.(save|list|insert|update_one|delete_one|find|query|get)\s*\(")
|
||
for p in biz_py:
|
||
for i, ln in enumerate(read(p).split("\n"), 1):
|
||
if fake_pat.search(ln):
|
||
fake.append("%s:%d" % (rel(p), i))
|
||
check(g, not fake, "无编造 sqlor API（save/list/insert/find…）", "命中=%s" % (fake or "无"))
|
||
|
||
|
||
# ============================================================== E. 注册同步
|
||
def group_e() -> None:
|
||
g = "E.注册同步"
|
||
pkg_init = read(os.path.join(PKG_DIR, "__init__.py"))
|
||
miss = [s for s in ERRORS_14 if not re.search(r"\b%s\b" % re.escape(s), pkg_init)]
|
||
check(g, not miss, "__init__.py 导出 errors 14 符号", "缺失=%s" % (miss or "无"))
|
||
check(g, os.path.isfile(os.path.join(PKG_DIR, "errors.py")),
|
||
"errors.py 存在（14 符号定义处）", rel(os.path.join(PKG_DIR, "errors.py")))
|
||
err_src = read(os.path.join(PKG_DIR, "errors.py"))
|
||
miss2 = [s for s in ERRORS_14 if not re.search(r"^(class|def)\s+%s\b|^%s\s*=" % (re.escape(s), re.escape(s)), err_src, re.M)]
|
||
check(g, not miss2, "errors.py 定义 14 符号", "缺失=%s" % (miss2 or "无"))
|
||
|
||
init_src = read(os.path.join(PKG_DIR, "init.py"))
|
||
check(g, "def load_pbl_blueprint(" in init_src,
|
||
"init.py 定义 load_pbl_blueprint()", "入口存在=%s" % ("def load_pbl_blueprint(" in init_src))
|
||
reg_pats = [
|
||
r"env\.[A-Za-z_][A-Za-z0-9_]*\s*=", # env.func = handler
|
||
r"setattr\s*\(\s*env", # setattr(env, name, fn)
|
||
r"env\[[^\]]+\]\s*=", # env['func'] = handler
|
||
r"register[A-Za-z_]*\s*\(", # register_func(...) / env.register(...)
|
||
r"env\.(?:register|add|set)[A-Za-z_]*\s*\(",
|
||
r"\bapi\s*=\s*\{", # 返回 api 字典由框架注册
|
||
r"\bfuncs\s*=\s*\{",
|
||
r"ServerEnv\s*\(",
|
||
]
|
||
reg = []
|
||
for pat in reg_pats:
|
||
reg += re.findall(pat, init_src)
|
||
check(g, len(reg) > 0, "init.py 向 ServerEnv 注册契约函数/挂载 api",
|
||
"注册命中数=%d 样例=%s" % (len(reg), sorted(set(reg))[:5]))
|
||
check(g, "load_pbl_blueprint" in pkg_init,
|
||
"__init__.py 重导出 load_pbl_blueprint（三处同步）",
|
||
"命中=%s" % ("load_pbl_blueprint" in pkg_init))
|
||
|
||
|
||
# ============================================================== F. RBAC 路径
|
||
def group_f() -> None:
|
||
"""以 import load_path 后的运行时 PATHS 元组为权威（不扫源码文本，避免把
|
||
生成器模板串 '/api/x/%s.dspy' 误判为通配路径）。"""
|
||
g = "F.RBAC路径"
|
||
check(g, os.path.isfile(LOAD_PATH), "load_path.py 存在", rel(LOAD_PATH))
|
||
|
||
mod = None
|
||
err = ""
|
||
try:
|
||
import importlib.util
|
||
spec = importlib.util.spec_from_file_location("_lp_m1a", LOAD_PATH)
|
||
mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(mod)
|
||
except Exception as e: # noqa: BLE001
|
||
err = "%s: %s" % (type(e).__name__, e)
|
||
check(g, mod is not None, "load_path.py 可导入执行", err or "导入成功")
|
||
if mod is None:
|
||
return
|
||
|
||
paths = list(getattr(mod, "PATHS", ()) or ())
|
||
check(g, len(paths) > 0, "RBAC 注册路径显式枚举（PATHS 元组）",
|
||
"路径数=%d" % len(paths))
|
||
|
||
wild = [p for p in paths if "%" in p or "*" in p]
|
||
check(g, not wild, "RBAC 路径无 %/* 通配符（运行时真实值）",
|
||
"通配路径=%s" % (wild[:5] or "无"))
|
||
|
||
nonstd = [p for p in paths
|
||
if not re.fullmatch(r"/api/[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+\.dspy", p)]
|
||
check(g, not nonstd, "RBAC 路径均为 /api/<表>/<动作>.dspy 规范字面量",
|
||
"异常=%s" % (nonstd[:5] or "无"))
|
||
|
||
dup = sorted({p for p in paths if paths.count(p) > 1})
|
||
check(g, not dup, "RBAC 路径无重复", "重复=%s" % (dup[:5] or "无"))
|
||
|
||
# 覆盖 json/*.json editable 的 3×11 条（超集合法：支撑表含额外动作）
|
||
need = set()
|
||
if os.path.isdir(JSON_DIR):
|
||
for fn in sorted(os.listdir(JSON_DIR)):
|
||
if not fn.endswith(".json"):
|
||
continue
|
||
d, _e = load_json(os.path.join(JSON_DIR, fn))
|
||
for u in ((d or {}).get("editable") or []):
|
||
if isinstance(u, str) and u.startswith("api/"):
|
||
need.add("/" + u)
|
||
missing = sorted(need - set(paths))
|
||
check(g, not missing,
|
||
"load_path.py 覆盖 json/*.json editable 全部路径（超集）",
|
||
"load_path=%d json_editable=%d 未覆盖=%s"
|
||
% (len(paths), len(need), missing[:5] or "无"))
|
||
|
||
# 孤儿路径：每条路径的表名必须归属主清单 11 表或支撑表 9 表
|
||
main_t = set(getattr(mod, "MAIN_TABLES", ()) or ())
|
||
sup_t = set(getattr(mod, "SUPPORT_TABLES", ()) or ())
|
||
known = main_t | sup_t
|
||
check(g, len(main_t) == 11, "load_path.MAIN_TABLES = 设计定稿 11 表",
|
||
"实际=%d" % len(main_t))
|
||
orphan = []
|
||
for p in paths:
|
||
m = re.match(r"/api/([a-z0-9_]+)/", p)
|
||
if not m or (known and m.group(1) not in known):
|
||
orphan.append(p)
|
||
check(g, not orphan, "load_path.py 无孤儿路径（均归属已知表）",
|
||
"孤儿=%s" % (orphan[:5] or "无"))
|
||
|
||
# 模块自带 validate() 必须通过
|
||
vfn = getattr(mod, "validate", None)
|
||
if callable(vfn):
|
||
verrs = vfn() or []
|
||
check(g, not verrs, "load_path.validate() 自检通过",
|
||
"错误=%s" % (verrs[:3] or "无"))
|
||
else:
|
||
check(g, False, "load_path.validate() 存在", "未定义 validate()")
|
||
|
||
|
||
# ============================================================== G. 租户 fail-closed
|
||
def group_g() -> None:
|
||
g = "G.租户fail-closed"
|
||
biz = py_files(exclude_dirs=("scripts", "tests"))
|
||
src_all = "\n".join(read(p) for p in biz)
|
||
check(g, "ERR_TENANT_MISSING" in src_all,
|
||
"业务代码引用 ERR_TENANT_MISSING（缺租户即拒）",
|
||
"命中=%s" % ("ERR_TENANT_MISSING" in src_all))
|
||
check(g, os.path.isfile(os.path.join(PKG_DIR, "tenant.py")),
|
||
"tenant.py 租户上下文模块存在", rel(os.path.join(PKG_DIR, "tenant.py")))
|
||
tsrc = read(os.path.join(PKG_DIR, "tenant.py"))
|
||
check(g, "tenant_id" in tsrc, "tenant.py 处理 tenant_id", "命中=%s" % ("tenant_id" in tsrc))
|
||
# 禁止默认租户兜底
|
||
bad = re.findall(r"tenant_id\s*=\s*['\"](?:default|DEFAULT|1|0|test)['\"]", src_all)
|
||
check(g, not bad, "无默认租户兜底（fail-closed）", "命中=%s" % (bad or "无"))
|
||
|
||
|
||
# ============================================================== H. DDL
|
||
def group_h() -> None:
|
||
g = "H.DDL"
|
||
sqls = sorted(f for f in os.listdir(SQL_DIR) if f.endswith(".sql")) if os.path.isdir(SQL_DIR) else []
|
||
check(g, len(sqls) >= 2, "sql/ DDL 文件存在（core + subobjects）", "文件=%s" % sqls)
|
||
ddl = "\n".join(read(os.path.join(SQL_DIR, f)) for f in sqls)
|
||
tables = sorted(set(re.findall(r"CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?`?([a-z0-9_]+)`?", ddl, re.I)))
|
||
check(g, len(tables) > 0, "DDL 含 CREATE TABLE 语句", "建表数=%d" % len(tables))
|
||
# 设计定稿 B1~B3 核心表（B4~B11 为独立建表，非泛化单表）
|
||
need_core = ["pbl_blueprint", "pbl_blueprint_version", "pbl_blueprint_approval"]
|
||
miss = [t for t in need_core if t not in tables]
|
||
check(g, not miss, "DDL 覆盖核心表 B1~B3（聚合根/版本/审批）",
|
||
"缺失=%s" % (miss or "无"))
|
||
# 设计定稿 B4~B11 八张子对象表必须独立建表
|
||
need_sub = ["pbl_learner", "pbl_learning_goal", "pbl_problem",
|
||
"pbl_driving_question", "pbl_project", "pbl_role",
|
||
"pbl_mission", "pbl_artifact_def"]
|
||
miss_sub = [t for t in need_sub if t not in tables]
|
||
check(g, not miss_sub, "DDL 覆盖子对象表 B4~B11（8 张独立建表）",
|
||
"缺失=%s 建表总数=%d" % (miss_sub or "无", len(tables)))
|
||
# models 表 -> 物理承载映射核验
|
||
mfiles = sorted(f[:-5] for f in os.listdir(MODELS_DIR) if f.endswith(".json")) \
|
||
if os.path.isdir(MODELS_DIR) else []
|
||
# 设计定稿：models 主清单 11 表 = DDL 独立建表，逐表必须有 CREATE TABLE
|
||
uncovered = [t for t in mfiles if t not in tables]
|
||
check(g, not uncovered, "models 主清单 11 表均有独立 CREATE TABLE 承载",
|
||
"未承载=%s models=%d ddl建表=%d" % (uncovered or "无", len(mfiles), len(tables)))
|
||
check(g, "tenant_id" in ddl, "DDL 各表含 tenant_id 列", "命中=%s" % ("tenant_id" in ddl))
|
||
|
||
|
||
# ============================================================== I. git 证据
|
||
def group_i() -> None:
|
||
g = "I.git证据"
|
||
def git(*args):
|
||
r = subprocess.run(["git", "-C", REPO_ROOT] + list(args),
|
||
capture_output=True, text=True)
|
||
return r.returncode, (r.stdout or "").strip(), (r.stderr or "").strip()
|
||
rc, out, _ = git("rev-parse", "--is-inside-work-tree")
|
||
check(g, rc == 0 and out == "true", "模块目录为 git 仓库", "rc=%d out=%s" % (rc, out))
|
||
rc, head, _ = git("log", "-1", "--format=%h %s")
|
||
check(g, rc == 0 and bool(head), "存在本地提交（HEAD）", head[:120])
|
||
rc, st, _ = git("status", "--porcelain")
|
||
dirty = [x for x in st.split("\n") if x.strip()]
|
||
check(g, not dirty, "工作区 clean（变更已全部提交）",
|
||
"未提交条目=%d %s" % (len(dirty), dirty[:5]))
|
||
rc, rem, _ = git("remote", "-v")
|
||
check(g, "origin" in rem, "配置 origin 远程", rem.split("\n")[0][:100] if rem else "无")
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--report", default="", help="报告输出路径")
|
||
args = ap.parse_args()
|
||
|
||
groups = [
|
||
("A.目录结构", group_a), ("B.表定义", group_b), ("C.CRUD契约", group_c),
|
||
("D.代码质量", group_d), ("E.注册同步", group_e), ("F.RBAC路径", group_f),
|
||
("G.租户fail-closed", group_g), ("H.DDL", group_h), ("I.git证据", group_i),
|
||
]
|
||
header = "=" * 78
|
||
LINES.append(header)
|
||
LINES.append("pbl_blueprint M1a 自查报告（selfcheck.py 重写版）")
|
||
LINES.append("仓库根: %s" % REPO_ROOT)
|
||
LINES.append(header)
|
||
print("\n".join(LINES[-4:]))
|
||
|
||
for name, fn in groups:
|
||
LINES.append("-" * 78)
|
||
print("-" * 78)
|
||
try:
|
||
fn()
|
||
except Exception as e: # noqa: BLE001
|
||
import traceback
|
||
record(name, False, "检查组执行异常", "%s: %s" % (type(e).__name__, e))
|
||
LINES.append(traceback.format_exc())
|
||
|
||
LINES.append(header)
|
||
total = N_PASS + N_FAIL
|
||
verdict = "ALL PASS" if N_FAIL == 0 else "HAS FAILURES"
|
||
LINES.append("结论: %s | PASS=%d FAIL=%d 总=%d" % (verdict, N_PASS, N_FAIL, total))
|
||
LINES.append(header)
|
||
print("\n".join(LINES[-3:]))
|
||
|
||
if args.report:
|
||
try:
|
||
os.makedirs(os.path.dirname(os.path.abspath(args.report)), exist_ok=True)
|
||
with open(args.report, "w", encoding="utf-8") as f:
|
||
f.write("\n".join(LINES) + "\n")
|
||
print("报告已写入: %s" % args.report)
|
||
except OSError as e:
|
||
print("报告写入失败: %s" % e)
|
||
|
||
# 门禁：有失败项必须非 0 退出
|
||
return 0 if N_FAIL == 0 else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|