approve: script_engine 模块开发（W-06 脚本引擎）

.gitignore

@@ -3,7 +3,6 @@ __pycache__/
 *.egg-info/
 build/
 dist/
-*.swp
-*.swo
 models/mysql.ddl.sql
-wwwroot/script_engine/
+wwwroot/script/
+.venv/

README.md

@@ -1,24 +1,35 @@
-# script_engine 模块
+# script_engine 脚本引擎模块（W-06）
 
-逻辑编程（脚本/规则引擎）模块：管理逻辑脚本/规则表 `script_engine`，提供脚本执行与语法校验接口。
+脚本引擎：脚本表 CRUD + `execute_script` / `validate_script` 接口。独立模块，无业务依赖，可与 world / scene / entity 链路并行开发。
 
 ## 功能
-- `script_engine` 表 CRUD：脚本名称、编码、所属世界/场景/绑定实体、脚本类型、触发事件、脚本内容、状态。
+- **script 表 CRUD**：新增 / 编辑 / 删除 / 详情 / 分页列表
-- `api/execute_script.dspy`：按 `script_type` 选择执行器执行脚本内容，返回 `{success, result}`。
+- **execute_script**：按脚本 id 取库内脚本（或直传 content），校验通过后执行
-- `api/validate_script.dspy`：按 `script_type` 做语法/格式校验，返回 `{valid, message}`。
+- **validate_script**：校验脚本内容（语法 + 受限语法 + 调用白名单），不落库不执行
+- **编码字典**：`script_type`（0=Python / 1=JavaScript / 2=规则表达式）、`script_status`（0=停用 / 1=启用）经 init/data.json（Format B appcodes）幂等落库
 
 ## 数据表
-- `script_engine`（逻辑脚本表），脚本内容 `content` 使用 `text` 字段。
+- `script`：id / script_name / script_type / content / description / status / created_at / updated_at
 
-## 脚本类型（appcodes script_type）
+## 接口约定
-- `0` = Python 脚本（受限命名空间执行，通过 `result` 变量返回结果）
+- REST 统一前缀：`/api/*`（宿主自动路由 `/script_engine/api/xxx.dspy`）
-- `1` = 规则(JSON)（解析 JSON 作为结果）
+- 错误结构：`{code, message, field, detail}`（code=0 成功，400 参数错误，404 不存在，500 内部错误，501 类型暂不支持）
-- `2` = 表达式（eval 求值）
+- 分页：`{list, total}`
+- 非法输入 100% 拦截，不落库、不执行
 
-## 挂载
+## 集成
-宿主应用入口 `init()` 中调用 `load_script_engine()`。
+1. `pip install .`
+2. 宿主应用 `from script_engine.init import load_script_engine` → `init()` 内调用 `load_script_engine()`
+3. `scripts/load_path.py` 登记 RBAC 路径；宿主 `load_path.py` 兜底
+4. `build.sh` 生成 DDL / CRUD UI / 链接 wwwroot
 
-## 安装
+## 接口清单（wwwroot/api/）
-```bash
+| 路径 | 说明 |
-pip install .
+|---|---|
-```
+| `/script_engine/api/script_create.dspy` | 新增脚本 |
+| `/script_engine/api/script_update.dspy` | 编辑脚本 |
+| `/script_engine/api/script_delete.dspy` | 删除脚本 |
+| `/script_engine/api/script_get.dspy` | 脚本详情 |
+| `/script_engine/api/script_list.dspy` | 分页列表（{list,total}） |
+| `/script_engine/api/script_execute.dspy` | 执行脚本 |
+| `/script_engine/api/script_validate.dspy` | 校验脚本 |

build.sh

