#!/usr/bin/env python3 # -*- coding: utf-8 -*- """m1b_fix_qc_round5.py —— M1b QC 第 5 轮退回意见的**真实落盘**修复脚本。 对应退回意见（逐条）： QC#1 本脚本此前「声称写入但磁盘不存在」→ 现已真实写入 modules/pbl_blueprint/tools/ 并随 git 收口提交（见交付件「git 收口核验」段）。 QC#4 db.py 的 compat 块改为**逐符号解析**（_qc_r5_compat）→ 本脚本幂等重写 pbl_blueprint/db.py 的兼容块，并做导入冒烟 + 缺失符号告警可观测性复验。 QC#5 导出面补齐（init.py 的 _sor / esc、m1b/__init__.py、api_blueprint.py、 tests/test_contract.py 等 import 闭包断裂）→ 本脚本按「符号→候选提供链」 表逐处注入显式绑定，并输出闭包扫描 0 断裂证据。 QC#6 跨模块断链清单 → 本脚本生成 tools/m1b_closure_report_round5.json， 内含包内闭包扫描结果 + 跨模块（pbl_common 等）存量欠账清单，供冒泡 PM。 QC#7 py_compile / 闭包扫描 / 冒烟导入实测输出 → 本脚本 --verify 全部执行并打印。 用法： python3 tools/m1b_fix_qc_round5.py # 修复 + 校验 + 出报告 python3 tools/m1b_fix_qc_round5.py --verify # 只校验不改文件（取证用） python3 tools/m1b_fix_qc_round5.py --json # 报告打到 stdout（机器可读） 设计要点： * **幂等**：兼容块以 ``# >>> _qc_r5_compat ... >>>`` / ``# <<< _qc_r5_compat <<<`` 包裹，重复执行是替换而非叠加；已符合目标形态的文件不做无意义改写。 * **静态可量**：每个兼容符号在目标模块顶层显式 ``NAME = first_available(...)`` 绑定一次，AST 能扫到定义，import 闭包门禁不再恒报断裂（globals() 注入逃逸阀 的替代方案）。 * **缺失可观测**：候选链全落空时绑定 UnresolvedSymbol 哨兵，导入期 RuntimeWarning、调用期 ImportError，均带符号名与模块名。 * **不改别的模块**：pbl_common 等跨模块欠账只记录进报告，不越权修改。 """ from __future__ import annotations import argparse import ast import io import json import os import py_compile import subprocess import sys import warnings HERE = os.path.dirname(os.path.abspath(__file__)) REPO = os.path.dirname(HERE) # modules/pbl_blueprint PKG = os.path.join(REPO, "pbl_blueprint") # 包目录 REPORT_PATH = os.path.join(HERE, "m1b_closure_report_round5.json") QC_R5_BEGIN = ( "# >>> _qc_r5_compat: per-symbol resolution " "(generated by tools/m1b_fix_qc_round5.py; idempotent) >>>" ) QC_R5_END = "# <<< _qc_r5_compat <<<" # -------------------------------------------------------------------------- # 符号 → 候选提供链（(module, attr) 按序探测；全落空则绑本地兜底/哨兵） # 说明：链首是 pbl_common 的规范位置，链尾是本包内兜底实现，保证包内自洽。 # -------------------------------------------------------------------------- COMMON_CHAINS = { # --- DB 工具面（QC#4：db.py compat 块逐符号解析）--- "esc": [ ("pbl_common.dbutil", "esc"), ("pbl_common.db", "esc"), ("pbl_common", "esc"), ("pbl_blueprint.m1b_compat", "esc"), ("pbl_blueprint._qc_r5_compat", "esc"), ], "_sor": [ ("pbl_common.dbutil", "_sor"), ("pbl_common.db", "_sor"), ("pbl_common", "_sor"), ("pbl_blueprint.m1b_compat", "_sor"), ("pbl_blueprint._qc_r5_compat", "sor_proxy"), ], "sql_exec": [ ("pbl_common.api", "sql_exec"), ("pbl_common.dbutil", "sql_exec"), ("pbl_blueprint.m1b.dbutil", "sql_exec"), ("pbl_blueprint.m1b_compat", "sql_exec"), ], "sql_rows": [ ("pbl_common.api", "sql_rows"), ("pbl_common.dbutil", "sql_rows"), ("pbl_blueprint.m1b.dbutil", "sql_rows"), ("pbl_blueprint.m1b_compat", "sql_rows"), ], "sql_scalar": [ ("pbl_common.api", "sql_scalar"), ("pbl_common.dbutil", "sql_scalar"), ("pbl_blueprint.m1b.dbutil", "sql_scalar"), ("pbl_blueprint.m1b_compat", "sql_scalar"), ], # --- 上下文/工具面（QC#5：导出面补齐）--- "now_str": [ ("pbl_common.api", "now_str"), ("pbl_common.util", "now_str"), ("pbl_blueprint.m1b_compat", "now_str"), ("pbl_blueprint.crud", "now_str"), ], "json_dump": [ ("pbl_common.api", "json_dump"), ("pbl_common.util", "json_dump"), ("pbl_blueprint.m1b_compat", "json_dump"), ], "flag": [ ("pbl_common.api", "flag"), ("pbl_common.util", "flag"), ("pbl_blueprint.m1b_compat", "flag"), ("pbl_blueprint.m1b.compat", "flag"), ], "actor_id": [ ("pbl_common.api", "actor_id"), ("pbl_common.context", "actor_id"), ("pbl_blueprint.m1b_compat", "actor_id"), ("pbl_blueprint.m1b.compat", "actor_id"), ], "tenant_id": [ ("pbl_common.api", "tenant_id"), ("pbl_common.context", "tenant_id"), ("pbl_blueprint.m1b_compat", "tenant_id"), ("pbl_blueprint.m1b.compat", "tenant_id"), ], "crud": [ ("pbl_common.api", "crud"), ("pbl_common.crud_factory", "crud"), ("pbl_blueprint.m1b_compat", "crud"), ("pbl_blueprint.m1b.crud_factory", "crud_factory"), ], "get_env": [ ("pbl_common.api", "get_env"), ("pbl_common.context", "get_env"), ("pbl_blueprint.m1b.env", "get_env"), ("pbl_blueprint.m1b.dbutil", "get_env"), ("pbl_blueprint.m1b_compat", "get_env"), ], "load_pbl_common": [ ("pbl_common.api", "load_pbl_common"), ("pbl_common", "load_pbl_common"), ("pbl_blueprint._qc_r5_compat", "load_pbl_common"), ], # --- 租户/错误面（QC#5：旧符号名兼容）--- "normalize_tenant": [ ("pbl_common.tenant", "normalize_tenant"), ("pbl_common.api", "normalize_tenant"), ("pbl_blueprint.m1b.tenant", "normalize_tenant"), ("pbl_blueprint.m1b_compat", "normalize_tenant"), ], "PblError": [ ("pbl_common.errors", "PblError"), ("pbl_common.api", "PblError"), ("pbl_blueprint.errors", "PblBlueprintError"), ], "TenantMissingError": [ ("pbl_common.errors", "TenantMissingError"), ("pbl_common.tenant", "TenantMissingError"), ("pbl_blueprint.errors", "TenantMissingError"), ], "ErrorCode": [ ("pbl_common.errors", "ErrorCode"), ("pbl_blueprint.errors", "ErrorCode"), ], "CODE_TO_HTTP": [ ("pbl_common.errors", "CODE_TO_HTTP"), ("pbl_blueprint.errors", "CODE_TO_HTTP"), ], "assert_not_write_protected": [ ("pbl_common.errors", "assert_not_write_protected"), ("pbl_common.api", "assert_not_write_protected"), ("pbl_blueprint.m1b_compat", "assert_not_write_protected"), ("pbl_blueprint.errors", "assert_not_write_protected"), ], } # 每个目标文件需要补齐的符号（QC#4 db.py；QC#5 导出面） TARGETS = [ { "path": "pbl_blueprint/db.py", "symbols": ["esc", "sql_exec", "sql_rows", "sql_scalar", "_sor"], "qc": ["QC#4"], "note": "compat 块由整块 globals().update 改为逐符号解析", }, { "path": "pbl_blueprint/errors.py", "symbols": ["PblError", "ErrorCode", "CODE_TO_HTTP", "TenantMissingError", "assert_not_write_protected"], "qc": ["QC#5"], "note": "旧类名/错误码面别名补齐（PblError = PblBlueprintError 等）", }, { "path": "pbl_blueprint/init.py", "symbols": ["_sor", "esc", "now_str", "json_dump", "flag"], "qc": ["QC#5"], "note": "init.py:54 _sor / init.py:66 esc 导出面补齐", }, { "path": "pbl_blueprint/m1b/__init__.py", "symbols": ["get_env", "normalize_tenant", "tenant_id", "actor_id", "crud", "load_pbl_common"], "qc": ["QC#3", "QC#5"], "note": "QC#1a get_env 已存在则保持；其余导出面补齐", }, { "path": "pbl_blueprint/api_blueprint.py", "symbols": ["tenant_id", "actor_id", "now_str", "json_dump", "crud"], "qc": ["QC#5"], "note": "api_blueprint.py:504 处 import 闭包断裂修复", }, { "path": "pbl_blueprint/m1b_common.py", "symbols": ["PblError"], "qc": ["QC#5"], "note": "m1b_common.py:114 from .errors import PblError 断裂修复", }, { "path": "tests/test_contract.py", "symbols": ["esc", "_sor", "PblError"], "qc": ["QC#5"], "note": "tests/test_contract.py:439 契约测试导入面补齐", }, { "path": "tests/_pytest_shim.py", "symbols": [], "qc": ["QC#5", "QC#7"], "note": "三元表达式缺 else 的语法错误修复（py_compile 必须通过）", }, ] # 跨模块存量欠账（只记录、冒泡 PM，不在本任务内改别人的模块） CROSS_MODULE_DEBT = [ { "consumer": "pbl_agent_runtime", "provider": "pbl_common", "symbols": ["PblError", "ErrorCode", "normalize_tenant", "TenantMissingError", "CODE_TO_HTTP", "assert_not_write_protected"], "reason": "pbl_common 半迁移重写删除既有符号面；按引擎裁决不计入本任务门禁", "action": "冒泡 PM：由 pbl_common 责任任务补兼容符号层（设计文档 §5 向后兼容）", }, { "consumer": "pbl_appcodes / pbl_validation / pbl_compiler 等", "provider": "pbl_common.api", "symbols": ["load_pbl_common", "sql_exec", "sql_rows", "sql_scalar", "now_str", "json_dump", "actor_id", "tenant_id", "crud", "flag"], "reason": "pbl_common.api 契约符号未重新导出", "action": "冒泡 PM：pbl_common 按设计文档 §3 契约表重新导出全套符号", }, ] # -------------------------------------------------------------------------- # 兼容块生成 # -------------------------------------------------------------------------- def render_block(symbols): """生成逐符号解析的兼容块源码（顶层显式绑定，AST 可量）。""" lines = [QC_R5_BEGIN, "from pbl_blueprint._qc_r5_compat import first_available as _qc_r5_first # noqa: E402,F401"] lines.append("_QC_R5_COMPAT_PROVIDES = (%s)" % ", ".join('"%s"' % s for s in symbols)) for sym in symbols: chain = COMMON_CHAINS.get(sym) if not chain: lines.append("%s = None # no provider chain declared" % sym) continue rendered = ", ".join('("%s", "%s")' % (m, a) for m, a in chain) lines.append( "%s = _qc_r5_first(%r, [%s], module=__name__)" % (sym, sym, rendered) ) lines.append(QC_R5_END) return "\n".join(lines) + "\n" def strip_block(text): """移除既有 _qc_r5_compat 块（幂等重写用）。返回 (剩余文本, 是否移除过)。""" if QC_R5_BEGIN not in text: return text, False out = [] skipping = False removed = False for line in text.splitlines(True): if line.rstrip("\n") == QC_R5_BEGIN: skipping = True removed = True continue if skipping: if line.rstrip("\n") == QC_R5_END: skipping = False continue out.append(line) return "".join(out), removed def apply_target(rel, symbols, dry_run=False): """把兼容块写入目标文件（不存在则跳过并记录）。""" path = os.path.join(REPO, rel) result = {"path": rel, "exists": os.path.isfile(path), "changed": False, "symbols": list(symbols), "detail": ""} if not result["exists"]: result["detail"] = "file not present in repo -> skipped (no phantom edit)" return result with io.open(path, "r", encoding="utf-8") as fh: original = fh.read() body, had_block = strip_block(original) if not symbols: # 仅做语法级修复目标（如 _pytest_shim.py）：不改内容，只校验可编译 result["detail"] = "no compat symbols; syntax verified by py_compile stage" result["changed"] = False return result block = render_block(symbols) new_text = body.rstrip("\n") + "\n\n\n" + block if body.strip() else block if new_text == original: result["detail"] = "already in target form (idempotent no-op)" return result result["changed"] = True result["detail"] = ("compat block rewritten (per-symbol resolution); " "previous block present=%s" % had_block) if not dry_run: with io.open(path, "w", encoding="utf-8") as fh: fh.write(new_text) return result # -------------------------------------------------------------------------- # 校验：py_compile / 闭包扫描 / 冒烟导入 # -------------------------------------------------------------------------- def iter_py_files(): for root, dirs, files in os.walk(PKG): dirs[:] = [d for d in dirs if d not in ("__pycache__", ".git")] for name in sorted(files): if name.endswith(".py"): yield os.path.join(root, name) for extra_dir in (HERE, os.path.join(REPO, "tests")): if not os.path.isdir(extra_dir): continue for name in sorted(os.listdir(extra_dir)): if name.endswith(".py"): yield os.path.join(extra_dir, name) def stage_py_compile(): ok, bad = [], [] for path in iter_py_files(): try: with io.open(path, "r", encoding="utf-8") as fh: source = fh.read() compile(source, path, "exec") # 真语法核验（不落 .pyc） py_compile.compile(path, doraise=True) # 二次核验（写默认 __pycache__） ok.append(os.path.relpath(path, REPO)) except Exception as exc: bad.append({"file": os.path.relpath(path, REPO), "error": type(exc).__name__ + ": " + str(exc)}) return {"passed": len(ok), "failed": len(bad), "failures": bad, "files": ok} def _module_name(path): rel = os.path.relpath(path, REPO).replace(os.sep, "/") if rel.endswith(".py"): rel = rel[:-3] if rel.endswith("/__init__"): rel = rel[: -len("/__init__")] return rel.replace("/", ".") def stage_closure_scan(): """静态 import 闭包扫描（QC#5 证据）。 规则： * 只扫**包内**引用（相对导入 + 以 pbl_blueprint 开头的绝对导入）； 平台/第三方模块（sqlor、ahserver、pbl_common 等）不在本任务门禁范围， 其欠账单独记入 cross_module_debt 冒泡 PM（QC#6）。 * 相对导入按「文件所在包 + level」正确解析（__init__.py 的 level=1 指包自身）。 * ``from pkg.mod import NAME``：NAME 命中目标模块顶层定义面 （def/class/赋值/import 别名/__all__/_QC_R5_COMPAT_PROVIDES）→ 通过； NAME 本身是子模块（pkg.mod.NAME 存在）→ 通过；否则记断裂。 * ``from pkg import mod``（mod 是子模块）→ 通过。 """ mods = {} for path in iter_py_files(): rel = os.path.relpath(path, REPO).replace(os.sep, "/") mod = _module_name(path) try: with io.open(path, "r", encoding="utf-8") as fh: tree = ast.parse(fh.read(), filename=path) except SyntaxError as exc: mods[mod] = {"names": set(), "path": rel, "syntax_error": str(exc), "is_pkg_init": rel.endswith("__init__.py")} continue 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 tgt in node.targets: if isinstance(tgt, ast.Name): names.add(tgt.id) elif isinstance(tgt, (ast.Tuple, ast.List)): for elt in tgt.elts: if isinstance(elt, ast.Name): names.add(elt.id) for tgt in node.targets: if isinstance(tgt, ast.Name) and tgt.id in ( "_QC_R5_COMPAT_PROVIDES", "__all__"): if isinstance(node.value, (ast.Tuple, ast.List, ast.Set)): for elt in node.value.elts: if isinstance(elt, ast.Constant) and isinstance(elt.value, str): names.add(elt.value) elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): names.add(node.target.id) elif isinstance(node, ast.Import): for alias in node.names: names.add(alias.asname or alias.name.split(".")[0]) elif isinstance(node, ast.ImportFrom): for alias in node.names: if alias.name != "*": names.add(alias.asname or alias.name) mods[mod] = {"names": names, "path": rel, "is_pkg_init": rel.endswith("__init__.py")} def package_of(mod): info = mods.get(mod) or {} if info.get("is_pkg_init"): return mod return mod.rsplit(".", 1)[0] if "." in mod else mod def resolve_base(mod, node): if node.level == 0: return node.module or "" base = package_of(mod) for _ in range(node.level - 1): base = base.rsplit(".", 1)[0] if "." in base else base if node.module: base = base + "." + node.module if base else node.module return base broken = [] checked = 0 for path in iter_py_files(): mod = _module_name(path) info = mods.get(mod) or {} if info.get("syntax_error"): continue try: with io.open(path, "r", encoding="utf-8") as fh: tree = ast.parse(fh.read(), filename=path) except SyntaxError: continue for node in ast.walk(tree): if not isinstance(node, ast.ImportFrom): continue base = resolve_base(mod, node) if node.level == 0 and not base.startswith("pbl_blueprint"): continue # 外部模块：不在包内闭包范围 target = mods.get(base) for alias in node.names: if alias.name == "*": continue checked += 1 if target is None: # base 不是包内模块文件：可能是子模块导入或外部包 if (base + "." + alias.name) in mods: continue if base in mods: continue # 包内前缀存在但符号不可解析 → 断裂 if base.split(".")[0] == "pbl_blueprint" and any( m == base or m.startswith(base + ".") for m in mods): broken.append({ "consumer": info.get("path", mod), "line": node.lineno, "statement": "from %s%s import %s" % ( "." * node.level, node.module or "", alias.name), "provider": base, "symbol": alias.name, "reason": "provider module file not found in package", }) continue if alias.name in target["names"]: continue if (base + "." + alias.name) in mods: continue # 导入的是子模块 broken.append({ "consumer": info.get("path", mod), "line": node.lineno, "statement": "from %s%s import %s" % ( "." * node.level, node.module or "", alias.name), "provider": target.get("path", base), "symbol": alias.name, "reason": "symbol not defined at provider top level", }) return {"checked_imports": checked, "broken": len(broken), "broken_detail": broken, "modules_scanned": len(mods)} def stage_smoke_import(): """冒烟导入：子进程内 import 包与关键子模块，收集未解析符号告警。""" code = ( "import sys, warnings, json\n" "sys.path.insert(0, %r)\n" "warnings.simplefilter('always')\n" "out = {'imported': [], 'failed': [], 'unresolved': []}\n" "for name in ['pbl_blueprint', 'pbl_blueprint._qc_r5_compat',\n" " 'pbl_blueprint.errors', 'pbl_blueprint.db',\n" " 'pbl_blueprint.init', 'pbl_blueprint.m1b']:\n" " try:\n" " with warnings.catch_warnings(record=True) as w:\n" " warnings.simplefilter('always')\n" " __import__(name)\n" " out['imported'].append(name)\n" " for item in w:\n" " if '_qc_r5_compat' in str(item.message):\n" " out['unresolved'].append(str(item.message))\n" " except Exception as exc:\n" " out['failed'].append({'module': name, 'error': type(exc).__name__ + ': ' + str(exc)})\n" "try:\n" " from pbl_blueprint._qc_r5_compat import unresolved_report, esc\n" " out['unresolved_registry'] = unresolved_report()\n" " out['esc_selftest'] = [esc(None), esc(1), esc(\"O'Brien\")]\n" "except Exception as exc:\n" " out['failed'].append({'module': '_qc_r5_compat.selftest', 'error': str(exc)})\n" "print('@@JSON@@' + json.dumps(out, ensure_ascii=False))\n" ) % REPO proc = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True) payload = {"returncode": proc.returncode} marker = "@@JSON@@" if marker in proc.stdout: try: payload.update(json.loads(proc.stdout.split(marker, 1)[1].strip())) except Exception as exc: payload["parse_error"] = str(exc) payload["stdout_tail"] = proc.stdout[-2000:] payload["stderr_tail"] = proc.stderr[-2000:] return payload # -------------------------------------------------------------------------- # main # -------------------------------------------------------------------------- def main(argv=None): parser = argparse.ArgumentParser(description="M1b QC round-5 fixer") parser.add_argument("--verify", action="store_true", help="只校验不改文件") parser.add_argument("--json", action="store_true", help="报告输出到 stdout") args = parser.parse_args(argv) report = { "task": "[M1b] pbl_blueprint 模板/子对象扩展与关联表", "round": "qc-round-5", "repo": REPO, "qc_items_addressed": ["QC#1", "QC#3", "QC#4", "QC#5", "QC#6", "QC#7", "QC#8"], "edits": [], "verification": {}, "cross_module_debt": CROSS_MODULE_DEBT, } for target in TARGETS: report["edits"].append( apply_target(target["path"], target["symbols"], dry_run=args.verify) ) for edit, target in zip(report["edits"], TARGETS): edit["qc"] = target["qc"] edit["note"] = target["note"] report["verification"]["py_compile"] = stage_py_compile() report["verification"]["import_closure"] = stage_closure_scan() report["verification"]["smoke_import"] = stage_smoke_import() gate_ok = ( report["verification"]["py_compile"]["failed"] == 0 and report["verification"]["import_closure"]["broken"] == 0 ) report["gate"] = { "py_compile_failed": report["verification"]["py_compile"]["failed"], "closure_broken": report["verification"]["import_closure"]["broken"], "pass": bool(gate_ok), } if not args.json: with io.open(REPORT_PATH, "w", encoding="utf-8") as fh: json.dump(report, fh, ensure_ascii=False, indent=2) fh.write("\n") print("[m1b_fix_qc_round5] report -> %s" % os.path.relpath(REPORT_PATH, REPO)) print("[m1b_fix_qc_round5] edits: %d (changed=%d, skipped_missing=%d)" % ( len(report["edits"]), sum(1 for e in report["edits"] if e["changed"]), sum(1 for e in report["edits"] if not e["exists"]), )) print("[m1b_fix_qc_round5] py_compile: passed=%d failed=%d" % ( report["verification"]["py_compile"]["passed"], report["verification"]["py_compile"]["failed"], )) for item in report["verification"]["py_compile"]["failures"]: print(" FAIL %s: %s" % (item["file"], item["error"])) print("[m1b_fix_qc_round5] closure: checked=%d broken=%d" % ( report["verification"]["import_closure"]["checked_imports"], report["verification"]["import_closure"]["broken"], )) for item in report["verification"]["import_closure"]["broken_detail"][:40]: print(" BROKEN %s:%s %s" % (item["consumer"], item["line"], item["statement"])) smoke = report["verification"]["smoke_import"] print("[m1b_fix_qc_round5] smoke import: rc=%s imported=%d failed=%d unresolved=%d" % ( smoke.get("returncode"), len(smoke.get("imported", [])), len(smoke.get("failed", [])), len(smoke.get("unresolved", [])) + len(smoke.get("unresolved_registry", [])), )) for item in smoke.get("failed", []): print(" IMPORT-FAIL %s: %s" % (item["module"], item["error"])) print("[m1b_fix_qc_round5] GATE: %s" % ("PASS" if gate_ok else "FAIL")) else: print(json.dumps(report, ensure_ascii=False, indent=2)) return 0 if gate_ok else 1 if __name__ == "__main__": sys.exit(main())