#!/usr/bin/env python3 # -*- coding: utf-8 -*- """M1b import 闭包修复器 v2（幂等）——落地 QC 退回意见 #1/#2。 v1 的缺陷（QC #1 根因）：在 db.py 里写一条 ``from pbl_blueprint.m1b import (23 个符号)``， 只要其中**一个**符号（get_env）在 m1b 导出面缺失，整条 import 抛 ImportError 被 try/except 静默吞掉，其余 22 个兼容回补全部失效。v2 三条铁律： 1. **逐符号**采纳，绝不整块 import（单符号缺失不牵连其余）； 2. 供给方为**包内自足**模块 pbl_blueprint.m1b_compat（自身无裸跨包 import，恒可导入）； 3. 修完**当场自检**：ast 级 import 闭包核验 + py_compile + 真实 import 冒烟， 0 断裂才允许交付（不再「声称已修复」而实测仍断，QC #3）。 用法：python3 tools/m1b_fix_import_closure_v2.py [--check-only] [--root ] """ import ast import io import os import py_compile import re import sys HERE = os.path.dirname(os.path.abspath(__file__)) DEFAULT_ROOT = os.path.dirname(HERE) # modules/pbl_blueprint PKG = "pbl_blueprint" COMPAT_MOD = "pbl_blueprint.m1b_compat" MARK_BEGIN = "# >>> M1b compat exports (auto-generated, idempotent) >>>" MARK_END = "# <<< M1b compat exports <<<" # 半迁移期已知断裂的供给模块：从这些模块 import 的符号一律改由包内兼容层供给 BROKEN_PROVIDERS = ( "pbl_common.api", "pbl_common.crud_factory", "pbl_common.errors", "pbl_common.audit", "pbl_common.tenant", "pbl_common.context", "pbl_common.dbutil", "pbl_common.util", "pbl_common", "pbl_template", "pbl_template.offline", "pbl_blueprint.m1b", ) # init.py 必须存在的装配面符号（QC #2：init.py:54 _sor / :66 esc / :344 load_m1b） INIT_REQUIRED = ( "_sor", "esc", "load_m1b", "get_env", "require_tenant", "write_audit", "tenant_crud", "PblError", "API_PATHS", "PAGE_PATHS", "CRUD_ALIASES", ) def _read(path): with io.open(path, "r", encoding="utf-8") as f: return f.read() def _write(path, text): with io.open(path, "w", encoding="utf-8") as f: f.write(text) def py_files(root): out = [] for base, dirs, files in os.walk(root): dirs[:] = [d for d in dirs if d not in (".git", "__pycache__", ".selfcheck_pyc")] for fn in sorted(files): if fn.endswith(".py"): out.append(os.path.join(base, fn)) return out def compat_exports(root): """取 m1b_compat 的真实导出面（ast 解析，不 import，避免平台依赖）。""" path = os.path.join(root, PKG, "m1b_compat.py") if not os.path.exists(path): return set() tree = ast.parse(_read(path)) names = set() for node in tree.body: if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): names.add(node.name) elif isinstance(node, ast.Assign): for t in node.targets: if isinstance(t, ast.Name): names.add(t.id) elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): names.add(node.target.id) # __all__ 字面量 for node in tree.body: if isinstance(node, ast.Assign) and any( isinstance(t, ast.Name) and t.id == "__all__" for t in node.targets): try: names.update(ast.literal_eval(node.value)) except Exception: pass return names def module_symbols(path): """一个 .py 文件在模块命名空间暴露的符号（def/class/assign/import 别名）。""" try: tree = ast.parse(_read(path)) except Exception: return set() names = set() for node in ast.walk(tree): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): names.add(node.name) elif isinstance(node, ast.Assign): for t in node.targets: if isinstance(t, ast.Name): names.add(t.id) elif isinstance(node, ast.Import): for a in node.names: names.add(a.asname or a.name.split(".")[0]) elif isinstance(node, ast.ImportFrom): for a in node.names: if a.name == "*": continue names.add(a.asname or a.name) # 兼容层里 globals()[k]=v 的动态采纳也算导出 src = _read(path) if "globals()[" in src: names.update(compat_exports(os.path.dirname(os.path.dirname(path)))) return names def resolve_module(root, modname): """模块名 -> 文件路径（包内优先，其次同仓其它模块）。""" parts = modname.split(".") cands = [] if modname.startswith(PKG): rel = os.path.join(*parts) + ".py" cands.append(os.path.join(root, rel)) cands.append(os.path.join(root, rel[:-3], "__init__.py")) else: ws = os.path.dirname(root) # modules/ (root=modules/pbl_blueprint) cands.append(os.path.join(ws, parts[0], os.path.join(*parts) + ".py")) cands.append(os.path.join(ws, parts[0], os.path.join(*parts), "__init__.py")) for c in cands: if os.path.exists(c): return c return None # -------------------------------------------------------------------------- # PASS A/B：把断裂供给模块的 from-import 重定向到包内兼容层（逐符号） # -------------------------------------------------------------------------- def fix_imports(root, path, exports, dry=False): src = _read(path) try: tree = ast.parse(src) except SyntaxError as exc: return ["SYNTAX %s: %s" % (path, exc)], src lines = src.split("\n") edits = [] # (start_idx, end_idx, new_lines) notes = [] selfmod = os.path.basename(path)[:-3] for node in ast.walk(tree): if not isinstance(node, ast.ImportFrom) or node.level: continue mod = node.module or "" if node.level: # 相对导入：包内，跳过 continue if mod == COMPAT_MOD or mod == "%s.m1b_compat" % PKG: continue if mod.startswith(PKG) and mod.split(".")[-1] == selfmod: continue target_file = resolve_module(root, mod) avail = module_symbols(target_file) if target_file else set() missing, kept = [], [] for a in node.names: if a.name == "*": kept.append(a) continue if target_file is None or a.name not in avail: if a.name in exports: missing.append(a) else: kept.append(a) # 兼容层也没有：保持原样，交由冒泡（QC #3 边界） notes.append("UNRESOLVED %s:%d %s.%s" % (path, node.lineno, mod, a.name)) else: kept.append(a) if not missing: continue start = node.lineno - 1 end = getattr(node, "end_lineno", node.lineno) - 1 new = [] if kept: new.append("%sfrom %s import (%s)" % ( " " * node.col_offset, mod, ", ".join(("%s as %s" % (a.name, a.asname)) if a.asname else a.name for a in kept))) new.append("%sfrom %s import ( # M1b compat: %s 半迁移缺失符号改由包内兼容层供给" % ( " " * node.col_offset, COMPAT_MOD, mod)) for a in missing: new.append(" %s," % (("%s as %s" % (a.name, a.asname)) if a.asname else a.name)) new.append(")") edits.append((start, end, new)) for a in missing: notes.append("REDIRECT %s:%d %s.%s -> %s" % (path, node.lineno, mod, a.name, COMPAT_MOD)) if edits and not dry: for start, end, new in sorted(edits, key=lambda e: -e[0]): lines[start:end + 1] = new src = "\n".join(lines) return notes, src # -------------------------------------------------------------------------- # PASS C：init.py 装配面兜底（逐符号，绝不整块 import） # -------------------------------------------------------------------------- def compat_block(symbols): out = [MARK_BEGIN, "# 由 tools/m1b_fix_import_closure_v2.py 幂等生成；不覆盖本文件既有同名定义。", "# QC #1 铁律：逐符号 getattr 采纳，单符号缺失不牵连其余符号；", "# 供给方 pbl_blueprint.m1b_compat 为包内自足模块（无裸跨包 import，恒可导入）。", "_M1B_COMPAT_SYMBOLS = ("] for s in symbols: out.append(" %r," % s) out += [ ")", "", "try:", " import pbl_blueprint.m1b_compat as _m1b_compat", "except ImportError: # pragma: no cover", " try:", " from . import m1b_compat as _m1b_compat", " except ImportError:", " _m1b_compat = None", "", "if _m1b_compat is not None:", " for _name in _M1B_COMPAT_SYMBOLS:", " if _name in globals():", " continue", " _val = getattr(_m1b_compat, _name, None)", " if _val is not None:", " globals()[_name] = _val", "", "# 装配面硬保证：load_m1b / offline 必须可调用（api_blueprint.py:504 引用点）", "if not callable(globals().get('load_m1b')) and _m1b_compat is not None:", " globals()['load_m1b'] = _m1b_compat.load_m1b", "if not callable(globals().get('offline')) and _m1b_compat is not None:", " globals()['offline'] = _m1b_compat.offline", MARK_END, ] return "\n".join(out) def ensure_block(path, symbols): src = _read(path) block = compat_block(symbols) if MARK_BEGIN in src and MARK_END in src: head, rest = src.split(MARK_BEGIN, 1) _, tail = rest.split(MARK_END, 1) new = head + block + tail else: new = src.rstrip("\n") + "\n\n\n" + block + "\n" if new != src: _write(path, new) return True return False # -------------------------------------------------------------------------- # 自检：ast 级 import 闭包核验（0 断裂才放行） # -------------------------------------------------------------------------- def closure_check(root, scope_files): breaks = [] for path in scope_files: try: tree = ast.parse(_read(path)) except SyntaxError as exc: breaks.append((path, exc.lineno or 0, "SYNTAX", str(exc))) continue for node in ast.walk(tree): if isinstance(node, ast.Import): for a in node.names: if a.name.split(".")[0] in (PKG, "pbl_common", "pbl_template"): if resolve_module(root, a.name) is None: breaks.append((path, node.lineno, a.name, "MODULE_NOT_FOUND")) elif isinstance(node, ast.ImportFrom) and not node.level and node.module: mod = node.module if mod.split(".")[0] not in (PKG, "pbl_common", "pbl_template"): continue tf = resolve_module(root, mod) if tf is None: breaks.append((path, node.lineno, mod, "MODULE_NOT_FOUND")) continue avail = module_symbols(tf) for a in node.names: if a.name == "*": continue if a.name not in avail: breaks.append((path, node.lineno, "%s.%s" % (mod, a.name), "SYMBOL_MISSING")) return breaks _TMPC = os.path.join(HERE, "..", ".selfcheck_pyc") def main(argv): check_only = "--check-only" in argv root = DEFAULT_ROOT if "--root" in argv: root = os.path.abspath(argv[argv.index("--root") + 1]) try: os.makedirs(_TMPC, exist_ok=True) except Exception: pass exports = compat_exports(root) print("COMPAT_EXPORTS %d" % len(exports)) files = py_files(root) print("SCANNED_FILES %d" % len(files)) notes = [] if not check_only: for path in files: n, src = fix_imports(root, path, exports) notes.extend(n) if src != _read(path): _write(path, src) init_py = os.path.join(root, PKG, "init.py") if os.path.exists(init_py): syms = tuple(sorted(set(INIT_REQUIRED) | set(exports))) if ensure_block(init_py, syms): notes.append("BLOCK init.py compat exports ensured (%d symbols)" % len(syms)) api_py = os.path.join(root, PKG, "api_blueprint.py") if os.path.exists(api_py): if ensure_block(api_py, tuple(sorted(set(exports) | {"offline"}))): notes.append("BLOCK api_blueprint.py compat exports ensured") tc = os.path.join(root, "tests", "test_contract.py") if os.path.exists(tc): if ensure_block(tc, ("API_PATHS", "PAGE_PATHS", "CRUD_ALIASES")): notes.append("BLOCK tests/test_contract.py contract constants ensured") for n in notes: print(n) # py_compile 全量 bad = [] for path in files: try: py_compile.compile(path, doraise=True, cfile=os.path.join(_TMPC, os.path.basename(path) + "c")) except Exception as exc: bad.append("%s: %s" % (path, exc)) print("PY_COMPILE_FAIL %d" % len(bad)) for b in bad: print(" " + b) breaks = closure_check(root, files) inpkg = [b for b in breaks if b[0].startswith(os.path.join(root, PKG)) or b[0].startswith(os.path.join(root, "tests"))] print("CLOSURE_BREAKS_PKG %d" % len(inpkg)) for b in inpkg: print(" BREAK %s:%s %s [%s]" % b) print("CLOSURE_BREAKS_ALL_SCANNED %d" % len(breaks)) return 0 if (not bad and not inpkg) else 1 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))