diff --git a/models/pbl_entity_ref.json b/models/pbl_entity_ref.json new file mode 100644 index 0000000..9bb973f --- /dev/null +++ b/models/pbl_entity_ref.json @@ -0,0 +1,30 @@ +{ + "summary": "PBL 实体关联表（薄扩展）：把基表 entity 绑定到 PBL 租户/班级/团队 + scene/world 父级归属，不改 entity 基表结构（Q-OPEN-3）", + "tblname": "pbl_entity_ref", + "fields": [ + {"name": "id", "type": "bigint", "primary": true, "auto": true, "comment": "主键"}, + {"name": "tenant_id", "type": "varchar(64)", "null": false, "comment": "PBL 租户ID（强制打头）"}, + {"name": "entity_id", "type": "bigint", "null": false, "comment": "基表 entity.id（只读引用，不建外键）"}, + {"name": "scene_id", "type": "bigint", "null": false, "default": 0, "comment": "父级归属 scene.id，0=未指定"}, + {"name": "world_id", "type": "bigint", "null": false, "default": 0, "comment": "父级归属 world.id，0=未指定"}, + {"name": "class_id", "type": "bigint", "null": false, "default": 0, "comment": "归属班级 pbl_class.id"}, + {"name": "team_id", "type": "bigint", "null": false, "default": 0, "comment": "归属团队 pbl_team.id"}, + {"name": "visibility", "type": "varchar(16)", "null": false, "default": "team", "comment": "可见性 team/class/tenant"}, + {"name": "role_code", "type": "varchar(64)", "null": false, "default": "", "comment": "角色绑定（该实体扮演的蓝图角色），空=通用"}, + {"name": "source", "type": "varchar(32)", "null": false, "default": "manual", "comment": "来源 manual/compiler/import"}, + {"name": "ext_json", "type": "text", "null": true, "comment": "PBL 侧扩展属性 JSON（实体在 PBL 语义下的额外属性，不回写基表）"}, + {"name": "created_by", "type": "varchar(64)", "null": true, "comment": "创建人"}, + {"name": "created_at", "type": "datetime", "null": false, "default": "CURRENT_TIMESTAMP", "comment": "创建时间"}, + {"name": "updated_at", "type": "datetime", "null": true, "comment": "更新时间"} + ], + "indexes": [ + {"name": "uk_pbl_entity_ref", "unique": true, "fields": ["tenant_id", "entity_id", "team_id", "role_code"], "comment": "幂等唯一键"}, + {"name": "idx_pbl_entity_ref_scene", "unique": false, "fields": ["tenant_id", "scene_id"], "comment": "按场景查实体"}, + {"name": "idx_pbl_entity_ref_world", "unique": false, "fields": ["tenant_id", "world_id"], "comment": "按世界查实体"}, + {"name": "idx_pbl_entity_ref_team", "unique": false, "fields": ["tenant_id", "team_id"], "comment": "按团队查"} + ], + "codes": { + "visibility": {"team": "团队可见", "class": "班级可见", "tenant": "租户可见"}, + "source": {"manual": "人工绑定", "compiler": "Compiler 物化", "import": "批量导入"} + } +} diff --git a/models/pbl_scene_ref.json b/models/pbl_scene_ref.json new file mode 100644 index 0000000..d628a84 --- /dev/null +++ b/models/pbl_scene_ref.json @@ -0,0 +1,29 @@ +{ + "summary": "PBL 场景关联表（薄扩展）：把基表 scene 绑定到 PBL 租户/班级/团队 + world 父级归属，不改 scene 基表结构（Q-OPEN-3）", + "tblname": "pbl_scene_ref", + "fields": [ + {"name": "id", "type": "bigint", "primary": true, "auto": true, "comment": "主键"}, + {"name": "tenant_id", "type": "varchar(64)", "null": false, "comment": "PBL 租户ID（强制打头）"}, + {"name": "scene_id", "type": "bigint", "null": false, "comment": "基表 scene.id（只读引用，不建外键）"}, + {"name": "world_id", "type": "bigint", "null": false, "default": 0, "comment": "父级归属 world.id，0=未指定"}, + {"name": "class_id", "type": "bigint", "null": false, "default": 0, "comment": "归属班级 pbl_class.id"}, + {"name": "team_id", "type": "bigint", "null": false, "default": 0, "comment": "归属团队 pbl_team.id"}, + {"name": "visibility", "type": "varchar(16)", "null": false, "default": "team", "comment": "可见性 team/class/tenant"}, + {"name": "role_code", "type": "varchar(64)", "null": false, "default": "", "comment": "角色绑定，空=通用"}, + {"name": "source", "type": "varchar(32)", "null": false, "default": "manual", "comment": "来源 manual/compiler/import"}, + {"name": "ext_json", "type": "text", "null": true, "comment": "PBL 侧扩展属性 JSON"}, + {"name": "created_by", "type": "varchar(64)", "null": true, "comment": "创建人"}, + {"name": "created_at", "type": "datetime", "null": false, "default": "CURRENT_TIMESTAMP", "comment": "创建时间"}, + {"name": "updated_at", "type": "datetime", "null": true, "comment": "更新时间"} + ], + "indexes": [ + {"name": "uk_pbl_scene_ref", "unique": true, "fields": ["tenant_id", "scene_id", "team_id", "role_code"], "comment": "幂等唯一键"}, + {"name": "idx_pbl_scene_ref_world", "unique": false, "fields": ["tenant_id", "world_id"], "comment": "按世界查场景"}, + {"name": "idx_pbl_scene_ref_class", "unique": false, "fields": ["tenant_id", "class_id"], "comment": "按班级查"}, + {"name": "idx_pbl_scene_ref_team", "unique": false, "fields": ["tenant_id", "team_id"], "comment": "按团队查"} + ], + "codes": { + "visibility": {"team": "团队可见", "class": "班级可见", "tenant": "租户可见"}, + "source": {"manual": "人工绑定", "compiler": "Compiler 物化", "import": "批量导入"} + } +} diff --git a/models/pbl_world_ref.json b/models/pbl_world_ref.json new file mode 100644 index 0000000..3c68a26 --- /dev/null +++ b/models/pbl_world_ref.json @@ -0,0 +1,28 @@ +{ + "summary": "PBL 世界关联表（薄扩展）：把基表 world 绑定到 PBL 租户/班级/团队，不改 world 基表结构（Q-OPEN-3）", + "tblname": "pbl_world_ref", + "fields": [ + {"name": "id", "type": "bigint", "primary": true, "auto": true, "comment": "主键"}, + {"name": "tenant_id", "type": "varchar(64)", "null": false, "comment": "PBL 租户ID（强制打头，所有查询必带）"}, + {"name": "world_id", "type": "bigint", "null": false, "comment": "基表 world.id（只读引用，不建外键以免锁基表）"}, + {"name": "class_id", "type": "bigint", "null": false, "default": 0, "comment": "归属班级 pbl_class.id，0=未指定"}, + {"name": "team_id", "type": "bigint", "null": false, "default": 0, "comment": "归属团队 pbl_team.id，0=未指定"}, + {"name": "visibility", "type": "varchar(16)", "null": false, "default": "team", "comment": "可见性 team/class/tenant"}, + {"name": "role_code", "type": "varchar(64)", "null": false, "default": "", "comment": "角色绑定（teacher/student/observer 或蓝图角色码），空=通用"}, + {"name": "source", "type": "varchar(32)", "null": false, "default": "manual", "comment": "来源 manual/compiler/import"}, + {"name": "ext_json", "type": "text", "null": true, "comment": "PBL 侧扩展属性 JSON（不改基表的前提下承载额外字段）"}, + {"name": "created_by", "type": "varchar(64)", "null": true, "comment": "创建人"}, + {"name": "created_at", "type": "datetime", "null": false, "default": "CURRENT_TIMESTAMP", "comment": "创建时间"}, + {"name": "updated_at", "type": "datetime", "null": true, "comment": "更新时间"} + ], + "indexes": [ + {"name": "uk_pbl_world_ref", "unique": true, "fields": ["tenant_id", "world_id", "team_id", "role_code"], "comment": "幂等唯一键"}, + {"name": "idx_pbl_world_ref_class", "unique": false, "fields": ["tenant_id", "class_id"], "comment": "按班级查"}, + {"name": "idx_pbl_world_ref_team", "unique": false, "fields": ["tenant_id", "team_id"], "comment": "按团队查"}, + {"name": "idx_pbl_world_ref_world", "unique": false, "fields": ["world_id"], "comment": "反查基表归属"} + ], + "codes": { + "visibility": {"team": "团队可见", "class": "班级可见", "tenant": "租户可见"}, + "source": {"manual": "人工绑定", "compiler": "Compiler 物化", "import": "批量导入"} + } +} diff --git a/pbl_domain_ext/__init__.py b/pbl_domain_ext/__init__.py index dd067f7..0b08738 100644 --- a/pbl_domain_ext/__init__.py +++ b/pbl_domain_ext/__init__.py @@ -1,27 +1,42 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- -"""pbl_domain_ext —— world/scene/entity 薄扩展：租户/班级/团队关联（M8，不改基表） +"""pbl_domain_ext —— PBL 基础域薄扩展（world/scene/entity 关联表 + 查询契约，M8）。 -注册三处同步之 ②：必须导出 init.py 里的全部契约函数，漏一行 .dspy 调用即 NameError。 +对外暴露： +- load_pbl_domain_ext()：模块挂载入口（应用 init() 调用） +- api：契约层（pbl_*_ref_upsert / pbl_*_ref_list / pbl_domain_scope_resolve / ...） +- assoc：读写内核（不依赖 Sage 运行时，可离线单测） + +铁律：world / scene / entity 三张基表零 ALTER、零写入（Q-OPEN-3）。 """ -from pbl_domain_ext.init import load_pbl_domain_ext -from pbl_domain_ext.api import ( - pbl_tenant_upsert, - pbl_class_save, - pbl_class_list, - pbl_team_save, - pbl_team_list, - pbl_domain_materialize_game_definition, +from __future__ import annotations + +from . import api, assoc +from .init import ( + API_NAMES, + MODULE_NAME, + OWN_TABLES, + READONLY_BASE_TABLES, + get_dbname, + load_cruds, + load_models, + load_module, + load_pbl_domain_ext, ) __all__ = [ - 'load_pbl_domain_ext', - 'pbl_tenant_upsert', - 'pbl_class_save', - 'pbl_class_list', - 'pbl_team_save', - 'pbl_team_list', - 'pbl_domain_materialize_game_definition', - + "api", + "assoc", + "load_pbl_domain_ext", + "load_module", + "load_models", + "load_cruds", + "get_dbname", + "MODULE_NAME", + "OWN_TABLES", + "READONLY_BASE_TABLES", + "API_NAMES", ] + +__version__ = "2.0.0" diff --git a/pbl_domain_ext/api.py b/pbl_domain_ext/api.py index b8092f8..7bf741a 100644 --- a/pbl_domain_ext/api.py +++ b/pbl_domain_ext/api.py @@ -1,139 +1,464 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- -"""pbl_domain_ext.api —— world/scene/entity 薄扩展（M8，Q-OPEN-3：不改基表，只加关联表）。 +"""pbl_domain_ext.api —— world/scene/entity 薄扩展契约面（M8，Q-OPEN-3：不改基表）。 -规则： -- 基表 world/scene/entity **零 ALTER**（对既有 yuanjing 项目零影响），PBL 侧一律通过 - 关联表 `pbl_tenant`/`pbl_class`/`pbl_team` + 本模块的映射函数建立溯源； -- Game Definition 物化（materialize）= 把编译产物写入 world/scene/entity 的**只新增行**路径， - 通过基座模块契约函数（ServerEnv 上已注册的 load_world/load_scene/load_entity 能力）调用， - 不直接拼基表 SQL；基座函数不在位时降级为 pbl 侧影子表并明确回报 degraded（不静默假成功）。 +铁律： +- 基表 world / scene / entity **零 ALTER、零写入**；PBL 侧归属关系一律落在本模块 + 自有表（pbl_tenant / pbl_class / pbl_team + pbl_world_ref / pbl_scene_ref / pbl_entity_ref）； +- 所有契约 tenant_id 强制打头：显式入参 > pbl_common 上下文租户 > fail-closed 报错， + 绝不静默兜底成「默认租户」； +- 业务规则全部在 assoc.py（可离线单测），本层只做「解包 → 调内核 → 统一响应体」。 + +历史版 api.py 依赖 pbl_common.api 中并不存在的符号（actor_id/crud/json_dump/sql_rows/ +sql_exec/sql_scalar/tenant_id），import 即 ImportError。本版改为只依赖 assoc 内核 + +可选的 pbl_common 上下文，import 闭包闭合，可离线单测。 """ + +from __future__ import annotations + import json +from typing import Any, Dict, List, Optional -from pbl_common.api import (PblError, actor_id, crud, json_dump, now_str, sql_exec, - sql_rows, sql_scalar, tenant_id) +from . import assoc +from .assoc import ( + ENTITY_REF, + SCENE_REF, + WORLD_REF, + DomainExtError, + default_db, +) -TEN = crud('pbl_tenant', 'pbl', ['tenant_uid', 'name', 'kind', 'plan', 'seat_limit', 'status', - 'config_json']) -CLS = crud('pbl_class', 'pbl', ['class_uid', 'name', 'grade', 'teacher_id', 'student_count', - 'roster_json', 'status']) -TEAM = crud('pbl_team', 'pbl', ['team_uid', 'class_id', 'name', 'mission_id', 'members_json', - 'capacity', 'status', 'session_id']) +__all__ = [ + "pbl_tenant_upsert", + "pbl_class_save", + "pbl_class_list", + "pbl_team_save", + "pbl_team_list", + "pbl_world_ref_upsert", + "pbl_scene_ref_upsert", + "pbl_entity_ref_upsert", + "pbl_world_ref_list", + "pbl_scene_ref_list", + "pbl_entity_ref_list", + "pbl_domain_ref_get", + "pbl_domain_ref_delete", + "pbl_domain_scope_resolve", + "pbl_domain_stats", + "pbl_domain_materialize_game_definition", + "call_api", + "API_NAMES", +] + +# ---------------------------------------------------------------- 可选依赖（缺失不致命） +try: # pragma: no cover - 取决于运行环境 + from pbl_common.context import get_tenant as _ctx_tenant +except Exception: # pragma: no cover + _ctx_tenant = None + +try: # pragma: no cover + from pbl_common.errors import PBLError as _PBLError +except Exception: # pragma: no cover + class _PBLError(Exception): + code = "PBL_ERROR" + + def __init__(self, code: str = "PBL_ERROR", message: str = "") -> None: + super().__init__(message) + self.code = code + self.message = message -def _loads(v, d=None): - if isinstance(v, (dict, list)): - return v +# ---------------------------------------------------------------- 归属主数据表字段 +TENANT_FIELDS = ["tenant_uid", "name", "kind", "plan", "seat_limit", "status", "config_json"] +CLASS_FIELDS = ["class_uid", "name", "grade", "teacher_id", "student_count", "roster_json", "status"] +TEAM_FIELDS = ["team_uid", "class_id", "name", "mission_id", "members_json", "capacity", + "status", "session_id"] + +_TABLE_ALIASES: Dict[str, str] = { + "world": WORLD_REF, "world_ref": WORLD_REF, "pbl_world_ref": WORLD_REF, + "scene": SCENE_REF, "scene_ref": SCENE_REF, "pbl_scene_ref": SCENE_REF, + "entity": ENTITY_REF, "entity_ref": ENTITY_REF, "pbl_entity_ref": ENTITY_REF, +} + +_KEY_FIELD = {WORLD_REF: "world_id", SCENE_REF: "scene_id", ENTITY_REF: "entity_id"} + + +# ---------------------------------------------------------------- 响应体 / 上下文 +def _ok(data: Any = None, **extra: Any) -> Dict[str, Any]: + payload: Dict[str, Any] = {"ok": True, "code": "OK"} + if data is not None: + payload["data"] = data + payload.update(extra) + return payload + + +def _fail(code: str, message: str, **detail: Any) -> Dict[str, Any]: + payload: Dict[str, Any] = {"ok": False, "code": code, "message": message} + if detail: + payload["detail"] = detail + return payload + + +def _guard(func): + """统一异常 → 响应体（DomainExtError / PBLError / ValueError）。""" + + def wrapper(params: Dict[str, Any]) -> Dict[str, Any]: + try: + return func(params) + except DomainExtError as exc: + return _fail(exc.code, exc.message, **exc.detail) + except _PBLError as exc: + return _fail(getattr(exc, "code", "PBL_ERROR"), str(getattr(exc, "message", exc))) + except ValueError as exc: + return _fail(assoc.ERR_PARAM_REQUIRED, "参数解析失败: %s" % (exc,)) + + wrapper.__name__ = func.__name__ + wrapper.__doc__ = func.__doc__ + return wrapper + + +def _db(params: Optional[Dict[str, Any]] = None): + """测试可注入 _db；生产走 sqlor + ServerEnv().get_module_dbname('pbl_domain_ext')。""" + injected = (params or {}).get("_db") + return injected if injected is not None else default_db() + + +def _resolve_tenant(params: Dict[str, Any]) -> str: + """租户解析：显式入参优先，其次 pbl_common 上下文，都没有 → fail-closed。""" + raw = params.get("tenant_id") + if raw is not None and str(raw).strip(): + return str(raw).strip() + if _ctx_tenant is not None: + try: + ctx = _ctx_tenant() + if isinstance(ctx, str) and ctx.strip(): + return ctx.strip() + if isinstance(ctx, dict): + value = ctx.get("tenant_id") or ctx.get("tenantid") + if value and str(value).strip(): + return str(value).strip() + if ctx is not None and not isinstance(ctx, (str, dict)): + value = getattr(ctx, "tenant_id", None) + if value and str(value).strip(): + return str(value).strip() + except Exception: + pass + raise DomainExtError(assoc.ERR_TENANT_REQUIRED, + "缺少 tenant_id：入参未给且无 PBL 租户上下文（拒绝默认租户兜底）") + + +def _loads(value: Any, default: Any = None): + if isinstance(value, (dict, list)): + return value + if value is None or (isinstance(value, str) and not value.strip()): + return default if default is not None else {} try: - return json.loads(v) if v else (d or {}) + return json.loads(value) except ValueError: - return d or {} + return default if default is not None else {} -async def _tid(): - return await tenant_id() +def _dumps(value: Any) -> str: + if value is None: + return "" + if isinstance(value, str): + return value + return json.dumps(value, ensure_ascii=False) -async def pbl_tenant_upsert(**kw): - payload = dict(kw) - payload['config_json'] = json_dump(kw.get('config_json')) - payload.setdefault('tenant_uid', kw.get('tenant_uid') or await _tid()) - payload.setdefault('status', 'active') - r = await TEN['upsert'](['tenant_uid'], **payload) - r['base_tables_altered'] = False - return r +# ================================================================ 归属主数据（tenant/class/team） +def _upsert_master(db, table: str, tenant: str, unique_field: str, fields: List[str], + params: Dict[str, Any]) -> Dict[str, Any]: + """pbl_tenant / pbl_class / pbl_team 的幂等 upsert（按 tenant_id + 业务唯一码）。""" + table = assoc._safe_col(table) + unique = assoc._safe_col(unique_field) + uid = str(params.get(unique_field) or "").strip() + if not uid: + raise DomainExtError(assoc.ERR_PARAM_REQUIRED, "缺少必填 %s" % unique_field) + cols = [c for c in fields if c != unique_field] + values = {c: params.get(c) for c in cols} + + rows = db.select( + "SELECT * FROM %s WHERE tenant_id=%%s AND %s=%%s ORDER BY id LIMIT 1" % (table, unique), + (tenant, uid), + ) + if rows: + assigns = ", ".join("%s=%%s" % assoc._safe_col(c) for c in cols) + sql = "UPDATE %s SET %s, updated_at=NOW() WHERE tenant_id=%%s AND id=%%s" % (table, assigns) + db.execute(sql, tuple([values[c] for c in cols] + [tenant, int(rows[0].get("id", 0))])) + return {"table": table, "action": "updated", "id": int(rows[0].get("id", 0)), + "tenant_id": tenant, unique_field: uid} + + all_cols = ["tenant_id", unique] + cols + sql = "INSERT INTO %s (%s) VALUES (%s)" % ( + table, ", ".join(all_cols), ", ".join(["%s"] * len(all_cols))) + new_id = db.insert(sql, tuple([tenant, uid] + [values[c] for c in cols])) + return {"table": table, "action": "created", "id": int(new_id or 0), + "tenant_id": tenant, unique_field: uid} -async def pbl_class_save(**kw): - payload = dict(kw) - roster = _loads(kw.get('roster_json'), []) - payload['roster_json'] = json_dump(roster) - payload['student_count'] = len(roster) if roster else kw.get('student_count', 0) - payload.setdefault('class_uid', 'CL%s' % now_str().replace('-', '').replace(':', '') - .replace(' ', '')) - if kw.get('id'): - return await CLS['update'](**payload) - return await CLS['create'](**payload) +def _list_master(db, table: str, tenant: str, params: Dict[str, Any]) -> Dict[str, Any]: + table = assoc._safe_col(table) + conds = [("tenant_id", tenant)] + for field in ("class_id", "status", "kind", "session_id", "mission_id"): + if field in params and params[field] not in (None, ""): + conds.append((assoc._safe_col(field), params[field])) + items = assoc._select(db, table, conds, order_by="id", + limit=params.get("limit", assoc.DEFAULT_LIMIT), + offset=params.get("offset", 0)) + return { + "table": table, + "tenant_id": tenant, + "items": items, + "total": assoc._count(db, table, conds), + "limit": assoc._clamp_limit(params.get("limit", assoc.DEFAULT_LIMIT)), + "offset": assoc._clamp_offset(params.get("offset", 0)), + } -async def pbl_class_list(**kw): - return await CLS['list'](**kw) +@_guard +def pbl_tenant_upsert(params: Dict[str, Any]) -> Dict[str, Any]: + """租户主数据幂等写入（tenant_uid 唯一）。基表零改动。""" + db = _db(params) + tenant = _resolve_tenant(params) + payload = dict(params) + payload["config_json"] = _dumps(_loads(params.get("config_json"), {})) + payload.setdefault("status", "active") + payload.setdefault("tenant_uid", tenant) + data = _upsert_master(db, "pbl_tenant", tenant, "tenant_uid", TENANT_FIELDS, payload) + data["base_tables_altered"] = False + return _ok(data) -async def pbl_team_save(**kw): - tid = await _tid() - members = _loads(kw.get('members_json'), []) - capacity = int(kw.get('capacity') or 8) - if len(members) > capacity: - raise PblError('PBL_TEAM_OVER_CAPACITY', '成员 %d 超容量 %d（单会话 2~8 人）' - % (len(members), capacity)) - payload = dict(kw) - payload['members_json'] = json_dump(members) - payload.setdefault('team_uid', 'TM%s' % now_str().replace('-', '').replace(':', '') - .replace(' ', '')) - payload.setdefault('status', 'active') - if kw.get('id'): - return await TEAM['update'](**payload) - return await TEAM['create'](**payload) +@_guard +def pbl_class_save(params: Dict[str, Any]) -> Dict[str, Any]: + """班级写入：roster_json 决定 student_count，class_uid 缺省自动生成。""" + db = _db(params) + tenant = _resolve_tenant(params) + payload = dict(params) + roster = _loads(params.get("roster_json"), []) + payload["roster_json"] = _dumps(roster) + if isinstance(roster, list) and roster: + payload["student_count"] = len(roster) + payload.setdefault("class_uid", "CL%s" % assoc._norm_tenant(tenant).replace("-", "")[:8] + + str(int(params.get("id") or 0))) + payload.setdefault("status", "active") + data = _upsert_master(db, "pbl_class", tenant, "class_uid", CLASS_FIELDS, payload) + return _ok(data) -async def pbl_team_list(**kw): - r = await TEAM['list'](**kw) - for row in r['data']: - row['members_json'] = _loads(row.get('members_json'), []) - row['member_count'] = len(row['members_json']) - return r +@_guard +def pbl_class_list(params: Dict[str, Any]) -> Dict[str, Any]: + db = _db(params) + tenant = _resolve_tenant(params) + data = _list_master(db, "pbl_class", tenant, params) + for row in data.get("items", []): + row["roster_json"] = _loads(row.get("roster_json"), []) + return _ok(data) -async def pbl_domain_materialize_game_definition(**kw): - """Game Definition → world/scene/entity（新增行 + 溯源映射），供 pbl_compiler 调用。""" - from ahserver.serverenv import ServerEnv - tid = await _tid() - gd_id = kw.get('id') - rows = await sql_rows('SELECT * FROM `pbl_game_definition` WHERE `tenant_id` = ${t}$' - ' AND `id` = ${id}$ LIMIT 1', {'t': tid, 'id': gd_id}, 'pbl') - if not rows: - raise PblError('PBL_GD_NOT_FOUND', 'Game Definition 不存在') - gd = rows[0] - definition = _loads(gd['definition_json'], {}) - if gd.get('world_id'): - return {'ok': True, 'already': True, 'world_id': gd['world_id'], - 'scene_id': gd['scene_id'], 'degraded': False} - env = ServerEnv() - world_fn = getattr(env, 'create_world', None) or getattr(env, 'create_worlds', None) - scene_fn = getattr(env, 'create_scene', None) or getattr(env, 'create_scenes', None) - entity_fn = getattr(env, 'create_entity', None) or getattr(env, 'create_entities', None) - degraded = not all(callable(f) for f in (world_fn, scene_fn, entity_fn)) - title = 'PBL-%s-%s' % (gd['blueprint_id'], str(gd['content_fingerprint'])[:8]) - if degraded: - # 基座函数未挂载（如离线单测）：影子登记，明确回报，不假成功 - world_id = await sql_scalar( - 'SELECT MAX(`id`) AS m FROM `pbl_game_definition` WHERE `tenant_id` = ${t}$', - {'t': tid}, 'pbl', default=0) - return {'ok': True, 'world_id': None, 'scene_id': None, 'degraded': True, - 'reason': 'base module loaders (world/scene/entity) not mounted', - 'shadow_ref': world_id, 'title': title} - w = await _maybe(world_fn(name=title, description='compiled from game_definition %s' % gd_id, - mode=definition.get('world', {}).get('mode', 'shared'))) - world_id = _pk(w) - s = await _maybe(scene_fn(name=title + '-scene', world_id=world_id)) - scene_id = _pk(s) - for ent in definition.get('entities', []): - await _maybe(entity_fn(name=ent['entity_id'], world_id=world_id, scene_id=scene_id, - entity_type=ent['kind'], - capability_key=ent.get('capability_key'))) - return {'ok': True, 'world_id': world_id, 'scene_id': scene_id, 'degraded': False, - 'entity_rows': len(definition.get('entities', [])), - 'base_tables_altered': False, - 'note': '仅新增行；world/scene/entity 表结构未改（Q-OPEN-3）'} +@_guard +def pbl_team_save(params: Dict[str, Any]) -> Dict[str, Any]: + """团队写入：成员数不得超 capacity（单会话 2~8 人约束由蓝图侧校验，这里只挡超容量）。""" + db = _db(params) + tenant = _resolve_tenant(params) + members = _loads(params.get("members_json"), []) + capacity = int(params.get("capacity") or 8) + if isinstance(members, list) and len(members) > capacity: + raise DomainExtError(assoc.ERR_CONFLICT, + "成员 %d 超容量 %d" % (len(members), capacity), + member_count=len(members), capacity=capacity) + payload = dict(params) + payload["members_json"] = _dumps(members) + payload["capacity"] = capacity + payload.setdefault("team_uid", "TM%s" % str(params.get("name") or "").strip()[:8] + + str(int(params.get("class_id") or 0))) + payload.setdefault("status", "active") + data = _upsert_master(db, "pbl_team", tenant, "team_uid", TEAM_FIELDS, payload) + return _ok(data) -async def _maybe(v): - return await v if hasattr(v, '__await__') else v +@_guard +def pbl_team_list(params: Dict[str, Any]) -> Dict[str, Any]: + db = _db(params) + tenant = _resolve_tenant(params) + data = _list_master(db, "pbl_team", tenant, params) + for row in data.get("items", []): + row["members_json"] = _loads(row.get("members_json"), []) + row["member_count"] = len(row["members_json"]) if isinstance(row["members_json"], list) else 0 + return _ok(data) -def _pk(res): - if isinstance(res, dict): - return res.get('id') or res.get('data') or res.get('world_id') or res.get('scene_id') - return res +# ================================================================ 三张关联表契约 +@_guard +def pbl_world_ref_upsert(params: Dict[str, Any]) -> Dict[str, Any]: + """把基表 world 绑定到 PBL 租户/班级/团队（不写 world 基表）。""" + db = _db(params) + tenant = _resolve_tenant(params) + data = assoc.upsert_world_ref( + db, tenant, params.get("world_id"), + class_id=params.get("class_id", 0), team_id=params.get("team_id", 0), + visibility=params.get("visibility"), role_code=params.get("role_code", ""), + source=params.get("source", assoc.SOURCE_MANUAL), ext=params.get("ext"), + created_by=params.get("created_by", ""), + ) + data["base_tables_altered"] = False + return _ok(data) + + +@_guard +def pbl_scene_ref_upsert(params: Dict[str, Any]) -> Dict[str, Any]: + """把基表 scene 绑定到 PBL 归属，可带 world_id 父级（不写 scene 基表）。""" + db = _db(params) + tenant = _resolve_tenant(params) + data = assoc.upsert_scene_ref( + db, tenant, params.get("scene_id"), world_id=params.get("world_id", 0), + class_id=params.get("class_id", 0), team_id=params.get("team_id", 0), + visibility=params.get("visibility"), role_code=params.get("role_code", ""), + source=params.get("source", assoc.SOURCE_MANUAL), ext=params.get("ext"), + created_by=params.get("created_by", ""), + ) + data["base_tables_altered"] = False + return _ok(data) + + +@_guard +def pbl_entity_ref_upsert(params: Dict[str, Any]) -> Dict[str, Any]: + """把基表 entity 绑定到 PBL 归属，可带 scene_id/world_id 父级（不写 entity 基表）。""" + db = _db(params) + tenant = _resolve_tenant(params) + data = assoc.upsert_entity_ref( + db, tenant, params.get("entity_id"), scene_id=params.get("scene_id", 0), + world_id=params.get("world_id", 0), class_id=params.get("class_id", 0), + team_id=params.get("team_id", 0), visibility=params.get("visibility"), + role_code=params.get("role_code", ""), source=params.get("source", assoc.SOURCE_MANUAL), + ext=params.get("ext"), created_by=params.get("created_by", ""), + ) + data["base_tables_altered"] = False + return _ok(data) + + +def _ref_list(params: Dict[str, Any], table: str) -> Dict[str, Any]: + db = _db(params) + tenant = _resolve_tenant(params) + filters = params.get("filters") + if not isinstance(filters, dict): + filters = {k: params[k] for k in ("world_id", "scene_id", "entity_id", "class_id", + "team_id", "visibility", "role_code", "source") + if k in params} + return _ok(assoc.list_refs(db, table, tenant, filters, + limit=params.get("limit", assoc.DEFAULT_LIMIT), + offset=params.get("offset", 0))) + + +@_guard +def pbl_world_ref_list(params: Dict[str, Any]) -> Dict[str, Any]: + return _ref_list(params, WORLD_REF) + + +@_guard +def pbl_scene_ref_list(params: Dict[str, Any]) -> Dict[str, Any]: + return _ref_list(params, SCENE_REF) + + +@_guard +def pbl_entity_ref_list(params: Dict[str, Any]) -> Dict[str, Any]: + return _ref_list(params, ENTITY_REF) + + +@_guard +def pbl_domain_ref_get(params: Dict[str, Any]) -> Dict[str, Any]: + """按 (table, tenant_id, 业务主键, team_id, role_code) 取单条关联记录。""" + db = _db(params) + tenant = _resolve_tenant(params) + key = str(params.get("table") or "").strip().lower() + if key not in _TABLE_ALIASES: + raise DomainExtError(assoc.ERR_PARAM_REQUIRED, "未知关联表别名: %r" % (params.get("table"),)) + table = _TABLE_ALIASES[key] + key_col = _KEY_FIELD[table] + row = assoc.get_ref(db, table, tenant, params.get(key_col, params.get("id_value")), + team_id=params.get("team_id", 0), role_code=params.get("role_code", ""), + strict=bool(params.get("strict", True))) + return _ok(row, found=row is not None) + + +@_guard +def pbl_domain_ref_delete(params: Dict[str, Any]) -> Dict[str, Any]: + """删除单条关联记录（只删 PBL 关联表，绝不动基表）。""" + db = _db(params) + tenant = _resolve_tenant(params) + key = str(params.get("table") or "").strip().lower() + if key not in _TABLE_ALIASES: + raise DomainExtError(assoc.ERR_PARAM_REQUIRED, "未知关联表别名: %r" % (params.get("table"),)) + table = _TABLE_ALIASES[key] + affected = assoc.delete_ref(db, table, tenant, params.get("id")) + return _ok({"table": table, "deleted": int(affected or 0), "base_tables_altered": False}) + + +@_guard +def pbl_domain_scope_resolve(params: Dict[str, Any]) -> Dict[str, Any]: + """entity→scene→world 逐级回溯解析 PBL 归属作用域（供 runtime/assessment/kdb_ext 用）。""" + db = _db(params) + tenant = _resolve_tenant(params) + return _ok(assoc.resolve_scope(db, tenant, world_id=params.get("world_id"), + scene_id=params.get("scene_id"), + entity_id=params.get("entity_id"))) + + +@_guard +def pbl_domain_stats(params: Dict[str, Any]) -> Dict[str, Any]: + """租户维度关联计数（治理看板 / QC 核对）。""" + db = _db(params) + return _ok(assoc.domain_stats(db, _resolve_tenant(params))) + + +@_guard +def pbl_domain_materialize_game_definition(params: Dict[str, Any]) -> Dict[str, Any]: + """Game Definition → 三张关联表归属记录（幂等物化，供 pbl_compiler 调用）。 + + 与历史版差异：历史版试图往 world/scene/entity 基表「只新增行」，违反 Q-OPEN-3 + 「不改基表」决议且依赖不存在的 sql_rows；本版只写 PBL 关联表，基表零写入。 + """ + db = _db(params) + tenant = _resolve_tenant(params) + definition = params.get("game_definition") + if definition is None: + definition = params.get("definition") + data = assoc.bind_game_definition( + db, tenant, definition, class_id=params.get("class_id", 0), + team_id=params.get("team_id", 0), source=params.get("source", assoc.SOURCE_COMPILER), + created_by=params.get("created_by", ""), + ) + data["base_tables_altered"] = False + return _ok(data) + + +# ================================================================ dspy 统一入口 +_API_MAP = { + "pbl_tenant_upsert": pbl_tenant_upsert, + "pbl_class_save": pbl_class_save, + "pbl_class_list": pbl_class_list, + "pbl_team_save": pbl_team_save, + "pbl_team_list": pbl_team_list, + "pbl_world_ref_upsert": pbl_world_ref_upsert, + "pbl_scene_ref_upsert": pbl_scene_ref_upsert, + "pbl_entity_ref_upsert": pbl_entity_ref_upsert, + "pbl_world_ref_list": pbl_world_ref_list, + "pbl_scene_ref_list": pbl_scene_ref_list, + "pbl_entity_ref_list": pbl_entity_ref_list, + "pbl_domain_ref_get": pbl_domain_ref_get, + "pbl_domain_ref_delete": pbl_domain_ref_delete, + "pbl_domain_scope_resolve": pbl_domain_scope_resolve, + "pbl_domain_stats": pbl_domain_stats, + "pbl_domain_materialize_game_definition": pbl_domain_materialize_game_definition, +} + +API_NAMES: List[str] = sorted(_API_MAP.keys()) + + +def call_api(name: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + """按接口名分发（dspy 端点一行接入：`return call_api('pbl_world_ref_list', params_kw)`）。""" + handler = _API_MAP.get(str(name or "").strip()) + if handler is None: + return _fail(assoc.ERR_PARAM_REQUIRED, + "未知接口: %r（可用 %s）" % (name, ", ".join(API_NAMES))) + return handler(dict(params or {})) diff --git a/pbl_domain_ext/assoc.py b/pbl_domain_ext/assoc.py new file mode 100644 index 0000000..9bcd5c1 --- /dev/null +++ b/pbl_domain_ext/assoc.py @@ -0,0 +1,735 @@ +"""PBL 基础域薄扩展 —— world/scene/entity 关联表读写内核（M8）。 + +设计约束（Q-OPEN-3 决议，见 projects/pbls/docs/01-design/modules/pbl_domain_ext.md）： + +1. **不改基表**：world / scene / entity 三张基表结构与数据一律不动，本模块只新增 + 三张 PBL 侧关联表（pbl_world_ref / pbl_scene_ref / pbl_entity_ref）承载归属关系； +2. **租户强制打头**：任何读写都必须带 tenant_id，缺失/为空立即 fail-closed 抛错， + 不做「默认租户」兜底； +3. **薄扩展**：关联表只存归属（tenant/class/team）+ 可见性 + 角色绑定 + 扩展 JSON， + 业务状态仍留在基表与 scense_runtime，不在本模块复制； +4. **SQL 白名单**：表名/列名一律经 _safe_col 校验，值一律走占位符，杜绝拼接注入。 + +本文件不依赖 Sage 运行时（sqlor/ServerEnv 仅在 SqlorDb 适配器里延迟导入）， +因此可被 tests/ 下的内存 DB 直接单测。 +""" + +from __future__ import annotations + +import json +import re +from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple + +MODULE_NAME = "pbl_domain_ext" + +# ---------------------------------------------------------------- 关联表常量 +WORLD_REF = "pbl_world_ref" +SCENE_REF = "pbl_scene_ref" +ENTITY_REF = "pbl_entity_ref" +REF_TABLES: Tuple[str, ...] = (WORLD_REF, SCENE_REF, ENTITY_REF) + +# 基表（只读引用，本模块绝不写入 / 绝不 DDL） +BASE_TABLES: Tuple[str, ...] = ("world", "scene", "entity") + +REF_SCHEMA: Dict[str, Dict[str, Any]] = { + WORLD_REF: {"key": "world_id", "parents": (), "base": "world"}, + SCENE_REF: {"key": "scene_id", "parents": ("world_id",), "base": "scene"}, + ENTITY_REF: {"key": "entity_id", "parents": ("scene_id", "world_id"), "base": "entity"}, +} + +# ---------------------------------------------------------------- 可见性 / 来源 +VISIBILITY_TEAM = "team" +VISIBILITY_CLASS = "class" +VISIBILITY_TENANT = "tenant" +VISIBILITY_VALUES: Tuple[str, ...] = (VISIBILITY_TEAM, VISIBILITY_CLASS, VISIBILITY_TENANT) + +SOURCE_MANUAL = "manual" +SOURCE_COMPILER = "compiler" +SOURCE_IMPORT = "import" +SOURCE_VALUES: Tuple[str, ...] = (SOURCE_MANUAL, SOURCE_COMPILER, SOURCE_IMPORT) + +# ---------------------------------------------------------------- 错误码 +ERR_TENANT_REQUIRED = "PBL_DOMAIN_EXT_TENANT_REQUIRED" +ERR_PARAM_REQUIRED = "PBL_DOMAIN_EXT_PARAM_REQUIRED" +ERR_BAD_VISIBILITY = "PBL_DOMAIN_EXT_BAD_VISIBILITY" +ERR_NOT_FOUND = "PBL_DOMAIN_EXT_NOT_FOUND" +ERR_CONFLICT = "PBL_DOMAIN_EXT_CONFLICT" +ERR_DB_UNAVAILABLE = "PBL_DOMAIN_EXT_DB_UNAVAILABLE" + +MAX_LIMIT = 200 +DEFAULT_LIMIT = 50 + +_COL_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") +_SOURCE_RE = re.compile(r"^[A-Za-z0-9_\-]{1,32}$") +# role_code 允许蓝图角色码常见字符；显式禁掉引号/分号/空白，纵深防御（值本身仍走占位符） +_ROLE_RE = re.compile(r"^[A-Za-z0-9_\-.:@]{0,64}$") + + +class DomainExtError(Exception): + """PBL 基础域薄扩展统一异常（fail-closed，带机器可读 code）。""" + + def __init__(self, code: str, message: str, **detail: Any) -> None: + super().__init__(message) + self.code = code + self.message = message + self.detail: Dict[str, Any] = dict(detail) + + def to_dict(self) -> Dict[str, Any]: + payload: Dict[str, Any] = {"code": self.code, "message": self.message} + if self.detail: + payload["detail"] = self.detail + return payload + + +# ================================================================ DB 适配层 +class DbPort(object): + """关联表读写所需的最小 DB 契约（便于单测注入内存实现）。""" + + def select(self, sql: str, params: Sequence[Any] = ()) -> List[Dict[str, Any]]: + raise NotImplementedError + + def insert(self, sql: str, params: Sequence[Any] = ()) -> int: + raise NotImplementedError + + def execute(self, sql: str, params: Sequence[Any] = ()) -> int: + raise NotImplementedError + + +class SqlorDb(DbPort): + """sqlor 适配器：库名走 ServerEnv().get_module_dbname，禁止硬编码 DBNAME。""" + + def __init__(self, dbname: Optional[str] = None) -> None: + self._dbname = dbname + self._sor: Any = None + + @property + def sor(self) -> Any: + if self._sor is None: + try: + import sqlor as _sqlor # type: ignore + except Exception as exc: # pragma: no cover - 运行环境缺失 + raise DomainExtError(ERR_DB_UNAVAILABLE, "sqlor 不可用: %s" % (exc,)) + self._sor = _sqlor + return self._sor + + @property + def dbname(self) -> str: + if self._dbname: + return self._dbname + try: + from appbase import ServerEnv # type: ignore + except Exception: # pragma: no cover + ServerEnv = None # type: ignore + if ServerEnv is not None: + try: + name = ServerEnv().get_module_dbname(MODULE_NAME) + if name: + return str(name) + except Exception: # pragma: no cover + pass + raise DomainExtError( + ERR_DB_UNAVAILABLE, + "无法解析模块库名：ServerEnv().get_module_dbname('%s')" % MODULE_NAME, + ) + + def _run(self, sql: str, params: Sequence[Any]) -> Any: + args = tuple(params or ()) + try: + return self.sor.sqlExe(self.dbname, sql, args) + except TypeError: + return self.sor.sqlExe(sql, args) + + def select(self, sql: str, params: Sequence[Any] = ()) -> List[Dict[str, Any]]: + return _normalize_rows(self._run(sql, params)) + + def insert(self, sql: str, params: Sequence[Any] = ()) -> int: + result = self._run(sql, params) + if isinstance(result, (list, tuple)) and result: + head = result[0] + if isinstance(head, dict): + for key in ("lastrowid", "insert_id", "id"): + if key in head: + return int(head[key] or 0) + try: + return int(head) + except (TypeError, ValueError): + return 0 + if isinstance(result, dict): + for key in ("lastrowid", "insert_id", "id"): + if key in result: + return int(result[key] or 0) + try: + return int(result or 0) + except (TypeError, ValueError): + return 0 + + def execute(self, sql: str, params: Sequence[Any] = ()) -> int: + result = self._run(sql, params) + if isinstance(result, (list, tuple)): + return len(result) + try: + return int(result or 0) + except (TypeError, ValueError): + return 0 + + +def _normalize_rows(result: Any) -> List[Dict[str, Any]]: + """把 sqlor 各种返回形态统一成 list[dict]。""" + if result is None: + return [] + if isinstance(result, dict): + for key in ("rows", "data", "result", "items"): + if key in result: + return _normalize_rows(result[key]) + return [result] + if isinstance(result, (list, tuple)): + rows: List[Dict[str, Any]] = [] + for item in result: + if isinstance(item, dict): + rows.append(item) + elif isinstance(item, (list, tuple)): + rows.append({str(index): value for index, value in enumerate(item)}) + return rows + return [] + + +def default_db() -> DbPort: + """缺省 DB（生产路径）：sqlor + ServerEnv 库名映射。""" + return SqlorDb() + + +# ================================================================ 归一化 / 校验 +def _norm_tenant(tenant_id: Any) -> str: + if tenant_id is None: + raise DomainExtError(ERR_TENANT_REQUIRED, "缺少 tenant_id（PBL 全域强制租户打头）") + text = str(tenant_id).strip() + if not text: + raise DomainExtError(ERR_TENANT_REQUIRED, "tenant_id 不能为空字符串") + if len(text) > 64: + raise DomainExtError(ERR_TENANT_REQUIRED, "tenant_id 超长（>64）", length=len(text)) + return text + + +def _norm_id(value: Any, field: str, required: bool = True) -> int: + if value is None or (isinstance(value, str) and not value.strip()): + if required: + raise DomainExtError(ERR_PARAM_REQUIRED, "缺少必填参数 %s" % field, field=field) + return 0 + try: + ivalue = int(str(value).strip()) + except (TypeError, ValueError): + raise DomainExtError(ERR_PARAM_REQUIRED, "参数 %s 必须为整数，实际 %r" % (field, value), field=field) + if ivalue < 0: + raise DomainExtError(ERR_PARAM_REQUIRED, "参数 %s 不能为负数，实际 %d" % (field, ivalue), field=field) + if ivalue == 0 and required: + raise DomainExtError(ERR_PARAM_REQUIRED, "参数 %s 必须为正整数" % field, field=field) + return ivalue + + +def _norm_visibility(value: Any) -> str: + text = str(value or "").strip().lower() + if not text: + return VISIBILITY_TEAM + if text not in VISIBILITY_VALUES: + raise DomainExtError( + ERR_BAD_VISIBILITY, + "visibility 非法: %r，允许值 %s" % (value, "/".join(VISIBILITY_VALUES)), + ) + return text + + +def _norm_role(role_code: Any) -> str: + text = str(role_code or "").strip() + if len(text) > 64: + raise DomainExtError(ERR_PARAM_REQUIRED, "role_code 超长（>64）", length=len(text)) + if text and not _ROLE_RE.match(text): + raise DomainExtError( + ERR_PARAM_REQUIRED, + "role_code 含非法字符: %r（限 [A-Za-z0-9_-.:@]，≤64）" % (role_code,), + ) + return text + + +def _norm_source(value: Any) -> str: + text = str(value or SOURCE_MANUAL).strip() + if not _SOURCE_RE.match(text): + raise DomainExtError(ERR_PARAM_REQUIRED, "source 非法: %r（限 [A-Za-z0-9_-]{1,32}）" % (value,)) + return text + + +def _auto_visibility(class_id: int, team_id: int) -> str: + if team_id > 0: + return VISIBILITY_TEAM + if class_id > 0: + return VISIBILITY_CLASS + return VISIBILITY_TENANT + + +def _check_scope(visibility: str, class_id: int, team_id: int) -> None: + if visibility == VISIBILITY_TEAM and team_id <= 0: + raise DomainExtError(ERR_PARAM_REQUIRED, "visibility=team 时必须提供 team_id>0") + if visibility == VISIBILITY_CLASS and class_id <= 0: + raise DomainExtError(ERR_PARAM_REQUIRED, "visibility=class 时必须提供 class_id>0") + + +def _dump_ext(ext: Any) -> str: + if ext is None: + return "" + if isinstance(ext, str): + text = ext.strip() + if not text: + return "" + try: + json.loads(text) + except ValueError as exc: + raise DomainExtError(ERR_PARAM_REQUIRED, "ext 必须是合法 JSON 字符串: %s" % (exc,)) + return text + try: + return json.dumps(ext, ensure_ascii=False, sort_keys=True) + except (TypeError, ValueError) as exc: + raise DomainExtError(ERR_PARAM_REQUIRED, "ext 无法序列化为 JSON: %s" % (exc,)) + + +def _safe_col(name: Any) -> str: + text = str(name) + if not _COL_RE.match(text): + raise DomainExtError(ERR_PARAM_REQUIRED, "非法标识符（表名/列名）: %r" % (name,)) + return text + + +def _clamp_limit(limit: Any) -> int: + try: + value = int(limit) + except (TypeError, ValueError): + return DEFAULT_LIMIT + if value <= 0: + return DEFAULT_LIMIT + return min(value, MAX_LIMIT) + + +def _clamp_offset(offset: Any) -> int: + try: + value = int(offset) + except (TypeError, ValueError): + return 0 + return max(0, min(value, 1000000)) + + +def _columns(table: str) -> List[str]: + schema = REF_SCHEMA[table] + cols = ["tenant_id", schema["key"]] + cols.extend(schema["parents"]) + cols.extend(["class_id", "team_id", "visibility", "role_code", "source", "ext_json", "created_by"]) + return cols + + +# ================================================================ SQL 构造（受限子集） +def _build_where(conds: Sequence[Tuple[str, Any]]) -> Tuple[str, List[Any]]: + clauses: List[str] = [] + params: List[Any] = [] + for name, value in conds: + clauses.append("%s=%%s" % _safe_col(name)) + params.append(value) + where = (" WHERE " + " AND ".join(clauses)) if clauses else "" + return where, params + + +def _select(db: DbPort, table: str, conds: Sequence[Tuple[str, Any]], + order_by: str = "id", limit: Any = DEFAULT_LIMIT, offset: Any = 0, + cols: str = "*") -> List[Dict[str, Any]]: + where, params = _build_where(conds) + sql = "SELECT %s FROM %s%s ORDER BY %s LIMIT %d OFFSET %d" % ( + cols, _safe_col(table), where, _safe_col(order_by), _clamp_limit(limit), _clamp_offset(offset), + ) + return db.select(sql, tuple(params)) + + +def _count(db: DbPort, table: str, conds: Sequence[Tuple[str, Any]]) -> int: + where, params = _build_where(conds) + sql = "SELECT COUNT(*) AS cnt FROM %s%s" % (_safe_col(table), where) + rows = db.select(sql, tuple(params)) + if not rows: + return 0 + first = rows[0] + for key in ("cnt", "count", "COUNT(*)"): + if key in first: + return int(first[key] or 0) + values = list(first.values()) + return int(values[0] or 0) if values else 0 + + +def _require_table(table: str) -> str: + name = _safe_col(table) + if name not in REF_SCHEMA: + raise DomainExtError(ERR_PARAM_REQUIRED, "未知关联表: %s（允许 %s）" % (table, "/".join(REF_TABLES))) + return name + + +# ================================================================ 通用读写 +def _find_ref(db: DbPort, table: str, tenant: str, key_id: int, team_id: int, role_code: str) -> Optional[Dict[str, Any]]: + key_col = REF_SCHEMA[table]["key"] + conds = [("tenant_id", tenant), (key_col, key_id), ("team_id", team_id), ("role_code", role_code)] + rows = _select(db, table, conds, order_by="id", limit=1, offset=0) + return rows[0] if rows else None + + +def upsert_ref(db: DbPort, table: str, tenant_id: Any, key_value: Any, + parents: Optional[Dict[str, Any]] = None, class_id: Any = 0, team_id: Any = 0, + visibility: Any = None, role_code: Any = "", source: Any = SOURCE_MANUAL, + ext: Any = None, created_by: Any = "") -> Dict[str, Any]: + """关联表幂等写入：唯一键 (tenant_id, 业务主键, team_id, role_code)。 + + 存在则更新归属/可见性/扩展，不存在则插入；返回 action=created|updated。 + """ + name = _require_table(table) + schema = REF_SCHEMA[name] + tenant = _norm_tenant(tenant_id) + key_col = schema["key"] + key_id = _norm_id(key_value, key_col) + + given = dict(parents or {}) + parent_values: Dict[str, int] = {} + for pname in schema["parents"]: + parent_values[pname] = _norm_id(given.get(pname, 0), pname, required=False) + for extra in given: + if extra not in schema["parents"]: + raise DomainExtError(ERR_PARAM_REQUIRED, "%s 不支持父级列 %s" % (name, extra)) + + cid = _norm_id(class_id, "class_id", required=False) + tid = _norm_id(team_id, "team_id", required=False) + vis = _norm_visibility(visibility if visibility is not None else _auto_visibility(cid, tid)) + _check_scope(vis, cid, tid) + role = _norm_role(role_code) + src = _norm_source(source) + ext_text = _dump_ext(ext) + + existing = _find_ref(db, name, tenant, key_id, tid, role) + base_result: Dict[str, Any] = { + "table": name, + "tenant_id": tenant, + "class_id": cid, + "team_id": tid, + "visibility": vis, + "role_code": role, + "source": src, + } + base_result[key_col] = key_id + base_result.update(parent_values) + + if existing is not None: + assigns = ["class_id=%s", "visibility=%s", "source=%s", "ext_json=%s", "updated_at=NOW()"] + for pname in schema["parents"]: + assigns.append("%s=%%s" % _safe_col(pname)) + sql = "UPDATE %s SET %s WHERE tenant_id=%%s AND id=%%s" % (name, ", ".join(assigns)) + params: List[Any] = [cid, vis, src, ext_text] + for pname in schema["parents"]: + params.append(parent_values[pname]) + params.append(tenant) + params.append(int(existing.get("id", 0))) + db.execute(sql, tuple(params)) + base_result["action"] = "updated" + base_result["id"] = int(existing.get("id", 0)) + return base_result + + cols = _columns(name) + sql = "INSERT INTO %s (%s) VALUES (%s)" % (name, ", ".join(cols), ", ".join(["%s"] * len(cols))) + params = [tenant, key_id] + for pname in schema["parents"]: + params.append(parent_values[pname]) + params.extend([cid, tid, vis, role, src, ext_text, str(created_by or "")]) + new_id = db.insert(sql, tuple(params)) + base_result["action"] = "created" + base_result["id"] = int(new_id or 0) + return base_result + + +def list_refs(db: DbPort, table: str, tenant_id: Any, filters: Optional[Dict[str, Any]] = None, + limit: Any = DEFAULT_LIMIT, offset: Any = 0) -> Dict[str, Any]: + """关联表分页查询：tenant_id 强制打头，过滤字段走白名单。""" + name = _require_table(table) + schema = REF_SCHEMA[name] + tenant = _norm_tenant(tenant_id) + conds: List[Tuple[str, Any]] = [("tenant_id", tenant)] + allowed = set(schema["parents"]) | {"class_id", "team_id", "visibility", "role_code", "source", schema["key"]} + for field, value in dict(filters or {}).items(): + if field not in allowed: + raise DomainExtError(ERR_PARAM_REQUIRED, "不支持的过滤字段: %s（允许 %s）" % (field, sorted(allowed))) + if value is None or (isinstance(value, str) and not value.strip()): + continue + conds.append((field, value)) + total = _count(db, name, conds) + items = _select(db, name, conds, order_by="id", limit=limit, offset=offset) + return { + "table": name, + "tenant_id": tenant, + "total": total, + "limit": _clamp_limit(limit), + "offset": _clamp_offset(offset), + "items": items, + } + + +def get_ref(db: DbPort, table: str, tenant_id: Any, key_value: Any, team_id: Any = 0, + role_code: Any = "", strict: bool = True) -> Optional[Dict[str, Any]]: + name = _require_table(table) + tenant = _norm_tenant(tenant_id) + key_col = REF_SCHEMA[name]["key"] + key_id = _norm_id(key_value, key_col) + tid = _norm_id(team_id, "team_id", required=False) + role = _norm_role(role_code) + row = _find_ref(db, name, tenant, key_id, tid, role) + if row is None and strict: + raise DomainExtError(ERR_NOT_FOUND, "关联记录不存在: %s.%s=%d" % (name, key_col, key_id), table=name) + return row + + +def delete_ref(db: DbPort, table: str, tenant_id: Any, ref_id: Any) -> int: + name = _require_table(table) + tenant = _norm_tenant(tenant_id) + rid = _norm_id(ref_id, "id") + sql = "DELETE FROM %s WHERE tenant_id=%%s AND id=%%s" % name + return int(db.execute(sql, (tenant, rid)) or 0) + + +# ================================================================ 三表薄封装 +def upsert_world_ref(db: DbPort, tenant_id: Any, world_id: Any, class_id: Any = 0, team_id: Any = 0, + visibility: Any = None, role_code: Any = "", source: Any = SOURCE_MANUAL, + ext: Any = None, created_by: Any = "") -> Dict[str, Any]: + return upsert_ref(db, WORLD_REF, tenant_id, world_id, None, class_id, team_id, + visibility, role_code, source, ext, created_by) + + +def upsert_scene_ref(db: DbPort, tenant_id: Any, scene_id: Any, world_id: Any = 0, class_id: Any = 0, + team_id: Any = 0, visibility: Any = None, role_code: Any = "", + source: Any = SOURCE_MANUAL, ext: Any = None, created_by: Any = "") -> Dict[str, Any]: + return upsert_ref(db, SCENE_REF, tenant_id, scene_id, {"world_id": world_id}, class_id, team_id, + visibility, role_code, source, ext, created_by) + + +def upsert_entity_ref(db: DbPort, tenant_id: Any, entity_id: Any, scene_id: Any = 0, world_id: Any = 0, + class_id: Any = 0, team_id: Any = 0, visibility: Any = None, role_code: Any = "", + source: Any = SOURCE_MANUAL, ext: Any = None, created_by: Any = "") -> Dict[str, Any]: + return upsert_ref(db, ENTITY_REF, tenant_id, entity_id, {"scene_id": scene_id, "world_id": world_id}, + class_id, team_id, visibility, role_code, source, ext, created_by) + + +def list_world_refs(db: DbPort, tenant_id: Any, filters: Optional[Dict[str, Any]] = None, + limit: Any = DEFAULT_LIMIT, offset: Any = 0) -> Dict[str, Any]: + return list_refs(db, WORLD_REF, tenant_id, filters, limit, offset) + + +def list_scene_refs(db: DbPort, tenant_id: Any, filters: Optional[Dict[str, Any]] = None, + limit: Any = DEFAULT_LIMIT, offset: Any = 0) -> Dict[str, Any]: + return list_refs(db, SCENE_REF, tenant_id, filters, limit, offset) + + +def list_entity_refs(db: DbPort, tenant_id: Any, filters: Optional[Dict[str, Any]] = None, + limit: Any = DEFAULT_LIMIT, offset: Any = 0) -> Dict[str, Any]: + return list_refs(db, ENTITY_REF, tenant_id, filters, limit, offset) + + +def domain_stats(db: DbPort, tenant_id: Any) -> Dict[str, Any]: + """租户维度的关联表计数（供治理看板 / QC 核对）。""" + tenant = _norm_tenant(tenant_id) + stats: Dict[str, Any] = {"tenant_id": tenant} + for name in REF_TABLES: + stats[name] = _count(db, name, [("tenant_id", tenant)]) + stats["total"] = sum(int(stats[name]) for name in REF_TABLES) + return stats + + +# ================================================================ 作用域解析 +def _absorb(target: Dict[str, Any], row: Dict[str, Any]) -> None: + """把关联行的归属信息吸收进解析结果（已确定的值不被覆盖：越具体越优先）。""" + for field in ("team_id", "class_id"): + if not target.get(field) and row.get(field) is not None: + try: + value = int(row.get(field) or 0) + except (TypeError, ValueError): + value = 0 + if value > 0: + target[field] = value + if not target.get("visibility") and row.get("visibility"): + target["visibility"] = str(row["visibility"]) + + +def resolve_scope(db: DbPort, tenant_id: Any, world_id: Any = None, scene_id: Any = None, + entity_id: Any = None) -> Dict[str, Any]: + """由 entity → scene → world 逐级回溯，解析出 PBL 归属作用域。 + + 返回 {tenant_id, world, scene, entity, world_id, scene_id, entity_id, + team_id, class_id, visibility, resolved_from}；三级都查不到即 fail-closed。 + """ + tenant = _norm_tenant(tenant_id) + wid = _norm_id(world_id, "world_id", required=False) + sid = _norm_id(scene_id, "scene_id", required=False) + eid = _norm_id(entity_id, "entity_id", required=False) + if wid <= 0 and sid <= 0 and eid <= 0: + raise DomainExtError(ERR_PARAM_REQUIRED, "world_id / scene_id / entity_id 至少提供一个正整数") + + result: Dict[str, Any] = { + "tenant_id": tenant, "world": None, "scene": None, "entity": None, + "world_id": wid, "scene_id": sid, "entity_id": eid, + "team_id": 0, "class_id": 0, "visibility": "", "resolved_from": "", + } + + if eid > 0: + row = get_ref(db, ENTITY_REF, tenant, eid, strict=False) + if row is not None: + result["entity"] = row + result["resolved_from"] = "entity" + _absorb(result, row) + if sid <= 0: + sid = _norm_id(row.get("scene_id"), "scene_id", required=False) + result["scene_id"] = sid + if wid <= 0: + wid = _norm_id(row.get("world_id"), "world_id", required=False) + result["world_id"] = wid + + if sid > 0: + row = get_ref(db, SCENE_REF, tenant, sid, strict=False) + if row is not None: + result["scene"] = row + if not result["resolved_from"]: + result["resolved_from"] = "scene" + _absorb(result, row) + if wid <= 0: + wid = _norm_id(row.get("world_id"), "world_id", required=False) + result["world_id"] = wid + + if wid > 0: + row = get_ref(db, WORLD_REF, tenant, wid, strict=False) + if row is not None: + result["world"] = row + if not result["resolved_from"]: + result["resolved_from"] = "world" + _absorb(result, row) + + if not result["resolved_from"]: + raise DomainExtError( + ERR_NOT_FOUND, + "未找到任何 PBL 关联记录（world=%s scene=%s entity=%s）" % (wid, sid, eid), + world_id=wid, scene_id=sid, entity_id=eid, + ) + return result + + +# ================================================================ Game Definition 物化 +_ID_KEYS: Dict[str, Tuple[str, ...]] = { + WORLD_REF: ("world_id", "worldId", "world"), + SCENE_REF: ("scene_id", "sceneId", "scene", "scenes"), + ENTITY_REF: ("entity_id", "entityId", "entity", "entities"), +} + + +def _walk(node: Any) -> Iterable[Any]: + if isinstance(node, dict): + yield node + for value in node.values(): + for sub in _walk(value): + yield sub + elif isinstance(node, (list, tuple)): + for item in node: + for sub in _walk(item): + yield sub + + +def _as_id_list(value: Any) -> List[int]: + out: List[int] = [] + if isinstance(value, (list, tuple)): + for item in value: + out.extend(_as_id_list(item)) + return out + if isinstance(value, dict): + for key in ("id", "world_id", "scene_id", "entity_id"): + if key in value: + out.extend(_as_id_list(value[key])) + break + return out + if isinstance(value, bool) or value is None: + return out + try: + ivalue = int(str(value).strip()) + except (TypeError, ValueError): + return out + if ivalue > 0: + out.append(ivalue) + return out + + +def collect_ids(game_definition: Any, table: str) -> List[int]: + """从 Compiler 产出的 Game Definition 里递归抽取某类基表 id（去重保序）。""" + name = _require_table(table) + if isinstance(game_definition, (str, bytes)): + game_definition = json.loads(game_definition) + ids: List[int] = [] + seen = set() + for node in _walk(game_definition): + for key in _ID_KEYS[name]: + if key in node: + for cand in _as_id_list(node[key]): + if cand not in seen: + seen.add(cand) + ids.append(cand) + return ids + + +def bind_game_definition(db: DbPort, tenant_id: Any, game_definition: Any, class_id: Any = 0, + team_id: Any = 0, source: Any = SOURCE_COMPILER, + created_by: Any = "") -> Dict[str, Any]: + """把 Compiler 产出的 Game Definition 物化成三张关联表的归属记录（幂等）。 + + 只写 PBL 关联表，不回写 world/scene/entity 基表。 + """ + tenant = _norm_tenant(tenant_id) + if isinstance(game_definition, (str, bytes)): + text = game_definition.decode("utf-8") if isinstance(game_definition, bytes) else game_definition + if not text.strip(): + raise DomainExtError(ERR_PARAM_REQUIRED, "game_definition 为空") + try: + game_definition = json.loads(text) + except ValueError as exc: + raise DomainExtError(ERR_PARAM_REQUIRED, "game_definition 不是合法 JSON: %s" % (exc,)) + if not isinstance(game_definition, dict): + raise DomainExtError(ERR_PARAM_REQUIRED, "game_definition 必须是对象或 JSON 字符串") + + cid = _norm_id(class_id, "class_id", required=False) + tid = _norm_id(team_id, "team_id", required=False) + vis = _auto_visibility(cid, tid) + src = _norm_source(source) + + world_ids = collect_ids(game_definition, WORLD_REF) + scene_ids = collect_ids(game_definition, SCENE_REF) + entity_ids = collect_ids(game_definition, ENTITY_REF) + + summary: Dict[str, Any] = { + "tenant_id": tenant, "class_id": cid, "team_id": tid, "visibility": vis, "source": src, + "world_refs": 0, "scene_refs": 0, "entity_refs": 0, "created": 0, "updated": 0, "skipped": [], + } + if not world_ids and not scene_ids and not entity_ids: + summary["skipped"].append("game_definition 中未发现 world/scene/entity id") + return summary + + primary_world = world_ids[0] if world_ids else 0 + if len(world_ids) > 1: + summary["skipped"].append("发现多个 world_id=%s，父级归属取首个 %d" % (world_ids, primary_world)) + primary_scene = scene_ids[0] if len(scene_ids) == 1 else 0 + if len(scene_ids) > 1: + summary["skipped"].append("发现多个 scene_id=%s，entity 的 scene 父级留空(0)待人工确认" % (scene_ids,)) + + for wid in world_ids: + res = upsert_world_ref(db, tenant, wid, cid, tid, vis, "", src, None, created_by) + summary["world_refs"] += 1 + summary[res["action"]] = summary.get(res["action"], 0) + 1 + + for sid in scene_ids: + res = upsert_scene_ref(db, tenant, sid, primary_world, cid, tid, vis, "", src, None, created_by) + summary["scene_refs"] += 1 + summary[res["action"]] = summary.get(res["action"], 0) + 1 + + for eid in entity_ids: + res = upsert_entity_ref(db, tenant, eid, primary_scene, primary_world, cid, tid, vis, "", src, None, created_by) + summary["entity_refs"] += 1 + summary[res["action"]] = summary.get(res["action"], 0) + 1 + + return summary diff --git a/pbl_domain_ext/init.py b/pbl_domain_ext/init.py index aede6c1..d8b15aa 100644 --- a/pbl_domain_ext/init.py +++ b/pbl_domain_ext/init.py @@ -1,29 +1,124 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- -"""`load_pbl_domain_ext()` —— pbl_domain_ext 模块唯一挂载入口。 +"""`load_pbl_domain_ext()` —— pbl_domain_ext 模块唯一挂载入口（M8）。 -注册三处同步之 ③：env.<契约名> = <契约名>（① 定义在 api.py，② 导出在 __init__.py）。 +三处同步注册（module-development-spec）： +① 表定义 models/*.json（pbl_tenant/pbl_class/pbl_team + pbl_world_ref/pbl_scene_ref/pbl_entity_ref） +② CRUD 定义 json/*.json +③ 契约端点 wwwroot/api/*.dspy → 本文件把 api.py 的契约函数挂到 ServerEnv + +铁律： +- **不改基表**：world / scene / entity 三张基表零 ALTER、零写入（Q-OPEN-3 决议）， + PBL 归属关系只落在本模块自有关联表； +- 库名走 ServerEnv().get_module_dbname('pbl_domain_ext')，禁止硬编码 DBNAME； +- 所有契约 tenant_id 强制打头，缺失即 fail-closed。 + +本文件对 ahserver 的导入是**延迟且容错**的：无 Sage 运行时（离线单测 / QC 静态核验） +时 load_pbl_domain_ext() 仍返回挂载摘要并把错误记入 summary['env_error']，不抛异常。 """ -from ahserver.serverenv import ServerEnv -from pbl_domain_ext.api import ( - pbl_tenant_upsert, - pbl_class_save, - pbl_class_list, - pbl_team_save, - pbl_team_list, - pbl_domain_materialize_game_definition, +from __future__ import annotations -) +import json +import os +from typing import Any, Dict, List, Optional + +from . import api, assoc + +MODULE_NAME = assoc.MODULE_NAME +MODULE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + +# 本模块自有表（归属主数据 + 三张关联表） +OWN_TABLES: List[str] = [ + "pbl_tenant", + "pbl_class", + "pbl_team", + assoc.WORLD_REF, + assoc.SCENE_REF, + assoc.ENTITY_REF, +] + +# 只读引用的基表（本模块绝不 DDL / 绝不写入） +READONLY_BASE_TABLES: List[str] = list(assoc.BASE_TABLES) + +# 契约清单（与 wwwroot/api/*.dspy 一一对应） +API_NAMES: List[str] = list(api.API_NAMES) -def load_pbl_domain_ext(): - env = ServerEnv() - env.pbl_tenant_upsert = pbl_tenant_upsert - env.pbl_class_save = pbl_class_save - env.pbl_class_list = pbl_class_list - env.pbl_team_save = pbl_team_save - env.pbl_team_list = pbl_team_list - env.pbl_domain_materialize_game_definition = pbl_domain_materialize_game_definition +def get_module_dir() -> str: + return MODULE_DIR - return 'pbl_domain_ext' + +def load_models() -> List[Dict[str, Any]]: + """读取 models/*.json 表定义（供应用建表 / QC 核对字段）。""" + models: List[Dict[str, Any]] = [] + model_dir = os.path.join(MODULE_DIR, "models") + if not os.path.isdir(model_dir): + return models + for fname in sorted(os.listdir(model_dir)): + if not fname.endswith(".json"): + continue + with open(os.path.join(model_dir, fname), "r", encoding="utf-8") as fp: + models.append(json.load(fp)) + return models + + +def load_cruds() -> List[Dict[str, Any]]: + """读取 json/*.json CRUD 定义。""" + cruds: List[Dict[str, Any]] = [] + crud_dir = os.path.join(MODULE_DIR, "json") + if not os.path.isdir(crud_dir): + return cruds + for fname in sorted(os.listdir(crud_dir)): + if not fname.endswith(".json"): + continue + with open(os.path.join(crud_dir, fname), "r", encoding="utf-8") as fp: + cruds.append(json.load(fp)) + return cruds + + +def get_dbname() -> str: + """模块库名：ServerEnv 映射优先，禁止硬编码。""" + return assoc.SqlorDb().dbname + + +def load_pbl_domain_ext(env: Optional[Any] = None) -> Dict[str, Any]: + """挂载入口：应用 init() 里 `load_pbl_domain_ext()` 调用，返回挂载摘要。 + + env 缺省时尝试取 ahserver.ServerEnv()；取不到（离线环境）则只返回摘要不抛错。 + """ + summary: Dict[str, Any] = { + "module": MODULE_NAME, + "module_dir": MODULE_DIR, + "own_tables": list(OWN_TABLES), + "readonly_base_tables": list(READONLY_BASE_TABLES), + "apis": list(API_NAMES), + "models": len(load_models()), + "cruds": len(load_cruds()), + "registered": [], + } + + target = env + if target is None: + try: + from ahserver.serverenv import ServerEnv # type: ignore + target = ServerEnv() + except Exception as exc: # pragma: no cover - 离线环境 + summary["env_error"] = "ahserver.ServerEnv 不可用: %s" % (exc,) + + if target is not None: + for name in API_NAMES: + setattr(target, name, getattr(api, name)) + summary["registered"].append(name) + + try: + summary["dbname"] = get_dbname() + except assoc.DomainExtError as exc: + summary["dbname"] = None + summary["dbname_error"] = exc.to_dict() + + return summary + + +# 兼容旧命名 +load_module = load_pbl_domain_ext diff --git a/sql/pbl_domain_ext_assoc.sql b/sql/pbl_domain_ext_assoc.sql new file mode 100644 index 0000000..b504ffd --- /dev/null +++ b/sql/pbl_domain_ext_assoc.sql @@ -0,0 +1,69 @@ +-- ============================================================ +-- pbl_domain_ext —— M8 基础域薄扩展 DDL（world/scene/entity 关联表） +-- 铁律：不修改 world / scene / entity 三张基表（Q-OPEN-3 决议） +-- 关联表不建外键（避免锁基表 / 跨库），只存 id 引用 +-- 方言：MariaDB / MySQL（BIGINT AUTO_INCREMENT） +-- ============================================================ + +CREATE TABLE IF NOT EXISTS pbl_world_ref ( + id BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键', + tenant_id VARCHAR(64) NOT NULL COMMENT 'PBL 租户ID（强制打头）', + world_id BIGINT NOT NULL COMMENT '基表 world.id（只读引用）', + class_id BIGINT NOT NULL DEFAULT 0 COMMENT '归属班级 pbl_class.id', + team_id BIGINT NOT NULL DEFAULT 0 COMMENT '归属团队 pbl_team.id', + visibility VARCHAR(16) NOT NULL DEFAULT 'team' COMMENT '可见性 team/class/tenant', + role_code VARCHAR(64) NOT NULL DEFAULT '' COMMENT '角色绑定，空=通用', + source VARCHAR(32) NOT NULL DEFAULT 'manual' COMMENT '来源 manual/compiler/import', + ext_json TEXT NULL COMMENT 'PBL 侧扩展属性 JSON', + created_by VARCHAR(64) NULL COMMENT '创建人', + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', + updated_at DATETIME NULL COMMENT '更新时间', + PRIMARY KEY (id), + UNIQUE KEY uk_pbl_world_ref (tenant_id, world_id, team_id, role_code), + KEY idx_pbl_world_ref_class (tenant_id, class_id), + KEY idx_pbl_world_ref_team (tenant_id, team_id), + KEY idx_pbl_world_ref_world (world_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='PBL 世界关联表（薄扩展，不改 world 基表）'; + +CREATE TABLE IF NOT EXISTS pbl_scene_ref ( + id BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键', + tenant_id VARCHAR(64) NOT NULL COMMENT 'PBL 租户ID（强制打头）', + scene_id BIGINT NOT NULL COMMENT '基表 scene.id（只读引用）', + world_id BIGINT NOT NULL DEFAULT 0 COMMENT '父级归属 world.id', + class_id BIGINT NOT NULL DEFAULT 0 COMMENT '归属班级 pbl_class.id', + team_id BIGINT NOT NULL DEFAULT 0 COMMENT '归属团队 pbl_team.id', + visibility VARCHAR(16) NOT NULL DEFAULT 'team' COMMENT '可见性 team/class/tenant', + role_code VARCHAR(64) NOT NULL DEFAULT '' COMMENT '角色绑定，空=通用', + source VARCHAR(32) NOT NULL DEFAULT 'manual' COMMENT '来源 manual/compiler/import', + ext_json TEXT NULL COMMENT 'PBL 侧扩展属性 JSON', + created_by VARCHAR(64) NULL COMMENT '创建人', + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', + updated_at DATETIME NULL COMMENT '更新时间', + PRIMARY KEY (id), + UNIQUE KEY uk_pbl_scene_ref (tenant_id, scene_id, team_id, role_code), + KEY idx_pbl_scene_ref_world (tenant_id, world_id), + KEY idx_pbl_scene_ref_class (tenant_id, class_id), + KEY idx_pbl_scene_ref_team (tenant_id, team_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='PBL 场景关联表（薄扩展，不改 scene 基表）'; + +CREATE TABLE IF NOT EXISTS pbl_entity_ref ( + id BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键', + tenant_id VARCHAR(64) NOT NULL COMMENT 'PBL 租户ID（强制打头）', + entity_id BIGINT NOT NULL COMMENT '基表 entity.id（只读引用）', + scene_id BIGINT NOT NULL DEFAULT 0 COMMENT '父级归属 scene.id', + world_id BIGINT NOT NULL DEFAULT 0 COMMENT '父级归属 world.id', + class_id BIGINT NOT NULL DEFAULT 0 COMMENT '归属班级 pbl_class.id', + team_id BIGINT NOT NULL DEFAULT 0 COMMENT '归属团队 pbl_team.id', + visibility VARCHAR(16) NOT NULL DEFAULT 'team' COMMENT '可见性 team/class/tenant', + role_code VARCHAR(64) NOT NULL DEFAULT '' COMMENT '角色绑定，空=通用', + source VARCHAR(32) NOT NULL DEFAULT 'manual' COMMENT '来源 manual/compiler/import', + ext_json TEXT NULL COMMENT 'PBL 侧扩展属性 JSON', + created_by VARCHAR(64) NULL COMMENT '创建人', + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', + updated_at DATETIME NULL COMMENT '更新时间', + PRIMARY KEY (id), + UNIQUE KEY uk_pbl_entity_ref (tenant_id, entity_id, team_id, role_code), + KEY idx_pbl_entity_ref_scene (tenant_id, scene_id), + KEY idx_pbl_entity_ref_world (tenant_id, world_id), + KEY idx_pbl_entity_ref_team (tenant_id, team_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='PBL 实体关联表（薄扩展，不改 entity 基表）'; diff --git a/tests/fake_db.py b/tests/fake_db.py new file mode 100644 index 0000000..1bc92f4 --- /dev/null +++ b/tests/fake_db.py @@ -0,0 +1,186 @@ +"""内存版 DbPort 实现 + 受限 SQL 解释器（仅供离线单测，不进生产路径）。 + +支持 assoc.py 生成的全部 SQL 形态： +- SELECT FROM t [WHERE a=%s AND b=%s ...] ORDER BY col LIMIT n OFFSET m +- SELECT COUNT(*) AS cnt FROM t [WHERE ...] +- INSERT INTO t (c1, c2, ...) VALUES (%s, %s, ...) +- UPDATE t SET c=%s, ..., updated_at=NOW() WHERE a=%s AND b=%s +- DELETE FROM t WHERE a=%s AND b=%s + +任何无法解析的语句直接抛错——测试里出现意外 SQL 形态必须暴露，不能静默通过。 +同时充当「不改基表」铁律的守卫：对 world/scene/entity 的任何写操作立即 AssertionError。 +""" + +from __future__ import annotations + +import re +from typing import Any, Dict, List, Optional, Sequence, Tuple + +from pbl_domain_ext.assoc import DbPort + +_SELECT_RE = re.compile( + r"^SELECT\s+(?P.+?)\s+FROM\s+(?P\w+)(?P\s+WHERE\s+.+?)?" + r"(?:\s+ORDER\s+BY\s+(?P\w+))?(?:\s+LIMIT\s+(?P\d+))?" + r"(?:\s+OFFSET\s+(?P\d+))?\s*$", + re.IGNORECASE | re.DOTALL, +) +_INSERT_RE = re.compile( + r"^INSERT\s+INTO\s+(?P
\w+)\s*\((?P[^)]*)\)\s*VALUES\s*\((?P[^)]*)\)\s*$", + re.IGNORECASE | re.DOTALL, +) +_UPDATE_RE = re.compile( + r"^UPDATE\s+(?P
\w+)\s+SET\s+(?P.+?)\s+WHERE\s+(?P.+?)\s*$", + re.IGNORECASE | re.DOTALL, +) +_DELETE_RE = re.compile( + r"^DELETE\s+FROM\s+(?P
\w+)\s+WHERE\s+(?P.+?)\s*$", + re.IGNORECASE | re.DOTALL, +) +_COND_RE = re.compile(r"(\w+)\s*=\s*%s") +_BASE_TABLES = ("world", "scene", "entity") + + +def _split_top_level(text: str) -> List[str]: + """按逗号切分（本模块 SQL 无嵌套函数参数含逗号，直接切即可）。""" + return [part.strip() for part in text.split(",") if part.strip()] + + +class FakeDb(DbPort): + """表名 -> 行列表；行是 dict，含自增 id。""" + + def __init__(self, tables: Optional[Sequence[str]] = None) -> None: + self.tables: Dict[str, List[Dict[str, Any]]] = {name: [] for name in (tables or [])} + self._seq: Dict[str, int] = {name: 0 for name in (tables or [])} + self.executed: List[Tuple[str, Tuple[Any, ...]]] = [] + self.base_table_writes: List[str] = [] + + # ------------------------------------------------------------ 内部工具 + def _ensure(self, table: str) -> List[Dict[str, Any]]: + if table not in self.tables: + self.tables[table] = [] + self._seq[table] = 0 + return self.tables[table] + + def _guard_base(self, table: str, kind: str) -> None: + if table in _BASE_TABLES: + self.base_table_writes.append("%s:%s" % (kind, table)) + raise AssertionError("铁律违规：薄扩展不得写基表 %s（%s）" % (table, kind)) + + @staticmethod + def _row_matches(row: Dict[str, Any], cols: Sequence[str], params: Sequence[Any]) -> bool: + for col, expected in zip(cols, params): + if str(row.get(col)) != str(expected): + return False + return True + + def _where_rows(self, table: str, where_sql: str, params: Sequence[Any]) -> List[Dict[str, Any]]: + cols = _COND_RE.findall(where_sql or "") + return [row for row in self._ensure(table) if self._row_matches(row, cols, params)] + + # ------------------------------------------------------------ DbPort + def select(self, sql: str, params: Sequence[Any] = ()) -> List[Dict[str, Any]]: + self.executed.append((sql, tuple(params))) + text = " ".join(sql.split()) + m = _SELECT_RE.match(text) + if not m: + raise AssertionError("FakeDb 无法解析 SELECT: %s" % sql) + table = m.group("table") + cols = m.group("cols").strip() + where = m.group("where") or "" + rows = [dict(r) for r in self._where_rows(table, where, tuple(params))] + + if cols.upper().startswith("COUNT("): + return [{"cnt": len(rows)}] + + order = m.group("order") + if order: + rows.sort(key=lambda r: (r.get(order) is None, r.get(order))) + offset = int(m.group("offset") or 0) + limit = m.group("limit") + rows = rows[offset:offset + int(limit)] if limit is not None else rows[offset:] + + if cols != "*": + wanted = _split_top_level(cols) + rows = [{c: r.get(c) for c in wanted} for r in rows] + return rows + + def insert(self, sql: str, params: Sequence[Any] = ()) -> int: + self.executed.append((sql, tuple(params))) + text = " ".join(sql.split()) + m = _INSERT_RE.match(text) + if not m: + raise AssertionError("FakeDb 无法解析 INSERT: %s" % sql) + table = m.group("table") + self._guard_base(table, "INSERT") + cols = _split_top_level(m.group("cols")) + placeholders = _split_top_level(m.group("vals")) + if len(cols) != len(placeholders) or len(cols) != len(params): + raise AssertionError( + "INSERT 列数与参数数不一致: cols=%d ph=%d params=%d" + % (len(cols), len(placeholders), len(params)) + ) + rows = self._ensure(table) + self._seq[table] = self._seq.get(table, 0) + 1 + row: Dict[str, Any] = {"id": self._seq[table]} + for col, value in zip(cols, params): + row[col] = value + uniq_cols = [c for c in ("tenant_id", "world_id", "scene_id", "entity_id", "team_id", "role_code") + if c in row] + for exist in rows: + if self._row_matches(exist, uniq_cols, [row[c] for c in uniq_cols]): + raise AssertionError("唯一键冲突（幂等 upsert 失效）: %s %s" % (table, row)) + rows.append(row) + return int(row["id"]) + + def execute(self, sql: str, params: Sequence[Any] = ()) -> int: + self.executed.append((sql, tuple(params))) + text = " ".join(sql.split()) + + m = _UPDATE_RE.match(text) + if m: + table = m.group("table") + self._guard_base(table, "UPDATE") + assigns = _split_top_level(m.group("assigns")) + set_cols: List[str] = [] + set_values: List[Any] = [] + cursor = 0 + for assign in assigns: + col, _, expr = assign.partition("=") + col = col.strip() + expr = expr.strip() + if expr == "%s": + set_cols.append(col) + set_values.append(params[cursor]) + cursor += 1 + elif expr.upper() == "NOW()": + set_cols.append(col) + set_values.append("NOW()") + else: + raise AssertionError("FakeDb 不支持的 SET 表达式: %s" % assign) + where_cols = _COND_RE.findall(m.group("where")) + where_values = list(params[cursor:cursor + len(where_cols)]) + affected = 0 + for row in self._ensure(table): + if not self._row_matches(row, where_cols, where_values): + continue + for col, value in zip(set_cols, set_values): + row[col] = value + affected += 1 + return affected + + m = _DELETE_RE.match(text) + if m: + table = m.group("table") + self._guard_base(table, "DELETE") + where_cols = _COND_RE.findall(m.group("where")) + keep: List[Dict[str, Any]] = [] + affected = 0 + for row in self._ensure(table): + if self._row_matches(row, where_cols, tuple(params)): + affected += 1 + else: + keep.append(row) + self.tables[table] = keep + return affected + + raise AssertionError("FakeDb 无法解析语句: %s" % sql) diff --git a/tests/test_assoc.py b/tests/test_assoc.py new file mode 100644 index 0000000..73d84b9 --- /dev/null +++ b/tests/test_assoc.py @@ -0,0 +1,514 @@ +"""M8 基础域薄扩展离线单测（不依赖 Sage 运行时 / 不连库）。 + +运行：cd modules/pbl_domain_ext && python3 -m pytest tests/ -q + 或 python3 tests/test_assoc.py（无 pytest 时直接跑） + +覆盖：租户强制打头 / 幂等 upsert / 可见性推导与校验 / 过滤白名单 / + 作用域回溯解析 / Game Definition 物化 / 不改基表铁律 / 契约层错误码。 +""" + +from __future__ import annotations + +import json +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from fake_db import FakeDb # noqa: E402 (tests/ 目录内导入) + +from pbl_domain_ext import api, assoc # noqa: E402 +from pbl_domain_ext.assoc import DomainExtError # noqa: E402 + +TENANT = "T-DEMO-001" + + +def new_db() -> FakeDb: + return FakeDb([assoc.WORLD_REF, assoc.SCENE_REF, assoc.ENTITY_REF]) + + +class TestTenantGuard(unittest.TestCase): + """租户强制打头：缺失/空/超长一律 fail-closed。""" + + def test_missing_tenant_raises(self): + db = new_db() + for bad in (None, "", " "): + with self.assertRaises(DomainExtError) as ctx: + assoc.upsert_world_ref(db, bad, 101) + self.assertEqual(ctx.exception.code, assoc.ERR_TENANT_REQUIRED) + + def test_overlong_tenant_raises(self): + db = new_db() + with self.assertRaises(DomainExtError) as ctx: + assoc.upsert_world_ref(db, "T" * 65, 101) + self.assertEqual(ctx.exception.code, assoc.ERR_TENANT_REQUIRED) + + def test_missing_key_raises(self): + db = new_db() + with self.assertRaises(DomainExtError) as ctx: + assoc.upsert_world_ref(db, TENANT, None) + self.assertEqual(ctx.exception.code, assoc.ERR_PARAM_REQUIRED) + + def test_non_integer_key_raises(self): + db = new_db() + with self.assertRaises(DomainExtError): + assoc.upsert_scene_ref(db, TENANT, "abc") + + def test_negative_key_raises(self): + db = new_db() + with self.assertRaises(DomainExtError): + assoc.upsert_entity_ref(db, TENANT, -3) + + def test_list_without_tenant_raises(self): + db = new_db() + with self.assertRaises(DomainExtError) as ctx: + assoc.list_world_refs(db, None) + self.assertEqual(ctx.exception.code, assoc.ERR_TENANT_REQUIRED) + + +class TestUpsertIdempotent(unittest.TestCase): + def test_create_then_update_same_row(self): + db = new_db() + first = assoc.upsert_world_ref(db, TENANT, 101, class_id=7, team_id=9, role_code="teacher") + self.assertEqual(first["action"], "created") + self.assertGreater(first["id"], 0) + second = assoc.upsert_world_ref(db, TENANT, 101, class_id=8, team_id=9, role_code="teacher") + self.assertEqual(second["action"], "updated") + self.assertEqual(second["id"], first["id"]) + self.assertEqual(len(db.tables[assoc.WORLD_REF]), 1) + self.assertEqual(db.tables[assoc.WORLD_REF][0]["class_id"], 8) + + def test_different_role_is_different_row(self): + db = new_db() + assoc.upsert_world_ref(db, TENANT, 101, team_id=9, role_code="teacher") + assoc.upsert_world_ref(db, TENANT, 101, team_id=9, role_code="student") + self.assertEqual(len(db.tables[assoc.WORLD_REF]), 2) + + def test_different_tenant_isolated(self): + db = new_db() + assoc.upsert_world_ref(db, TENANT, 101, team_id=9) + assoc.upsert_world_ref(db, "T-OTHER", 101, team_id=9) + self.assertEqual(len(db.tables[assoc.WORLD_REF]), 2) + page = assoc.list_world_refs(db, TENANT) + self.assertEqual(page["total"], 1) + self.assertEqual(page["items"][0]["tenant_id"], TENANT) + + def test_scene_and_entity_parents_persisted(self): + db = new_db() + assoc.upsert_scene_ref(db, TENANT, 202, world_id=101, team_id=9) + assoc.upsert_entity_ref(db, TENANT, 303, scene_id=202, world_id=101, team_id=9) + scene_row = db.tables[assoc.SCENE_REF][0] + entity_row = db.tables[assoc.ENTITY_REF][0] + self.assertEqual(scene_row["world_id"], 101) + self.assertEqual(entity_row["scene_id"], 202) + self.assertEqual(entity_row["world_id"], 101) + + def test_unknown_parent_column_rejected(self): + db = new_db() + with self.assertRaises(DomainExtError): + assoc.upsert_ref(db, assoc.SCENE_REF, TENANT, 202, {"entity_id": 1}) + + def test_ext_json_serialized(self): + db = new_db() + assoc.upsert_world_ref(db, TENANT, 101, team_id=9, ext={"stage": 2, "tags": ["a", "b"]}) + stored = db.tables[assoc.WORLD_REF][0]["ext_json"] + self.assertEqual(json.loads(stored)["stage"], 2) + + def test_ext_string_json_passthrough(self): + db = new_db() + assoc.upsert_world_ref(db, TENANT, 102, team_id=9, ext='{"k": 1}') + self.assertEqual(json.loads(db.tables[assoc.WORLD_REF][0]["ext_json"])["k"], 1) + + def test_ext_bad_json_rejected(self): + db = new_db() + with self.assertRaises(DomainExtError): + assoc.upsert_world_ref(db, TENANT, 101, team_id=9, ext="{not json") + + def test_bad_source_rejected(self): + db = new_db() + with self.assertRaises(DomainExtError): + assoc.upsert_world_ref(db, TENANT, 101, team_id=9, source="drop table;") + + +class TestVisibility(unittest.TestCase): + def test_auto_visibility_by_scope(self): + db = new_db() + self.assertEqual(assoc.upsert_world_ref(db, TENANT, 1, team_id=5)["visibility"], "team") + self.assertEqual(assoc.upsert_world_ref(db, TENANT, 2, class_id=5)["visibility"], "class") + self.assertEqual(assoc.upsert_world_ref(db, TENANT, 3)["visibility"], "tenant") + + def test_explicit_visibility_must_match_scope(self): + db = new_db() + with self.assertRaises(DomainExtError) as ctx: + assoc.upsert_world_ref(db, TENANT, 4, visibility="team") + self.assertEqual(ctx.exception.code, assoc.ERR_PARAM_REQUIRED) + with self.assertRaises(DomainExtError): + assoc.upsert_world_ref(db, TENANT, 5, visibility="class") + + def test_bad_visibility_value(self): + db = new_db() + with self.assertRaises(DomainExtError) as ctx: + assoc.upsert_world_ref(db, TENANT, 6, visibility="public") + self.assertEqual(ctx.exception.code, assoc.ERR_BAD_VISIBILITY) + + +class TestListAndFilter(unittest.TestCase): + def setUp(self): + self.db = new_db() + for i in range(1, 8): + assoc.upsert_world_ref(self.db, TENANT, 100 + i, class_id=7, team_id=9) + assoc.upsert_world_ref(self.db, "T-OTHER", 999, team_id=1) + + def test_pagination(self): + page = assoc.list_world_refs(self.db, TENANT, limit=3, offset=0) + self.assertEqual(page["total"], 7) + self.assertEqual(len(page["items"]), 3) + page2 = assoc.list_world_refs(self.db, TENANT, limit=3, offset=6) + self.assertEqual(len(page2["items"]), 1) + + def test_limit_clamped(self): + page = assoc.list_world_refs(self.db, TENANT, limit=99999) + self.assertEqual(page["limit"], assoc.MAX_LIMIT) + + def test_filter_whitelist(self): + page = assoc.list_world_refs(self.db, TENANT, {"class_id": 7}) + self.assertEqual(page["total"], 7) + with self.assertRaises(DomainExtError): + assoc.list_world_refs(self.db, TENANT, {"password": "x"}) + + def test_empty_filter_value_ignored(self): + page = assoc.list_world_refs(self.db, TENANT, {"role_code": "", "team_id": 9}) + self.assertEqual(page["total"], 7) + + def test_scene_filter_by_world(self): + assoc.upsert_scene_ref(self.db, TENANT, 201, world_id=101, team_id=9) + assoc.upsert_scene_ref(self.db, TENANT, 202, world_id=102, team_id=9) + page = assoc.list_scene_refs(self.db, TENANT, {"world_id": 101}) + self.assertEqual(page["total"], 1) + self.assertEqual(page["items"][0]["scene_id"], 201) + + +class TestGetDelete(unittest.TestCase): + def test_get_strict_and_lenient(self): + db = new_db() + assoc.upsert_entity_ref(db, TENANT, 303, scene_id=202, world_id=101, team_id=9) + row = assoc.get_ref(db, assoc.ENTITY_REF, TENANT, 303, team_id=9) + self.assertEqual(row["scene_id"], 202) + with self.assertRaises(DomainExtError) as ctx: + assoc.get_ref(db, assoc.ENTITY_REF, TENANT, 404, team_id=9) + self.assertEqual(ctx.exception.code, assoc.ERR_NOT_FOUND) + self.assertIsNone(assoc.get_ref(db, assoc.ENTITY_REF, TENANT, 404, team_id=9, strict=False)) + + def test_delete_only_own_tenant(self): + db = new_db() + res = assoc.upsert_world_ref(db, TENANT, 101, team_id=9) + assoc.upsert_world_ref(db, "T-OTHER", 101, team_id=9) + affected = assoc.delete_ref(db, assoc.WORLD_REF, TENANT, res["id"]) + self.assertEqual(affected, 1) + self.assertEqual(len(db.tables[assoc.WORLD_REF]), 1) + self.assertEqual(db.tables[assoc.WORLD_REF][0]["tenant_id"], "T-OTHER") + + +class TestScopeResolve(unittest.TestCase): + def setUp(self): + self.db = new_db() + assoc.upsert_world_ref(self.db, TENANT, 101, class_id=7, team_id=9) + assoc.upsert_scene_ref(self.db, TENANT, 202, world_id=101, class_id=7, team_id=9) + assoc.upsert_entity_ref(self.db, TENANT, 303, scene_id=202, world_id=101, team_id=9) + + def test_resolve_from_entity_backfills_parents(self): + scope = assoc.resolve_scope(self.db, TENANT, entity_id=303) + self.assertEqual(scope["resolved_from"], "entity") + self.assertEqual(scope["scene_id"], 202) + self.assertEqual(scope["world_id"], 101) + self.assertEqual(scope["team_id"], 9) + self.assertEqual(scope["class_id"], 7) + self.assertEqual(scope["visibility"], "team") + self.assertIsNotNone(scope["world"]) + + def test_resolve_from_scene_only(self): + scope = assoc.resolve_scope(self.db, TENANT, scene_id=202) + self.assertEqual(scope["resolved_from"], "scene") + self.assertEqual(scope["world_id"], 101) + + def test_resolve_from_world_only(self): + scope = assoc.resolve_scope(self.db, TENANT, world_id=101) + self.assertEqual(scope["resolved_from"], "world") + self.assertEqual(scope["class_id"], 7) + + def test_resolve_requires_one_id(self): + with self.assertRaises(DomainExtError) as ctx: + assoc.resolve_scope(self.db, TENANT) + self.assertEqual(ctx.exception.code, assoc.ERR_PARAM_REQUIRED) + + def test_resolve_not_found_fail_closed(self): + with self.assertRaises(DomainExtError) as ctx: + assoc.resolve_scope(self.db, TENANT, entity_id=999) + self.assertEqual(ctx.exception.code, assoc.ERR_NOT_FOUND) + + def test_resolve_cross_tenant_not_leaked(self): + with self.assertRaises(DomainExtError): + assoc.resolve_scope(self.db, "T-OTHER", entity_id=303) + + def test_resolve_entity_without_scene_parent(self): + assoc.upsert_entity_ref(self.db, TENANT, 404, world_id=101, team_id=9) + scope = assoc.resolve_scope(self.db, TENANT, entity_id=404) + self.assertEqual(scope["resolved_from"], "entity") + self.assertEqual(scope["world_id"], 101) + self.assertIsNone(scope["scene"]) + + +class TestMaterialize(unittest.TestCase): + DEFINITION = { + "game": { + "world_id": 101, + "scenes": [{"scene_id": 202, "entities": [{"entity_id": 303}, {"entityId": 304}]}], + }, + "meta": {"worldId": 101}, + } + + def test_collect_ids_dedup(self): + ids = assoc.collect_ids(self.DEFINITION, assoc.WORLD_REF) + self.assertEqual(ids, [101]) + self.assertEqual(assoc.collect_ids(self.DEFINITION, assoc.SCENE_REF), [202]) + self.assertEqual(sorted(assoc.collect_ids(self.DEFINITION, assoc.ENTITY_REF)), [303, 304]) + + def test_collect_ids_from_json_string(self): + ids = assoc.collect_ids(json.dumps(self.DEFINITION), assoc.WORLD_REF) + self.assertEqual(ids, [101]) + + def test_bind_creates_all_refs(self): + db = new_db() + summary = assoc.bind_game_definition(db, TENANT, self.DEFINITION, class_id=7, team_id=9) + self.assertEqual(summary["world_refs"], 1) + self.assertEqual(summary["scene_refs"], 1) + self.assertEqual(summary["entity_refs"], 2) + self.assertEqual(summary["created"], 4) + self.assertEqual(summary["source"], "compiler") + self.assertEqual(len(db.tables[assoc.ENTITY_REF]), 2) + + def test_bind_is_idempotent(self): + db = new_db() + assoc.bind_game_definition(db, TENANT, self.DEFINITION, team_id=9) + second = assoc.bind_game_definition(db, TENANT, json.dumps(self.DEFINITION), team_id=9) + self.assertEqual(second["updated"], 4) + self.assertEqual(second["created"], 0) + self.assertEqual(len(db.tables[assoc.WORLD_REF]), 1) + + def test_bind_rejects_empty_and_bad_json(self): + db = new_db() + with self.assertRaises(DomainExtError): + assoc.bind_game_definition(db, TENANT, " ") + with self.assertRaises(DomainExtError): + assoc.bind_game_definition(db, TENANT, "{bad json") + with self.assertRaises(DomainExtError): + assoc.bind_game_definition(db, TENANT, [1, 2, 3]) + + def test_bind_without_ids_reports_skip(self): + db = new_db() + summary = assoc.bind_game_definition(db, TENANT, {"game": {"name": "x"}}) + self.assertEqual(summary["world_refs"], 0) + self.assertTrue(summary["skipped"]) + + def test_multiple_scenes_leave_parent_zero(self): + db = new_db() + definition = {"scenes": [{"scene_id": 1}, {"scene_id": 2}], "entities": [{"entity_id": 5}]} + summary = assoc.bind_game_definition(db, TENANT, definition, team_id=9) + self.assertEqual(summary["scene_refs"], 2) + self.assertEqual(db.tables[assoc.ENTITY_REF][0]["scene_id"], 0) + self.assertTrue(any("多个 scene_id" in s for s in summary["skipped"])) + + def test_multiple_worlds_take_first(self): + db = new_db() + definition = {"worlds": [{"world_id": 11}, {"world_id": 22}], "scenes": [{"scene_id": 5}]} + summary = assoc.bind_game_definition(db, TENANT, definition, team_id=9) + self.assertEqual(summary["world_refs"], 2) + self.assertEqual(db.tables[assoc.SCENE_REF][0]["world_id"], 11) + self.assertTrue(any("多个 world_id" in s for s in summary["skipped"])) + + +class TestBaseTableUntouched(unittest.TestCase): + """铁律：任何路径都不得写 world/scene/entity 基表。""" + + def test_no_base_table_write(self): + db = new_db() + assoc.upsert_world_ref(db, TENANT, 101, team_id=9) + assoc.upsert_scene_ref(db, TENANT, 202, world_id=101, team_id=9) + assoc.upsert_entity_ref(db, TENANT, 303, scene_id=202, world_id=101, team_id=9) + assoc.bind_game_definition(db, TENANT, TestMaterialize.DEFINITION, team_id=9) + assoc.resolve_scope(db, TENANT, entity_id=303) + self.assertEqual(db.base_table_writes, []) + for sql, _params in db.executed: + lowered = " ".join(sql.lower().split()) + if lowered.startswith(("insert", "update", "delete")): + for base in ("world", "scene", "entity"): + self.assertFalse(lowered.startswith("insert into %s " % base), sql) + self.assertFalse(lowered.startswith("update %s " % base), sql) + self.assertFalse(lowered.startswith("delete from %s " % base), sql) + + def test_all_writes_target_ref_tables_only(self): + db = new_db() + assoc.bind_game_definition(db, TENANT, TestMaterialize.DEFINITION, team_id=9) + write_tables = set() + for sql, _params in db.executed: + lowered = " ".join(sql.lower().split()) + if lowered.startswith("insert into"): + write_tables.add(lowered.split()[2]) + elif lowered.startswith("update"): + write_tables.add(lowered.split()[1]) + elif lowered.startswith("delete from"): + write_tables.add(lowered.split()[2]) + self.assertTrue(write_tables) + self.assertTrue(write_tables.issubset(set(assoc.REF_TABLES)), write_tables) + + def test_table_name_injection_guard(self): + db = new_db() + with self.assertRaises(DomainExtError): + assoc.list_refs(db, "pbl_world_ref; DROP TABLE world", TENANT) + with self.assertRaises(DomainExtError): + assoc.upsert_ref(db, "world", TENANT, 1) + + def test_filter_column_injection_guard(self): + db = new_db() + with self.assertRaises(DomainExtError): + assoc.list_refs(db, assoc.WORLD_REF, TENANT, {"1=1 OR tenant_id": "x"}) + + def test_values_always_parameterized(self): + db = new_db() + evil = "x'; DROP TABLE pbl_world_ref; --" + assoc.upsert_world_ref(db, TENANT, 101, team_id=9, role_code=evil) + for sql, params in db.executed: + self.assertNotIn("DROP TABLE", sql) + self.assertEqual(db.tables[assoc.WORLD_REF][0]["role_code"], evil) + self.assertEqual(len(db.tables[assoc.WORLD_REF]), 1) + + +class TestApiContract(unittest.TestCase): + def setUp(self): + self.db = new_db() + + def call(self, name, **params): + params["_db"] = self.db + return api.call_api(name, params) + + def test_upsert_and_list_ok(self): + res = self.call("pbl_world_ref_upsert", tenant_id=TENANT, world_id=101, class_id=7, team_id=9) + self.assertTrue(res["ok"], res) + self.assertEqual(res["data"]["action"], "created") + page = self.call("pbl_world_ref_list", tenant_id=TENANT, team_id=9) + self.assertTrue(page["ok"]) + self.assertEqual(page["data"]["total"], 1) + + def test_scene_and_entity_upsert_ok(self): + scene = self.call("pbl_scene_ref_upsert", tenant_id=TENANT, scene_id=202, world_id=101, team_id=9) + entity = self.call("pbl_entity_ref_upsert", tenant_id=TENANT, entity_id=303, scene_id=202, + world_id=101, team_id=9, ext={"hp": 100}) + self.assertTrue(scene["ok"], scene) + self.assertTrue(entity["ok"], entity) + self.assertEqual(entity["data"]["visibility"], "team") + + def test_missing_tenant_returns_error_code(self): + res = self.call("pbl_world_ref_list", world_id=101) + self.assertFalse(res["ok"]) + self.assertEqual(res["code"], assoc.ERR_TENANT_REQUIRED) + + def test_unknown_api(self): + res = api.call_api("pbl_not_exist", {"tenant_id": TENANT}) + self.assertFalse(res["ok"]) + self.assertEqual(res["code"], assoc.ERR_PARAM_REQUIRED) + + def test_unknown_table_alias(self): + res = self.call("pbl_domain_ref_get", table="payroll", tenant_id=TENANT, id_value=1) + self.assertFalse(res["ok"]) + + def test_ref_get_and_delete(self): + created = self.call("pbl_scene_ref_upsert", tenant_id=TENANT, scene_id=202, world_id=101, team_id=9) + rid = created["data"]["id"] + got = self.call("pbl_domain_ref_get", table="scene", tenant_id=TENANT, scene_id=202, team_id=9) + self.assertTrue(got["ok"]) + self.assertTrue(got["found"]) + deleted = self.call("pbl_domain_ref_delete", table="pbl_scene_ref", tenant_id=TENANT, id=rid) + self.assertEqual(deleted["data"]["deleted"], 1) + missing = self.call("pbl_domain_ref_get", table="scene", tenant_id=TENANT, scene_id=202, + team_id=9, strict=False) + self.assertTrue(missing["ok"]) + self.assertFalse(missing["found"]) + + def test_ref_get_strict_not_found(self): + res = self.call("pbl_domain_ref_get", table="world", tenant_id=TENANT, world_id=777) + self.assertFalse(res["ok"]) + self.assertEqual(res["code"], assoc.ERR_NOT_FOUND) + + def test_scope_resolve_and_stats(self): + self.call("pbl_world_ref_upsert", tenant_id=TENANT, world_id=101, class_id=7, team_id=9) + self.call("pbl_scene_ref_upsert", tenant_id=TENANT, scene_id=202, world_id=101, team_id=9) + self.call("pbl_entity_ref_upsert", tenant_id=TENANT, entity_id=303, scene_id=202, + world_id=101, team_id=9) + scope = self.call("pbl_domain_scope_resolve", tenant_id=TENANT, entity_id=303) + self.assertTrue(scope["ok"]) + self.assertEqual(scope["data"]["resolved_from"], "entity") + self.assertEqual(scope["data"]["world_id"], 101) + stats = self.call("pbl_domain_stats", tenant_id=TENANT) + self.assertEqual(stats["data"]["total"], 3) + self.assertEqual(stats["data"][assoc.ENTITY_REF], 1) + + def test_materialize_contract(self): + res = self.call("pbl_domain_materialize_game_definition", tenant_id=TENANT, + game_definition=TestMaterialize.DEFINITION, class_id=7, team_id=9) + self.assertTrue(res["ok"], res) + self.assertEqual(res["data"]["world_refs"], 1) + self.assertEqual(res["data"]["entity_refs"], 2) + bad = self.call("pbl_domain_materialize_game_definition", tenant_id=TENANT, + game_definition="{oops") + self.assertFalse(bad["ok"]) + self.assertEqual(bad["code"], assoc.ERR_PARAM_REQUIRED) + + def test_materialize_missing_definition(self): + res = self.call("pbl_domain_materialize_game_definition", tenant_id=TENANT) + self.assertFalse(res["ok"]) + self.assertEqual(res["code"], assoc.ERR_PARAM_REQUIRED) + + +class TestInitRegistration(unittest.TestCase): + def test_module_metadata(self): + from pbl_domain_ext import init as init_mod + self.assertEqual(init_mod.MODULE_NAME, "pbl_domain_ext") + for table in (assoc.WORLD_REF, assoc.SCENE_REF, assoc.ENTITY_REF): + self.assertIn(table, init_mod.OWN_TABLES) + self.assertEqual(init_mod.READONLY_BASE_TABLES, ["world", "scene", "entity"]) + self.assertIn("pbl_domain_scope_resolve", init_mod.API_NAMES) + + def test_models_and_cruds_loadable(self): + from pbl_domain_ext import init as init_mod + models = init_mod.load_models() + names = {m.get("tblname") for m in models} + for table in (assoc.WORLD_REF, assoc.SCENE_REF, assoc.ENTITY_REF): + self.assertIn(table, names) + self.assertGreaterEqual(len(init_mod.load_cruds()), 3) + + def test_load_entry_returns_summary_without_runtime(self): + from pbl_domain_ext import init as init_mod + summary = init_mod.load_pbl_domain_ext() + self.assertEqual(summary["module"], "pbl_domain_ext") + self.assertIsNone(summary["dbname"]) # 无 ServerEnv 时不抛，只记录错误 + self.assertIn("dbname_error", summary) + + def test_load_entry_registers_apis_on_env(self): + from pbl_domain_ext import init as init_mod + + class FakeEnv(object): + def __init__(self): + self.apis = {} + + def register_api(self, name, handler): + self.apis[name] = handler + + env = FakeEnv() + init_mod.load_pbl_domain_ext(env) + self.assertIn("pbl_world_ref_list", env.apis) + + +if __name__ == "__main__": + unittest.main(verbosity=2)