diff --git a/scripts/verify_import_closure.py b/scripts/verify_import_closure.py new file mode 100644 index 0000000..aedb1ff --- /dev/null +++ b/scripts/verify_import_closure.py @@ -0,0 +1,512 @@ +#!/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: + missing = getattr(exc, "name", "") or "" + if missing.split(".")[0] in BASE_TOPLEVEL: + return { + "module": name, "code": "RUNTIME_IMPORT_SKIP_BASE_ABSENT", + "severity": "SKIP", + "detail": "基座未安装（Q5 未解除）：%s" % exc, + "path": repo, + } + return { + "module": name, "code": "RUNTIME_IMPORT_FAIL", "severity": "FAIL", + "detail": str(exc), "path": repo, + } + except ImportError as exc: + return { + "module": name, "code": "RUNTIME_IMPORT_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())