pbls/scripts/verify_import_closure.py
agent.develop c3252391e2 chore(pbls): track pre-existing untracked tool scripts/verify_import_closure.py (Q6 companion, not Q4 scope)
Repo hygiene only - no code change by this task.

scripts/verify_import_closure.py (512 lines, py_compile OK) was already present but
untracked in the apps/pbls working tree when the Q4 task started. It is the companion
verifier for deploy_test round 3 question Q6 (28-module import closure: IMPORT_OK 15/28),
i.e. it belongs to a different task, not to Q4 (build.sh gate 6.4).

It was deliberately kept OUT of the Q4 commit bc4aaa2 (selective git add, never
git add -A) so the Q4 change set stays exactly build.sh + scripts/ddl_gate_check.py +
scripts/ddl/pbls_tables.sql.

Committed separately here because leaving a dirty working tree blocks the develop-stage
git closure gate. Content is untouched; authorship of the tool remains with the Q6 work.
2026-09-25 08:42:40 +08:00

513 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/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())