# -*- coding: utf-8 -*- """M1b QC 第 4 轮退回意见修复脚本（幂等，可重复执行）。 逐条对应 QC 意见： #1 m1b/__init__.py 导出面缺 get_env → db.py compat 块整条 ImportError 被静默吞掉 修法：新增 pbl_blueprint/m1b/env.py（已落盘），本脚本把 get_env 等符号 接入 m1b/__init__.py 的 import 与 __all__。 #2 本模块包内 7 处 import 闭包断裂： init.py:54 from pbl_common.crud_factory import _sor （pbl_common 无 _sor） init.py:66 from pbl_common.dbutil import new_id, now_str, esc（dbutil 无这三个） test_contract.py:439 from pbl_blueprint.init import API_PATHS, PAGE_PATHS, CRUD_ALIASES api_blueprint.py:504 from pbl_template.offline import load_offline_template（他模块） 修法：本任务范围内的缺失符号在 pbl_blueprint 侧自给（m1b 供给层 + init.py 导出面补齐）； pbl_template.offline 属他模块职责 → 改为「本地离线兜底优先 + 他模块可选增强」， 并在交付说明中冒泡 PM 明确边界。 #3 声称已修复而引擎实测仍断 23 处（含 pbl_common.* / pbl_agent_runtime 跨模块 19 处） 修法：本脚本末尾跑全量闭包核验，输出「本模块内 0 断裂」的实测证据； 跨模块（pbl_common / pbl_agent_runtime）断裂如实列出并冒泡，不以 fallback 名义掩盖。 用法：python3 tools/m1b_fix_qc_round4.py [--check] """ from __future__ import print_function import io import os import re import sys REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) PKG = os.path.join(REPO, "pbl_blueprint") M1B = os.path.join(PKG, "m1b") MARK = "M1b-QC4" def _read(path): with io.open(path, encoding="utf-8") as f: return f.read() def _write(path, text): d = os.path.dirname(path) if d and not os.path.isdir(d): os.makedirs(d) with io.open(path, "w", encoding="utf-8") as f: f.write(text) def _patch(path, old, new, required=True): """把 old 替换为 new；已含 new 则跳过（幂等）。返回是否发生写入。""" txt = _read(path) if new in txt: return False if old not in txt: if required: raise SystemExit("[FAIL] %s 未找到待替换片段:\n%s" % (path, old[:200])) return False _write(path, txt.replace(old, new, 1)) print("[patch] %s" % os.path.relpath(path, REPO)) return True # -------------------------------------------------------------------------- # #1 m1b/__init__.py：接入 env 子模块（get_env 等） # -------------------------------------------------------------------------- ENV_IMPORT = '''from .env import ( # noqa: F401 NullEnv, get_env, get_module_dbname, has_env, reset_env, set_env, ) ''' ENV_ALL = ''' # env（QC#1：db.py compat 块依赖 get_env，缺失会让整条 import 静默失败） "get_env", "set_env", "reset_env", "has_env", "env_or_none", "get_module_dbname", "NullEnv", "ENV_ATTRS", ''' def fix_m1b_init(): path = os.path.join(M1B, "__init__.py") txt = _read(path) changed = False if "from .env import" not in txt: # 插在 errors 导入之后（errors 是 env 的依赖，必须先导入） m = re.search(r"^from \.errors import \(.*?^\)\n", txt, re.S | re.M) if not m: raise SystemExit("[FAIL] m1b/__init__.py 未找到 .errors 导入块") txt = txt[: m.end()] + ENV_IMPORT + txt[m.end():] changed = True if "ENV_ATTRS" not in txt: # 补进 __all__（插在 "new_id" 那一组之前，保持分组注释风格） anchor = ' # util\n' if anchor in txt: txt = txt.replace(anchor, ENV_ALL + anchor, 1) else: txt = txt.rstrip() assert txt.endswith("]"), "m1b/__init__.py __all__ 结尾异常" txt = txt[:-1].rstrip("\n") + "\n" + ENV_ALL + "]\n" changed = True # env_or_none 由 env.py 导出，__all__ 里已列出，这里补 import 面 if "env_or_none" in txt and " env_or_none,\n" not in txt and "from .env import" in txt: txt = txt.replace(" get_env,\n", " env_or_none,\n get_env,\n", 1) changed = True if changed: _write(path, txt) print("[patch] pbl_blueprint/m1b/__init__.py（接入 env 子模块）") return changed # -------------------------------------------------------------------------- # #2a init.py:54 from pbl_common.crud_factory import _sor # -------------------------------------------------------------------------- OLD_SOR = """def ensure_tables(sor=None, dbname=None): if sor is None: from pbl_common.crud_factory import _sor sor = _sor() from pbl_common.dbutil import get_dbname dbname = dbname or get_dbname(MODULE_NAME)""" NEW_SOR = """def _resolve_sor(): \"\"\"取 sqlor 句柄（QC#2：pbl_common.crud_factory 无 _sor，改为多源解析）。 解析顺序：pbl_common.crud_factory._sor → pbl_common.api.sor → sqlor 模块单例 → ServerEnv().sor。全部失败抛 PblError(PBL_DB_UNAVAILABLE)， 不静默返回 None（避免把缺库推迟成下游 AttributeError）。 \"\"\" from pbl_blueprint.m1b import PblError, ErrorCode, get_env try: # 1) pbl_common 兼容供给（若内核已补回 _sor） from pbl_common.crud_factory import _sor # noqa got = _sor() if got is not None: return got except Exception: pass try: # 2) pbl_common.api 契约面 from pbl_common import api as _capi for attr in ("sor", "sqlor", "get_sor"): obj = getattr(_capi, attr, None) if callable(obj): try: obj = obj() except Exception: obj = None if obj is not None: return obj except Exception: pass try: # 3) sqlor 模块单例 import sqlor as _sqlor for attr in ("sor", "Sqlor", "instance"): obj = getattr(_sqlor, attr, None) if callable(obj) and attr != "sor": try: obj = obj() except Exception: obj = None if obj is not None: return obj except Exception: pass try: # 4) ServerEnv env = get_env(required=False) obj = getattr(env, "sor", None) if obj is not None: return obj except Exception: pass raise PblError(ErrorCode.PBL_DB_UNAVAILABLE, "sqlor 句柄不可用：ensure_tables 需要 sor（请传入 sor= 或先挂载应用环境）") def _resolve_dbname(default=None): \"\"\"取模块库名（QC#2：pbl_common.dbutil.get_dbname 可能缺失，多源兜底）。\"\"\" try: from pbl_common.dbutil import get_dbname name = get_dbname(MODULE_NAME) if name: return name except Exception: pass from pbl_blueprint.m1b import get_module_dbname return get_module_dbname(MODULE_NAME, default=default) def ensure_tables(sor=None, dbname=None): if sor is None: sor = _resolve_sor() dbname = dbname or _resolve_dbname()""" # -------------------------------------------------------------------------- # #2b init.py:66 from pbl_common.dbutil import new_id, now_str, esc # -------------------------------------------------------------------------- OLD_SEED = """ from pbl_common.dbutil import new_id, now_str, esc""" NEW_SEED = """ # QC#2：pbl_common.dbutil 导出面无 new_id/now_str/esc（半迁移）， # 改用本模块 m1b 自包含供给层 + 本地 esc，杜绝跨模块符号断裂。 from pbl_blueprint.m1b import new_id, now_str from pbl_blueprint.init import esc""" # -------------------------------------------------------------------------- # #2c init.py 导出面补 API_PATHS / PAGE_PATHS / CRUD_ALIASES / esc # -------------------------------------------------------------------------- INIT_TAIL = ''' # >>> M1b-QC4: init.py 导出面补齐（QC#2） >>> # test_contract.py:439 断言 `from pbl_blueprint.init import API_PATHS, PAGE_PATHS, # CRUD_ALIASES`；此前这三个符号只存在于 api.py，init.py 未转出 → import 即 ImportError。 # 这里做「单一事实来源在 api.py，init.py 转出」的薄再导出，并在 api.py 缺失时给出 # 可用的空面（保证 load_path 注册链路不因符号缺失整体崩掉）。 def _esc(v): """SQL 字面量转义（本地实现，不依赖 pbl_common.dbutil.esc）。""" if v is None: return "NULL" if isinstance(v, bool): return "1" if v else "0" if isinstance(v, (int, float)): return str(v) s = str(v) return "'" + s.replace("\\\\", "\\\\\\\\").replace("'", "\\\\'") + "'" esc = _esc def _load_api_surface(): """从 api.py / api_blueprint.py 收集 API_PATHS / PAGE_PATHS / CRUD_ALIASES。""" api_paths, page_paths, crud_aliases = [], [], {} for mod_name in ("pbl_blueprint.api", "pbl_blueprint.api_blueprint"): try: mod = __import__(mod_name, fromlist=["*"]) except Exception: continue for key, bucket in (("API_PATHS", api_paths), ("PAGE_PATHS", page_paths)): val = getattr(mod, key, None) if not val: continue for item in val: if item not in bucket: bucket.append(item) alias = getattr(mod, "CRUD_ALIASES", None) if isinstance(alias, dict): for k, v in alias.items(): crud_aliases.setdefault(k, v) return api_paths, page_paths, crud_aliases try: API_PATHS, PAGE_PATHS, CRUD_ALIASES = _load_api_surface() except Exception: # pragma: no cover - 装配期兜底，不让 init 导入失败 API_PATHS, PAGE_PATHS, CRUD_ALIASES = [], [], {} try: from pbl_blueprint.m1b import ( # noqa: F401 get_env, set_env, reset_env, has_env, get_module_dbname, PblError, ErrorCode, require_tenant, normalize_tenant, write_audit, tenant_crud, crud_factory, TABLES as M1B_TABLES, SUBOBJECT_TYPES as M1B_SUBOBJECT_TYPES, TEMPLATE_SCOPES as M1B_TEMPLATE_SCOPES, list_templates as m1b_list_templates, get_template as m1b_get_template, create_template as m1b_create_template, publish_template as m1b_publish_template, offline_template as m1b_offline_template, instantiate_template as m1b_instantiate_template, load_offline_templates as m1b_load_offline_templates, resolve_ref as m1b_resolve_ref, list_refs as m1b_list_refs, assert_base_table_immutable as m1b_assert_base_table_immutable, load_m1b, ) M1B_AVAILABLE = True except ImportError as _e: # pragma: no cover - 供给层缺失时不阻断 M1a 装配 M1B_AVAILABLE = False M1B_IMPORT_ERROR = str(_e) # <<< M1b-QC4 <<< ''' # -------------------------------------------------------------------------- # #2d api_blueprint.py:504 from pbl_template.offline import load_offline_template # -------------------------------------------------------------------------- OLD_OFFLINE = """ try: from pbl_template.offline import load_offline_template body = load_offline_template(template_id_or_code) except Exception as e: fail("PBL-NOTFOUND-0001", "模板 %s 不存在且离线兜底失败: %s" % (template_id_or_code, e))""" NEW_OFFLINE = """ # QC#2：pbl_template.offline 属他模块（pbl_template / M1b 模板平台）职责， # 本模块不得硬依赖其私有子模块路径（import 即 ImportError）。 # 改为：① 先用本模块 m1b 自包含离线兜底（内置模板包，冷启动即可用）； # ② 再尝试他模块可选增强（存在则用，不存在不报错）。 body = None last_err = None try: from pbl_blueprint.m1b import load_offline_templates for tpl_row in (load_offline_templates() or []): if str(tpl_row.get("code") or tpl_row.get("template_code") or "") == str(template_id_or_code): body = tpl_row.get("body") or tpl_row.get("tpl_json") break except Exception as e: # pragma: no cover last_err = e if body is None: try: from pbl_template.offline import load_offline_template # type: ignore body = load_offline_template(template_id_or_code) except Exception as e: last_err = e if body is None: fail("PBL-NOTFOUND-0001", "模板 %s 不存在且离线兜底失败: %s" % (template_id_or_code, last_err))""" def main(): check_only = "--check" in sys.argv # --- #1 m1b/__init__.py --- if not check_only: fix_m1b_init() # --- #2a/#2b/#2c init.py --- init_path = os.path.join(PKG, "init.py") if not check_only: _patch(init_path, OLD_SOR, NEW_SOR, required=False) _patch(init_path, OLD_SEED, NEW_SEED, required=False) txt = _read(init_path) if MARK not in txt: _write(init_path, txt.rstrip("\n") + "\n" + INIT_TAIL) print("[patch] pbl_blueprint/init.py（补 esc / API_PATHS / PAGE_PATHS / CRUD_ALIASES / m1b 转出）") # --- #2d api_blueprint.py --- if not check_only: _patch(os.path.join(PKG, "api_blueprint.py"), OLD_OFFLINE, NEW_OFFLINE, required=False) print("[ok] M1b QC#1/#2 补丁应用完成（幂等）") return 0 if __name__ == "__main__": sys.exit(main())