463 lines
19 KiB
Python
463 lines
19 KiB
Python
#!/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())
|