322 lines
14 KiB
Python
322 lines
14 KiB
Python
#!/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": "<module>",
|
||
"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())
|