@@ -0,0 +1,41 @@
+#!/usr/bin/env bash
+# script_engine 模块构建脚本（由宿主应用 build.sh 集成调用）
+set -e
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+
+if [ -d "$SCRIPT_DIR/models" ]; then
+  if command -v json2ddl >/dev/null 2>&1; then
+    json2ddl mysql "$SCRIPT_DIR/models" > "$SCRIPT_DIR/models/mysql.ddl.sql"
+  else
+    echo "[script_engine] json2ddl 不可用，跳过 DDL 生成"
+  fi
+fi
+
+if [ -d "$SCRIPT_DIR/json" ]; then
+  if command -v xls2ui >/dev/null 2>&1; then
+    (cd "$SCRIPT_DIR" && xls2ui -m models -o wwwroot script_engine json/*.json)
+  else
+    echo "[script_engine] xls2ui 不可用，跳过 CRUD UI 生成"
+  fi
+fi
+
+SAGE_ROOT=""
+for candidate in "$SCRIPT_DIR/../.." "$HOME/repos/sage" "$HOME/sage"; do
+  if [ -d "$candidate/wwwroot" ] && [ -d "$candidate/py3/bin" ]; then
+    SAGE_ROOT="$(cd "$candidate" && pwd)"
+    break
+  fi
+done
+if [ -n "$SAGE_ROOT" ]; then
+  mkdir -p "$SAGE_ROOT/wwwroot/script_engine"
+  ln -sfn "$SCRIPT_DIR/wwwroot/index.ui" "$SAGE_ROOT/wwwroot/script_engine/index.ui"
+  mkdir -p "$SAGE_ROOT/wwwroot/script_engine/api"
+  for f in "$SCRIPT_DIR"/wwwroot/api/*.dspy; do
+    ln -sfn "$f" "$SAGE_ROOT/wwwroot/script_engine/api/$(basename "$f")"
+  done
+  echo "[script_engine] wwwroot 已链接到 $SAGE_ROOT/wwwroot/script_engine"
+else
+  echo "[script_engine] 未找到宿主 wwwroot，跳过链接"
+fi
+
+echo "[script_engine] build.sh 完成"

init/data.json

@@ -1,21 +1,6 @@
 {
     "appcodes": [
-        {
+        {"parentid": "script_type", "parentname": "脚本类型", "items": [{"k": "0", "v": "Python"}, {"k": "1", "v": "JavaScript"}, {"k": "2", "v": "规则表达式"}]},
-            "parentid": "script_type",
+        {"parentid": "script_status", "parentname": "脚本状态", "items": [{"k": "0", "v": "停用"}, {"k": "1", "v": "启用"}]}
-            "parentname": "脚本类型",
-            "items": [
-                {"k": "0", "v": "Python脚本"},
-                {"k": "1", "v": "规则(JSON)"},
-                {"k": "2", "v": "表达式"}
-            ]
-        },
-        {
-            "parentid": "script_status",
-            "parentname": "脚本状态",
-            "items": [
-                {"k": "0", "v": "禁用"},
-                {"k": "1", "v": "启用"}
-            ]
-        }
     ]
-}
+}

json/script.json

@@ -0,0 +1,14 @@
+{
+    "tblname": "script",
+    "title": "脚本管理",
+    "params": {
+        "sortby": ["created_at desc"],
+        "browserfields": {"exclouded": ["content"]},
+        "editexclouded": ["id", "created_at", "updated_at"],
+        "editable": {
+            "new_data_url": "{{entire_url('../api/script_create.dspy')}}",
+            "update_data_url": "{{entire_url('../api/script_update.dspy')}}",
+            "delete_data_url": "{{entire_url('../api/script_delete.dspy')}}"
+        }
+    }
+}

models/script.json

@@ -1,28 +1,28 @@
 {
-  "summary": [{"name": "script", "title": "脚本定义表", "primary": ["id"], "catelog": "entity"}],
+    "summary": [
-  "fields": [
+        {
-    {"name": "id", "title": "主键ID", "type": "str", "length": 32, "nullable": "no"},
+            "name": "script",
-    {"name": "world_id", "title": "所属世界ID", "type": "str", "length": 32, "nullable": "no"},
+            "title": "脚本表",
-    {"name": "name", "title": "脚本名称", "type": "str", "length": 255, "nullable": "no"},
+            "primary": ["id"],
-    {"name": "code", "title": "脚本编码", "type": "str", "length": 64, "nullable": "no"},
+            "catelog": "entity"
-    {"name": "script_type", "title": "脚本类型", "type": "str", "length": 16, "nullable": "no", "default": "lua"},
+        }
-    {"name": "bind_type", "title": "绑定对象类型", "type": "str", "length": 16, "nullable": "no", "default": "world"},
+    ],
-    {"name": "bind_id", "title": "绑定对象ID", "type": "str", "length": 32, "nullable": "yes"},
+    "fields": [
-    {"name": "trigger_event", "title": "触发事件", "type": "str", "length": 64, "nullable": "yes"},
+        {"name": "id", "title": "主键ID", "type": "str", "length": 32, "nullable": "no"},
-    {"name": "content", "title": "脚本内容", "type": "text", "nullable": "yes"},
+        {"name": "script_name", "title": "脚本名称", "type": "str", "length": 100, "nullable": "no"},
-    {"name": "enabled", "title": "是否启用", "type": "str", "length": 1, "nullable": "no", "default": "0"},
+        {"name": "script_type", "title": "脚本类型", "type": "str", "length": 32, "nullable": "no", "default": "0"},
-    {"name": "created_by", "title": "创建人ID", "type": "str", "length": 32, "nullable": "yes"},
+        {"name": "content", "title": "脚本内容", "type": "text", "nullable": "no"},
-    {"name": "created_at", "title": "创建时间", "type": "timestamp", "nullable": "no"},
+        {"name": "description", "title": "描述", "type": "str", "length": 255},
-    {"name": "updated_at", "title": "更新时间", "type": "timestamp", "nullable": "no"}
+        {"name": "status", "title": "状态", "type": "str", "length": 32, "nullable": "no", "default": "1"},
-  ],
+        {"name": "created_at", "title": "创建时间", "type": "timestamp", "nullable": "no"},
-  "indexes": [
+        {"name": "updated_at", "title": "更新时间", "type": "timestamp", "nullable": "no"}
-    {"name": "idx_script_code", "idxtype": "unique", "idxfields": ["world_id", "code"]},
+    ],
-    {"name": "idx_script_world", "idxtype": "index", "idxfields": ["world_id"]},
+    "indexes": [
-    {"name": "idx_script_bind", "idxtype": "index", "idxfields": ["bind_type", "bind_id"]}
+        {"name": "idx_script_name", "idxtype": "index", "idxfields": ["script_name"]},
-  ],
+        {"name": "idx_script_type", "idxtype": "index", "idxfields": ["script_type"]}
-  "codes": [
+    ],
-    {"field": "world_id", "table": "world", "valuefield": "id", "textfield": "name"},
+    "codes": [
-    {"field": "script_type", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='script_type'"},
+        {"field": "script_type", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='script_type'"},
-    {"field": "bind_type", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='bind_type'"}
+        {"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='script_status'"}
-  ]
+    ]
-}
+}

pyproject.toml

@@ -5,6 +5,7 @@ build-backend = "setuptools.build_meta"
 [project]
 name = "script_engine"
 version = "1.0.0"
+description = "脚本引擎模块（W-06）：脚本 CRUD + execute_script/validate_script"
 requires-python = ">=3.8"
 dependencies = ["sqlor", "bricks_for_python"]
 

script_engine/__init__.py

@@ -1,18 +1,6 @@
-"""script_engine 包 —— 逻辑编程（脚本/规则引擎）模块实现。"""
+# -*- coding: utf-8 -*-
-from .init import (
+"""script_engine 模块——脚本引擎（W-06）。"""
-    load_script_engine,
+from script_engine.engine import (validate_script as validate_script, execute_script_content as execute_script_content)
-    execute_script,
+from script_engine.init import (load_script_engine as load_script_engine, create_script as create_script, update_script as update_script, delete_script as delete_script, list_scripts as list_scripts, get_script as get_script, execute_script as execute_script, validate_script_api as validate_script_api)
-    validate_script,
-    create_script,
-    update_script,
-    delete_script,
-)
 
-__all__ = [
+__all__ = ['load_script_engine','create_script','update_script','delete_script','list_scripts','get_script','execute_script','validate_script_api','validate_script','execute_script_content']
-    'load_script_engine',
-    'execute_script',
-    'validate_script',
-    'create_script',
-    'update_script',
-    'delete_script',
-]

script_engine/engine.py

@@ -0,0 +1,107 @@
+# -*- coding: utf-8 -*-
+"""script_engine.engine —— 脚本引擎核心（纯逻辑，无 DB 依赖）。"""
+import ast
+import builtins
+
+SCRIPT_TYPE_PYTHON = '0'
+SCRIPT_TYPE_JS = '1'
+SCRIPT_TYPE_RULE = '2'
+VALID_SCRIPT_TYPES = (SCRIPT_TYPE_PYTHON, SCRIPT_TYPE_JS, SCRIPT_TYPE_RULE)
+
+_ALLOWED_BUILTINS = frozenset(['abs','all','any','bool','dict','divmod','enumerate','filter','float','format','frozenset','int','isinstance','issubclass','iter','len','list','map','max','min','next','object','pow','range','repr','reversed','round','set','slice','sorted','str','sum','tuple','zip'])
+
+_FORBIDDEN_NODES = (ast.Import, ast.ImportFrom, ast.Global, ast.Nonlocal, ast.Lambda, ast.ClassDef, ast.Yield, ast.YieldFrom, ast.AsyncFunctionDef, ast.Await)
+
+
+def _pair_check(text, open_ch, close_ch):
+    depth = 0
+    for ch in text:
+        if ch == open_ch:
+            depth += 1
+        elif ch == close_ch:
+            depth -= 1
+            if depth < 0:
+                return False
+    return depth == 0
+
+
+def validate_python(content):
+    if not content or not content.strip():
+        return False, '脚本内容不能为空'
+    try:
+        tree = ast.parse(content, mode='exec')
+    except SyntaxError as e:
+        return False, '语法错误: %s (第 %s 行)' % (e.msg or 'unknown', e.lineno or 0)
+    for node in ast.walk(tree):
+        if isinstance(node, _FORBIDDEN_NODES):
+            return False, '禁止使用的语法: %s (第 %s 行)' % (type(node).__name__, getattr(node, 'lineno', 0))
+        if isinstance(node, ast.Call):
+            func = node.func
+            if isinstance(func, ast.Name):
+                if func.id not in _ALLOWED_BUILTINS:
+                    return False, '禁止调用的函数: %s (第 %s 行)' % (func.id, getattr(node, 'lineno', 0))
+            elif isinstance(func, ast.Attribute):
+                return False, '禁止调用对象方法: %s (第 %s 行)' % (func.attr, getattr(node, 'lineno', 0))
+    return True, ''
+
+
+def validate_js(content):
+    if not content or not content.strip():
+        return False, '脚本内容不能为空'
+    if not _pair_check(content, '{', '}'):
+        return False, '大括号不配对'
+    if not _pair_check(content, '(', ')'):
+        return False, '圆括号不配对'
+    if not _pair_check(content, '[', ']'):
+        return False, '方括号不配对'
+    return True, ''
+
+
+def validate_rule(content):
+    if not content or not content.strip():
+        return False, '脚本内容不能为空'
+    text = content.strip()
+    if text.startswith('{') or text.startswith('['):
+        try:
+            import json as _json
+            _json.loads(text)
+            return True, ''
+        except Exception as e:
+            return False, '规则 JSON 解析失败: %s' % str(e)
+    if '->' not in text:
+        return False, '规则格式应为 JSON 或 "条件 -> 动作"'
+    return True, ''
+
+
+def validate_script(content, script_type):
+    stype = str(script_type or SCRIPT_TYPE_PYTHON)
+    if stype not in VALID_SCRIPT_TYPES:
+        return False, '不支持的脚本类型: %s' % stype
+    if stype == SCRIPT_TYPE_PYTHON:
+        return validate_python(content)
+    if stype == SCRIPT_TYPE_JS:
+        return validate_js(content)
+    return validate_rule(content)
+
+
+def execute_python(content, params):
+    safe_builtins = {}
+    for name in _ALLOWED_BUILTINS:
+        if hasattr(builtins, name):
+            safe_builtins[name] = getattr(builtins, name)
+    namespace = {'__name__': '__script_engine__', '__builtins__': safe_builtins, 'params': params}
+    code = compile(content, '<script_engine>', 'exec')
+    exec(code, namespace)
+    main_fn = namespace.get('main')
+    if callable(main_fn):
+        return main_fn(params)
+    return {'executed': True, 'scope': {k: v for k, v in namespace.items() if not k.startswith('__') and not callable(v)}}
+
+
+def execute_script_content(content, script_type, params):
+    stype = str(script_type or SCRIPT_TYPE_PYTHON)
+    if stype == SCRIPT_TYPE_PYTHON:
+        return execute_python(content, params or {})
+    if stype == SCRIPT_TYPE_JS:
+        raise NotImplementedError('JavaScript 类型暂不支持执行')
+    raise NotImplementedError('规则表达式类型暂不支持执行')

script_engine/init.py

@@ -1,134 +1,235 @@
-"""script_engine 模块初始化。
+# -*- coding: utf-8 -*-
-
+"""script_engine.init —— 模块初始化与业务函数注册。"""
-逻辑编程（脚本/规则引擎）领域模块，通过 load_script_engine() 挂载到宿主应用，
-向 ServerEnv 注册业务函数：
-- execute_script(script_id, context_json)  执行脚本
-- validate_script(content, script_type)    校验脚本语法
-- create_script / update_script / delete_script  CRUD 业务逻辑
-"""
-import json as _json
-
-from appPublic.uniqueID import getID
-from appPublic.timeUtils import curDateString
 from ahserver.serverenv import ServerEnv
 from sqlor.dbpools import DBPools
-
+from appPublic.uniqueID import getID
-_MODULE = 'script_engine'
+from appPublic.timeUtils import curDateString
+from script_engine.engine import validate_script as _validate_content, execute_script_content as _execute_content, VALID_SCRIPT_TYPES
 
 
 def _dbname():
-    """取模块库名（禁止硬编码，由宿主应用 get_module_dbname 决定）。"""
+    return ServerEnv().get_module_dbname('script_engine')
-    return ServerEnv().get_module_dbname(_MODULE)
 
 
-async def _get_script(sor, script_id):
+def _ok(data=None, message='ok'):
-    rows = await sor.sqlExe(
+    return {'code': 0, 'message': message, 'field': '', 'detail': '', 'data': data}
-        'SELECT * FROM script_engine WHERE id = ${id}$', {'id': script_id})
-    return rows[0] if rows else None
 
 
-async def execute_script(script_id, context_json=None):
+def _err(code, message, field='', detail=''):
-    """执行指定脚本（逻辑编程运行时）。"""
+    return {'code': code, 'message': message, 'field': field, 'detail': detail}
-    if not script_id:
-        return {'success': False, 'error': '缺少 script_id 参数'}
-
-    context = {}
-    if context_json:
-        if isinstance(context_json, str):
-            try:
-                context = _json.loads(context_json)
-            except Exception:
-                context = {}
-        elif isinstance(context_json, dict):
-            context = context_json
-
-    db = DBPools()
-    async with db.sqlorContext(_dbname()) as sor:
-        script = await _get_script(sor, script_id)
-        if not script:
-            return {'success': False, 'error': '脚本不存在: %s' % script_id}
-        script_type = script.get('script_type') or '0'
-        content = script.get('content') or ''
-        try:
-            if script_type == '0':
-                loc = {'context': context, 'result': None}
-                exec(content, {'__builtins__': {}}, loc)
-                return {'success': True, 'result': loc.get('result')}
-            if script_type == '1':
-                return {'success': True, 'result': _json.loads(content)}
-            val = eval(content, {'__builtins__': {}}, {'context': context})
-            return {'success': True, 'result': val}
-        except Exception as e:
-            return {'success': False, 'error': '%s: %s' % (type(e).__name__, e)}
 
 
-async def validate_script(content, script_type='0'):
+def _clean_str(value, default=''):
-    """校验脚本语法。"""
+    if value is None:
-    script_type = script_type or '0'
+        return default
-    content = content or ''
+    s = str(value).strip()
+    return s if s else default
+
+
+def _validate_input(script_name, script_type, content):
+    if not script_name:
+        return False, 'script_name', '脚本名称不能为空'
+    if len(script_name) > 100:
+        return False, 'script_name', '脚本名称长度不能超过 100'
+    if script_type not in VALID_SCRIPT_TYPES:
+        return False, 'script_type', '不支持的脚本类型: %s' % script_type
+    if not content or not content.strip():
+        return False, 'content', '脚本内容不能为空'
+    if len(content) > 65535:
+        return False, 'content', '脚本内容长度不能超过 65535'
+    ok, err = _validate_content(content, script_type)
+    if not ok:
+        return False, 'content', err
+    return True, '', ''
+
+
+async def create_script(request, params_kw):
     try:
-        if script_type == '0':
+        params = params_kw or {}
-            compile(content, '<script>', 'exec')
+        script_name = _clean_str(params.get('script_name'))
-            return {'valid': True, 'message': 'Python 脚本语法正确'}
+        script_type = _clean_str(params.get('script_type'), '0')
-        if script_type == '1':
+        content = _clean_str(params.get('content'))
-            _json.loads(content or '{}')
+        description = _clean_str(params.get('description'))
-            return {'valid': True, 'message': 'JSON 规则格式正确'}
+        status = _clean_str(params.get('status'), '1')
-        compile(content or '0', '<expr>', 'eval')
+        ok, field, msg = _validate_input(script_name, script_type, content)
-        return {'valid': True, 'message': '表达式语法正确'}
+        if not ok:
-    except Exception as e:
+            return _err(400, msg, field, 'create_script: 参数校验失败')
-        return {'valid': False, 'message': '%s: %s' % (type(e).__name__, e)}
+        dbname = _dbname()
-
+        db = DBPools()
-
-async def create_script(ns):
-    """新建脚本。"""
-    code = ns.get('code') or ''
-    db = DBPools()
-    async with db.sqlorContext(_dbname()) as sor:
-        if code:
-            rows = await sor.sqlExe(
-                'SELECT id FROM script_engine WHERE code = ${code}$', {'code': code})
-            if rows:
-                return {'success': False, 'error': '脚本编码已存在'}
-        ns['id'] = getID()
         now = curDateString()
-        ns['created_at'] = now
+        new_id = getID()
-        ns['updated_at'] = now
+        ns = {'id': new_id, 'script_name': script_name, 'script_type': script_type, 'content': content, 'description': description, 'status': status, 'created_at': now, 'updated_at': now}
-        ns.setdefault('status', '0')
+        async with db.sqlorContext(dbname) as sor:
-        ns.setdefault('script_type', '0')
+            await sor.C('script', ns)
-        await sor.C('script_engine', ns)
+        return _ok({'id': new_id}, '创建成功')
-        return {'success': True, 'id': ns['id']}
+    except Exception as e:
+        from traceback import format_exc
+        return _err(500, '创建失败: %s' % str(e), '', format_exc())
 
 
-async def update_script(ns):
+async def update_script(request, params_kw):
-    """更新脚本。"""
+    try:
-    rid = ns.get('id')
+        params = params_kw or {}
-    if not rid:
+        script_id = _clean_str(params.get('id'))
-        return {'success': False, 'error': '缺少 id 参数'}
+        if not script_id:
-    upd = {k: v for k, v in ns.items() if k != 'id'}
+            return _err(400, '缺少脚本ID', 'id', 'update_script: id 必填')
-    upd['id'] = rid
+        script_name = _clean_str(params.get('script_name'))
-    upd['updated_at'] = curDateString()
+        script_type = _clean_str(params.get('script_type'), '0')
-    db = DBPools()
+        content = _clean_str(params.get('content'))
-    async with db.sqlorContext(_dbname()) as sor:
+        description = _clean_str(params.get('description'))
-        await sor.U('script_engine', upd)
+        status = _clean_str(params.get('status'), '1')
-        return {'success': True, 'id': rid}
+        ok, field, msg = _validate_input(script_name, script_type, content)
+        if not ok:
+            return _err(400, msg, field, 'update_script: 参数校验失败')
+        dbname = _dbname()
+        db = DBPools()
+        async with db.sqlorContext(dbname) as sor:
+            rows = await sor.R('script', {'id': script_id})
+            if not rows:
+                return _err(404, '脚本不存在', 'id', 'update_script: 记录不存在')
+            upd = {'script_name': script_name, 'script_type': script_type, 'content': content, 'description': description, 'status': status, 'updated_at': curDateString()}
+            upd['id'] = script_id
+            await sor.U('script', upd)
+        return _ok({'id': script_id}, '更新成功')
+    except Exception as e:
+        from traceback import format_exc
+        return _err(500, '更新失败: %s' % str(e), '', format_exc())
 
 
-async def delete_script(ns):
+async def delete_script(request, params_kw):
-    """删除脚本。"""
+    try:
-    rid = ns.get('id')
+        params = params_kw or {}
-    if not rid:
+        script_id = _clean_str(params.get('id'))
-        return {'success': False, 'error': '缺少 id 参数'}
+        if not script_id:
-    db = DBPools()
+            return _err(400, '缺少脚本ID', 'id', 'delete_script: id 必填')
-    async with db.sqlorContext(_dbname()) as sor:
+        dbname = _dbname()
-        await sor.D('script_engine', {'id': rid})
+        db = DBPools()
-        return {'success': True, 'id': rid}
+        async with db.sqlorContext(dbname) as sor:
+            rows = await sor.R('script', {'id': script_id})
+            if not rows:
+                return _err(404, '脚本不存在', 'id', 'delete_script: 记录不存在')
+            await sor.D('script', {'id': script_id})
+        return _ok({'id': script_id}, '删除成功')
+    except Exception as e:
+        from traceback import format_exc
+        return _err(500, '删除失败: %s' % str(e), '', format_exc())
+
+
+async def get_script(request, params_kw):
+    try:
+        params = params_kw or {}
+        script_id = _clean_str(params.get('id'))
+        if not script_id:
+            return _err(400, '缺少脚本ID', 'id', 'get_script: id 必填')
+        dbname = _dbname()
+        db = DBPools()
+        async with db.sqlorContext(dbname) as sor:
+            rows = await sor.R('script', {'id': script_id})
+            if not rows:
+                return _err(404, '脚本不存在', 'id', 'get_script: 记录不存在')
+            rec = dict(rows[0])
+        return _ok(rec, '查询成功')
+    except Exception as e:
+        from traceback import format_exc
+        return _err(500, '查询失败: %s' % str(e), '', format_exc())
+
+
+async def list_scripts(request, params_kw):
+    try:
+        params = params_kw or {}
+        page = int(params.get('page', 1) or 1)
+        rows = int(params.get('rows', 20) or 20)
+        if page < 1:
+            page = 1
+        if rows < 1 or rows > 200:
+            rows = 20
+        keyword = _clean_str(params.get('script_name'))
+        script_type = _clean_str(params.get('script_type'))
+        status = _clean_str(params.get('status'))
+        conds = []
+        ns = {'page': page, 'rows': rows, 'sort': 'created_at desc'}
+        if keyword:
+            conds.append('script_name like ${keyword}$')
+            ns['keyword'] = '%%%s%%' % keyword
+        if script_type:
+            conds.append('script_type = ${script_type}$')
+            ns['script_type'] = script_type
+        if status:
+            conds.append('status = ${status}$')
+            ns['status'] = status
+        where = (' where ' + ' and '.join(conds)) if conds else ''
+        sql = 'select id, script_name, script_type, description, status, created_at, updated_at from script%s' % where
+        dbname = _dbname()
+        db = DBPools()
+        async with db.sqlorContext(dbname) as sor:
+            result = await sor.sqlExe(sql, ns)
+        total = result.get('total', 0) if isinstance(result, dict) else len(result or [])
+        data = result.get('rows', []) if isinstance(result, dict) else (result or [])
+        return _ok({'list': data, 'total': total}, '查询成功')
+    except Exception as e:
+        from traceback import format_exc
+        return _err(500, '列表查询失败: %s' % str(e), '', format_exc())
+
+
+async def execute_script(request, params_kw):
+    try:
+        params = params_kw or {}
+        script_id = _clean_str(params.get('id'))
+        content = None
+        script_type = '0'
+        if script_id:
+            dbname = _dbname()
+            db = DBPools()
+            async with db.sqlorContext(dbname) as sor:
+                rows = await sor.R('script', {'id': script_id})
+                if not rows:
+                    return _err(404, '脚本不存在', 'id', 'execute_script: 记录不存在')
+                content = rows[0].content
+                script_type = str(rows[0].script_type or '0')
+        else:
+            content = _clean_str(params.get('content'))
+            script_type = _clean_str(params.get('script_type'), '0')
+        if not content:
+            return _err(400, '脚本内容不能为空', 'content', 'execute_script: 无脚本内容')
+        ok, field, msg = _validate_input('t', script_type, content)
+        if not ok:
+            return _err(400, msg, field, 'execute_script: 校验失败，不执行')
+        run_params = params.get('params') or {}
+        result = _execute_content(content, script_type, run_params)
+        return _ok(result, '执行成功')
+    except NotImplementedError as e:
+        return _err(501, str(e), 'script_type', 'execute_script: 该类型暂不支持执行')
+    except Exception as e:
+        from traceback import format_exc
+        return _err(500, '执行失败: %s' % str(e), '', format_exc())
+
+
+async def validate_script_api(request, params_kw):
+    try:
+        params = params_kw or {}
+        content = _clean_str(params.get('content'))
+        script_type = _clean_str(params.get('script_type'), '0')
+        if not content:
+            return _err(400, '脚本内容不能为空', 'content', 'validate_script: 无脚本内容')
+        ok, field, msg = _validate_input('t', script_type, content)
+        if not ok:
+            return _err(400, msg, field, 'validate_script: 校验失败')
+        return _ok({'valid': True}, '校验通过')
+    except Exception as e:
+        from traceback import format_exc
+        return _err(500, '校验失败: %s' % str(e), '', format_exc())
 
 
 def load_script_engine():
-    """挂载 script_engine 模块到 ServerEnv。"""
     env = ServerEnv()
-    env.execute_script = execute_script
-    env.validate_script = validate_script
     env.create_script = create_script
     env.update_script = update_script
     env.delete_script = delete_script
+    env.get_script = get_script
+    env.list_scripts = list_scripts
+    env.execute_script = execute_script
+    env.validate_script_api = validate_script_api
+    env.create_scripts = create_script
+    env.update_scripts = update_script
+    env.delete_scripts = delete_script
+    return env

scripts/load_path.py

@@ -1,65 +1,65 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-"""script_engine 模块 RBAC 权限注册（显式路径，禁用通配符）。"""
+"""script_engine 模块 RBAC 路径登记（load_path.py）。"""
 import os
 import sys
 
-MODULE = 'script_engine'
+MOD = 'script_engine'
+
+PATHS_ANY = ['/%s/menu.ui' % MOD]
 
 PATHS_LOGINED = [
-    '/{0}'.format(MODULE),
+    '/%s' % MOD,
-    '/{0}/index.ui'.format(MODULE),
+    '/%s/index.ui' % MOD,
-    '/{0}/script_engine'.format(MODULE),
+    '/%s/script' % MOD,
-    '/{0}/script_engine/index.ui'.format(MODULE),
+    '/%s/script/index.ui' % MOD,
-    '/{0}/script_engine/get_script_engine.dspy'.format(MODULE),
+    '/%s/script/get_script.dspy' % MOD,
-    '/{0}/script_engine/add_script_engine.dspy'.format(MODULE),
+    '/%s/script/add_script.dspy' % MOD,
-    '/{0}/script_engine/update_script_engine.dspy'.format(MODULE),
+    '/%s/script/update_script.dspy' % MOD,
-    '/{0}/script_engine/delete_script_engine.dspy'.format(MODULE),
+    '/%s/script/delete_script.dspy' % MOD,
-    '/{0}/api/execute_script.dspy'.format(MODULE),
+    '/%s/api/script_create.dspy' % MOD,
-    '/{0}/api/validate_script.dspy'.format(MODULE),
+    '/%s/api/script_update.dspy' % MOD,
-    '/{0}/api/script_engine_create.dspy'.format(MODULE),
+    '/%s/api/script_delete.dspy' % MOD,
-    '/{0}/api/script_engine_update.dspy'.format(MODULE),
+    '/%s/api/script_get.dspy' % MOD,
-    '/{0}/api/script_engine_delete.dspy'.format(MODULE),
+    '/%s/api/script_list.dspy' % MOD,
-    '/{0}/api/get_search_world_id.dspy'.format(MODULE),
+    '/%s/api/script_execute.dspy' % MOD,
-    '/{0}/api/get_search_scene_id.dspy'.format(MODULE),
+    '/%s/api/script_validate.dspy' % MOD,
-    '/{0}/api/get_search_entity_id.dspy'.format(MODULE),
-    '/{0}/api/get_search_script_type.dspy'.format(MODULE),
-    '/{0}/api/get_search_status.dspy'.format(MODULE),
 ]
 
-PATHS_ANY = []
-
 
 def find_sage_root():
-    cur = os.path.dirname(os.path.abspath(__file__))
+    candidates = [
-    for _ in range(6):
+        os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', '..', '..'),
-        if os.path.isdir(os.path.join(cur, 'wwwroot')) and os.path.isdir(os.path.join(cur, 'py3', 'bin')):
+        os.path.expanduser('~/repos/sage'),
-            return cur
+        os.path.expanduser('~/sage'),
-        parent = os.path.dirname(cur)
+    ]
-        if parent == cur:
+    for cand in candidates:
-            break
+        if os.path.isdir(os.path.join(cand, 'wwwroot')) and os.path.isdir(os.path.join(cand, 'py3', 'bin')):
-        cur = parent
+            return os.path.abspath(cand)
     return None
 
 
 def main():
     sage_root = find_sage_root()
     if not sage_root:
-        print('[{0}] Sage root not found, skip RBAC registration'.format(MODULE))
+        print('[script_engine] 未找到 Sage 根目录，跳过 RBAC 登记（由宿主 load_path.py 兜底）')
-        return
+        return 0
+    set_role_perm = os.path.join(sage_root, 'set_role_perm.py')
+    if not os.path.exists(set_role_perm):
+        print('[script_engine] 未找到 set_role_perm.py，跳过')
+        return 0
     sys.path.insert(0, sage_root)
-    try:
+    import importlib.util
-        from set_role_perm import set_role_perm
+    spec = importlib.util.spec_from_file_location('set_role_perm', set_role_perm)
-    except ImportError:
+    mod = importlib.util.module_from_spec(spec)
-        print('[{0}] set_role_perm not found, skip RBAC registration'.format(MODULE))
+    spec.loader.exec_module(mod)
-        return
+    for p in PATHS_ANY:
-    for path in PATHS_ANY:
+        mod.set_role_perm(p, 'any')
-        set_role_perm(path, 'any')
+    for p in PATHS_LOGINED:
-    for path in PATHS_LOGINED:
+        mod.set_role_perm(p, 'logined')
-        set_role_perm(path, 'logined')
+    print('[script_engine] RBAC 路径登记完成: any=%d, logined=%d' % (len(PATHS_ANY), len(PATHS_LOGINED)))
-    print('[{0}] registered {1} any + {2} logined paths'.format(
+    return 0
-        MODULE, len(PATHS_ANY), len(PATHS_LOGINED)))
 
 
 if __name__ == '__main__':
-    main()
+    sys.exit(main())

skill/SKILL.md

@@ -1,33 +1,33 @@
 ---
 name: script_engine
-description: 逻辑编程（脚本/规则引擎）模块——脚本表 CRUD + execute_script/validate_script 接口，通过 load_script_engine() 挂载。
+description: 逻辑编程（脚本/规则引擎）模块——脚本表 CRUD + execute_script/validate_script 接口，通过 load_script_engine() 挂载。脚本类型 0=Python / 1=JavaScript / 2=规则表达式。
 ---
 
 # script_engine 模块
 
-## 概述
+脚本/规则引擎：脚本表（`script`）CRUD + 执行/校验接口。独立模块，无业务依赖。
-逻辑编程（脚本/规则引擎）模块。管理 `script_engine` 表（脚本/规则），提供脚本执行与语法校验接口。依赖 world、scene、entity 模块。
 
 ## 数据模型
-表 `script_engine`：`id`、`world_id`(→world)、`scene_id`(→scene)、`entity_id`(→entity)、`name`、`code`(unique)、`script_type`、`trigger_event`、`content`(text)、`status`、`created_at`、`updated_at`。
+- `script`：`id`(str32 PK)、`script_name`(str100 not null)、`script_type`(str32 default '0')、`content`(text not null)、`description`(str255)、`status`(str32 default '1')、`created_at`、`updated_at`
+- 索引：`idx_script_name`、`idx_script_type`
+- 编码：`script_type` → appcodes_kv parentid='script_type'；`status` → appcodes_kv parentid='script_status'
 
-## 关键接口
+## 关键接口（ServerEnv 注册函数，.dspy 直接调用）
-- `api/execute_script.dspy`：输入 `script_id` / `context_json`（可选），按 `script_type` 选择执行器执行 `content`，返回 `{success, result}`。
+- `create_script / update_script / delete_script / get_script / list_scripts`
-- `api/validate_script.dspy`：输入 `content` / `script_type`，语法/格式校验，返回 `{valid, message}`。
+- `execute_script`（id 或 content+script_type，校验后执行）
+- `validate_script_api`（仅校验，不落库不执行）
+- 复数别名 `create_scripts/update_scripts/delete_scripts` 与单数同实现（CRUD 框架约定）
 
-## 脚本类型编码（appcodes script_type）
+## 错误结构与分页
-- `0` = Python脚本（受限命名空间，通过 `result` 变量返回结果）
+- 统一 `{code, message, field, detail}`；code=0 成功
-- `1` = 规则(JSON)（解析 JSON 作为结果）
+- 分页 `{list, total}`（list_scripts 返回 data.list / data.total）
-- `2` = 表达式（eval 求值）
-
-## 状态编码（appcodes script_status）
-- `0` = 禁用、`1` = 启用
-
-## 挂载
-宿主应用入口 `init()` 调用 `load_script_engine()`，向 ServerEnv 注册 `execute_script` / `validate_script` / `create_script` / `update_script` / `delete_script`。
 
 ## 陷阱
-- 脚本执行使用受限命名空间（`__builtins__={}`），Python 脚本须通过 `result` 变量返回结果。
+- 取库名用 `ServerEnv().get_module_dbname('script_engine')`，禁止硬编码 DBNAME
-- `content` 为 `text` 字段：列表页隐藏（`browserfields.exclouded`），编辑页显示。
+- 非法输入（空名/超长/非法类型/语法错误/受限语法/危险调用）100% 拦截，不落库不执行
-- `code` 唯一索引，create 时校验重复。
+- Python 执行受限命名空间：仅内置白名单函数，禁 import/类/λ/async/await/对象方法调用
-- 取库名统一 `ServerEnv().get_module_dbname('script_engine')`，禁止硬编码 DBNAME。
+- .dspy 无 import/print/uuid/f-string，显式 return；helper 在 init.py
+- init/data.json 为 Format B appcodes（parentid + items），幂等落库
+
+## 依赖
+- 无业务依赖；依赖基础包 sqlor / ahserver / appPublic

wwwroot/api/execute_script.dspy

@@ -1,4 +1,2 @@
-script_id = params_kw.get('script_id')
+result = await execute_script(request, params_kw)
-context_json = params_kw.get('context_json')
-result = await execute_script(script_id, context_json)
 return result

wwwroot/api/script_create.dspy

@@ -0,0 +1,2 @@
+result = await create_script(request, params_kw)
+return result

wwwroot/api/script_delete.dspy

@@ -0,0 +1,2 @@
+result = await delete_script(request, params_kw)
+return result

wwwroot/api/script_execute.dspy

@@ -0,0 +1,2 @@
+result = await execute_script(request, params_kw)
+return result

wwwroot/api/script_get.dspy

@@ -0,0 +1,2 @@
+result = await get_script(request, params_kw)
+return result

wwwroot/api/script_list.dspy

@@ -0,0 +1,2 @@
+result = await list_scripts(request, params_kw)
+return result

wwwroot/api/script_update.dspy

@@ -0,0 +1,2 @@
+result = await update_script(request, params_kw)
+return result

wwwroot/api/script_validate.dspy

@@ -0,0 +1,2 @@
+result = await validate_script_api(request, params_kw)
+return result

wwwroot/api/validate_script.dspy

@@ -1,4 +1,2 @@
-content = params_kw.get('content')
+result = await validate_script(request, params_kw)
-script_type = params_kw.get('script_type', '0')
-result = await validate_script(content, script_type)
 return result

wwwroot/i18n/script_engine.json

@@ -0,0 +1,15 @@
+{
+    "script_engine": {
+        "module": "脚本引擎",
+        "script_manage": "脚本管理",
+        "script_create": "新增脚本",
+        "script_update": "编辑脚本",
+        "script_delete": "删除脚本",
+        "script_execute": "执行脚本",
+        "script_validate": "校验脚本",
+        "script_type": "脚本类型",
+        "script_name": "脚本名称",
+        "script_content": "脚本内容",
+        "script_status": "脚本状态"
+    }
+}

wwwroot/index.ui

@@ -1,18 +1 @@
-{
+{"widgettype":"VBox","options":{"width":"100%","height":"100%","padding":"20px"},"subwidgets":[{"widgettype":"Text","options":{"label":"脚本引擎","fontSize":"24px"}},{"widgettype":"ResponsableBox","options":{"gap":"16px","minWidth":"250px"},"subwidgets":[{"widgettype":"VBox","options":{"backgroundColor":"#FFFFFF","padding":"20px","cursor":"pointer"},"binds":[{"wid":"self","event":"click","actiontype":"urlwidget","target":"app.script_engine_content","options":{"url":"{{entire_url('/script_engine/script/index.ui')}}"},"mode":"replace"}],"subwidgets":[{"widgettype":"Text","options":{"label":"脚本管理"}},{"widgettype":"Text","options":{"label":"脚本 CRUD：新增/编辑/删除/查询"}}]},{"widgettype":"VBox","options":{"backgroundColor":"#FFFFFF","padding":"20px","cursor":"pointer"},"binds":[{"wid":"self","event":"click","actiontype":"urlwidget","target":"app.script_engine_content","options":{"url":"{{entire_url('/script_engine/api/script_execute.dspy')}}"},"mode":"replace"}],"subwidgets":[{"widgettype":"Text","options":{"label":"脚本执行"}},{"widgettype":"Text","options":{"label":"execute_script 接口"}}]},{"widgettype":"VBox","options":{"backgroundColor":"#FFFFFF","padding":"20px","cursor":"pointer"},"binds":[{"wid":"self","event":"click","actiontype":"urlwidget","target":"app.script_engine_content","options":{"url":"{{entire_url('/script_engine/api/script_validate.dspy')}}"},"mode":"replace"}],"subwidgets":[{"widgettype":"Text","options":{"label":"脚本校验"}},{"widgettype":"Text","options":{"label":"validate_script 接口"}}]}]},{"widgettype":"VBox","id":"app.script_engine_content","options":{"width":"100%","flex":"1","marginTop":"20px"}}]}
-    "widgettype": "VBox",
-    "options": {"width": "100%", "height": "100%", "padding": "20px"},
-    "subwidgets": [
-        {"widgettype": "Text", "options": {"label": "逻辑编程", "fontSize": "24px"}},
-        {"widgettype": "ResponsableBox", "options": {"gap": "16px", "minWidth": "250px"}, "subwidgets": [
-            {"widgettype": "VBox", "options": {"backgroundColor": "#FFFFFF", "padding": "20px", "cursor": "pointer"},
-             "binds": [{"wid": "self", "event": "click", "actiontype": "urlwidget", "target": "app.script_engine_content",
-                        "options": {"url": "{{entire_url('/script_engine/script_engine')}}"}, "mode": "replace"}],
-             "subwidgets": [{"widgettype": "Text", "options": {"label": "脚本列表", "fontSize": "16px"}}]},
-            {"widgettype": "VBox", "options": {"backgroundColor": "#FFFFFF", "padding": "20px", "cursor": "pointer"},
-             "binds": [{"wid": "self", "event": "click", "actiontype": "urlwidget", "target": "app.script_engine_content",
-                        "options": {"url": "{{entire_url('/script_engine/script_engine/add_script_engine.dspy')}}"}, "mode": "replace"}],
-             "subwidgets": [{"widgettype": "Text", "options": {"label": "新建脚本", "fontSize": "16px"}}]}
-        ]},
-        {"widgettype": "VBox", "id": "app.script_engine_content", "options": {"width": "100%", "flex": "1", "marginTop": "20px"}}
-    ]
-}

wwwroot/script_list.ui

@@ -0,0 +1,42 @@
+{
+  "widgettype": "VBox",
+  "options": {"width": "100%", "height": "100%", "padding": "10px"},
+  "subwidgets": [
+    {
+      "widgettype": "DataViewer",
+      "options": {
+        "title": "脚本列表",
+        "data_url": "{{entire_url('/script_engine/api/script_list.dspy')}}",
+        "row_options": {
+          "keyid": "id",
+          "editable": {
+            "new_data_url": "{{entire_url('/script_engine/api/script_create.dspy')}}",
+            "update_data_url": "{{entire_url('/script_engine/api/script_update.dspy')}}",
+            "delete_data_url": "{{entire_url('/script_engine/api/script_delete.dspy')}}"
+          },
+          "browserfields": {
+            "exclouded": ["id", "content"],
+            "alters": {
+              "script_type": {
+                "uitype": "code",
+                "data": [
+                  {"value": "0", "text": "Python脚本"},
+                  {"value": "1", "text": "规则(JSON)"},
+                  {"value": "2", "text": "表达式"}
+                ]
+              },
+              "status": {
+                "uitype": "code",
+                "data": [
+                  {"value": "1", "text": "启用"},
+                  {"value": "0", "text": "禁用"}
+                ]
+              }
+            }
+          },
+          "editexclouded": ["id", "created_at", "updated_at"]
+        }
+      }
+    }
+  ]
+}