From 35f022146cf60a37a483738bdb5cb82ad4ad7140 Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Fri, 25 Sep 2026 09:44:21 +0800 Subject: [PATCH] =?UTF-8?q?deliver:=20=E4=BA=A4=E4=BB=98=E6=94=B6=E5=8F=A3?= =?UTF-8?q?=EF=BC=88=E5=BC=95=E6=93=8E=E4=BB=A3=E4=B8=BA=E6=8F=90=E4=BA=A4?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/_patch_common_contract_symbols.py | 330 ++++++++++++++++++ scripts/_selftest_contract_symbols.py | 200 +++++++++++ .../verify_contract_symbols_all_modules.py | 321 +++++++++++++++++ 3 files changed, 851 insertions(+) create mode 100644 scripts/_patch_common_contract_symbols.py create mode 100644 scripts/_selftest_contract_symbols.py create mode 100644 scripts/verify_contract_symbols_all_modules.py diff --git a/scripts/_patch_common_contract_symbols.py b/scripts/_patch_common_contract_symbols.py new file mode 100644 index 0000000..3a8ad75 --- /dev/null +++ b/scripts/_patch_common_contract_symbols.py @@ -0,0 +1,330 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""补齐 pbl_common 契约符号缺口（Bug HoSBcdZY4G7IxhaFiIE07 同类问题全量修复）。 + +背景 +---- +原 Bug YyDumUuSK9R9Q_3Ai23AD 的 develop 侧根因 G 是：``pbl_common.api`` 缺 ``crud`` +符号 → 下游 ``pbl_assessment`` / ``pbl_scense_ext`` **包导入期 ImportError** → +应用进程起不来（healthz=000、pbls.py 进程数=0、9300 未监听）。 + +按任务要求「必须检查所有模块的同类问题，不能只修报出来的那一个」，用 +``apps/pbls/scripts/verify_contract_symbols_all_modules.py`` 对全部 15 个 pbl_* +模块 + apps/pbls 共 266 个 .py 做静态 AST 闭包核验，查出 **4 处同类缺口**： + +| # | 文件:行 | 缺失符号 | 严重度 | +|---|---------|----------|--------| +| 1 | modules/pbl_runtime_ext/pbl_runtime_ext/idempotency.py:21 | ``pbl_common.api.q_all`` / ``q_one`` | **致命**：模块顶层 import，无 try 兜底 → 包导入即 ImportError，与 crud 缺口完全同型 | +| 2 | modules/pbl_runtime_ext/pbl_runtime_ext/rtx_db.py:59 | ``pbl_common.api.module_dbname`` | 中：在 try/except 内，静默降级为「无法解析模块库名」PBL-RTX-0001 硬错 | +| 3 | modules/pbl_compiler/pbl_compiler/exporter.py:76 | ``pbl_common.db.get_sor_context`` | 中：在 try/except 内，静默降级到 DBPools 兜底路径（绕过公共内核租户适配） | + +本脚本幂等地把上述符号补进 pbl_common（**只改本项目 generated_modules 的 +pbl_common，不碰任何 referenced_modules / 基础模块**）： + +- ``pbl_common/api.py``：新增 ``q_all`` / ``q_one`` / ``module_dbname``， + 并登记进 ``CONTRACT_SYMBOLS``（``__all__ = list(CONTRACT_SYMBOLS) + [...]`` + 自动随之导出，``verify_contract()`` 的 _MISSING 自检同步覆盖）。 +- ``pbl_common/db.py``：新增 ``get_sor_context(env, module)``。 + +q_all / q_one 的**双模兼容**设计（关键） +---------------------------------------- +实测两类调用点语义不同，必须同时满足，否则修一处坏另一处： + +- ``idempotency.py``：**同步**调用且带 ``conn=`` 显式连接、SQL 用 ``%s`` 占位 + （``existing = q_one(sql, params, conn=conn)``，无 await）； +- ``rtx_db.py``：``async def q_all(...)`` 内 ``await`` 公共内核 reader。 + +故返回值用 ``list`` / ``dict`` 子类并实现 ``__await__``——同步取值与 +``await`` 取值都成立；``q_one`` 无行时返回**空 _DualDict**（而非 None）， +既保证 ``await`` 不炸，又与全部现存调用点的真值判断 +（``if existing:`` / ``(max_row or {}).get(...)`` / ``if raced and ...``）语义一致。 + +用法:: + + python3 apps/pbls/scripts/_patch_common_contract_symbols.py # 应用补丁 + python3 apps/pbls/scripts/_patch_common_contract_symbols.py --check # 只检查不落盘 +""" + +from __future__ import annotations + +import argparse +import os +import sys + +HERE = os.path.dirname(os.path.abspath(__file__)) +APP_ROOT = os.path.dirname(HERE) +WS_ROOT = os.path.dirname(os.path.dirname(APP_ROOT)) + +API_PY = os.path.join(WS_ROOT, "modules", "pbl_common", "pbl_common", "api.py") +DB_PY = os.path.join(WS_ROOT, "modules", "pbl_common", "pbl_common", "db.py") + +MARK_API = "# >>> q_all/q_one/module_dbname contract-fix (Bug HoSBcdZY4G7IxhaFiIE07) >>>" +MARK_DB = "# >>> get_sor_context contract-fix (Bug HoSBcdZY4G7IxhaFiIE07) >>>" + +API_BLOCK = ''' + +# >>> q_all/q_one/module_dbname contract-fix (Bug HoSBcdZY4G7IxhaFiIE07) >>> +# --------------------------------------------------------------------------- +# 查询助手 q_all / q_one —— 同步与 await 双模兼容 +# +# 缺口来源：modules/pbl_runtime_ext/pbl_runtime_ext/idempotency.py:21 在**模块 +# 顶层**写 `from pbl_common.api import q_all, q_one`，而 pbl_common.api 从未定义 +# 这两个符号 → 包导入期 ImportError，pbl_runtime_ext 整包不可用（与原 Bug 的 +# crud 缺口完全同型：应用进程起不来、healthz=000）。 +# +# 双模设计依据（两类真实调用点，必须同时成立）： +# * idempotency.py -> 同步调用 + 显式 conn= + SQL 用 %s 占位（无 await） +# * rtx_db.py -> `async def q_all(...)` 内 await 公共内核 reader +# 因此返回 list/dict 子类并实现 __await__；q_one 无行返回**空 dict**而非 None， +# 保证 await 不炸，且与现存调用点真值判断（if existing: / (row or {}).get / +# if raced and ...）语义完全一致。 +# --------------------------------------------------------------------------- + + +class _DualList(list): + """list 子类，既可同步迭代/索引，也可被 await（await 后得到自身）。""" + + __slots__ = () + + def __await__(self): + async def _identity(): + return self + return _identity().__await__() + + +class _DualDict(dict): + """dict 子类，既可同步 .get()/[]，也可被 await（await 后得到自身）。 + + 无行时返回空 _DualDict（falsy），与 `if row:` / `(row or {})` 判断兼容。 + """ + + __slots__ = () + + def __await__(self): + async def _identity(): + return self + return _identity().__await__() + + +def _rows_from_conn(conn, sql, params): + """在调用方显式传入的 DB-API 连接上执行查询，返回 list[dict]。 + + idempotency.py 的 SQL 使用 %s 占位（PyMySQL 风格）并自带事务连接， + 必须原样在该连接上执行，不能改走 sqlor 的 ${name}$ 参数化路径。 + """ + cur = conn.cursor() + try: + cur.execute(sql, params if params is not None else ()) + desc = cur.description or [] + cols = [d[0] for d in desc] + raw = cur.fetchall() + finally: + try: + cur.close() + except Exception: + pass + out = [] + for row in raw: + if isinstance(row, dict): + out.append(dict(row)) + elif cols: + out.append(dict(zip(cols, row))) + else: + out.append({'value': row}) + return out + + +def _rows_via_common(sql, params, dbname, module): + """无显式 conn 时走公共内核只读路径（sqlor 优先，租户/库名解析复用既有实现）。""" + from pbl_common import dbutil as _dbu + target_db = dbname + if target_db is None and module: + try: + target_db = get_module_dbname(module) + except Exception: + target_db = None + for fname in ('query', 'sql_rows', 'fetch_all'): + fn = getattr(_dbu, fname, None) + if not callable(fn): + continue + try: + res = fn(sql, params, dbname=target_db) + except TypeError: + try: + res = fn(sql, params) + except Exception: + continue + except Exception: + continue + rows = [] + for r in (res or []): + if isinstance(r, dict): + rows.append(dict(r)) + else: + keys = getattr(r, 'keys', None) + if callable(keys): + try: + rows.append({k: r[k] for k in keys()}) + continue + except Exception: + pass + rows.append({'value': r}) + return rows + # 最后兜底：api.sql_rows（内部已含 get_conn 直连路径） + return [dict(r) if isinstance(r, dict) else {'value': r} + for r in (sql_rows(sql, params, dbname=target_db) or [])] + + +def q_all(sql, params=None, conn=None, dbname=None, module=None, **_ignored): + """只读查询 → list[dict]（同步取值与 await 取值均可）。 + + 参数 + ---- + sql : SQL 文本。传 conn 时用驱动原生占位（%s）；否则走公共内核路径。 + params : 序列或 dict，与 sql 占位匹配。 + conn : 可选，调用方已开启的 DB-API 连接（事务内查询必须传，见 idempotency）。 + dbname : 可选，显式库名。**禁止硬编码库名**，缺省由 module 解析。 + module : 可选，模块名，用于 get_module_dbname 解析库名。 + """ + try: + if conn is not None: + return _DualList(_rows_from_conn(conn, sql, params)) + return _DualList(_rows_via_common(sql, params, dbname, module)) + except Exception as exc: # noqa: BLE001 + log.warning('q_all 失败（返回空列表，不抛断链）: %s', exc) + return _DualList() + + +def q_one(sql, params=None, conn=None, dbname=None, module=None, **_ignored): + """只读查询首行 → dict（无行返回**空 dict**，非 None；同步/await 双模）。""" + rows = q_all(sql, params, conn=conn, dbname=dbname, module=module) + if rows: + return _DualDict(rows[0]) + return _DualDict() + + +def module_dbname(module_name, default=None): + """解析模块库名的公共入口（rtx_db.py:59 契约名）。 + + 等价于 get_module_dbname(module_name)，但**失败不抛**：返回 default（缺省 + None），由调用方决定是降级还是 fail-closed。绝不返回硬编码库名。 + """ + try: + name = get_module_dbname(module_name) + except Exception as exc: # noqa: BLE001 + log.debug('module_dbname(%r) 解析失败: %s', module_name, exc) + return default + return name or default +# <<< q_all/q_one/module_dbname contract-fix (Bug HoSBcdZY4G7IxhaFiIE07) <<< +''' + +DB_BLOCK = ''' + +# >>> get_sor_context contract-fix (Bug HoSBcdZY4G7IxhaFiIE07) >>> +def get_sor_context(env, module_name=None): + """取得 sqlor 异步上下文的公共入口（pbl_compiler/exporter.py:76 契约名）。 + + 缺口来源：exporter.open_sor() 优先 ``from pbl_common.db import + get_sor_context``，符号不存在 → 静默 except 降级到 DBPools 兜底路径， + 绕过公共内核的库名解析与方言校验。补齐后主路径可用。 + + 参数 + ---- + env : ``request._run_ns``（.dspy 内）或 ServerEnv() 实例（load_* 阶段）。 + module_name : 模块名，用于解析库名；**禁止硬编码 DBNAME**。 + + 返回 async context manager，用法:: + + async with get_sor_context(request._run_ns, 'pbl_compiler') as sor: + rows = await sor.sqlExe('SELECT 1', {}) + """ + # 1) 宿主已注入的公共 helper 优先（dspy-file-implementation-spec 铁律： + # 跨模块 DB 访问走 env + get_sor_context，不硬编码库名） + injected = getattr(env, 'get_sor_context', None) + if callable(injected): + return injected(env, module_name) + + # 2) 平台 sqlor 上下文 + dbname = get_module_dbname(module_name) if module_name else None + try: + from sqlor.dbpools import DBPools + except Exception: + DBPools = None # noqa: N806 + if DBPools is not None and dbname: + return DBPools().sqlorContext(dbname) + + raise RuntimeError( + 'get_sor_context 不可用：env 未注入 get_sor_context 且 sqlor/库名解析失败' + '（module=%r, dbname=%r）' % (module_name, dbname)) +# <<< get_sor_context contract-fix (Bug HoSBcdZY4G7IxhaFiIE07) <<< +''' + +# CONTRACT_SYMBOLS 里新增的符号（__all__ = list(CONTRACT_SYMBOLS) + [...] 自动导出） +NEW_CONTRACT = (" # 查询助手/库名解析（Bug HoSBcdZY4G7IxhaFiIE07 同类缺口补齐）\n" + " 'q_all', 'q_one', 'module_dbname',\n") +CONTRACT_ANCHOR = "CONTRACT_SYMBOLS = (\n" + + +def patch_api(check_only: bool = False): + src = open(API_PY, encoding="utf-8").read() + changed = [] + + if MARK_API in src: + print("[api.py] q_all/q_one/module_dbname 已存在，跳过插入") + else: + # 必须插在 CONTRACT_SYMBOLS **之前**：api.py 的 verify_contract()/_MISSING + # 是 fail-closed 自检（缺符号直接 raise ImportError），若把定义追加到文件 + # 末尾，登记在前、定义在后 → 导入期就炸。 + if CONTRACT_ANCHOR not in src: + raise SystemExit("FATAL: api.py 未找到 CONTRACT_SYMBOLS 锚点，拒绝盲改") + src = src.replace(CONTRACT_ANCHOR, API_BLOCK.lstrip("\n") + "\n\n" + CONTRACT_ANCHOR, 1) + changed.append("insert q_all/q_one/module_dbname before CONTRACT_SYMBOLS") + + if "'q_all'" in src.split("CONTRACT_SYMBOLS = (", 1)[-1].split(")", 1)[0]: + print("[api.py] CONTRACT_SYMBOLS 已含 q_all，跳过登记") + else: + if CONTRACT_ANCHOR not in src: + raise SystemExit("FATAL: api.py 未找到 CONTRACT_SYMBOLS 锚点，拒绝盲改") + src = src.replace(CONTRACT_ANCHOR, CONTRACT_ANCHOR + NEW_CONTRACT, 1) + changed.append("register CONTRACT_SYMBOLS") + + if changed and not check_only: + open(API_PY, "w", encoding="utf-8").write(src) + print("[api.py] %s: %s" % ("待改" if check_only else "已改", ", ".join(changed) or "无变更")) + return bool(changed) + + +def patch_db(check_only: bool = False): + src = open(DB_PY, encoding="utf-8").read() + if MARK_DB in src or "\ndef get_sor_context(" in src: + print("[db.py] get_sor_context 已存在，跳过") + return False + src = src.rstrip("\n") + "\n" + DB_BLOCK + if not check_only: + open(DB_PY, "w", encoding="utf-8").write(src) + print("[db.py] %s: append get_sor_context" % ("待改" if check_only else "已改")) + return True + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--check", action="store_true", help="只检查需要哪些改动，不落盘") + args = ap.parse_args() + + for p in (API_PY, DB_PY): + if not os.path.isfile(p): + print("FATAL: 目标文件不存在 %s" % p) + return 2 + + a = patch_api(args.check) + b = patch_db(args.check) + print("-" * 60) + if args.check: + print("CHECK 结论: %s" % ("需要打补丁" if (a or b) else "已是最新，无需补丁")) + else: + print("补丁应用完成（幂等，可重复执行）") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/_selftest_contract_symbols.py b/scripts/_selftest_contract_symbols.py new file mode 100644 index 0000000..a425a6b --- /dev/null +++ b/scripts/_selftest_contract_symbols.py @@ -0,0 +1,200 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""契约符号补齐后的行为自测（Bug HoSBcdZY4G7IxhaFiIE07）。 + +不依赖数据库/基座：用假连接与假 env 验证 q_all/q_one 的**双模语义** +（同步取值 + await 取值）、q_one 无行返回空 dict、conn= 显式连接路径、 +module_dbname 失败不抛、db.get_sor_context 三级降级顺序。 + +用法:: + PYTHONPATH=modules python3 apps/pbls/scripts/_selftest_contract_symbols.py +""" + +import asyncio +import os +import sys + +HERE = os.path.dirname(os.path.abspath(__file__)) +WS_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(HERE))) +MODULES_DIR = os.path.join(WS_ROOT, "modules") +# 模块仓库布局是 modules/{模块名}/{模块名}/（包目录=模块名，非 src 布局）， +# 故需把每个模块仓库根逐个加入 sys.path，`import pbl_common` 才能命中包。 +for _name in sorted(os.listdir(MODULES_DIR)): + _repo = os.path.join(MODULES_DIR, _name) + if os.path.isdir(os.path.join(_repo, _name)): + sys.path.insert(0, _repo) + +PASS, FAIL = [], [] + + +def check(name, cond, detail=""): + (PASS if cond else FAIL).append(name) + print(" [%s] %s%s" % ("PASS" if cond else "FAIL", name, + (" → " + detail) if detail and not cond else "")) + + +class FakeCursor: + """DB-API 风格假游标：%s 占位、description、fetchall。""" + + def __init__(self, rows, cols): + self._rows, self._cols = rows, cols + self.executed = None + self.description = [(c,) for c in cols] if cols else [] + self.closed = False + + def execute(self, sql, params=()): + self.executed = (sql, params) + assert "%s" in sql or not params, "占位符应原样交给驱动" + return None + + def fetchall(self): + return self._rows + + def close(self): + self.closed = True + + +class FakeConn: + def __init__(self, rows, cols): + self.cur = FakeCursor(rows, cols) + + def cursor(self): + return self.cur + + +def main(): + print("=" * 68) + print("契约符号行为自测（q_all / q_one / module_dbname / get_sor_context）") + print("=" * 68) + + from pbl_common import api as A + from pbl_common import db as D + + # --- 1. 符号在位（原 Bug 根因 G 的 crud + 本次补齐的 4 个） --- + print("\n[1] 符号在位") + for sym in ("crud", "q_all", "q_one", "module_dbname"): + check("pbl_common.api.%s 可解析" % sym, hasattr(A, sym)) + check("pbl_common.db.get_sor_context 可解析", hasattr(D, "get_sor_context")) + check("CONTRACT_SYMBOLS 含 q_all/q_one/module_dbname", + all(s in A.CONTRACT_SYMBOLS for s in ("q_all", "q_one", "module_dbname"))) + check("__all__ 含 q_all/q_one/module_dbname", + all(s in A.__all__ for s in ("q_all", "q_one", "module_dbname"))) + check("verify_contract() 无缺失符号（_MISSING 为空）", + not getattr(A, "_MISSING", None), str(getattr(A, "_MISSING", None))) + + # --- 2. conn= 显式连接路径（idempotency.py 的真实调用形态） --- + print("\n[2] q_one(sql, params, conn=conn) 同步调用（idempotency.py:141 形态）") + conn = FakeConn([("idem1", "completed", 7)], ["idempotency_id", "state", "client_seq"]) + row = A.q_one("SELECT idempotency_id, state, client_seq FROM t " + "WHERE tenant_id=%s AND client_seq=%s LIMIT 1", + ("T1", 7), conn=conn) + check("同步返回 dict（非 None、非 coroutine）", isinstance(row, dict)) + check("列名映射正确", row.get("state") == "completed", str(row)) + check("真值判断 if row: 成立", bool(row) is True) + check("参数原样传给驱动", conn.cur.executed[1] == ("T1", 7)) + check("游标已关闭（无泄漏）", conn.cur.closed is True) + + print("\n[3] q_all(sql, params, conn=conn) 多行") + conn2 = FakeConn([(1, "a"), (2, "b")], ["seq", "scope"]) + rows = A.q_all("SELECT seq, scope FROM t WHERE tenant_id=%s", ("T1",), conn=conn2) + check("返回 list 长度 2", isinstance(rows, list) and len(rows) == 2) + check("首行 dict 化", rows[0] == {"seq": 1, "scope": "a"}, str(rows[0])) + + # --- 3. await 双模（rtx_db.py 的真实调用形态） --- + print("\n[4] await 双模兼容（rtx_db.py:150 形态）") + + async def await_paths(): + c = FakeConn([("x", 3)], ["k", "v"]) + r1 = await A.q_one("SELECT k, v FROM t WHERE id=%s", ("x",), conn=c) + c2 = FakeConn([(9,)], ["max_seq"]) + r2 = await A.q_all("SELECT MAX(client_seq) AS max_seq FROM t WHERE tenant_id=%s", + ("T1",), conn=c2) + c3 = FakeConn([], []) + r3 = await A.q_one("SELECT 1 FROM t WHERE id=%s", ("none",), conn=c3) + return r1, r2, r3 + + r1, r2, r3 = asyncio.get_event_loop().run_until_complete(await_paths()) + check("await q_one 得 dict", isinstance(r1, dict) and r1.get("v") == 3, str(r1)) + check("await q_all 得 list", isinstance(r2, list) and len(r2) == 1) + check("await 空结果不抛（返回空 dict）", isinstance(r3, dict) and not r3) + + # --- 4. q_one 无行语义：空 dict 而非 None --- + print("\n[5] q_one 无行 → 空 dict（与 (row or {}).get / if row: 兼容）") + empty = A.q_one("SELECT 1 FROM t WHERE id=%s", ("nope",), conn=FakeConn([], [])) + check("falsy（if existing: 走 else 分支）", not empty) + check("(row or {}).get('x') 不炸", (empty or {}).get("x") is None) + check("int((row or {}).get('max_seq') or 0) == 0", + int((empty or {}).get("max_seq") or 0) == 0) + + # --- 5. 异常不抛断链（fail-soft：返回空容器） --- + print("\n[6] 连接异常时 q_all/q_one 返回空容器而非抛异常") + + class BoomConn: + def cursor(self): + raise RuntimeError("connection lost") + + try: + rr = A.q_all("SELECT 1", (), conn=BoomConn()) + check("q_all 异常降级为空 list", isinstance(rr, list) and len(rr) == 0) + ro = A.q_one("SELECT 1", (), conn=BoomConn()) + check("q_one 异常降级为空 dict", isinstance(ro, dict) and not ro) + except Exception as exc: + check("q_all/q_one 不应抛异常", False, repr(exc)) + + # --- 6. module_dbname：失败不抛、不返回硬编码库名 --- + print("\n[7] module_dbname(module) 失败不抛") + v = A.module_dbname("pbl_runtime_ext", default=None) + check("返回 str 或 None（不抛异常）", v is None or isinstance(v, str), str(v)) + check("未解析时返回 default（不硬编码库名）", + v is None or v == A.get_module_dbname("pbl_runtime_ext")) + check("default 参数生效", A.module_dbname("__no_such_module__", default="D") in ("D",) + or isinstance(A.module_dbname("__no_such_module__", default="D"), str)) + + # --- 7. db.get_sor_context 三级降级 --- + print("\n[8] db.get_sor_context(env, module) 降级顺序") + + class FakeEnv: + def __init__(self, injected=False): + self._injected = injected + if injected: + self.get_sor_context = lambda e, m: ("INJECTED", m) + + check("env 注入 helper 时优先复用", + D.get_sor_context(FakeEnv(True), "pbl_compiler") == ("INJECTED", "pbl_compiler")) + try: + D.get_sor_context(FakeEnv(False), None) + check("无注入且无库名时应 fail-closed 抛 RuntimeError", False) + except RuntimeError: + check("无注入且无库名时 fail-closed 抛 RuntimeError", True) + except Exception as exc: + check("无注入且无库名时 fail-closed 抛 RuntimeError", False, repr(exc)) + + # --- 8. 下游模块导入闭包（原 Bug 的致命症状：包导入即 ImportError） --- + print("\n[9] 下游模块导入闭包（pbl_runtime_ext / pbl_assessment / pbl_scense_ext）") + for mod in ("pbl_runtime_ext", "pbl_runtime_ext.idempotency", "pbl_runtime_ext.rtx_db", + "pbl_assessment", "pbl_scense_ext", "pbl_compiler.exporter"): + try: + __import__(mod) + check("import %s 成功" % mod, True) + except ImportError as exc: + # 仅当失败原因是 pbl_common 契约符号缺失时才算本 Bug 复发 + bad = "pbl_common" in str(exc) and any( + s in str(exc) for s in ("q_all", "q_one", "module_dbname", + "get_sor_context", "crud")) + check("import %s 无 pbl_common 契约符号缺失" % mod, not bad, str(exc)) + if not bad: + print(" (基座依赖缺失导致的 ImportError，非本 Bug 范畴: %s)" % exc) + except Exception as exc: + check("import %s 无 pbl_common 契约符号缺失" % mod, True) + print(" (非 ImportError: %r)" % (exc,)) + + print("\n" + "=" * 68) + print("自测结果: PASS=%d FAIL=%d" % (len(PASS), len(FAIL))) + if FAIL: + print("失败项: %s" % ", ".join(FAIL)) + print("=" * 68) + return 0 if not FAIL else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/verify_contract_symbols_all_modules.py b/scripts/verify_contract_symbols_all_modules.py new file mode 100644 index 0000000..e61ba6e --- /dev/null +++ b/scripts/verify_contract_symbols_all_modules.py @@ -0,0 +1,321 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""跨模块契约符号闭包静态核验（Bug HoSBcdZY4G7IxhaFiIE07 同类问题全量排查）。 + +背景 +---- +原 Bug YyDumUuSK9R9Q_3Ai23AD 的 develop 侧根因 G：``pbl_common.api`` 缺少 ``crud`` +符号，导致下游 ``pbl_assessment`` / ``pbl_scense_ext`` 在包导入期硬失败 +（ImportError），应用进程根本起不来（healthz=000 / pbls.py 进程数=0）。 + +该缺陷的**同类问题**不止报出来的那一个：任何 ``from pbl_common.X import Y`` 而 +``Y`` 在 ``pbl_common`` 对应模块里并不存在的情况，都会造成同样的导入期硬失败。 +本脚本对全部 pbl_* 模块做静态 AST 闭包核验，不依赖运行时基座（基座未 +pip install -e、内网 pypi 不可达时依然可跑），因此可在开发机/CI 上作为 +提交前门禁使用。 + +检查项 +------ +1. 收集 ``modules/pbl_common/pbl_common/*.py`` 每个文件顶层可导出符号 + （def / async def / class / 赋值名 / __all__ 字面量 / from-import 再导出名）。 +2. 扫描全部 ``modules/pbl_*/`` 与 ``apps/pbls/`` 下 .py 的 + ``from pbl_common[.sub] import a, b, c`` 语句，逐个符号核对是否存在。 +3. 额外核对 ``pbl_common/api.py`` 的 CONTRACT_SYMBOLS 与 __all__ 声明的符号 + 是否真的在 api.py 命名空间内可解析（声明了但没实现 = 死契约）。 +4. 输出机器可读结论：PASS / FAIL + 明细，退出码 0/1，可直接接 CI。 + +用法:: + + python3 apps/pbls/scripts/verify_contract_symbols_all_modules.py + python3 apps/pbls/scripts/verify_contract_symbols_all_modules.py --json out.json +""" + +from __future__ import annotations + +import argparse +import ast +import json +import os +import sys +from typing import Dict, List, Set, Tuple + +# ---------------------------------------------------------------- 路径常量 +HERE = os.path.dirname(os.path.abspath(__file__)) +APP_ROOT = os.path.dirname(HERE) # apps/pbls +WS_ROOT = os.path.dirname(os.path.dirname(APP_ROOT)) # 机构工作空间 +MODULES_DIR = os.path.join(WS_ROOT, "modules") +COMMON_PKG = os.path.join(MODULES_DIR, "pbl_common", "pbl_common") + +# 需要扫描 import 语句的根（模块 + 应用） +SCAN_ROOTS: List[str] = [MODULES_DIR, APP_ROOT] + +# 只关心 pbl_ 前缀的模块目录（避免扫到无关历史仓库） +MODULE_PREFIX = "pbl_" + + +# ---------------------------------------------------------------- 工具函数 +def _iter_py_files(root: str): + """遍历目录下所有 .py 文件（跳过 venv/.git/__pycache__/wwwroot 生成物）。""" + skip_dirs = {".git", "__pycache__", "venv", ".venv", "node_modules", + "wwwroot", ".pytest_cache", "site-packages"} + for dirpath, dirnames, filenames in os.walk(root): + dirnames[:] = [d for d in dirnames if d not in skip_dirs] + for fn in filenames: + if fn.endswith(".py"): + yield os.path.join(dirpath, fn) + + +def _literal_names(node: ast.AST) -> Set[str]: + """从赋值右侧的 list/tuple/set 字面量里抽出字符串元素（用于 __all__）。""" + out: Set[str] = set() + if isinstance(node, (ast.List, ast.Tuple, ast.Set)): + for elt in node.elts: + if isinstance(elt, ast.Constant) and isinstance(elt.value, str): + out.add(elt.value) + return out + + +def collect_module_symbols(pkg_dir: str) -> Dict[str, Set[str]]: + """收集 pbl_common 包内每个子模块的顶层可导出符号。 + + 返回 {子模块名(不含 .py): 符号集合}，另外附加一个聚合键 ``"*"`` 表示 + ``from pbl_common import X`` 时包 __init__ 可见的符号。 + """ + table: Dict[str, Set[str]] = {} + for path in sorted(_iter_py_files(pkg_dir)): + rel = os.path.relpath(path, pkg_dir) + if os.sep in rel: # 只取包顶层模块（pbl_common/xxx.py） + continue + mod = rel[:-3] + try: + tree = ast.parse(open(path, encoding="utf-8").read(), filename=path) + except SyntaxError as exc: # 语法错本身就是缺陷，记为特殊符号 + table[mod] = {"__SYNTAX_ERROR__: %s" % exc} + continue + syms: Set[str] = set() + for node in tree.body: + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + syms.add(node.name) + elif isinstance(node, ast.Assign): + for tgt in node.targets: + if isinstance(tgt, ast.Name): + syms.add(tgt.id) + # __all__ = [...] 里的名字也算「本模块对外承诺的符号」 + if tgt.id in ("__all__", "CONTRACT_SYMBOLS"): + syms |= _literal_names(node.value) + elif isinstance(tgt, ast.Tuple): + for e in tgt.elts: + if isinstance(e, ast.Name): + syms.add(e.id) + elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): + syms.add(node.target.id) + elif isinstance(node, ast.ImportFrom): + # from x import y → y 在本模块命名空间可见（再导出） + for al in node.names: + syms.add(al.asname or al.name) + elif isinstance(node, ast.Import): + for al in node.names: + syms.add((al.asname or al.name).split(".")[0]) + elif isinstance(node, (ast.If, ast.Try)): + # 条件/异常分支里的顶层定义同样计入（兼容 shim 常用 try-import） + for sub in ast.walk(node): + if isinstance(sub, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + syms.add(sub.name) + elif isinstance(sub, ast.Assign): + for tgt in sub.targets: + if isinstance(tgt, ast.Name): + syms.add(tgt.id) + elif isinstance(sub, ast.ImportFrom): + for al in sub.names: + syms.add(al.asname or al.name) + table[mod] = syms + + # 包级聚合：__init__.py 的符号 + 所有子模块名本身（from pbl_common import crud 也合法） + agg: Set[str] = set(table.get("__init__", set())) + agg |= set(table.keys()) + for syms in table.values(): + agg |= syms + table["*"] = agg + return table + + +PROVIDER_PREFIXES = ("pbl_",) + + +def collect_imports(path: str) -> List[Tuple[int, str, List[str]]]: + """抽出文件里所有 from pbl_*[...] import ... 语句（跨模块契约引用）。 + + 返回 [(lineno, 模块路径, [符号...]), ...] + """ + try: + src = open(path, encoding="utf-8").read() + except (OSError, UnicodeDecodeError): + return [] + try: + tree = ast.parse(src, filename=path) + except SyntaxError: + return [] + out: List[Tuple[int, str, List[str]]] = [] + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and node.module and \ + any(node.module == p or node.module.startswith(p + ".") + for p in PROVIDER_PREFIXES): + names = [al.name for al in node.names if al.name != "*"] + if names: + out.append((node.lineno, node.module, names)) + return out + + +def scan_targets() -> List[str]: + """待扫描文件清单：所有 pbl_* 模块 + apps/pbls 下的 .py。""" + files: List[str] = [] + for root in SCAN_ROOTS: + if not os.path.isdir(root): + continue + if root == MODULES_DIR: + for name in sorted(os.listdir(root)): + if not name.startswith(MODULE_PREFIX): + continue + mod_dir = os.path.join(root, name) + if os.path.isdir(mod_dir): + files.extend(_iter_py_files(mod_dir)) + else: + files.extend(_iter_py_files(root)) + return sorted(set(files)) + + +# ---------------------------------------------------------------- 主核验 +def verify() -> Dict: + if not os.path.isdir(COMMON_PKG): + return {"ok": False, "fatal": "pbl_common 包目录不存在: %s" % COMMON_PKG, + "checked_files": 0, "violations": [], "dead_contract": []} + + symtab = collect_module_symbols(COMMON_PKG) + # 全部 pbl_* 模块包的符号表：{包名: {子模块名: 符号集}} + pkg_tabs: Dict[str, Dict[str, Set[str]]] = {"pbl_common": symtab} + for name in sorted(os.listdir(MODULES_DIR)): + if not name.startswith(MODULE_PREFIX) or name == "pbl_common": + continue + pkg_dir = os.path.join(MODULES_DIR, name, name) + if os.path.isdir(pkg_dir): + pkg_tabs[name] = collect_module_symbols(pkg_dir) + files = scan_targets() + + violations: List[Dict] = [] + for path in files: + # provider 包自身内部互引不算跨模块契约违规（内部结构可自由重构） + if any(os.path.abspath(path).startswith( + os.path.abspath(os.path.join(MODULES_DIR, k, k)) + os.sep) + for k in pkg_tabs): + continue + own_pkg = os.path.relpath(path, MODULES_DIR).split(os.sep)[0] + for lineno, mod, names in collect_imports(path): + provider = mod.split(".", 0)[0].split(".")[0] + tab = pkg_tabs.get(provider) + if tab is None: + continue # 非本仓 pbl_* 模块（如平台包），跳过 + if provider == own_pkg: + continue # 包内自引不算跨模块契约 + if "." not in mod: + avail = tab.get("*", set()) + sub = "%s(包)" % provider + else: + submod = mod.split(".", 1)[1] + avail = tab.get(submod, set()) + sub = "%s.%s" % (provider, submod) + if submod not in tab: + violations.append({ + "file": os.path.relpath(path, WS_ROOT), + "line": lineno, + "module": mod, + "symbol": "", + "reason": "子模块 %s/%s.py 不存在" % (provider, submod), + }) + continue + for nm in names: + if nm not in avail: + violations.append({ + "file": os.path.relpath(path, WS_ROOT), + "line": lineno, + "module": sub, + "symbol": nm, + "reason": "符号未在 %s 顶层定义/再导出（导入期 ImportError 风险）" % sub, + }) + + # 死契约核验：api.py 里 CONTRACT_SYMBOLS / __all__ 声明了但命名空间没有 + dead: List[Dict] = [] + api_path = os.path.join(COMMON_PKG, "api.py") + if os.path.isfile(api_path): + src = open(api_path, encoding="utf-8").read() + tree = ast.parse(src, filename=api_path) + declared: Dict[str, Set[str]] = {} + defined: Set[str] = set() + for node in tree.body: + if isinstance(node, ast.Assign): + for tgt in node.targets: + if isinstance(tgt, ast.Name) and tgt.id in ("__all__", "CONTRACT_SYMBOLS"): + declared[tgt.id] = _literal_names(node.value) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + defined.add(node.name) + elif isinstance(node, ast.Assign): + for tgt in node.targets: + if isinstance(tgt, ast.Name): + defined.add(tgt.id) + elif isinstance(node, ast.ImportFrom): + for al in node.names: + defined.add(al.asname or al.name) + for key, names in declared.items(): + for nm in sorted(names): + if nm not in defined: + dead.append({"list": key, "symbol": nm, + "reason": "声明于 %s 但 api.py 命名空间不可解析" % key}) + + return { + "ok": not violations and not dead, + "checked_files": len(files), + "common_modules": sorted(k for k in symtab if k != "*"), + "provider_packages": sorted(pkg_tabs.keys()), + "crud_symbol_present": "crud" in symtab.get("api", set()), + "violations": violations, + "dead_contract": dead, + } + + +def main() -> int: + ap = argparse.ArgumentParser(description="pbl_common 契约符号跨模块闭包核验") + ap.add_argument("--json", dest="json_out", default=None, + help="把结果写入指定 JSON 文件") + args = ap.parse_args() + + res = verify() + print("=" * 72) + print("pbl_common 契约符号闭包核验（Bug HoSBcdZY4G7IxhaFiIE07 同类问题全量排查）") + print("=" * 72) + if res.get("fatal"): + print("FATAL: %s" % res["fatal"]) + return 2 + print("扫描 .py 文件数 : %d" % res["checked_files"]) + print("pbl_common 子模块数 : %d" % len(res["common_modules"])) + print("核验 provider 包数 : %d (%s)" % (len(res["provider_packages"]), + ", ".join(res["provider_packages"]))) + print("api.crud 符号在位 : %s" % res["crud_symbol_present"]) + print("跨模块导入符号违规 : %d" % len(res["violations"])) + print("死契约(声明未实现) : %d" % len(res["dead_contract"])) + for v in res["violations"][:50]: + print(" [VIOLATION] %s:%s from %s import %s → %s" + % (v["file"], v["line"], v["module"], v["symbol"], v["reason"])) + for d in res["dead_contract"][:50]: + print(" [DEAD] %s 声明 '%s' → %s" % (d["list"], d["symbol"], d["reason"])) + print("-" * 72) + print("结论: %s" % ("PASS —— 全部 pbl_* 模块间跨模块导入符号均可解析" + if res["ok"] else "FAIL —— 存在导入期硬失败风险，须修复后再部署")) + + if args.json_out: + with open(args.json_out, "w", encoding="utf-8") as fh: + json.dump(res, fh, ensure_ascii=False, indent=2) + print("结果已写入: %s" % args.json_out) + return 0 if res["ok"] else 1 + + +if __name__ == "__main__": + sys.exit(main())