feat(域模型v2+工作台): 聚合API game_create/game_workbench/scene_add/entity_add + 4个dspy端点

scense_game/aggregation.py

@@ -0,0 +1,140 @@
+# -*- coding: utf-8 -*-
+"""域模型 v2 聚合 API：game 为唯一聚合根。
+
+- game_create：建游戏 → 自动建 1:1 空世界（world.game_id 回填）
+- game_workbench：game 上下文全量查询（world/scenes/entities/scripts）
+- scene_add / entity_add：强归属创建（scene 必属 game 的 world；entity 必属该 world 的 scene 或世界级）
+"""
+import json
+
+from appPublic.uniqueID import getID
+from appPublic.log import info, error, exception
+from sqlor.dbpools import DBPools
+
+
+def _get_dbname():
+    from ahserver.serverenv import ServerEnv
+    return ServerEnv().get_module_dbname('scense_game')
+
+
+def _now():
+    import time
+    return time.strftime('%Y-%m-%d %H:%M:%S')
+
+
+async def game_create(ns):
+    """建游戏并自动创建 1:1 空世界。ns: {name, code?, description?, icon?}"""
+    name = (ns or {}).get('name') or ''
+    if not str(name).strip():
+        return {'code': 'PARAM_REQUIRED', 'message': '游戏名称必填', 'field': 'name'}
+    gid = getID()
+    wid = getID()
+    now = _now()
+    dbname = _get_dbname()
+    try:
+        async with DBPools().sqlorContext(dbname) as sor:
+            await sor.C('game', {
+                'id': gid, 'name': str(name).strip(),
+                'code': str((ns or {}).get('code') or ('G-' + gid[:8].upper())),
+                'description': str((ns or {}).get('description') or ''),
+                'icon': str((ns or {}).get('icon') or 'fa-gamepad'),
+                'status': 'draft', 'created_at': now, 'updated_at': now})
+            await sor.C('world', {
+                'id': wid, 'game_id': gid,
+                'name': str(name).strip() + ' 世界',
+                'code': 'W-' + gid[:8].upper(),
+                'description': '游戏「%s」的世界（1:1 自动创建）' % str(name).strip(),
+                'world_type': 'game', 'status': 'active',
+                'config_json': '{}', 'created_at': now, 'updated_at': now})
+        return {'success': True, 'data': {'game_id': gid, 'world_id': wid}}
+    except Exception as e:
+        exception('game_create: %s' % e)
+        return {'code': 'DB_ERROR', 'message': '创建游戏失败: %s' % e}
+
+
+async def game_workbench(ns):
+    """game 上下文全量：{game, world, scenes, entities, scripts}"""
+    gid = (ns or {}).get('game_id') or ''
+    if not gid:
+        return {'code': 'PARAM_REQUIRED', 'message': 'game_id 必填', 'field': 'game_id'}
+    dbname = _get_dbname()
+    try:
+        async with DBPools().sqlorContext(dbname) as sor:
+            g = await sor.R('game', {'id': gid})
+            if not g:
+                return {'code': 'NOT_FOUND', 'message': '游戏不存在'}
+            w = await sor.R('world', {'game_id': gid})
+            wid = w[0].id if w else None
+            scenes = await sor.R('scene', {'world_id': wid}) if wid else []
+            ents = await sor.R('entity', {'world_id': wid}) if wid else []
+            scripts = await sor.R('script_engine', {'game_id': gid}) if gid else []
+            return {'success': True, 'data': {
+                'game': dict(g[0]) if g else None,
+                'world': dict(w[0]) if w else None,
+                'scenes': [dict(s) for s in (scenes or [])],
+                'entities': [dict(e) for e in (ents or [])],
+                'scripts': [dict(s) for s in (scripts or [])]}}
+    except Exception as e:
+        exception('game_workbench: %s' % e)
+        return {'code': 'DB_ERROR', 'message': '查询失败: %s' % e}
+
+
+async def scene_add(ns):
+    """在游戏的世界下加场景。ns: {game_id, name, scene_type?}"""
+    gid = (ns or {}).get('game_id') or ''
+    name = (ns or {}).get('name') or ''
+    if not gid or not str(name).strip():
+        return {'code': 'PARAM_REQUIRED', 'message': 'game_id 与 name 必填'}
+    dbname = _get_dbname()
+    try:
+        async with DBPools().sqlorContext(dbname) as sor:
+            w = await sor.R('world', {'game_id': gid})
+            if not w:
+                return {'code': 'NOT_FOUND', 'message': '游戏无世界，先建游戏'}
+            sid = getID()
+            now = _now()
+            await sor.C('scene', {
+                'id': sid, 'world_id': w[0].id, 'name': str(name).strip(),
+                'code': 'S-' + sid[:8].upper(),
+                'description': str((ns or {}).get('description') or ''),
+                'scene_type': str((ns or {}).get('scene_type') or '0'),
+                'status': 'active', 'config_json': '{}',
+                'created_at': now, 'updated_at': now})
+            return {'success': True, 'data': {'scene_id': sid, 'world_id': w[0].id}}
+    except Exception as e:
+        exception('scene_add: %s' % e)
+        return {'code': 'DB_ERROR', 'message': '创建场景失败: %s' % e}
+
+
+async def entity_add(ns):
+    """在游戏世界内加实体。ns: {game_id, scene_id?(空=世界级实体), name, entity_type?, attributes_json?}"""
+    gid = (ns or {}).get('game_id') or ''
+    name = (ns or {}).get('name') or ''
+    if not gid or not str(name).strip():
+        return {'code': 'PARAM_REQUIRED', 'message': 'game_id 与 name 必填'}
+    dbname = _get_dbname()
+    try:
+        async with DBPools().sqlorContext(dbname) as sor:
+            w = await sor.R('world', {'game_id': gid})
+            if not w:
+                return {'code': 'NOT_FOUND', 'message': '游戏无世界'}
+            wid = w[0].id
+            scene_id = (ns or {}).get('scene_id') or ''
+            if scene_id:
+                sc = await sor.R('scene', {'id': scene_id})
+                if not sc or sc[0].world_id != wid:
+                    return {'code': 'NOT_FOUND', 'message': '场景不属于该游戏的世界'}
+            eid = getID()
+            now = _now()
+            await sor.C('entity', {
+                'id': eid, 'world_id': wid, 'scene_id': scene_id or None,
+                'name': str(name).strip(), 'code': 'E-' + eid[:8].upper(),
+                'entity_type': str((ns or {}).get('entity_type') or '0'),
+                'status': 'active',
+                'attributes_json': (ns or {}).get('attributes_json') or '{}',
+                'created_at': now, 'updated_at': now})
+            return {'success': True, 'data': {'entity_id': eid, 'world_id': wid,
+                                              'scene_id': scene_id or None}}
+    except Exception as e:
+        exception('entity_add: %s' % e)
+        return {'code': 'DB_ERROR', 'message': '创建实体失败: %s' % e}

