#!/usr/bin/env python3 # -*- coding: utf-8 -*- """跨模块契约符号闭包静态核验（Bug HoSBcdZY4G7IxhaFiIE07 同类问题全量排查）。 背景 ---- 原 Bug YyDumUuSK9R9Q_3Ai23AD 的 develop 侧根因 G：``pbl_common.api`` 缺少 ``crud`` 符号，导致下游 ``pbl_assessment`` / ``pbl_scense_ext`` 在包导入期硬失败 （ImportError），应用进程根本起不来（healthz=000 / pbls.py 进程数=0）。 该缺陷的**同类问题**不止报出来的那一个：任何 ``from pbl_common.X import Y`` 而 ``Y`` 在 ``pbl_common`` 对应模块里并不存在的情况，都会造成同样的导入期硬失败。 本脚本对全部 pbl_* 模块做静态 AST 闭包核验，不依赖运行时基座（基座未 pip install -e、内网 pypi 不可达时依然可跑），因此可在开发机/CI 上作为 提交前门禁使用。 检查项 ------ 1. 收集 ``modules/pbl_common/pbl_common/*.py`` 每个文件顶层可导出符号 （def / async def / class / 赋值名 / __all__ 字面量 / from-import 再导出名）。 2. 扫描全部 ``modules/pbl_*/`` 与 ``apps/pbls/`` 下 .py 的 ``from pbl_common[.sub] import a, b, c`` 语句，逐个符号核对是否存在。 3. 额外核对 ``pbl_common/api.py`` 的 CONTRACT_SYMBOLS 与 __all__ 声明的符号 是否真的在 api.py 命名空间内可解析（声明了但没实现 = 死契约）。 4. 输出机器可读结论：PASS / FAIL + 明细，退出码 0/1，可直接接 CI。 用法:: python3 apps/pbls/scripts/verify_contract_symbols_all_modules.py python3 apps/pbls/scripts/verify_contract_symbols_all_modules.py --json out.json """ from __future__ import annotations import argparse import ast import json import os import sys from typing import Dict, List, Set, Tuple # ---------------------------------------------------------------- 路径常量 HERE = os.path.dirname(os.path.abspath(__file__)) APP_ROOT = os.path.dirname(HERE) # apps/pbls WS_ROOT = os.path.dirname(os.path.dirname(APP_ROOT)) # 机构工作空间 MODULES_DIR = os.path.join(WS_ROOT, "modules") COMMON_PKG = os.path.join(MODULES_DIR, "pbl_common", "pbl_common") # 需要扫描 import 语句的根（模块 + 应用） SCAN_ROOTS: List[str] = [MODULES_DIR, APP_ROOT] # 只关心 pbl_ 前缀的模块目录（避免扫到无关历史仓库） MODULE_PREFIX = "pbl_" # ---------------------------------------------------------------- 工具函数 def _iter_py_files(root: str): """遍历目录下所有 .py 文件（跳过 venv/.git/__pycache__/wwwroot 生成物）。""" skip_dirs = {".git", "__pycache__", "venv", ".venv", "node_modules", "wwwroot", ".pytest_cache", "site-packages"} for dirpath, dirnames, filenames in os.walk(root): dirnames[:] = [d for d in dirnames if d not in skip_dirs] for fn in filenames: if fn.endswith(".py"): yield os.path.join(dirpath, fn) def _literal_names(node: ast.AST) -> Set[str]: """从赋值右侧的 list/tuple/set 字面量里抽出字符串元素（用于 __all__）。""" out: Set[str] = set() if isinstance(node, (ast.List, ast.Tuple, ast.Set)): for elt in node.elts: if isinstance(elt, ast.Constant) and isinstance(elt.value, str): out.add(elt.value) return out def collect_module_symbols(pkg_dir: str) -> Dict[str, Set[str]]: """收集 pbl_common 包内每个子模块的顶层可导出符号。 返回 {子模块名(不含 .py): 符号集合}，另外附加一个聚合键 ``"*"`` 表示 ``from pbl_common import X`` 时包 __init__ 可见的符号。 """ table: Dict[str, Set[str]] = {} for path in sorted(_iter_py_files(pkg_dir)): rel = os.path.relpath(path, pkg_dir) if os.sep in rel: # 只取包顶层模块（pbl_common/xxx.py） continue mod = rel[:-3] try: tree = ast.parse(open(path, encoding="utf-8").read(), filename=path) except SyntaxError as exc: # 语法错本身就是缺陷，记为特殊符号 table[mod] = {"__SYNTAX_ERROR__: %s" % exc} continue syms: Set[str] = set() for node in tree.body: if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): syms.add(node.name) elif isinstance(node, ast.Assign): for tgt in node.targets: if isinstance(tgt, ast.Name): syms.add(tgt.id) # __all__ = [...] 里的名字也算「本模块对外承诺的符号」 if tgt.id in ("__all__", "CONTRACT_SYMBOLS"): syms |= _literal_names(node.value) elif isinstance(tgt, ast.Tuple): for e in tgt.elts: if isinstance(e, ast.Name): syms.add(e.id) elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): syms.add(node.target.id) elif isinstance(node, ast.ImportFrom): # from x import y → y 在本模块命名空间可见（再导出） for al in node.names: syms.add(al.asname or al.name) elif isinstance(node, ast.Import): for al in node.names: syms.add((al.asname or al.name).split(".")[0]) elif isinstance(node, (ast.If, ast.Try)): # 条件/异常分支里的顶层定义同样计入（兼容 shim 常用 try-import） for sub in ast.walk(node): if isinstance(sub, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): syms.add(sub.name) elif isinstance(sub, ast.Assign): for tgt in sub.targets: if isinstance(tgt, ast.Name): syms.add(tgt.id) elif isinstance(sub, ast.ImportFrom): for al in sub.names: syms.add(al.asname or al.name) table[mod] = syms # 包级聚合：__init__.py 的符号 + 所有子模块名本身（from pbl_common import crud 也合法） agg: Set[str] = set(table.get("__init__", set())) agg |= set(table.keys()) for syms in table.values(): agg |= syms table["*"] = agg return table PROVIDER_PREFIXES = ("pbl_",) def collect_imports(path: str) -> List[Tuple[int, str, List[str]]]: """抽出文件里所有 from pbl_*[...] import ... 语句（跨模块契约引用）。 返回 [(lineno, 模块路径, [符号...]), ...] """ try: src = open(path, encoding="utf-8").read() except (OSError, UnicodeDecodeError): return [] try: tree = ast.parse(src, filename=path) except SyntaxError: return [] out: List[Tuple[int, str, List[str]]] = [] for node in ast.walk(tree): if isinstance(node, ast.ImportFrom) and node.module and \ any(node.module == p or node.module.startswith(p + ".") for p in PROVIDER_PREFIXES): names = [al.name for al in node.names if al.name != "*"] if names: out.append((node.lineno, node.module, names)) return out def scan_targets() -> List[str]: """待扫描文件清单：所有 pbl_* 模块 + apps/pbls 下的 .py。""" files: List[str] = [] for root in SCAN_ROOTS: if not os.path.isdir(root): continue if root == MODULES_DIR: for name in sorted(os.listdir(root)): if not name.startswith(MODULE_PREFIX): continue mod_dir = os.path.join(root, name) if os.path.isdir(mod_dir): files.extend(_iter_py_files(mod_dir)) else: files.extend(_iter_py_files(root)) return sorted(set(files)) # ---------------------------------------------------------------- 主核验 def verify() -> Dict: if not os.path.isdir(COMMON_PKG): return {"ok": False, "fatal": "pbl_common 包目录不存在: %s" % COMMON_PKG, "checked_files": 0, "violations": [], "dead_contract": []} symtab = collect_module_symbols(COMMON_PKG) # 全部 pbl_* 模块包的符号表：{包名: {子模块名: 符号集}} pkg_tabs: Dict[str, Dict[str, Set[str]]] = {"pbl_common": symtab} for name in sorted(os.listdir(MODULES_DIR)): if not name.startswith(MODULE_PREFIX) or name == "pbl_common": continue pkg_dir = os.path.join(MODULES_DIR, name, name) if os.path.isdir(pkg_dir): pkg_tabs[name] = collect_module_symbols(pkg_dir) files = scan_targets() violations: List[Dict] = [] for path in files: # provider 包自身内部互引不算跨模块契约违规（内部结构可自由重构） if any(os.path.abspath(path).startswith( os.path.abspath(os.path.join(MODULES_DIR, k, k)) + os.sep) for k in pkg_tabs): continue own_pkg = os.path.relpath(path, MODULES_DIR).split(os.sep)[0] for lineno, mod, names in collect_imports(path): provider = mod.split(".", 0)[0].split(".")[0] tab = pkg_tabs.get(provider) if tab is None: continue # 非本仓 pbl_* 模块（如平台包），跳过 if provider == own_pkg: continue # 包内自引不算跨模块契约 if "." not in mod: avail = tab.get("*", set()) sub = "%s(包)" % provider else: submod = mod.split(".", 1)[1] avail = tab.get(submod, set()) sub = "%s.%s" % (provider, submod) if submod not in tab: violations.append({ "file": os.path.relpath(path, WS_ROOT), "line": lineno, "module": mod, "symbol": "", "reason": "子模块 %s/%s.py 不存在" % (provider, submod), }) continue for nm in names: if nm not in avail: violations.append({ "file": os.path.relpath(path, WS_ROOT), "line": lineno, "module": sub, "symbol": nm, "reason": "符号未在 %s 顶层定义/再导出（导入期 ImportError 风险）" % sub, }) # 死契约核验：api.py 里 CONTRACT_SYMBOLS / __all__ 声明了但命名空间没有 dead: List[Dict] = [] api_path = os.path.join(COMMON_PKG, "api.py") if os.path.isfile(api_path): src = open(api_path, encoding="utf-8").read() tree = ast.parse(src, filename=api_path) declared: Dict[str, Set[str]] = {} defined: Set[str] = set() for node in tree.body: if isinstance(node, ast.Assign): for tgt in node.targets: if isinstance(tgt, ast.Name) and tgt.id in ("__all__", "CONTRACT_SYMBOLS"): declared[tgt.id] = _literal_names(node.value) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): defined.add(node.name) elif isinstance(node, ast.Assign): for tgt in node.targets: if isinstance(tgt, ast.Name): defined.add(tgt.id) elif isinstance(node, ast.ImportFrom): for al in node.names: defined.add(al.asname or al.name) for key, names in declared.items(): for nm in sorted(names): if nm not in defined: dead.append({"list": key, "symbol": nm, "reason": "声明于 %s 但 api.py 命名空间不可解析" % key}) return { "ok": not violations and not dead, "checked_files": len(files), "common_modules": sorted(k for k in symtab if k != "*"), "provider_packages": sorted(pkg_tabs.keys()), "crud_symbol_present": "crud" in symtab.get("api", set()), "violations": violations, "dead_contract": dead, } def main() -> int: ap = argparse.ArgumentParser(description="pbl_common 契约符号跨模块闭包核验") ap.add_argument("--json", dest="json_out", default=None, help="把结果写入指定 JSON 文件") args = ap.parse_args() res = verify() print("=" * 72) print("pbl_common 契约符号闭包核验（Bug HoSBcdZY4G7IxhaFiIE07 同类问题全量排查）") print("=" * 72) if res.get("fatal"): print("FATAL: %s" % res["fatal"]) return 2 print("扫描 .py 文件数 : %d" % res["checked_files"]) print("pbl_common 子模块数 : %d" % len(res["common_modules"])) print("核验 provider 包数 : %d (%s)" % (len(res["provider_packages"]), ", ".join(res["provider_packages"]))) print("api.crud 符号在位 : %s" % res["crud_symbol_present"]) print("跨模块导入符号违规 : %d" % len(res["violations"])) print("死契约(声明未实现) : %d" % len(res["dead_contract"])) for v in res["violations"][:50]: print(" [VIOLATION] %s:%s from %s import %s → %s" % (v["file"], v["line"], v["module"], v["symbol"], v["reason"])) for d in res["dead_contract"][:50]: print(" [DEAD] %s 声明 '%s' → %s" % (d["list"], d["symbol"], d["reason"])) print("-" * 72) print("结论: %s" % ("PASS —— 全部 pbl_* 模块间跨模块导入符号均可解析" if res["ok"] else "FAIL —— 存在导入期硬失败风险，须修复后再部署")) if args.json_out: with open(args.json_out, "w", encoding="utf-8") as fh: json.dump(res, fh, ensure_ascii=False, indent=2) print("结果已写入: %s" % args.json_out) return 0 if res["ok"] else 1 if __name__ == "__main__": sys.exit(main())