#!/usr/bin/env python3 # -*- coding: utf-8 -*- """pbls 28 模块 import 闭包核验器（Q6 修复配套工具）。 背景（projects/pbls/docs/04-deploy/deploy-test-questions.md Q6）： deploy_test 第 3 轮实测 `IMPORT_OK 15 / 28`，13 个模块报 `No module named 'appPublic'` / `No module named 'ahserver.serverenv'` / `cannot import name 'sor' from 'sqlor'` / `cannot import name 'load_entity' from 'entity.init'`。 根因两类： (A) 基座包未 `pip install -e`（Q5 外部依赖源缺口）→ 模块包顶层 import 基座即硬失败； (B) `modules/entity/` 仓库根同时存在 `__init__.py`（使根目录成为包）与 `init/` 种子数据目录 → `entity.init` 被解析成根下的命名空间目录，遮蔽真实实现 `entity/entity/init.py`，导致 `cannot import name 'load_entity'`。 本脚本做两层核验，**不依赖基座已安装**（Q5 未解除也能跑）： L1 静态闭包（ast）：扫描每个模块包内全部 .py 的 import 语句，把基座导入路径 （appPublic/ahserver/sqlor/appbase/rbac/accounting/bricks_for_python/xls2ddl） 逐条对齐到基座真实包结构（磁盘存在性核验），路径不存在 = FAIL。 L2 命名空间核验：检测「仓库根 __init__.py + 根级 init/ 目录」并存的歧义结构， 以及包入口 re-export 的符号能否在包内静态定位到定义。 L3 运行态闭包（可选）：把基座根目录压入 sys.path 后真实 `import {module}`， 基座缺失时以 SKIP 记录（不当 FAIL），基座可用后即成为 28/28 的硬判据。 用法： python3 apps/pbls/scripts/verify_import_closure.py # 核验 + 打印报告 python3 apps/pbls/scripts/verify_import_closure.py --json out.json # 落盘机器可读结果 python3 apps/pbls/scripts/verify_import_closure.py --strict # 任一 FAIL 退出码 1 退出码：0 = 无 FAIL；1 = 存在 FAIL（--strict 或默认均如此）。 """ from __future__ import annotations import argparse import ast import importlib import json import os import sys from typing import Dict, List, Optional, Tuple # ---------------------------------------------------------------- 路径常量 # 机构工作空间根 = apps/pbls/scripts/ 的上上上级 WORKSPACE = os.path.abspath( os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "..") ) MODULES_DIR = os.path.join(WORKSPACE, "modules") SPEC_FILE = os.path.join(WORKSPACE, "projects", "pbls", "pbls_spec.json") # 基座真实包结构探测根（按优先级）。Q5 解除后目标机上是 venv/site-packages， # 本地开发态用 pipeline-app/pkgs 与 binpkg_overlay 作为事实源。 BASE_SEARCH_ROOTS = [ "/d/pipeline/pipeline-app/pkgs/apppublic", "/d/pipeline/pipeline-app/pkgs/sqlor", "/d/pipeline/pipeline-app/pkgs/ahserver", "/d/pipeline/pipeline-app/pkgs/appbase", "/d/pipeline/pipeline-app/pkgs/rbac", "/d/pipeline/pipeline-app/pkgs/accounting", "/d/pipeline/pipeline-app/pkgs/bricks_for_python", "/d/pipeline/pipeline-app/pkgs/xls2ddl", "/d/pipeline/binpkg_overlay", os.path.join(WORKSPACE, "apps", "scense", "pkgs"), os.path.join(WORKSPACE, "apps", "pbls", "pkgs"), ] # 基座顶层包名（模块里 import 到这些名字即视为「基座导入」） BASE_TOPLEVEL = { "appPublic", "ahserver", "sqlor", "appbase", "rbac", "accounting", "bricks_for_python", "xls2ddl", } # 28 模块 load_order 兜底（spec 读不到时用，与 pbls_spec.json 保持一致） FALLBACK_LOAD_ORDER = [ "apppublic", "sqlor", "ahserver", "appbase", "rbac", "accounting", "world", "scene", "entity", "script_engine", "scense_runtime", "scense", "scense_game", "pbl_common", "pbl_appcodes", "pbl_governance", "pbl_blueprint", "pbl_template", "pbl_validation", "pbl_compiler", "pbl_agent_runtime", "pbl_evidence", "pbl_assessment", "pbl_domain_ext", "pbl_scense_ext", "pbl_runtime_ext", "pbl_kdb_ext", "pbl_analytics", ] # 基座模块（load_order 前 6 项）不在 modules/ 下开发，只核验其可导入性 BASE_MODULES = {"apppublic", "sqlor", "ahserver", "appbase", "rbac", "accounting"} # ---------------------------------------------------------------- 工具函数 def load_load_order() -> List[str]: """从 pbls_spec.json 读 load_order（28 项）；读不到用兜底常量。""" try: with open(SPEC_FILE, "r", encoding="utf-8") as fh: spec = json.load(fh) order = spec.get("load_order") or [] if isinstance(order, list) and order: return [str(x) for x in order] except Exception: pass return list(FALLBACK_LOAD_ORDER) def module_repo_dir(name: str) -> Optional[str]: """模块仓库目录。目录名可能是 name 或 name 的连字符/下划线变体。""" cands = [name, name.replace("_", "-"), name.replace("-", "_")] for c in cands: p = os.path.join(MODULES_DIR, c) if os.path.isdir(p): return p return None def package_dir(repo: str, name: str) -> Optional[str]: """真实 Python 包目录（module-development-spec：包目录名 = 模块名，非 src）。""" cands = [name, name.replace("_", "-"), name.replace("-", "_")] for c in cands: p = os.path.join(repo, c) if os.path.isdir(p) and os.path.isfile(os.path.join(p, "__init__.py")): return p # 退化：仓库根本身是包 if os.path.isfile(os.path.join(repo, "__init__.py")): return repo return None def iter_py_files(root: str): for dirpath, dirnames, filenames in os.walk(root): dirnames[:] = [ d for d in dirnames if d not in {".git", "__pycache__", "venv", "node_modules", "wwwroot"} ] for fn in filenames: if fn.endswith(".py"): yield os.path.join(dirpath, fn) def resolve_base_path(dotted: str) -> Tuple[bool, str]: """把基座点号路径对齐到磁盘真实文件/包。 返回 (ok, evidence)。ok=False 表示该导入路径在基座真实包结构中不存在 （= 需要修的「缺失基座导入路径」）。 """ parts = dotted.split(".") for base_root in BASE_SEARCH_ROOTS: if not os.path.isdir(base_root): continue # 情况 1：base_root 本身就是包目录（如 .../pkgs/apppublic/appPublic） # 情况 2：base_root 是仓库根，包在其下同名子目录 for holder in (base_root, os.path.join(base_root, parts[0])): if not os.path.isdir(holder): continue cur = holder ok = True for i, seg in enumerate(parts): # holder 已含 parts[0] 时跳过首段 if holder.endswith(parts[0]) and i == 0: continue as_pkg = os.path.join(cur, seg) as_mod = os.path.join(cur, seg + ".py") as_so = os.path.join(cur, seg + ".cpython-310-x86_64-linux-gnu.so") if os.path.isdir(as_pkg): cur = as_pkg elif os.path.isfile(as_mod): cur = as_mod break elif os.path.isfile(as_so): cur = as_so break else: ok = False break if ok and os.path.exists(cur): return True, cur return False, "" def collect_imports(pyfile: str) -> List[Tuple[str, int]]: """ast 解析单个 .py，返回 [(点号路径, 行号)]；语法错误返回空并记录。""" out: List[Tuple[str, int]] = [] try: with open(pyfile, "r", encoding="utf-8", errors="replace") as fh: src = fh.read() tree = ast.parse(src, filename=pyfile) except SyntaxError: return out for node in ast.walk(tree): if isinstance(node, ast.Import): for al in node.names: out.append((al.name, node.lineno)) elif isinstance(node, ast.ImportFrom): if node.level: # 相对导入，不属基座路径 continue mod = node.module or "" if mod: out.append((mod, node.lineno)) for al in node.names: out.append((mod + "." + al.name if mod else al.name, node.lineno)) return out def is_guarded(pyfile: str, lineno: int) -> bool: """判断某行 import 是否处于 try/except 保护块内（惰性/容错导入）。""" try: with open(pyfile, "r", encoding="utf-8", errors="replace") as fh: tree = ast.parse(fh.read(), filename=pyfile) except SyntaxError: return False for node in ast.walk(tree): if isinstance(node, ast.Try): for sub in ast.walk(node): if isinstance(sub, (ast.Import, ast.ImportFrom)) and sub.lineno == lineno: return True return False def is_toplevel(pyfile: str, lineno: int) -> bool: """判断 import 是否在模块顶层（顶层 = 包导入即执行 = 闭包硬依赖）。""" try: with open(pyfile, "r", encoding="utf-8", errors="replace") as fh: tree = ast.parse(fh.read(), filename=pyfile) except SyntaxError: return False toplevel_lines = set() for node in tree.body: for sub in ast.walk(node): if isinstance(sub, (ast.Import, ast.ImportFrom)): toplevel_lines.add(sub.lineno) if isinstance(node, (ast.Import, ast.ImportFrom)): toplevel_lines.add(node.lineno) # 函数/类体内的 import 不算顶层 nested = set() for node in ast.walk(tree): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): for sub in ast.walk(node): if isinstance(sub, (ast.Import, ast.ImportFrom)): nested.add(sub.lineno) return lineno in toplevel_lines and lineno not in nested # ---------------------------------------------------------------- 核验项 def check_namespace_ambiguity(name: str, repo: str) -> List[Dict]: """L2：命名空间歧义核验（Q6 缺陷 2）。 歧义结构 = 仓库根有 __init__.py（根成为包）且根下有 init/ 目录（种子数据）。 此时 `{name}.init` 会被解析为根下 init/ 命名空间目录，遮蔽 `{name}/{name}/init.py` 的真实实现 → `cannot import name 'load_xxx'`。 """ issues: List[Dict] = [] root_init = os.path.join(repo, "__init__.py") root_initdir = os.path.join(repo, "init") if os.path.isfile(root_init) and os.path.isdir(root_initdir): issues.append({ "module": name, "code": "NS_AMBIGUITY_ROOT_INIT_VS_INITDIR", "severity": "FAIL", "detail": ( "仓库根同时存在 __init__.py 与 init/ 种子目录：`%s.init` 会解析到根下 " "init/ 命名空间目录，遮蔽真实实现 %s/%s/init.py" % (name, name, name) ), "path": root_init, }) if os.path.isfile(root_init): issues.append({ "module": name, "code": "ROOT_INIT_SHIM_PRESENT", "severity": "WARN", "detail": "仓库根 __init__.py 使根目录成为包；pyproject packages.find 只装内层包，" "根 shim 在 pip install -e 后不可达，建议移除或改为纯文档说明", "path": root_init, }) return issues def check_base_imports(name: str, pkg: str) -> List[Dict]: """L1：基座导入路径对齐核验（Q6 缺陷 1）。""" issues: List[Dict] = [] seen = set() for pyfile in iter_py_files(pkg): for dotted, lineno in collect_imports(pyfile): top = dotted.split(".")[0] if top not in BASE_TOPLEVEL: continue key = (dotted, os.path.relpath(pyfile, pkg), lineno) if key in seen: continue seen.add(key) ok, evidence = resolve_base_path(dotted) guarded = is_guarded(pyfile, lineno) toplevel = is_toplevel(pyfile, lineno) if not ok: issues.append({ "module": name, "code": "BASE_PATH_MISSING", "severity": "FAIL", "detail": "基座导入路径在真实包结构中不存在: %s" % dotted, "path": "%s:%d" % (os.path.relpath(pyfile, WORKSPACE), lineno), }) elif toplevel and not guarded: issues.append({ "module": name, "code": "BASE_IMPORT_UNGUARDED_TOPLEVEL", "severity": "WARN", "detail": ( "顶层无保护基座导入 `%s`（基座未 pip install -e 时包导入即失败）；" "已解析到 %s" % (dotted, evidence) ), "path": "%s:%d" % (os.path.relpath(pyfile, WORKSPACE), lineno), }) return issues def check_runtime_import(name: str, repo: str, pkg: str) -> Dict: """L3：运行态 import 闭包（基座缺失记 SKIP，不当 FAIL）。""" parent = os.path.dirname(pkg) if parent not in sys.path: sys.path.insert(0, parent) for br in BASE_SEARCH_ROOTS: if os.path.isdir(br) and br not in sys.path: sys.path.append(br) import_name = os.path.basename(pkg) try: importlib.invalidate_caches() mod = importlib.import_module(import_name) loaded_from = getattr(mod, "__file__", "") or "" if not loaded_from.startswith(pkg): return { "module": name, "code": "RUNTIME_IMPORT_WRONG_TARGET", "severity": "FAIL", "detail": "import %s 解析到 %s（非本模块包 %s）" % (import_name, loaded_from, pkg), "path": loaded_from, } return { "module": name, "code": "RUNTIME_IMPORT_OK", "severity": "OK", "detail": "import %s 成功 → %s" % (import_name, loaded_from), "path": loaded_from, } except ModuleNotFoundError as exc: # 缺基座/第三方模块 = 依赖未 pip install -e（Q5 派生症状），非代码缺陷 -> SKIP return { "module": name, "code": "RUNTIME_IMPORT_SKIP_DEP_ABSENT", "severity": "SKIP", "detail": "依赖未安装（Q5 未解除；pip install -e 基座后复跑即转 OK）：%s" % exc, "path": repo, } except ImportError as exc: # cannot import name X = 符号级缺陷（如 entity 命名空间歧义）-> FAIL，必须修 return { "module": name, "code": "RUNTIME_IMPORT_SYMBOL_FAIL", "severity": "FAIL", "detail": str(exc), "path": repo, } except Exception as exc: # 基座副作用（无 DB/无配置）不计闭包失败 return { "module": name, "code": "RUNTIME_IMPORT_SKIP_SIDE_EFFECT", "severity": "SKIP", "detail": "%s: %s" % (type(exc).__name__, exc), "path": repo, } def check_entry_symbol(name: str, pkg: str) -> List[Dict]: """L2b：包入口 re-export 的 load_{module} 能否在包内静态定位到定义。""" issues: List[Dict] = [] init_py = os.path.join(pkg, "__init__.py") if not os.path.isfile(init_py): issues.append({ "module": name, "code": "PKG_INIT_MISSING", "severity": "FAIL", "detail": "包目录缺 __init__.py: %s" % pkg, "path": pkg, }) return issues loader = "load_" + name.replace("-", "_") found_def = False for pyfile in iter_py_files(pkg): try: with open(pyfile, "r", encoding="utf-8", errors="replace") as fh: tree = ast.parse(fh.read(), filename=pyfile) except SyntaxError: continue for node in ast.walk(tree): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == loader: found_def = True if not found_def: issues.append({ "module": name, "code": "LOADER_DEF_MISSING", "severity": "WARN", "detail": "包内未找到 %s() 定义（若由宿主直接注册可忽略）" % loader, "path": pkg, }) return issues # ---------------------------------------------------------------- 主流程 def main() -> int: ap = argparse.ArgumentParser(description="pbls 28 模块 import 闭包核验器") ap.add_argument("--json", dest="json_out", default="", help="机器可读结果落盘路径") ap.add_argument("--strict", action="store_true", help="任一 FAIL 退出码 1") ap.add_argument("--no-runtime", action="store_true", help="跳过 L3 运行态 import") args = ap.parse_args() order = load_load_order() report: Dict[str, object] = { "tool": "verify_import_closure", "spec": SPEC_FILE, "load_order_count": len(order), "modules": [], } total_ok = total_fail = total_warn = total_skip = 0 for name in order: entry: Dict[str, object] = {"module": name, "issues": []} if name in BASE_MODULES: ok, ev = resolve_base_path(name if name != "apppublic" else "appPublic") entry["kind"] = "base" entry["issues"].append({ "module": name, "code": "BASE_PKG_RESOLVED" if ok else "BASE_PKG_UNRESOLVED", "severity": "OK" if ok else "FAIL", "detail": ("基座包已解析: %s" % ev) if ok else "基座包在探测根中未找到（需 pip install -e，Q5）", "path": ev or name, }) else: entry["kind"] = "business" repo = module_repo_dir(name) if not repo: entry["issues"].append({ "module": name, "code": "REPO_MISSING", "severity": "FAIL", "detail": "modules/ 下未找到模块仓库", "path": name, }) else: entry["repo"] = os.path.relpath(repo, WORKSPACE) pkg = package_dir(repo, name) if not pkg: entry["issues"].append({ "module": name, "code": "PKG_DIR_MISSING", "severity": "FAIL", "detail": "未找到包目录（包目录名应 = 模块名，含 __init__.py）", "path": repo, }) else: entry["package"] = os.path.relpath(pkg, WORKSPACE) entry["issues"] += check_namespace_ambiguity(name, repo) entry["issues"] += check_base_imports(name, pkg) entry["issues"] += check_entry_symbol(name, pkg) if not args.no_runtime: entry["issues"].append(check_runtime_import(name, repo, pkg)) sev = [i["severity"] for i in entry["issues"]] entry["status"] = ( "FAIL" if "FAIL" in sev else ("WARN" if "WARN" in sev else ("SKIP" if "SKIP" in sev else "OK")) ) total_fail += sev.count("FAIL") total_warn += sev.count("WARN") total_skip += sev.count("SKIP") if entry["status"] == "OK": total_ok += 1 report["modules"].append(entry) report["summary"] = { "modules_total": len(order), "modules_ok": total_ok, "issues_fail": total_fail, "issues_warn": total_warn, "issues_skip": total_skip, } # ---- 打印 print("=" * 78) print("pbls import 闭包核验（load_order %d 项）" % len(order)) print("=" * 78) for entry in report["modules"]: print("[%s] %-20s %s" % ( entry["status"], entry["module"], entry.get("package") or entry.get("repo") or "")) for iss in entry["issues"]: if iss["severity"] in ("OK",): continue print(" %-34s %-4s %s" % (iss["code"], iss["severity"], iss["detail"])) print(" @ %s" % iss["path"]) print("-" * 78) s = report["summary"] print("模块 OK %d/%d | FAIL 项 %d | WARN 项 %d | SKIP 项 %d" % (s["modules_ok"], s["modules_total"], s["issues_fail"], s["issues_warn"], s["issues_skip"])) print("=" * 78) if args.json_out: outp = args.json_out if not os.path.isabs(outp): outp = os.path.join(WORKSPACE, outp) os.makedirs(os.path.dirname(outp), exist_ok=True) with open(outp, "w", encoding="utf-8") as fh: json.dump(report, fh, ensure_ascii=False, indent=2) print("JSON 结果已落盘: %s" % outp) if total_fail: return 1 return 0 if __name__ == "__main__": sys.exit(main())