668 lines
28 KiB
Python
668 lines
28 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""pbl_blueprint M1a 严格退出码门禁 G1~G6（真实可执行代码，非注释大纲）。
|
||
|
||
复跑方式（机构工作空间根）::
|
||
|
||
cd /d/pipeline/workspaces/0/sdlc_general
|
||
python3 modules/pbl_blueprint/scripts/verify_gate.py; echo "gate rc=$?"
|
||
|
||
退出码契约
|
||
- 任一门禁项失败：逐条打印失败原因 -> 汇总失败编号 -> 尾行 ``GATE RESULT: FAIL rc=1`` -> ``sys.exit(1)``
|
||
- 全部通过：尾行 ``GATE RESULT: PASS rc=0`` -> ``sys.exit(0)``
|
||
- 不首错即停：一次跑完 G1~G6，暴露全部问题（便于 QC 一轮看清）
|
||
|
||
门禁项
|
||
G1 py_compile 全量编译（doraise=True，cfile=os.devnull，不污染工作区）
|
||
G2 models/*.json 四段式(summary/fields/indexes/codes) + primary==["id"] + id str(32)
|
||
+ tenant_id 为首业务字段 + 金额字段 double(18,2)
|
||
G3 json/*.json CRUD 根键 tblname+params、禁自创 table/list 根键、
|
||
editable 场景 .dspy >= 3 处、browserfields 非空
|
||
G4 禁硬编码库名（AST 精确识别 DBNAME/DB_NAME/dbname='...' 字面量赋值与关键字实参，
|
||
放行 ServerEnv().get_module_dbname）+ sqlor API 白名单 {C,U,D,R,I,sqlExe}（.py 走 AST，.json 走正则）
|
||
G5 函数三处注册同步：init.py 的 env 注册 -> 包内 def 定义 -> __init__.py 导出；
|
||
反向核对幽灵导出（__all__ 声明但包内不存在）
|
||
G6 scripts/load_path.py RBAC 路径显式枚举、无 %/*/? 通配符
|
||
|
||
路径解析基于 ``os.path.dirname(os.path.abspath(__file__))`` 上溯，不依赖 cwd。
|
||
本脚本只读，不写任何文件、不改 git 状态。
|
||
"""
|
||
|
||
import ast
|
||
import json
|
||
import os
|
||
import py_compile
|
||
import re
|
||
import sys
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 路径与作用域常量
|
||
# --------------------------------------------------------------------------
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
MODULE_ROOT = os.path.dirname(HERE)
|
||
PKG_NAME = "pbl_blueprint"
|
||
PKG_DIR = os.path.join(MODULE_ROOT, PKG_NAME)
|
||
MODELS_DIR = os.path.join(PKG_DIR, "models")
|
||
JSON_DIR = os.path.join(PKG_DIR, "json")
|
||
LOAD_PATH_PY = os.path.join(HERE, "load_path.py")
|
||
INIT_PY = os.path.join(PKG_DIR, "init.py")
|
||
PKG_INIT_PY = os.path.join(PKG_DIR, "__init__.py")
|
||
|
||
SKIP_DIRS = frozenset([".git", "__pycache__", ".idea", ".vscode", "node_modules", ".pytest_cache"])
|
||
|
||
SQLOR_WHITELIST = frozenset(["C", "U", "D", "R", "I", "sqlExe"])
|
||
SQLOR_IGNORE = frozenset(["Sqlor", "sqlor", "__init__", "__all__", "py"])
|
||
|
||
FOUR_SECTIONS = ("summary", "fields", "indexes", "codes")
|
||
|
||
MONEY_TOKENS = ("amount", "price", "fee", "cost", "money", "budget", "balance", "payment")
|
||
NON_MONEY_SUFFIX = (
|
||
"_type", "_id", "_code", "_flag", "_name", "_at", "_by", "_no",
|
||
"_key", "_status", "_unit", "_desc", "_remark", "_json", "_str",
|
||
)
|
||
NUMERIC_TYPE_RE = re.compile(r"^(double|decimal|numeric|float|number|real|bigint|int|integer)\b", re.I)
|
||
STR32_RE = re.compile(r"^(str\s*\(\s*32\s*\)|str32|varchar\s*\(\s*32\s*\)|char\s*\(\s*32\s*\)|string\s*\(\s*32\s*\))$", re.I)
|
||
MONEY182_RE = re.compile(r"^(double|decimal|numeric)\s*\(\s*18\s*,\s*2\s*\)$", re.I)
|
||
DBNAME_TARGET_RE = re.compile(r"^(DBNAME|DB_NAME|DATABASE_NAME|dbname|db_name|database_name)$")
|
||
IDENT_RE = re.compile(r"^[a-z][a-z0-9_]{2,}$")
|
||
SQLOR_TEXT_RE = re.compile(r"\b(?:sor|sqlor)\.([A-Za-z_]\w*)")
|
||
PATH_LITERAL_RE = re.compile(r"""['"](/[A-Za-z0-9_./\-]+)['"]""")
|
||
|
||
# G5 噪声过滤：这些名字即使出现在 env 注册解析结果里也不作为「对外契约函数」核对，
|
||
# 避免把模块对象/配置项/框架钩子误判为缺 def、缺导出（防假阳性）。
|
||
NOISE = frozenset([
|
||
"get_module_dbname", "load_pbl_blueprint", "pbl_blueprint", "init", "main", "setup",
|
||
"install", "register", "env", "server", "app", "conf", "config", "settings", "options",
|
||
"params", "context", "ctx", "user", "session", "request", "response", "db", "dbname",
|
||
"sqlor", "sor", "i18n", "lang", "static", "wwwroot", "menu", "routes", "apis", "tables",
|
||
"models", "module", "name", "path", "root", "log", "logger", "audit", "errors", "crud",
|
||
"service", "api", "tenant", "subobject", "subobjects", "templates", "tables_def",
|
||
"pool", "conn", "cursor", "transaction", "lock", "cache", "hooks", "handlers",
|
||
"middleware", "blueprint_crud", "api_blueprint", "db_adapter", "sql", "json",
|
||
])
|
||
|
||
FAILURES = [] # [(gate, message)]
|
||
WARNS = [] # [(gate, message)]
|
||
STATS = {}
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 通用工具
|
||
# --------------------------------------------------------------------------
|
||
def rel(path):
|
||
"""转成相对模块仓库根的可读路径（失败则原样返回）。"""
|
||
try:
|
||
return os.path.relpath(path, MODULE_ROOT)
|
||
except ValueError:
|
||
return path
|
||
|
||
|
||
def fail(gate, message):
|
||
FAILURES.append((gate, message))
|
||
|
||
|
||
def warn(gate, message):
|
||
WARNS.append((gate, message))
|
||
|
||
|
||
def iter_files(root, suffix):
|
||
"""递归产出 root 下指定后缀的文件绝对路径，跳过 .git/__pycache__ 等目录。"""
|
||
if not os.path.isdir(root):
|
||
return
|
||
for dirpath, dirnames, filenames in os.walk(root):
|
||
dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
|
||
for fn in sorted(filenames):
|
||
if fn.endswith(suffix):
|
||
yield os.path.join(dirpath, fn)
|
||
|
||
|
||
def read_text(path):
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
return f.read()
|
||
except (IOError, OSError, UnicodeDecodeError) as e:
|
||
fail("G0", "文件读取失败 %s: %s" % (rel(path), e))
|
||
return None
|
||
|
||
|
||
def parse_ast(path):
|
||
"""返回 (tree_or_None, source_text)。语法错误由 G1 负责报错，此处不重复失败。"""
|
||
src = read_text(path)
|
||
if src is None:
|
||
return None, ""
|
||
try:
|
||
return ast.parse(src, filename=path), src
|
||
except SyntaxError:
|
||
return None, src
|
||
|
||
|
||
def load_json(path):
|
||
src = read_text(path)
|
||
if src is None:
|
||
return None
|
||
try:
|
||
return json.loads(src)
|
||
except ValueError as e:
|
||
fail("G0", "JSON 解析失败 %s: %s" % (rel(path), e))
|
||
return None
|
||
|
||
|
||
def const_str(node):
|
||
"""兼容 py3.7 的 ast.Str 与 py3.8+ 的 ast.Constant 取字符串值。"""
|
||
if isinstance(node, ast.Constant) and isinstance(node.value, str):
|
||
return node.value
|
||
if hasattr(ast, "Str") and isinstance(node, ast.Str): # pragma: no cover
|
||
return node.s
|
||
return None
|
||
|
||
|
||
def subscript_key(node):
|
||
"""兼容 py3.8 的 ast.Index 包装，取 Subscript 的键节点。"""
|
||
sl = node.slice
|
||
if hasattr(ast, "Index") and isinstance(sl, ast.Index): # pragma: no cover
|
||
return sl.value
|
||
return sl
|
||
|
||
|
||
def find_key_values(obj, target_lower):
|
||
"""在任意嵌套的 dict/list 里递归收集键名（小写比对）等于 target_lower 的值。"""
|
||
found = []
|
||
if isinstance(obj, dict):
|
||
for k, v in obj.items():
|
||
if str(k).lower() == target_lower:
|
||
found.append(v)
|
||
found.extend(find_key_values(v, target_lower))
|
||
elif isinstance(obj, list):
|
||
for it in obj:
|
||
found.extend(find_key_values(it, target_lower))
|
||
return found
|
||
|
||
|
||
def is_nonempty(v):
|
||
if v is None:
|
||
return False
|
||
if isinstance(v, (list, dict, str, tuple, set)):
|
||
return len(v) > 0
|
||
return True
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# G1 py_compile 全量编译
|
||
# --------------------------------------------------------------------------
|
||
def gate_g1():
|
||
pys = sorted(iter_files(MODULE_ROOT, ".py"))
|
||
if not pys:
|
||
fail("G1", "模块仓库内未找到任何 .py 文件：%s" % MODULE_ROOT)
|
||
STATS["G1"] = {"total": 0, "bad": 0}
|
||
return
|
||
bad = 0
|
||
for p in pys:
|
||
try:
|
||
py_compile.compile(p, cfile=os.devnull, doraise=True)
|
||
except py_compile.PyCompileError as e:
|
||
bad += 1
|
||
detail = str(e).strip().splitlines()
|
||
fail("G1", "编译失败 %s: %s" % (rel(p), detail[0] if detail else e))
|
||
except SyntaxError as e:
|
||
bad += 1
|
||
fail("G1", "语法错误 %s: line %s: %s" % (rel(p), e.lineno, e.msg))
|
||
except Exception as e: # noqa: BLE001 —— 门禁要把任何编译异常都暴露出来
|
||
bad += 1
|
||
fail("G1", "编译异常 %s: %s: %s" % (rel(p), type(e).__name__, e))
|
||
STATS["G1"] = {"total": len(pys), "bad": bad}
|
||
print(" G1 py_compile 全量编译 %d 个 .py，失败 %d 个" % (len(pys), bad))
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# G2 models/*.json 表定义四段式与硬约束
|
||
# --------------------------------------------------------------------------
|
||
def norm_fields(obj):
|
||
"""把 fields 段归一化为 [(name, typestr, meta_dict)]，兼容 list[dict] / dict 两种写法。"""
|
||
fields = obj.get("fields")
|
||
out = []
|
||
if isinstance(fields, list):
|
||
for it in fields:
|
||
if isinstance(it, dict):
|
||
nm = it.get("name") or it.get("field") or it.get("column") or it.get("colname") or it.get("key")
|
||
tp = it.get("type") or it.get("datatype") or it.get("dbtype") or it.get("coltype") or ""
|
||
if nm is None:
|
||
continue
|
||
out.append((str(nm), str(tp), it))
|
||
elif isinstance(it, str):
|
||
out.append((it, "", {}))
|
||
elif isinstance(fields, dict):
|
||
for k, v in fields.items():
|
||
if isinstance(v, dict):
|
||
tp = v.get("type") or v.get("datatype") or v.get("dbtype") or ""
|
||
out.append((str(k), str(tp), v))
|
||
else:
|
||
out.append((str(k), "" if v is None else str(v), {}))
|
||
return out
|
||
|
||
|
||
def find_primary(obj):
|
||
"""在顶层 / summary / indexes 三处发现主键声明，返回 list[str] 或 None。"""
|
||
keys = ("primary", "primarykey", "primary_key", "pk")
|
||
for k in keys:
|
||
if k in obj:
|
||
v = obj[k]
|
||
return [str(x)] if isinstance(v, str) else [str(x) for x in v] if isinstance(v, (list, tuple)) else None
|
||
summary = obj.get("summary")
|
||
if isinstance(summary, dict):
|
||
for k in keys:
|
||
if k in summary:
|
||
v = summary[k]
|
||
return [str(x)] if isinstance(v, str) else [str(x) for x in v] if isinstance(v, (list, tuple)) else None
|
||
idx = obj.get("indexes")
|
||
entries = idx if isinstance(idx, list) else (list(idx.values()) if isinstance(idx, dict) else [])
|
||
for e in entries:
|
||
if isinstance(e, dict) and (e.get("primary") is True or str(e.get("type", "")).lower() in ("primary", "pk")):
|
||
cols = e.get("fields") or e.get("columns") or e.get("cols") or e.get("name")
|
||
if isinstance(cols, str):
|
||
return [cols]
|
||
if isinstance(cols, (list, tuple)):
|
||
return [str(c) for c in cols]
|
||
return None
|
||
|
||
|
||
def gate_g2():
|
||
files = sorted(iter_files(MODELS_DIR, ".json"))
|
||
if not files:
|
||
fail("G2", "models 目录不存在或无表定义 JSON：%s" % rel(MODELS_DIR))
|
||
STATS["G2"] = {"tables": 0}
|
||
return
|
||
money_checked = 0
|
||
for p in files:
|
||
obj = load_json(p)
|
||
if obj is None:
|
||
continue
|
||
r = rel(p)
|
||
if not isinstance(obj, dict):
|
||
fail("G2", "%s 根节点不是 JSON 对象" % r)
|
||
continue
|
||
# (1) 四段式
|
||
missing = [s for s in FOUR_SECTIONS if s not in obj]
|
||
if missing:
|
||
fail("G2", "%s 缺四段式字段段：%s" % (r, "/".join(missing)))
|
||
fields = norm_fields(obj)
|
||
if not fields:
|
||
fail("G2", "%s fields 段为空或格式不可识别" % r)
|
||
continue
|
||
names = [n for n, _t, _m in fields]
|
||
types = dict((n, t) for n, t, _m in fields)
|
||
# (2) 主键 primary == ["id"]
|
||
pk = find_primary(obj)
|
||
if pk is None:
|
||
if "id" not in names:
|
||
fail("G2", "%s 未声明 primary 且 fields 中无 id 字段" % r)
|
||
else:
|
||
warn("G2", "%s 未显式声明 primary=[\"id\"]（已按 fields 含 id 放行，建议补声明）" % r)
|
||
elif pk != ["id"]:
|
||
fail("G2", "%s primary 必须为 [\"id\"]，实际 %s" % (r, pk))
|
||
# (3) id 为 str(32)
|
||
if "id" in names:
|
||
idt = (types.get("id") or "").strip()
|
||
if not STR32_RE.match(idt):
|
||
fail("G2", "%s 主键 id 类型必须为 str(32)，实际 %r" % (r, idt))
|
||
if names[0] != "id":
|
||
warn("G2", "%s id 未置于 fields 首位（实际首位 %s）" % (r, names[0]))
|
||
# (4) tenant_id 为首业务字段（fail-closed 多租户硬约束）
|
||
if "tenant_id" not in names:
|
||
fail("G2", "%s 缺 tenant_id 字段（所有读写租户强制打头）" % r)
|
||
else:
|
||
biz = names[1:] if names and names[0] == "id" else names
|
||
if not biz or biz[0] != "tenant_id":
|
||
fail("G2", "%s tenant_id 必须是首业务字段，实际首业务字段为 %s" % (r, biz[0] if biz else "无"))
|
||
# (5) 金额字段 double(18,2)
|
||
for n, t, _m in fields:
|
||
low = n.lower()
|
||
if not any(tok in low for tok in MONEY_TOKENS):
|
||
continue
|
||
if low.endswith(NON_MONEY_SUFFIX):
|
||
continue
|
||
t = (t or "").strip()
|
||
if not NUMERIC_TYPE_RE.match(t):
|
||
continue # 非数值型（如 fee_type 之类编码字段）不按金额约束
|
||
money_checked += 1
|
||
if not MONEY182_RE.match(t):
|
||
fail("G2", "%s 金额字段 %s 必须为 double(18,2)，实际 %r" % (r, n, t))
|
||
STATS["G2"] = {"tables": len(files), "money_fields_checked": money_checked}
|
||
print(" G2 表定义 %d 个（四段式/primary/id str32/tenant_id 首业务字段/金额 double(18,2)），金额字段核对 %d 个"
|
||
% (len(files), money_checked))
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# G3 json/*.json CRUD 契约
|
||
# --------------------------------------------------------------------------
|
||
def gate_g3():
|
||
files = sorted(iter_files(JSON_DIR, ".json"))
|
||
if not files:
|
||
fail("G3", "json 目录不存在或无 CRUD 定义：%s" % rel(JSON_DIR))
|
||
STATS["G3"] = {"crud": 0}
|
||
return
|
||
forbid_root = ("table", "list")
|
||
dspy_total = 0
|
||
for p in files:
|
||
raw = read_text(p)
|
||
obj = load_json(p)
|
||
if obj is None:
|
||
continue
|
||
r = rel(p)
|
||
if not isinstance(obj, dict):
|
||
fail("G3", "%s 根节点不是 JSON 对象" % r)
|
||
continue
|
||
# (1) 根键 tblname + params
|
||
if "tblname" not in obj:
|
||
fail("G3", "%s 缺根键 tblname" % r)
|
||
elif not is_nonempty(obj.get("tblname")):
|
||
fail("G3", "%s 根键 tblname 为空" % r)
|
||
if "params" not in obj:
|
||
fail("G3", "%s 缺根键 params" % r)
|
||
# (2) 禁自创 table/list 根键格式
|
||
for bad_key in forbid_root:
|
||
if bad_key in obj:
|
||
fail("G3", "%s 含自创根键 %r（平台 CRUD 契约只认 tblname+params）" % (r, bad_key))
|
||
# (3) .dspy 引用：含 editable 的必须 >= 3 处（增/改/删或列表/编辑/删除三入口）
|
||
n_dspy = (raw or "").count(".dspy")
|
||
dspy_total += n_dspy
|
||
has_editable = len(find_key_values(obj, "editable")) > 0
|
||
if has_editable:
|
||
if n_dspy < 3:
|
||
fail("G3", "%s 含 editable 但 .dspy 引用仅 %d 处（要求 >= 3）" % (r, n_dspy))
|
||
elif n_dspy < 1:
|
||
warn("G3", "%s 无 editable 且无 .dspy 引用（若为纯查询契约可忽略）" % r)
|
||
# (4) browserfields 非空
|
||
bfs = find_key_values(obj, "browserfields")
|
||
if not bfs:
|
||
warn("G3", "%s 未声明 browserfields（列表/浏览型契约应声明）" % r)
|
||
elif not any(is_nonempty(v) for v in bfs):
|
||
fail("G3", "%s browserfields 已声明但为空" % r)
|
||
STATS["G3"] = {"crud": len(files), "dspy_refs": dspy_total}
|
||
print(" G3 CRUD 契约 %d 个（tblname+params/禁 table,list/.dspy 引用共 %d 处/browserfields 非空）"
|
||
% (len(files), dspy_total))
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# G4 禁硬编码库名 + sqlor API 白名单
|
||
# --------------------------------------------------------------------------
|
||
def gate_g4():
|
||
pys = sorted(iter_files(MODULE_ROOT, ".py"))
|
||
hardcode = 0
|
||
for p in pys:
|
||
tree, _src = parse_ast(p)
|
||
if tree is None:
|
||
continue
|
||
for node in ast.walk(tree):
|
||
if isinstance(node, (ast.Assign, ast.AnnAssign)):
|
||
targets = node.targets if isinstance(node, ast.Assign) else [node.target]
|
||
for t in targets:
|
||
nm = None
|
||
if isinstance(t, ast.Name):
|
||
nm = t.id
|
||
elif isinstance(t, ast.Attribute):
|
||
nm = t.attr
|
||
if not nm or not DBNAME_TARGET_RE.match(nm):
|
||
continue
|
||
v = const_str(node.value) if node.value is not None else None
|
||
if v is not None:
|
||
hardcode += 1
|
||
fail("G4", "%s:%s 硬编码库名 %s = %r（必须用 ServerEnv().get_module_dbname('%s')）"
|
||
% (rel(p), getattr(node, "lineno", "?"), nm, v, PKG_NAME))
|
||
if isinstance(node, ast.Call):
|
||
for kw in (node.keywords or []):
|
||
if kw.arg and DBNAME_TARGET_RE.match(kw.arg):
|
||
v = const_str(kw.value)
|
||
if v is not None:
|
||
hardcode += 1
|
||
fail("G4", "%s:%s 调用实参硬编码库名 %s=%r" % (rel(p), node.lineno, kw.arg, v))
|
||
# sqlor 白名单：.py 走 AST（注释/文档字符串不会误报）
|
||
used = set()
|
||
illegal = []
|
||
for p in pys:
|
||
tree, _src = parse_ast(p)
|
||
if tree is None:
|
||
continue
|
||
for node in ast.walk(tree):
|
||
if isinstance(node, ast.Attribute) and isinstance(node.value, ast.Name) \
|
||
and node.value.id in ("sor", "sqlor"):
|
||
attr = node.attr
|
||
if attr in SQLOR_IGNORE:
|
||
continue
|
||
used.add(attr)
|
||
if attr not in SQLOR_WHITELIST:
|
||
illegal.append("%s:%s %s.%s" % (rel(p), node.lineno, node.value.id, attr))
|
||
# .json（CRUD/表定义/dspy 契约）里的 sqlor 引用走正则
|
||
for p in sorted(list(iter_files(JSON_DIR, ".json")) + list(iter_files(MODELS_DIR, ".json"))):
|
||
txt = read_text(p) or ""
|
||
for m in SQLOR_TEXT_RE.finditer(txt):
|
||
attr = m.group(1)
|
||
if attr in SQLOR_IGNORE:
|
||
continue
|
||
used.add(attr)
|
||
if attr not in SQLOR_WHITELIST:
|
||
illegal.append("%s sqlor.%s" % (rel(p), attr))
|
||
for item in sorted(set(illegal)):
|
||
fail("G4", "sqlor API 越白名单（仅允许 %s）：%s" % ("/".join(sorted(SQLOR_WHITELIST)), item))
|
||
# 正向断言：包内必须通过 get_module_dbname 取库名
|
||
joined = ""
|
||
for p in iter_files(PKG_DIR, ".py"):
|
||
joined += (read_text(p) or "")
|
||
if "get_module_dbname" not in joined:
|
||
fail("G4", "包内未出现 get_module_dbname 调用，库名来源不合规（禁硬编码 DBNAME）")
|
||
STATS["G4"] = {"py": len(pys), "hardcode": hardcode, "sqlor_used": sorted(used), "illegal": len(set(illegal))}
|
||
print(" G4 扫描 %d 个 .py：硬编码库名命中 %d，sqlor 使用 API=%s（白名单 %s）"
|
||
% (len(pys), hardcode, sorted(used) or "[]", "/".join(sorted(SQLOR_WHITELIST))))
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# G5 函数三处注册同步
|
||
# --------------------------------------------------------------------------
|
||
def module_stems():
|
||
return set(os.path.splitext(os.path.basename(p))[0] for p in iter_files(PKG_DIR, ".py"))
|
||
|
||
|
||
def registered_names():
|
||
"""从 init.py 解析 env 注册的对外函数名（AST，避免注释误报）。"""
|
||
names = set()
|
||
tree, _src = parse_ast(INIT_PY)
|
||
if tree is None:
|
||
return names
|
||
for node in ast.walk(tree):
|
||
if isinstance(node, (ast.Assign, ast.AnnAssign)):
|
||
targets = node.targets if isinstance(node, ast.Assign) else [node.target]
|
||
for t in targets:
|
||
# env.xxx = fn / _env.xxx = fn
|
||
if isinstance(t, ast.Attribute) and isinstance(t.value, ast.Name) \
|
||
and t.value.id.lower().endswith("env"):
|
||
names.add(t.attr)
|
||
# env['xxx'] = fn
|
||
elif isinstance(t, ast.Subscript) and isinstance(t.value, ast.Name) \
|
||
and t.value.id.lower().endswith("env"):
|
||
key = const_str(subscript_key(t))
|
||
if key:
|
||
names.add(key)
|
||
# API_FUNCS = ("a", "b") 之类的批量注册清单
|
||
elif isinstance(t, ast.Name) and re.search(r"(API|FUNC|ROUTE|EXPORT|HANDLER|CRUD)", t.id.upper()):
|
||
if isinstance(node.value, (ast.List, ast.Tuple, ast.Set)):
|
||
for el in node.value.elts:
|
||
s = const_str(el)
|
||
if s:
|
||
names.add(s)
|
||
if isinstance(node, ast.Call):
|
||
fn = node.func
|
||
fname = fn.attr if isinstance(fn, ast.Attribute) else (fn.id if isinstance(fn, ast.Name) else "")
|
||
if fname.startswith("register") or fname in ("add_api", "add_function", "add_route", "route", "setattr"):
|
||
if node.args:
|
||
s = const_str(node.args[0])
|
||
if s:
|
||
names.add(s)
|
||
return names
|
||
|
||
|
||
def pkg_symbols():
|
||
"""返回 (包内 def 名集合, 包内 import 进来的名字集合)。"""
|
||
defs = set()
|
||
imports = set()
|
||
for p in iter_files(PKG_DIR, ".py"):
|
||
tree, _src = parse_ast(p)
|
||
if tree is None:
|
||
continue
|
||
for node in tree.body:
|
||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||
defs.add(node.name)
|
||
for node in ast.walk(tree):
|
||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||
defs.add(node.name)
|
||
elif isinstance(node, ast.ImportFrom):
|
||
for a in node.names:
|
||
imports.add(a.asname or a.name)
|
||
elif isinstance(node, ast.Import):
|
||
for a in node.names:
|
||
imports.add((a.asname or a.name).split(".")[0])
|
||
return defs, imports
|
||
|
||
|
||
def pkg_exports():
|
||
"""__init__.py 的对外导出面：__all__ 字符串 + from .x import (...) 名字。"""
|
||
exports = set()
|
||
tree, _src = parse_ast(PKG_INIT_PY)
|
||
if tree is None:
|
||
return exports
|
||
for node in ast.walk(tree):
|
||
if isinstance(node, ast.Assign):
|
||
for t in node.targets:
|
||
if isinstance(t, ast.Name) and t.id == "__all__" and isinstance(node.value, (ast.List, ast.Tuple, ast.Set)):
|
||
for el in node.value.elts:
|
||
s = const_str(el)
|
||
if s:
|
||
exports.add(s)
|
||
if isinstance(node, ast.ImportFrom):
|
||
for a in node.names:
|
||
exports.add(a.asname or a.name)
|
||
return exports
|
||
|
||
|
||
def gate_g5():
|
||
if not os.path.isfile(INIT_PY):
|
||
fail("G5", "缺少 init.py（挂载入口）：%s" % rel(INIT_PY))
|
||
if not os.path.isfile(PKG_INIT_PY):
|
||
fail("G5", "缺少 __init__.py（包导出面）：%s" % rel(PKG_INIT_PY))
|
||
stems = module_stems()
|
||
reg = set()
|
||
for n in registered_names():
|
||
if not isinstance(n, str):
|
||
continue
|
||
n = n.strip()
|
||
if not IDENT_RE.match(n) or n in NOISE or n in stems:
|
||
continue
|
||
if n.startswith("_"):
|
||
continue
|
||
reg.add(n)
|
||
defs, imports = pkg_symbols()
|
||
exports = pkg_exports()
|
||
if not reg:
|
||
fail("G5", "init.py 未解析到任何 env 注册函数名，三处注册同步无法核对：%s" % rel(INIT_PY))
|
||
miss_def = sorted(n for n in reg if n not in defs and n not in imports)
|
||
miss_exp = sorted(n for n in reg if n not in exports)
|
||
for n in miss_def:
|
||
fail("G5", "注册函数 %s 在包内无 def 定义（三处同步之「定义」缺失）" % n)
|
||
for n in miss_exp:
|
||
fail("G5", "注册函数 %s 未在 __init__.py 导出（三处同步之「导出」缺失）" % n)
|
||
phantom = sorted(n for n in exports
|
||
if n not in defs and n not in imports
|
||
and isinstance(n, str) and IDENT_RE.match(n) and n not in NOISE and n not in stems)
|
||
for n in phantom:
|
||
fail("G5", "__init__.py 导出的 %s 在包内既无定义也无导入（幽灵导出）" % n)
|
||
STATS["G5"] = {"registered": len(reg), "defs": len(defs), "exports": len(exports),
|
||
"miss_def": miss_def, "miss_export": miss_exp, "phantom": phantom}
|
||
print(" G5 三处注册同步：env 注册 %d 个 / 包内 def %d 个 / __init__ 导出 %d 个；缺定义 %d、缺导出 %d、幽灵导出 %d"
|
||
% (len(reg), len(defs), len(exports), len(miss_def), len(miss_exp), len(phantom)))
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# G6 load_path.py RBAC 路径显式枚举
|
||
# --------------------------------------------------------------------------
|
||
def gate_g6():
|
||
if not os.path.isfile(LOAD_PATH_PY):
|
||
fail("G6", "缺少 scripts/load_path.py：%s" % rel(LOAD_PATH_PY))
|
||
STATS["G6"] = {"paths": 0, "wildcard": 0}
|
||
return
|
||
tree, src = parse_ast(LOAD_PATH_PY)
|
||
paths = []
|
||
if tree is not None:
|
||
for node in ast.walk(tree):
|
||
s = const_str(node)
|
||
if s and s.strip().startswith("/"):
|
||
paths.append(s.strip())
|
||
if not paths:
|
||
paths = PATH_LITERAL_RE.findall(src or "")
|
||
if not paths:
|
||
fail("G6", "load_path.py 未解析到任何显式 RBAC 路径（必须逐条枚举注册路径）")
|
||
uniq = sorted(set(paths))
|
||
wild = [p for p in uniq if ("%" in p or "*" in p or "?" in p)]
|
||
for p in wild:
|
||
fail("G6", "RBAC 路径含通配符（禁止 %%/*/?，必须显式枚举）：%s" % p)
|
||
no_slash = [p for p in uniq if not p.startswith("/")]
|
||
for p in no_slash:
|
||
fail("G6", "RBAC 路径未以 / 开头：%s" % p)
|
||
if len(uniq) < 3:
|
||
warn("G6", "显式枚举 RBAC 路径仅 %d 条，建议覆盖模块全部对外接口" % len(uniq))
|
||
STATS["G6"] = {"paths": len(uniq), "wildcard": len(wild), "sample": uniq[:5]}
|
||
print(" G6 load_path.py 显式 RBAC 路径 %d 条，通配符命中 %d 条（样例 %s）"
|
||
% (len(uniq), len(wild), uniq[:3]))
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 主流程
|
||
# --------------------------------------------------------------------------
|
||
GATES = (
|
||
("G1", "py_compile 全量编译", gate_g1),
|
||
("G2", "models 表定义四段式与硬约束", gate_g2),
|
||
("G3", "json CRUD 契约", gate_g3),
|
||
("G4", "禁硬编码库名 + sqlor 白名单", gate_g4),
|
||
("G5", "函数三处注册同步", gate_g5),
|
||
("G6", "load_path.py RBAC 路径显式枚举", gate_g6),
|
||
)
|
||
|
||
|
||
def main():
|
||
print("=" * 78)
|
||
print("pbl_blueprint M1a verify_gate —— 严格退出码门禁 G1~G6（真实断言，非注释大纲）")
|
||
print("MODULE_ROOT = %s" % MODULE_ROOT)
|
||
print("python = %s" % sys.version.split()[0])
|
||
print("=" * 78)
|
||
for gid, title, fn in GATES:
|
||
before = len(FAILURES)
|
||
print("[%s] %s" % (gid, title))
|
||
try:
|
||
fn()
|
||
except Exception as e: # noqa: BLE001 —— 门禁自身异常也必须判失败，不得静默 rc=0
|
||
fail(gid, "门禁执行异常 %s: %s: %s" % (gid, type(e).__name__, e))
|
||
print("[%s] %s" % (gid, "PASS" if len(FAILURES) == before else "FAIL"))
|
||
print("-" * 78)
|
||
|
||
if WARNS:
|
||
print("警告（不阻断，共 %d 条）：" % len(WARNS))
|
||
for g, m in WARNS:
|
||
print(" WARN [%s] %s" % (g, m))
|
||
print("-" * 78)
|
||
|
||
if FAILURES:
|
||
print("失败明细（共 %d 条）：" % len(FAILURES))
|
||
for g, m in FAILURES:
|
||
print(" FAIL [%s] %s" % (g, m))
|
||
gates_failed = sorted(set(g for g, _m in FAILURES))
|
||
print("失败门禁编号：%s" % ",".join(gates_failed))
|
||
print("STATS: %s" % json.dumps(STATS, ensure_ascii=False, sort_keys=True))
|
||
print("GATE RESULT: FAIL rc=1")
|
||
return 1
|
||
|
||
print("全部门禁通过：%s" % ",".join(g for g, _t, _f in GATES))
|
||
print("STATS: %s" % json.dumps(STATS, ensure_ascii=False, sort_keys=True))
|
||
print("GATE RESULT: PASS rc=0")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|