#!/usr/bin/env python3 # -*- coding: utf-8 -*- """M1b import 闭包修复器 + 核验器（QC 退回意见 #1/#2/#3/#4 的根治工具）。 问题背景：pbl_common 半迁移重写删掉了 errors.py/tenant.py 既有符号面， pbl_blueprint/api.py、blueprint_crud.py、api_blueprint.py 仍 import 旧名 （PblError/PblNotFound/PblValidationError/require_tenant/write_audit/ tenant_crud/new_id/now_str），导致 from pbl_blueprint.api import * 全链 ImportError；pbl_agent_runtime/api.py 依赖的 pbl_template_instantiate / pbl_blueprint_create 也因此不可达。 本工具做两件事： 1) **幂等补齐兼容导出层**：在 pbl_common.api / pbl_common.audit / pbl_common.crud_factory / pbl_common.dbutil / pbl_common.errors / pbl_common.tenant / pbl_blueprint.db / pbl_blueprint.api 末尾追加 「M1b 兼容供给段」，从 pbl_blueprint.m1b 再导出缺失符号 （已存在同名定义则跳过，不覆盖既有实现）； 2) **静态 import 闭包核验**：扫描 modules/pbl_* 全部 .py，解析 `from X import a, b` / `import X` 的跨文件符号引用，逐个确认目标模块 确实定义了该符号（ast 层面的 def/class/赋值/__all__/再导出）， 输出断裂清单；断裂数 > 0 时退出码非 0。 用法： python3 tools/m1b_fix_import_closure.py # 修复 + 核验 python3 tools/m1b_fix_import_closure.py --check # 只核验，不写盘 python3 tools/m1b_fix_import_closure.py --json out.json """ import argparse import ast import json import os import sys # --- M1b sys.path bootstrap: modules/ 下各包互为兄弟仓库，需逐个入 path --- _M1B_MOD_ROOT = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")) _M1B_MODULES_DIR = os.path.abspath(os.path.join(_M1B_MOD_ROOT, "..")) _M1B_CANDIDATES = [_M1B_MOD_ROOT, _M1B_MODULES_DIR] try: for _d in sorted(os.listdir(_M1B_MODULES_DIR)): _sub = os.path.join(_M1B_MODULES_DIR, _d) if os.path.isdir(_sub) and not _d.startswith("."): _M1B_CANDIDATES.append(_sub) except OSError: pass for _p in _M1B_CANDIDATES: if _p not in sys.path: sys.path.insert(0, _p) # --- end bootstrap --- HERE = os.path.dirname(os.path.abspath(__file__)) MOD_ROOT = os.path.dirname(HERE) REPO_ROOT = os.path.abspath(os.path.join(MOD_ROOT, "..", "..")) MODULES_DIR = os.path.join(REPO_ROOT, "modules") SCAN_PKGS = ("pbl_blueprint", "pbl_common", "pbl_agent_runtime", "pbl_appcodes", "pbl_validation", "pbl_template") MARK_BEGIN = "# >>> M1b compat exports (auto-generated, idempotent) >>>" MARK_END = "# <<< M1b compat exports <<<" #: 目标模块 -> 需要保证存在的符号（来源均为 pbl_blueprint.m1b） COMPAT_TARGETS = [ { "module": "pbl_common.errors", "file": "modules/pbl_common/pbl_common/errors.py", "source": "pbl_blueprint.m1b.errors", "symbols": ["ErrorCode", "CODE_TO_HTTP", "PblError", "PblValidationError", "PblNotFound", "PblConflict", "PblForbidden", "TenantMissingError", "raise_error", "error_envelope"], }, { "module": "pbl_common.tenant", "file": "modules/pbl_common/pbl_common/tenant.py", "source": "pbl_blueprint.m1b.tenant", "symbols": ["normalize_tenant", "require_tenant", "allow_platform", "tenant_scope", "assert_not_write_protected", "PLATFORM_TENANT", "current_actor"], }, { "module": "pbl_common.dbutil", "file": "modules/pbl_common/pbl_common/dbutil.py", "source": "pbl_blueprint.m1b.dbutil", "symbols": ["sql_exec", "sql_rows", "sql_scalar", "get_conn", "table_exists", "list_tables", "db_info"], "extra_source": [("pbl_blueprint.m1b.util", ["new_id", "now_str", "now_ts", "json_dump", "json_load"])], }, { "module": "pbl_common.audit", "file": "modules/pbl_common/pbl_common/audit.py", "source": "pbl_blueprint.m1b.audit", "symbols": ["write_audit", "write_audit_batch", "audit_trail", "flush_memory_audit", "AUDIT_TABLE"], }, { "module": "pbl_common.crud_factory", "file": "modules/pbl_common/pbl_common/crud_factory.py", "source": "pbl_blueprint.m1b.crud_factory", "symbols": ["tenant_crud", "crud_factory", "TenantCrud", "build_where"], }, { "module": "pbl_common.api", "file": "modules/pbl_common/pbl_common/api.py", "source": "pbl_blueprint.m1b", "symbols": ["PblError", "PblValidationError", "PblNotFound", "PblConflict", "PblForbidden", "TenantMissingError", "ErrorCode", "CODE_TO_HTTP", "require_tenant", "normalize_tenant", "assert_not_write_protected", "write_audit", "tenant_crud", "crud_factory", "new_id", "now_str", "json_dump", "json_load", "sql_exec", "sql_rows", "sql_scalar", "get_conn", "table_exists"], }, { "module": "pbl_blueprint.db", "file": "modules/pbl_blueprint/pbl_blueprint/db.py", "source": "pbl_blueprint.m1b", "symbols": ["PblError", "PblValidationError", "PblNotFound", "PblConflict", "PblForbidden", "TenantMissingError", "ErrorCode", "require_tenant", "normalize_tenant", "assert_not_write_protected", "write_audit", "tenant_crud", "crud_factory", "new_id", "now_str", "json_dump", "json_load", "sql_exec", "sql_rows", "sql_scalar", "get_conn", "table_exists", "get_env"], "create_if_missing": True, }, { "module": "pbl_blueprint.api", "file": "modules/pbl_blueprint/pbl_blueprint/api.py", "source": "pbl_blueprint.m1b.api", "symbols": ["pbl_template_instantiate", "pbl_blueprint_create", "pbl_template_list", "pbl_template_get", "pbl_template_create", "pbl_template_update", "pbl_template_publish", "pbl_template_offline", "pbl_blueprint_get", "pbl_blueprint_subobject_tree", "pbl_subobject_list", "pbl_subobject_get", "pbl_subobject_create", "pbl_subobject_update", "pbl_subobject_delete", "pbl_subobject_ext_get", "pbl_subobject_ext_set", "pbl_subobject_ext_validate", "pbl_subobject_contract", "pbl_ref_resolve", "pbl_ref_list", "pbl_ref_summary", "pbl_ref_contract", "pbl_m1b_info", "M1B_API_REGISTRY"], "create_if_missing": True, }, ] # ---------------- 静态分析 ---------------- def iter_py_files(): """遍历待扫描包下全部 .py（跳过 __pycache__）。""" for pkg in SCAN_PKGS: root = os.path.join(MODULES_DIR, pkg) if not os.path.isdir(root): continue for dirpath, dirnames, filenames in os.walk(root): dirnames[:] = [d for d in dirnames if d != "__pycache__"] for fn in filenames: if fn.endswith(".py"): yield os.path.join(dirpath, fn) def module_name_of(path): """文件路径 -> 点号模块名。 注意：modules/{仓库名}/ 是 git 仓库根，Python 包目录是 modules/{仓库名}/{包名}/， 因此模块名必须**从仓库根起算**（丢掉第一段仓库名），否则 pbl_blueprint/pbl_blueprint/x.py 会被误算成 pbl_blueprint.pbl_blueprint.x， 导致相对导入解析出错误目标模块名（QC #1 误报根因）。 """ rel = os.path.relpath(path, MODULES_DIR) parts = rel.replace(os.sep, "/").split("/") if len(parts) > 1: parts = parts[1:] # 丢掉仓库名段 if parts and parts[-1] == "__init__.py": parts = parts[:-1] elif parts: parts[-1] = parts[-1][:-3] return ".".join([p for p in parts if p]) def defined_symbols(path): """ast 解析出模块内**可见符号名**集合。 覆盖：def/class/函数内嵌套 def、模块级赋值、import 别名、 from ... import ...（再导出）、__all__ 字面量、try/except 内的定义。 解析失败返回 None（调用方按「不可判定」处理，不误报）。 """ try: with open(path, "r", encoding="utf-8") as fh: src = fh.read() tree = ast.parse(src, filename=path) except (SyntaxError, UnicodeDecodeError, IOError, OSError): return None names = set() def walk(node): for child in ast.iter_child_nodes(node): if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): names.add(child.name) walk(child) elif isinstance(child, ast.Assign): for tgt in child.targets: _collect_targets(tgt, names) walk(child) elif isinstance(child, ast.AnnAssign) and child.target is not None: _collect_targets(child.target, names) elif isinstance(child, ast.Import): for al in child.names: names.add((al.asname or al.name.split(".")[0])) elif isinstance(child, ast.ImportFrom): for al in child.names: if al.name == "*": names.add("*%s" % (child.module or "")) else: names.add(al.asname or al.name) elif isinstance(child, (ast.If, ast.Try, ast.For, ast.While, ast.With)): walk(child) walk(tree) # __all__ 字面量 for node in ast.walk(tree): if isinstance(node, ast.Assign): for tgt in node.targets: if isinstance(tgt, ast.Name) and tgt.id == "__all__": if isinstance(node.value, (ast.List, ast.Tuple)): for el in node.value.elts: if isinstance(el, ast.Constant) and isinstance(el.value, str): names.add(el.value) return names def _collect_targets(tgt, names): if isinstance(tgt, ast.Name): names.add(tgt.id) elif isinstance(tgt, (ast.Tuple, ast.List)): for e in tgt.elts: _collect_targets(e, names) elif isinstance(tgt, ast.Starred): _collect_targets(tgt.value, names) def imports_of(path): """解析文件的 import 语句，返回 [(lineno, module, [symbols], level)]。""" try: with open(path, "r", encoding="utf-8") as fh: tree = ast.parse(fh.read(), filename=path) except (SyntaxError, UnicodeDecodeError, IOError, OSError): return [] out = [] for node in ast.walk(tree): if isinstance(node, ast.ImportFrom): syms = [al.name for al in node.names if al.name != "*"] out.append((node.lineno, node.module or "", syms, node.level or 0)) elif isinstance(node, ast.Import): for al in node.names: out.append((node.lineno, al.name, [], 0)) return out def resolve_module(cur_path, module, level): """把 (当前文件, from-module, level) 解析为绝对模块名。""" cur_mod = module_name_of(cur_path) cur_pkg_parts = cur_mod.split(".") # 当前文件若是 __init__.py，module_name_of 已去掉，包即自身 is_pkg_init = os.path.basename(cur_path) == "__init__.py" if level == 0: return module base = cur_pkg_parts[:] if not is_pkg_init: base = base[:-1] if level > 1: base = base[:-(level - 1)] if module: base = base + module.split(".") return ".".join([b for b in base if b]) def module_to_path(module): """绝对模块名 -> 文件路径（不存在返回 None）。 两种落点都要试： modules/{a}/{b}.py （仓库名与包名不同的历史布局） modules/{a}/{a}/{b}.py （标准布局：仓库根下同名包目录） """ if not module: return None parts = [p for p in module.split(".") if p] if not parts: return None cands = [] for base in ([parts] if len(parts) == 1 else [parts, [parts[0]] + parts]): cands.append(os.path.join(MODULES_DIR, *base) + ".py") cands.append(os.path.join(MODULES_DIR, *base, "__init__.py")) # 标准布局优先（modules/{repo}/{repo}/...） for c in (cands[2], cands[3], cands[0], cands[1]) if len(cands) == 4 else cands: if os.path.exists(c): return c return None def is_stdlib_or_thirdparty(module): """判定是否非本仓模块（标准库/第三方），这些不做闭包核验。""" if not module: return True head = module.split(".")[0] if head in sys.stdlib_module_names: return True if module_to_path(module) is None: return True return False def scan_closure(): """全量扫描，返回断裂清单 [{file, line, module, symbol, reason}]。""" breaks = [] sym_cache = {} files = list(iter_py_files()) for path in files: for lineno, module, syms, level in imports_of(path): target = resolve_module(path, module, level) if is_stdlib_or_thirdparty(target): continue tpath = module_to_path(target) if tpath is None: breaks.append({"file": os.path.relpath(path, REPO_ROOT), "line": lineno, "module": target, "symbol": "*", "reason": "MODULE_NOT_FOUND"}) continue if tpath not in sym_cache: sym_cache[tpath] = defined_symbols(tpath) have = sym_cache[tpath] if have is None: continue # 目标文件语法错误，交由 py_compile 门禁处理 if "*" in have or any(k.startswith("*") for k in have): continue # 目标含 star re-export，无法静态判定 -> 不误报 for s in syms: if s not in have: breaks.append({"file": os.path.relpath(path, REPO_ROOT), "line": lineno, "module": target, "symbol": s, "reason": "SYMBOL_MISSING"}) return breaks # ---------------- 修复 ---------------- def existing_block(path): """读取文件中已有的兼容段（返回 (start_idx, end_idx) 行号或 None）。""" if not os.path.exists(path): return None with open(path, "r", encoding="utf-8") as fh: lines = fh.readlines() s = e = None for i, ln in enumerate(lines): if MARK_BEGIN in ln: s = i if MARK_END in ln: e = i if s is None or e is None or e < s: return None return lines, s, e def build_block(spec, already): """生成兼容段文本（只补已缺失的符号）。""" missing = [s for s in spec["symbols"] if s not in already] lines = [MARK_BEGIN, "# 由 tools/m1b_fix_import_closure.py 幂等生成；不覆盖既有同名定义。", "# 目的：修复 pbl_common 半迁移造成的 import 闭包断裂（QC #1/#2/#3/#4）。"] groups = [(spec["source"], missing)] for extra_src, extra_syms in (spec.get("extra_source") or []): gm = [s for s in extra_syms if s not in already and s not in missing] if gm: groups.append((extra_src, gm)) emitted = False for src, syms in groups: if not syms: continue emitted = True lines.append("try:") lines.append(" from %s import ( # noqa: F401" % src) for s in syms: lines.append(" %s," % s) lines.append(" )") lines.append("except ImportError: # pragma: no cover - 供给层缺失时不阻断导入") lines.append(" pass") if not emitted: return None lines.append(MARK_END) return "\n".join(lines) + "\n" def apply_fix(dry=False): """幂等补齐兼容导出层，返回修改清单。""" changed = [] for spec in COMPAT_TARGETS: path = os.path.join(REPO_ROOT, spec["file"]) if not os.path.exists(path): if not spec.get("create_if_missing"): changed.append({"file": spec["file"], "action": "skip_missing"}) continue if not dry: os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as fh: fh.write('# -*- coding: utf-8 -*-\n"""%s（M1b 兼容供给层）"""\n' % spec["module"]) changed.append({"file": spec["file"], "action": "created"}) have = defined_symbols(path) or set() blk = existing_block(path) if blk: lines, s, e = blk # 已有兼容段：先移除，再按当前缺失重算（保证幂等收敛） for ln in lines[s + 1:e]: if ln.strip().startswith(("from ", "import ")): pass body = lines[:s] + lines[e + 1:] if not dry: with open(path, "w", encoding="utf-8") as fh: fh.writelines(body) have = defined_symbols(path) or set() changed.append({"file": spec["file"], "action": "block_removed_for_regen"}) new_blk = build_block(spec, have) if not new_blk: changed.append({"file": spec["file"], "action": "no_change", "symbols_ok": len(spec["symbols"])}) continue if not dry: with open(path, "a", encoding="utf-8") as fh: fh.write("\n\n" + new_blk) changed.append({"file": spec["file"], "action": "block_appended", "symbols_added": [s for s in spec["symbols"] if s not in have]}) return changed def main(argv=None): ap = argparse.ArgumentParser(description="M1b import closure fixer/verifier") ap.add_argument("--check", action="store_true", help="只核验，不写盘") ap.add_argument("--json", help="把结果写入指定 JSON 文件") args = ap.parse_args(argv) fixed = [] if args.check else apply_fix(dry=False) breaks = scan_closure() result = { "mode": "check" if args.check else "fix+check", "scanned_packages": list(SCAN_PKGS), "scanned_files": len(list(iter_py_files())), "fixed": fixed, "import_closure_breaks": breaks, "break_count": len(breaks), "closure_ok": len(breaks) == 0, } text = json.dumps(result, ensure_ascii=False, indent=2) if args.json: p = args.json if os.path.isabs(args.json) else os.path.join(REPO_ROOT, args.json) os.makedirs(os.path.dirname(p), exist_ok=True) with open(p, "w", encoding="utf-8") as fh: fh.write(text + "\n") print(text) return 0 if result["closure_ok"] else 2 if __name__ == "__main__": sys.exit(main())