scense_game/init.py

@@ -1,5 +1,8 @@
 # -*- coding: utf-8 -*-
 """scense_game 游戏展示模块初始化：注册全部服务函数到 ServerEnv。"""
+from scense_game.aggregation import (
+    game_create, game_workbench, scene_add, entity_add,
+)
 from scense_game.core import (
     game_list_games, game_list_sessions, game_list_results,
     game_start_session, game_pause_session, game_resume_session,
@@ -14,6 +17,10 @@ def load_scense_game(env=None):
     from ahserver.serverenv import ServerEnv
     if env is None:
         env = ServerEnv()
+    env.game_create = game_create
+    env.game_workbench = game_workbench
+    env.scene_add = scene_add
+    env.entity_add = entity_add
     env.game_list_games = game_list_games
     env.game_list_sessions = game_list_sessions
     env.game_list_results = game_list_results

wwwroot/game/api/entity_add.dspy

@@ -0,0 +1,3 @@
+# -*- coding: utf-8 -*-
+result = await entity_add(dict(params_kw))
+return json.dumps(result, ensure_ascii=False, default=str)

wwwroot/game/api/game_create.dspy

@@ -0,0 +1,3 @@
+# -*- coding: utf-8 -*-
+result = await game_create(dict(params_kw))
+return json.dumps(result, ensure_ascii=False, default=str)

wwwroot/game/api/game_workbench.dspy

@@ -0,0 +1,3 @@
+# -*- coding: utf-8 -*-
+result = await game_workbench(dict(params_kw))
+return json.dumps(result, ensure_ascii=False, default=str)

wwwroot/game/api/scene_add.dspy

@@ -0,0 +1,3 @@
+# -*- coding: utf-8 -*-
+result = await scene_add(dict(params_kw))
+return json.dumps(result, ensure_ascii=False, default=str)