From 9050c6c832173ba481e257aeab03c2811ddeab37 Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Fri, 25 Sep 2026 08:54:39 +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_crud_symbol.py | 145 ++++++++++++++++++++++++++++++++++ 1 file changed, 145 insertions(+) create mode 100644 scripts/_patch_crud_symbol.py diff --git a/scripts/_patch_crud_symbol.py b/scripts/_patch_crud_symbol.py new file mode 100644 index 0000000..f337ca6 --- /dev/null +++ b/scripts/_patch_crud_symbol.py @@ -0,0 +1,145 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""一次性补丁：向 modules/pbl_common/pbl_common/api.py 注入缺失的 `crud` 工厂符号。 + +背景（Bug YyDumUuSK9R9Q_3Ai23AD / rfJ4V-ynABa7dpfJsaEA2 根因 G）： + deploy_test 第 3 轮实测 import 闭包 15/28，其中 pbl_assessment / pbl_scense_ext + 报 `cannot import name 'crud' from 'pbl_common.api'`。下游模块以 + RUB = crud('pbl_rubric', 'pbl', ['code', ...]) + await RUB['read'](id=..) / await RUB['create'](**payload) / await sc['upsert'](['page_key'], ...) + 的「dict 句柄 + 全 **kw」语义取用 CRUD，但 pbl_common.api 只导出了 + tenant_crud / crud_factory / make_crud（CrudBase 实例，双参 (row, where) 语义）， + 从未提供 `crud` 这个适配符号 → 包导入即硬失败。 + +本补丁在 api.py 的 crud_factory 导入块之后插入 `crud()` 适配工厂，并把 'crud' +登记进 CONTRACT_SYMBOLS 与 __all__，使静态契约面与运行时导出一致。幂等：已存在则跳过。 +""" +import io +import os +import sys + +API = os.path.join(os.path.dirname(__file__), '..', '..', '..', + 'modules', 'pbl_common', 'pbl_common', 'api.py') +API = os.path.abspath(API) + +CRUD_SRC = ''' + +# ========================================================================== +# crud —— dict 句柄适配工厂（下游 pbl_assessment / pbl_scense_ext 取用入口） +# ========================================================================== +def crud(table, module=None, columns=None, pk='id', **factory_kw): + """返回 dict 形态的租户安全 CRUD 句柄，把 CrudBase 的 (row, where) 双参语义 + 适配为下游模块使用的「全 **kw」单参语义。 + + 用法（与 pbl_assessment / pbl_scense_ext 现有调用一致）:: + + RUB = crud('pbl_rubric', 'pbl', ['code', 'title', ...]) + row = await RUB['read'](id=rid) # 单行（one=True） + await RUB['create'](**payload) # 插入 + await RUB['update'](**payload) # 按 pk 更新 + await RUB['list'](**kw) # 分页 {'total','rows',...} + await sc['upsert'](['page_key'], page_key=k, ...) # 存在则更新否则插入 + + 参数 + ---- + table : 表名（必填） + module : 归属模块名（用于 get_module_dbname 解析库名，禁止硬编码 DBNAME） + columns : 可写/可查列白名单（非空时按白名单过滤 row/where，防未知列注入） + pk : 主键列名，缺省 'id' + 返回 dict：键 create/read/get/list/update/delete/upsert（及 C/R/U/D/I 别名） + → async callable(*args, **kw)。 + """ + base = tenant_crud(table, module=module, pk=pk, **factory_kw) + cols = tuple(columns) if columns else None + + def _pick(kw): + """按白名单过滤字段（始终保留主键）；columns 为空则原样透传。""" + data = dict(kw or {}) + if not cols: + return data + return {k: v for k, v in data.items() if (k in cols) or (k == pk)} + + async def _create(*args, **kw): + return await base.C(_pick(kw)) + + async def _read(*args, **kw): + return await base.R(_pick(kw), one=True) + + async def _get(*args, **kw): + return await base.get(where=_pick(kw)) + + async def _list(*args, **kw): + kw = dict(kw or {}) + page = kw.pop('page', 1) + rows = kw.pop('rows', kw.pop('rows_per_page', 20)) + sort = kw.pop('sort', None) + order = kw.pop('order', 'asc') + return await base.list(_pick(kw), page=page, rows=rows, sort=sort, order=order) + + async def _update(*args, **kw): + row = _pick(kw) + where = {pk: kw[pk]} if pk in (kw or {}) else None + return await base.U(row, where=where) + + async def _delete(*args, **kw): + return await base.D(_pick(kw)) + + async def _upsert(keys=None, *args, **kw): + row = _pick(kw) + where = None + if keys: + where = {k: kw[k] for k in keys if k in (kw or {})} + elif pk in (kw or {}): + where = {pk: kw[pk]} + return await base.upsert(row, where=where) + + return { + 'create': _create, 'read': _read, 'get': _get, 'list': _list, + 'update': _update, 'delete': _delete, 'upsert': _upsert, + 'C': _create, 'R': _read, 'U': _update, 'D': _delete, 'I': _create, + } + +''' + +ANCHOR = "log = logging.getLogger('pbl_common.api')" + + +def main(): + with io.open(API, 'r', encoding='utf-8') as f: + src = f.read() + + if '\ndef crud(' in src: + print('[patch] crud() already present, skip injection (idempotent)') + else: + if ANCHOR not in src: + print('[patch][FATAL] anchor not found: %r' % ANCHOR) + return 2 + src = src.replace(ANCHOR, CRUD_SRC.lstrip('\n') + '\n' + ANCHOR, 1) + print('[patch] injected crud() factory before logger init') + + # 登记进 CONTRACT_SYMBOLS（crud 段） + if "'CrudBase', 'tenant_crud', 'crud_factory', 'make_crud', 'bulk_create'," in src \ + and "'crud'," not in src: + src = src.replace( + "'CrudBase', 'tenant_crud', 'crud_factory', 'make_crud', 'bulk_create',", + "'crud', 'CrudBase', 'tenant_crud', 'crud_factory', 'make_crud', 'bulk_create',", + 1) + print('[patch] registered crud into CONTRACT_SYMBOLS') + + # 登记进 __all__ 追加段（确保 from pbl_common.api import crud 与 * 导出都可见） + if "'clear_registry', 'AUDIT_ACTIONS', 'AUDIT_TABLE_MARKERS'," in src \ + and "'crud', 'clear_registry'" not in src: + src = src.replace( + "'clear_registry', 'AUDIT_ACTIONS', 'AUDIT_TABLE_MARKERS',", + "'crud', 'clear_registry', 'AUDIT_ACTIONS', 'AUDIT_TABLE_MARKERS',", + 1) + print('[patch] registered crud into __all__ tail') + + with io.open(API, 'w', encoding='utf-8') as f: + f.write(src) + print('[patch] written: %s' % API) + return 0 + + +if __name__ == '__main__': + sys.exit(main())