yumoqing/script_engine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

approve: 测试执行 - world 世界管理模块

Browse Source
This commit is contained in:
agent.develop 2026-08-29 12:44:23 +08:00
parent 58a27e52b4
commit 1bf2042769
18 changed files with 302 additions and 188 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
README.md
View File

@ -1,6 +1,6 @@
# script_engine 脚本引擎模块（W-06） # script_engine 脚本引擎模块（W-06）
脚本引擎：脚本表 CRUD + `execute_script` / `validate_script` 接口。独立模块，无业务依赖，可与 world / scene / entity 链路并行开发。 脚本引擎：脚本表 CRUD + `execute_script` / `validate_script` 接口。独立模块，无业务依赖。
## 功能 ## 功能
- **script 表 CRUD**：新增 / 编辑 / 删除 / 详情 / 分页列表 - **script 表 CRUD**：新增 / 编辑 / 删除 / 详情 / 分页列表
@ -10,9 +10,10 @@
## 数据表 ## 数据表
- `script`：id / script_name / script_type / content / description / status / created_at / updated_at - `script`：id / script_name / script_type / content / description / status / created_at / updated_at
- 索引：idx_script_name、idx_script_type
## 接口约定 ## 接口约定
- REST 统一前缀：`/api/*`（宿主自动路由 `/script_engine/api/xxx.dspy`） - REST 统一前缀：`/script_engine/api/*.dspy`
- 错误结构：`{code, message, field, detail}`（code=0 成功，400 参数错误，404 不存在，500 内部错误，501 类型暂不支持） - 错误结构：`{code, message, field, detail}`（code=0 成功，400 参数错误，404 不存在，500 内部错误，501 类型暂不支持）
- 分页：`{list, total}` - 分页：`{list, total}`
- 非法输入 100% 拦截，不落库、不执行 - 非法输入 100% 拦截，不落库、不执行
@ -21,7 +22,7 @@
1. `pip install .` 1. `pip install .`
2. 宿主应用 `from script_engine.init import load_script_engine` → `init()` 内调用 `load_script_engine()` 2. 宿主应用 `from script_engine.init import load_script_engine` → `init()` 内调用 `load_script_engine()`
3. `scripts/load_path.py` 登记 RBAC 路径；宿主 `load_path.py` 兜底 3. `scripts/load_path.py` 登记 RBAC 路径；宿主 `load_path.py` 兜底
4. `build.sh` 生成 DDL / CRUD UI / 链接 wwwroot 4. `build.sh` 四步：安装 xls2ddl → 生成 DDL → 生成 CRUD UI → 链接 wwwroot
## 接口清单（wwwroot/api/） ## 接口清单（wwwroot/api/）
| 路径 | 说明 | | 路径 | 说明 |

11
__init__.py
View File

@ -1,7 +1,6 @@
"""script_engine 模块 —— 逻辑编程（脚本/规则引擎）。 # -*- coding: utf-8 -*-
"""script_engine module -- script/rule engine (W-06)."""
from script_engine.engine import validate_script as validate_script, execute_script_content as execute_script_content
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)
通过 load_script_engine() 挂载到宿主应用。 __all__ = ['load_script_engine', 'create_script', 'update_script', 'delete_script', 'list_scripts', 'get_script', 'execute_script', 'validate_script_api', 'validate_script', 'execute_script_content']
"""
from .script_engine.init import load_script_engine
__all__ = ['load_script_engine']

39
build.sh
View File

@ -1,24 +1,36 @@
#!/usr/bin/env bash #!/usr/bin/env bash
# script_engine 模块构建脚本（由宿主应用 build.sh 集成调用） # script_engine 模块构建脚本 —— 四步安装（由宿主应用 build.sh 集成调用）
# ① 安装生成工具 ② 生成 DDL ③ 生成 CRUD UI ④ 链接 wwwroot
set -e set -e
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
MODULE="script_engine"
echo "[$MODULE] ==== Step 1/4: 安装生成工具 xls2ddl ===="
if ! command -v json2ddl >/dev/null 2>&1 || ! command -v xls2ui >/dev/null 2>&1; then
pip install xls2ddl >/dev/null 2>&1 || echo "[$MODULE] 警告: xls2ddl 安装失败（宿主已装则忽略）"
fi
echo "[$MODULE] ==== Step 2/4: 生成 DDL ===="
if [ -d "$SCRIPT_DIR/models" ]; then if [ -d "$SCRIPT_DIR/models" ]; then
if command -v json2ddl >/dev/null 2>&1; then if command -v json2ddl >/dev/null 2>&1; then
json2ddl mysql "$SCRIPT_DIR/models" > "$SCRIPT_DIR/models/mysql.ddl.sql" json2ddl mysql "$SCRIPT_DIR/models" > "$SCRIPT_DIR/models/mysql.ddl.sql"
echo "[$MODULE] DDL 已生成: models/mysql.ddl.sql"
else else
echo "[script_engine] json2ddl 不可用，跳过 DDL 生成" echo "[$MODULE] 警告: json2ddl 不可用，跳过 DDL 生成"
fi fi
fi fi
echo "[$MODULE] ==== Step 3/4: 生成 CRUD UI ===="
if [ -d "$SCRIPT_DIR/json" ]; then if [ -d "$SCRIPT_DIR/json" ]; then
if command -v xls2ui >/dev/null 2>&1; then if command -v xls2ui >/dev/null 2>&1; then
(cd "$SCRIPT_DIR" && xls2ui -m models -o wwwroot script_engine json/*.json) (cd "$SCRIPT_DIR" && xls2ui -m models -o wwwroot "$MODULE" json/*.json)
echo "[$MODULE] CRUD UI 已生成: wwwroot/$MODULE/"
else else
echo "[script_engine] xls2ui 不可用，跳过 CRUD UI 生成" echo "[$MODULE] 警告: xls2ui 不可用，跳过 CRUD UI 生成"
fi fi
fi fi
echo "[$MODULE] ==== Step 4/4: 链接 wwwroot 到宿主 ===="
SAGE_ROOT="" SAGE_ROOT=""
for candidate in "$SCRIPT_DIR/../.." "$HOME/repos/sage" "$HOME/sage"; do for candidate in "$SCRIPT_DIR/../.." "$HOME/repos/sage" "$HOME/sage"; do
if [ -d "$candidate/wwwroot" ] && [ -d "$candidate/py3/bin" ]; then if [ -d "$candidate/wwwroot" ] && [ -d "$candidate/py3/bin" ]; then
@ -27,15 +39,20 @@ for candidate in "$SCRIPT_DIR/../.." "$HOME/repos/sage" "$HOME/sage"; do
fi fi
done done
if [ -n "$SAGE_ROOT" ]; then if [ -n "$SAGE_ROOT" ]; then
mkdir -p "$SAGE_ROOT/wwwroot/script_engine" mkdir -p "$SAGE_ROOT/wwwroot/$MODULE/api"
ln -sfn "$SCRIPT_DIR/wwwroot/index.ui" "$SAGE_ROOT/wwwroot/script_engine/index.ui" ln -sfn "$SCRIPT_DIR/wwwroot/index.ui" "$SAGE_ROOT/wwwroot/$MODULE/index.ui"
mkdir -p "$SAGE_ROOT/wwwroot/script_engine/api"
for f in "$SCRIPT_DIR"/wwwroot/api/*.dspy; do for f in "$SCRIPT_DIR"/wwwroot/api/*.dspy; do
ln -sfn "$f" "$SAGE_ROOT/wwwroot/script_engine/api/$(basename "$f")" [ -f "$f" ] && ln -sfn "$f" "$SAGE_ROOT/wwwroot/$MODULE/api/$(basename "$f")"
done done
echo "[script_engine] wwwroot 已链接到 $SAGE_ROOT/wwwroot/script_engine" for d in "$SCRIPT_DIR"/wwwroot/*/; do
[ -d "$d" ] || continue
name="$(basename "$d")"
case "$name" in api|styles|scripts) continue ;; esac
ln -sfn "$d" "$SAGE_ROOT/wwwroot/$MODULE/$name"
done
echo "[$MODULE] wwwroot 已链接到 $SAGE_ROOT/wwwroot/$MODULE"
else else
echo "[script_engine] 未找到宿主 wwwroot，跳过链接" echo "[$MODULE] 未找到宿主 wwwroot，跳过链接"
fi fi
echo "[script_engine] build.sh 完成" echo "[$MODULE] ==== build.sh 四步安装完成 ===="

25
init/data.json
View File

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

50
json/script_engine.json
View File

@ -1,30 +1,26 @@
{ {
"tblname": "script_engine", "tblname": "script",
"title": "逻辑脚本", "params": {
"params": { "listname": "脚本列表",
"sortby": ["updated_at desc"], "browserfields": {
"editable": { "gridwidth": "100%",
"new_data_url": "{{entire_url('../api/script_engine_create.dspy')}}", "fields": [
"update_data_url": "{{entire_url('../api/script_engine_update.dspy')}}", {"field": "script_name", "label": "脚本名称", "width": 150, "searchable": true},
"delete_data_url": "{{entire_url('../api/script_engine_delete.dspy')}}" {"field": "script_type", "label": "脚本类型", "width": 100},
}, {"field": "status", "label": "状态", "width": 80},
"browserfields": { {"field": "description", "label": "描述", "width": 220},
"exclouded": ["id", "content"], {"field": "created_at", "label": "创建时间", "width": 150},
"alters": { {"field": "updated_at", "label": "更新时间", "width": 150}
"world_id": {"uitype": "code", "dataurl": "{{entire_url('../api/get_search_world_id.dspy')}}"}, ]
"scene_id": {"uitype": "code", "dataurl": "{{entire_url('../api/get_search_scene_id.dspy')}}"}, },
"entity_id": {"uitype": "code", "dataurl": "{{entire_url('../api/get_search_entity_id.dspy')}}"}, "editexclouded": ["script_name", "script_type", "content", "description", "status"],
"script_type": {"uitype": "code", "dataurl": "{{entire_url('../api/get_search_script_type.dspy')}}"}, "new_data_url": "{{entire_url('../api/script_create.dspy')}}",
"status": {"uitype": "code", "dataurl": "{{entire_url('../api/get_search_status.dspy')}}"} "update_data_url": "{{entire_url('../api/script_update.dspy')}}",
} "delete_data_url": "{{entire_url('../api/script_delete.dspy')}}",
}, "editable": {
"editexclouded": ["id", "created_at", "updated_at"], "new_data_url": "{{entire_url('../api/script_create.dspy')}}",
"data_filter": { "update_data_url": "{{entire_url('../api/script_update.dspy')}}",
"AND": [ "delete_data_url": "{{entire_url('../api/script_delete.dspy')}}"
{"field": "name", "op": "LIKE", "var": "name"},
{"field": "world_id", "op": "=", "var": "world_id"},
{"field": "entity_id", "op": "=", "var": "entity_id"}
]
}
} }
}
} }

60
models/script_engine.json
View File

@ -1,36 +1,28 @@
{ {
"summary": [ "summary": [
{ {
"name": "script_engine", "table": "script",
"title": "逻辑脚本表", "primary": ["id"],
"primary": ["id"], "name": "脚本表",
"catelog": "entity" "desc": "脚本/规则引擎脚本表（W-06 脚本引擎）"
} }
], ],
"fields": [ "fields": [
{"name": "id", "title": "主键ID", "type": "str", "length": 32, "nullable": "no"}, {"name": "id", "type": "str32", "notnull": true, "comment": "脚本ID（主键）"},
{"name": "world_id", "title": "所属世界", "type": "str", "length": 32, "nullable": "no"}, {"name": "script_name", "type": "str100", "notnull": true, "comment": "脚本名称"},
{"name": "scene_id", "title": "所属场景", "type": "str", "length": 32, "nullable": "yes"}, {"name": "script_type", "type": "str32", "default": "0", "comment": "脚本类型: 0=Python/1=JavaScript/2=规则表达式"},
{"name": "entity_id", "title": "绑定实体", "type": "str", "length": 32, "nullable": "yes"}, {"name": "content", "type": "text", "notnull": true, "comment": "脚本内容"},
{"name": "name", "title": "脚本名称", "type": "str", "length": 255, "nullable": "no"}, {"name": "description", "type": "str255", "comment": "脚本描述"},
{"name": "code", "title": "脚本编码", "type": "str", "length": 64, "nullable": "no"}, {"name": "status", "type": "str32", "default": "1", "comment": "状态: 0=停用/1=启用"},
{"name": "script_type", "title": "脚本类型", "type": "str", "length": 16, "nullable": "no", "default": "0"}, {"name": "created_at", "type": "datetime", "comment": "创建时间"},
{"name": "trigger_event", "title": "触发事件", "type": "str", "length": 64, "nullable": "yes"}, {"name": "updated_at", "type": "datetime", "comment": "更新时间"}
{"name": "content", "title": "脚本内容", "type": "text", "nullable": "no"}, ],
{"name": "status", "title": "状态", "type": "str", "length": 16, "nullable": "no", "default": "0"}, "indexes": [
{"name": "created_at", "title": "创建时间", "type": "timestamp", "nullable": "no"}, {"name": "idx_script_name", "fields": ["script_name"]},
{"name": "updated_at", "title": "更新时间", "type": "timestamp", "nullable": "yes"} {"name": "idx_script_type", "fields": ["script_type"]}
], ],
"indexes": [ "codes": [
{"name": "idx_script_code", "idxtype": "unique", "idxfields": ["code"]}, {"field": "script_type", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='script_type'"},
{"name": "idx_script_world", "idxtype": "index", "idxfields": ["world_id"]}, {"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='script_status'"}
{"name": "idx_script_entity", "idxtype": "index", "idxfields": ["entity_id"]} ]
],
"codes": [
{"field": "world_id", "table": "world", "valuefield": "id", "textfield": "name"},
{"field": "scene_id", "table": "scene", "valuefield": "id", "textfield": "name"},
{"field": "entity_id", "table": "entity", "valuefield": "id", "textfield": "name"},
{"field": "script_type", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='script_type'"},
{"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='script_status'"}
]
} }

9
script_engine/__init__.py
View File

@ -1,6 +1,9 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
"""script_engine 模块——脚本引擎（W-06）。""" """script_engine module -- script/rule engine (W-06).
from script_engine.engine import (validate_script as validate_script, execute_script_content as execute_script_content)
Mounted into a host application via load_script_engine().
"""
from script_engine.engine import validate_script as validate_script, execute_script_content as execute_script_content
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) 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)
__all__ = ['load_script_engine','create_script','update_script','delete_script','list_scripts','get_script','execute_script','validate_script_api','validate_script','execute_script_content'] __all__ = ['load_script_engine', 'create_script', 'update_script', 'delete_script', 'list_scripts', 'get_script', 'execute_script', 'validate_script_api', 'validate_script', 'execute_script_content']

36
script_engine/engine.py
View File

@ -1,14 +1,15 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
"""script_engine.engine —— 脚本引擎核心（纯逻辑，无 DB 依赖）。""" """script_engine.engine -- script engine core (pure logic, no DB dependency)."""
import ast import ast
import builtins import builtins
import json as _json
SCRIPT_TYPE_PYTHON = '0' SCRIPT_TYPE_PYTHON = '0'
SCRIPT_TYPE_JS = '1' SCRIPT_TYPE_JS = '1'
SCRIPT_TYPE_RULE = '2' SCRIPT_TYPE_RULE = '2'
VALID_SCRIPT_TYPES = (SCRIPT_TYPE_PYTHON, SCRIPT_TYPE_JS, SCRIPT_TYPE_RULE) 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']) _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) _FORBIDDEN_NODES = (ast.Import, ast.ImportFrom, ast.Global, ast.Nonlocal, ast.Lambda, ast.ClassDef, ast.Yield, ast.YieldFrom, ast.AsyncFunctionDef, ast.Await)
@ -27,56 +28,55 @@ def _pair_check(text, open_ch, close_ch):
def validate_python(content): def validate_python(content):
if not content or not content.strip(): if not content or not content.strip():
return False, '脚本内容不能为空' return False, 'script content must not be empty'
try: try:
tree = ast.parse(content, mode='exec') tree = ast.parse(content, mode='exec')
except SyntaxError as e: except SyntaxError as e:
return False, '语法错误: %s (第 %s 行)' % (e.msg or 'unknown', e.lineno or 0) return False, 'syntax error: %s (line %s)' % (e.msg or 'unknown', e.lineno or 0)
for node in ast.walk(tree): for node in ast.walk(tree):
if isinstance(node, _FORBIDDEN_NODES): if isinstance(node, _FORBIDDEN_NODES):
return False, '禁止使用的语法: %s (第 %s 行)' % (type(node).__name__, getattr(node, 'lineno', 0)) return False, 'forbidden syntax: %s (line %s)' % (type(node).__name__, getattr(node, 'lineno', 0))
if isinstance(node, ast.Call): if isinstance(node, ast.Call):
func = node.func func = node.func
if isinstance(func, ast.Name): if isinstance(func, ast.Name):
if func.id not in _ALLOWED_BUILTINS: if func.id not in _ALLOWED_BUILTINS:
return False, '禁止调用的函数: %s (第 %s 行)' % (func.id, getattr(node, 'lineno', 0)) return False, 'forbidden function call: %s (line %s)' % (func.id, getattr(node, 'lineno', 0))
elif isinstance(func, ast.Attribute): elif isinstance(func, ast.Attribute):
return False, '禁止调用对象方法: %s (第 %s 行)' % (func.attr, getattr(node, 'lineno', 0)) return False, 'forbidden method call: %s (line %s)' % (func.attr, getattr(node, 'lineno', 0))
return True, '' return True, ''
def validate_js(content): def validate_js(content):
if not content or not content.strip(): if not content or not content.strip():
return False, '脚本内容不能为空' return False, 'script content must not be empty'
if not _pair_check(content, '{', '}'): if not _pair_check(content, '{', '}'):
return False, '大括号不配对' return False, 'unbalanced braces'
if not _pair_check(content, '(', ')'): if not _pair_check(content, '(', ')'):
return False, '圆括号不配对' return False, 'unbalanced parentheses'
if not _pair_check(content, '[', ']'): if not _pair_check(content, '[', ']'):
return False, '方括号不配对' return False, 'unbalanced brackets'
return True, '' return True, ''
def validate_rule(content): def validate_rule(content):
if not content or not content.strip(): if not content or not content.strip():
return False, '脚本内容不能为空' return False, 'script content must not be empty'
text = content.strip() text = content.strip()
if text.startswith('{') or text.startswith('['): if text.startswith('{') or text.startswith('['):
try: try:
import json as _json
_json.loads(text) _json.loads(text)
return True, '' return True, ''
except Exception as e: except Exception as e:
return False, '规则 JSON 解析失败: %s' % str(e) return False, 'rule json parse failed: %s' % str(e)
if '->' not in text: if '->' not in text:
return False, '规则格式应为 JSON 或 "条件 -> 动作"' return False, 'rule format should be json or "condition -> action"'
return True, '' return True, ''
def validate_script(content, script_type): def validate_script(content, script_type):
stype = str(script_type or SCRIPT_TYPE_PYTHON) stype = str(script_type or SCRIPT_TYPE_PYTHON)
if stype not in VALID_SCRIPT_TYPES: if stype not in VALID_SCRIPT_TYPES:
return False, '不支持的脚本类型: %s' % stype return False, 'unsupported script type: %s' % stype
if stype == SCRIPT_TYPE_PYTHON: if stype == SCRIPT_TYPE_PYTHON:
return validate_python(content) return validate_python(content)
if stype == SCRIPT_TYPE_JS: if stype == SCRIPT_TYPE_JS:
@ -103,5 +103,5 @@ def execute_script_content(content, script_type, params):
if stype == SCRIPT_TYPE_PYTHON: if stype == SCRIPT_TYPE_PYTHON:
return execute_python(content, params or {}) return execute_python(content, params or {})
if stype == SCRIPT_TYPE_JS: if stype == SCRIPT_TYPE_JS:
raise NotImplementedError('JavaScript 类型暂不支持执行') raise NotImplementedError('JavaScript type execution is not supported yet')
raise NotImplementedError('规则表达式类型暂不支持执行') raise NotImplementedError('rule expression type execution is not supported yet')

89
script_engine/init.py
View File

@ -1,11 +1,13 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
"""script_engine.init —— 模块初始化与业务函数注册。""" """script_engine.init -- module initialization and business function registration."""
from ahserver.serverenv import ServerEnv from ahserver.serverenv import ServerEnv
from sqlor.dbpools import DBPools from sqlor.dbpools import DBPools
from appPublic.uniqueID import getID from appPublic.uniqueID import getID
from appPublic.timeUtils import curDateString from appPublic.timeUtils import curDateString
from script_engine.engine import validate_script as _validate_content, execute_script_content as _execute_content, VALID_SCRIPT_TYPES from script_engine.engine import validate_script as _validate_content, execute_script_content as _execute_content, VALID_SCRIPT_TYPES
_SCRIPT_FIELDS = ('id', 'script_name', 'script_type', 'content', 'description', 'status', 'created_at', 'updated_at')
def _dbname(): def _dbname():
return ServerEnv().get_module_dbname('script_engine') return ServerEnv().get_module_dbname('script_engine')
@ -26,17 +28,24 @@ def _clean_str(value, default=''):
return s if s else default return s if s else default
def _row_to_dict(row):
out = {}
for f in _SCRIPT_FIELDS:
out[f] = getattr(row, f, None)
return out
def _validate_input(script_name, script_type, content): def _validate_input(script_name, script_type, content):
if not script_name: if not script_name:
return False, 'script_name', '脚本名称不能为空' return False, 'script_name', 'script name must not be empty'
if len(script_name) > 100: if len(script_name) > 100:
return False, 'script_name', '脚本名称长度不能超过 100' return False, 'script_name', 'script name length must not exceed 100'
if script_type not in VALID_SCRIPT_TYPES: if script_type not in VALID_SCRIPT_TYPES:
return False, 'script_type', '不支持的脚本类型: %s' % script_type return False, 'script_type', 'unsupported script type: %s' % script_type
if not content or not content.strip(): if not content or not content.strip():
return False, 'content', '脚本内容不能为空' return False, 'content', 'script content must not be empty'
if len(content) > 65535: if len(content) > 65535:
return False, 'content', '脚本内容长度不能超过 65535' return False, 'content', 'script content length must not exceed 65535'
ok, err = _validate_content(content, script_type) ok, err = _validate_content(content, script_type)
if not ok: if not ok:
return False, 'content', err return False, 'content', err
@ -53,7 +62,7 @@ async def create_script(request, params_kw):
status = _clean_str(params.get('status'), '1') status = _clean_str(params.get('status'), '1')
ok, field, msg = _validate_input(script_name, script_type, content) ok, field, msg = _validate_input(script_name, script_type, content)
if not ok: if not ok:
return _err(400, msg, field, 'create_script: 参数校验失败') return _err(400, msg, field, 'create_script: input validation failed')
dbname = _dbname() dbname = _dbname()
db = DBPools() db = DBPools()
now = curDateString() now = curDateString()
@ -61,10 +70,10 @@ async def create_script(request, params_kw):
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 = {'id': new_id, 'script_name': script_name, 'script_type': script_type, 'content': content, 'description': description, 'status': status, 'created_at': now, 'updated_at': now}
async with db.sqlorContext(dbname) as sor: async with db.sqlorContext(dbname) as sor:
await sor.C('script', ns) await sor.C('script', ns)
return _ok({'id': new_id}, '创建成功') return _ok({'id': new_id}, 'created')
except Exception as e: except Exception as e:
from traceback import format_exc from traceback import format_exc
return _err(500, '创建失败: %s' % str(e), '', format_exc()) return _err(500, 'create failed: %s' % str(e), '', format_exc())
async def update_script(request, params_kw): async def update_script(request, params_kw):
@ -72,7 +81,7 @@ async def update_script(request, params_kw):
params = params_kw or {} params = params_kw or {}
script_id = _clean_str(params.get('id')) script_id = _clean_str(params.get('id'))
if not script_id: if not script_id:
return _err(400, '缺少脚本ID', 'id', 'update_script: id 必填') return _err(400, 'script id is required', 'id', 'update_script: id required')
script_name = _clean_str(params.get('script_name')) script_name = _clean_str(params.get('script_name'))
script_type = _clean_str(params.get('script_type'), '0') script_type = _clean_str(params.get('script_type'), '0')
content = _clean_str(params.get('content')) content = _clean_str(params.get('content'))
@ -80,20 +89,19 @@ async def update_script(request, params_kw):
status = _clean_str(params.get('status'), '1') status = _clean_str(params.get('status'), '1')
ok, field, msg = _validate_input(script_name, script_type, content) ok, field, msg = _validate_input(script_name, script_type, content)
if not ok: if not ok:
return _err(400, msg, field, 'update_script: 参数校验失败') return _err(400, msg, field, 'update_script: input validation failed')
dbname = _dbname() dbname = _dbname()
db = DBPools() db = DBPools()
async with db.sqlorContext(dbname) as sor: async with db.sqlorContext(dbname) as sor:
rows = await sor.R('script', {'id': script_id}) rows = await sor.R('script', {'id': script_id})
if not rows: if not rows:
return _err(404, '脚本不存在', 'id', 'update_script: 记录不存在') return _err(404, 'script not found', 'id', 'update_script: record missing')
upd = {'script_name': script_name, 'script_type': script_type, 'content': content, 'description': description, 'status': status, 'updated_at': curDateString()} upd = {'id': script_id, '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) await sor.U('script', upd)
return _ok({'id': script_id}, '更新成功') return _ok({'id': script_id}, 'updated')
except Exception as e: except Exception as e:
from traceback import format_exc from traceback import format_exc
return _err(500, '更新失败: %s' % str(e), '', format_exc()) return _err(500, 'update failed: %s' % str(e), '', format_exc())
async def delete_script(request, params_kw): async def delete_script(request, params_kw):
@ -101,18 +109,18 @@ async def delete_script(request, params_kw):
params = params_kw or {} params = params_kw or {}
script_id = _clean_str(params.get('id')) script_id = _clean_str(params.get('id'))
if not script_id: if not script_id:
return _err(400, '缺少脚本ID', 'id', 'delete_script: id 必填') return _err(400, 'script id is required', 'id', 'delete_script: id required')
dbname = _dbname() dbname = _dbname()
db = DBPools() db = DBPools()
async with db.sqlorContext(dbname) as sor: async with db.sqlorContext(dbname) as sor:
rows = await sor.R('script', {'id': script_id}) rows = await sor.R('script', {'id': script_id})
if not rows: if not rows:
return _err(404, '脚本不存在', 'id', 'delete_script: 记录不存在') return _err(404, 'script not found', 'id', 'delete_script: record missing')
await sor.D('script', {'id': script_id}) await sor.D('script', {'id': script_id})
return _ok({'id': script_id}, '删除成功') return _ok({'id': script_id}, 'deleted')
except Exception as e: except Exception as e:
from traceback import format_exc from traceback import format_exc
return _err(500, '删除失败: %s' % str(e), '', format_exc()) return _err(500, 'delete failed: %s' % str(e), '', format_exc())
async def get_script(request, params_kw): async def get_script(request, params_kw):
@ -120,18 +128,18 @@ async def get_script(request, params_kw):
params = params_kw or {} params = params_kw or {}
script_id = _clean_str(params.get('id')) script_id = _clean_str(params.get('id'))
if not script_id: if not script_id:
return _err(400, '缺少脚本ID', 'id', 'get_script: id 必填') return _err(400, 'script id is required', 'id', 'get_script: id required')
dbname = _dbname() dbname = _dbname()
db = DBPools() db = DBPools()
async with db.sqlorContext(dbname) as sor: async with db.sqlorContext(dbname) as sor:
rows = await sor.R('script', {'id': script_id}) rows = await sor.R('script', {'id': script_id})
if not rows: if not rows:
return _err(404, '脚本不存在', 'id', 'get_script: 记录不存在') return _err(404, 'script not found', 'id', 'get_script: record missing')
rec = dict(rows[0]) rec = _row_to_dict(rows[0])
return _ok(rec, '查询成功') return _ok(rec, 'ok')
except Exception as e: except Exception as e:
from traceback import format_exc from traceback import format_exc
return _err(500, '查询失败: %s' % str(e), '', format_exc()) return _err(500, 'query failed: %s' % str(e), '', format_exc())
async def list_scripts(request, params_kw): async def list_scripts(request, params_kw):
@ -162,13 +170,14 @@ async def list_scripts(request, params_kw):
dbname = _dbname() dbname = _dbname()
db = DBPools() db = DBPools()
async with db.sqlorContext(dbname) as sor: async with db.sqlorContext(dbname) as sor:
result = await sor.sqlExe(sql, ns) result = await sor.sqlPaging(sql, ns)
total = result.get('total', 0) if isinstance(result, dict) else len(result or []) total = result.get('total', 0)
data = result.get('rows', []) if isinstance(result, dict) else (result or []) rows_data = result.get('rows', []) or []
return _ok({'list': data, 'total': total}, '查询成功') data = [_row_to_dict(r) for r in rows_data]
return _ok({'list': data, 'total': total}, 'ok')
except Exception as e: except Exception as e:
from traceback import format_exc from traceback import format_exc
return _err(500, '列表查询失败: %s' % str(e), '', format_exc()) return _err(500, 'list failed: %s' % str(e), '', format_exc())
async def execute_script(request, params_kw): async def execute_script(request, params_kw):
@ -183,25 +192,25 @@ async def execute_script(request, params_kw):
async with db.sqlorContext(dbname) as sor: async with db.sqlorContext(dbname) as sor:
rows = await sor.R('script', {'id': script_id}) rows = await sor.R('script', {'id': script_id})
if not rows: if not rows:
return _err(404, '脚本不存在', 'id', 'execute_script: 记录不存在') return _err(404, 'script not found', 'id', 'execute_script: record missing')
content = rows[0].content content = rows[0].content
script_type = str(rows[0].script_type or '0') script_type = str(rows[0].script_type or '0')
else: else:
content = _clean_str(params.get('content')) content = _clean_str(params.get('content'))
script_type = _clean_str(params.get('script_type'), '0') script_type = _clean_str(params.get('script_type'), '0')
if not content: if not content:
return _err(400, '脚本内容不能为空', 'content', 'execute_script: 无脚本内容') return _err(400, 'script content must not be empty', 'content', 'execute_script: no content')
ok, field, msg = _validate_input('t', script_type, content) ok, field, msg = _validate_input('t', script_type, content)
if not ok: if not ok:
return _err(400, msg, field, 'execute_script: 校验失败，不执行') return _err(400, msg, field, 'execute_script: validation failed, not executed')
run_params = params.get('params') or {} run_params = params.get('params') or {}
result = _execute_content(content, script_type, run_params) result = _execute_content(content, script_type, run_params)
return _ok(result, '执行成功') return _ok(result, 'executed')
except NotImplementedError as e: except NotImplementedError as e:
return _err(501, str(e), 'script_type', 'execute_script: 该类型暂不支持执行') return _err(501, str(e), 'script_type', 'execute_script: type not supported')
except Exception as e: except Exception as e:
from traceback import format_exc from traceback import format_exc
return _err(500, '执行失败: %s' % str(e), '', format_exc()) return _err(500, 'execute failed: %s' % str(e), '', format_exc())
async def validate_script_api(request, params_kw): async def validate_script_api(request, params_kw):
@ -210,14 +219,14 @@ async def validate_script_api(request, params_kw):
content = _clean_str(params.get('content')) content = _clean_str(params.get('content'))
script_type = _clean_str(params.get('script_type'), '0') script_type = _clean_str(params.get('script_type'), '0')
if not content: if not content:
return _err(400, '脚本内容不能为空', 'content', 'validate_script: 无脚本内容') return _err(400, 'script content must not be empty', 'content', 'validate_script: no content')
ok, field, msg = _validate_input('t', script_type, content) ok, field, msg = _validate_input('t', script_type, content)
if not ok: if not ok:
return _err(400, msg, field, 'validate_script: 校验失败') return _err(400, msg, field, 'validate_script: validation failed')
return _ok({'valid': True}, '校验通过') return _ok({'valid': True}, 'valid')
except Exception as e: except Exception as e:
from traceback import format_exc from traceback import format_exc
return _err(500, '校验失败: %s' % str(e), '', format_exc()) return _err(500, 'validate failed: %s' % str(e), '', format_exc())
def load_script_engine(): def load_script_engine():

74
scripts/load_path.py
View File

@ -1,65 +1,59 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
"""script_engine 模块 RBAC 路径登记（load_path.py）。""" """script_engine 模块 RBAC 显式路径注册（禁止通配符）。
运行方式（宿主应用内）:
cd <SAGE_ROOT> && ./py3/bin/python <module>/scripts/load_path.py
"""
import os import os
import sys import sys
MOD = 'script_engine' MODULE = 'script_engine'
PATHS_ANY = ['/%s/menu.ui' % MOD]
# 显式路径清单（无通配符）
PATHS_LOGINED = [ PATHS_LOGINED = [
'/%s' % MOD, '/script_engine',
'/%s/index.ui' % MOD, '/script_engine/index.ui',
'/%s/script' % MOD, '/script_engine/api/script_create.dspy',
'/%s/script/index.ui' % MOD, '/script_engine/api/script_update.dspy',
'/%s/script/get_script.dspy' % MOD, '/script_engine/api/script_delete.dspy',
'/%s/script/add_script.dspy' % MOD, '/script_engine/api/script_get.dspy',
'/%s/script/update_script.dspy' % MOD, '/script_engine/api/script_list.dspy',
'/%s/script/delete_script.dspy' % MOD, '/script_engine/api/script_execute.dspy',
'/%s/api/script_create.dspy' % MOD, '/script_engine/api/script_validate.dspy',
'/%s/api/script_update.dspy' % MOD,
'/%s/api/script_delete.dspy' % MOD,
'/%s/api/script_get.dspy' % MOD,
'/%s/api/script_list.dspy' % MOD,
'/%s/api/script_execute.dspy' % MOD,
'/%s/api/script_validate.dspy' % MOD,
] ]
def find_sage_root(): def _find_sage_root():
script_dir = os.path.dirname(os.path.abspath(__file__))
candidates = [ candidates = [
os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', '..', '..'), os.path.normpath(os.path.join(script_dir, '..', '..')),
os.path.expanduser('~/repos/sage'), os.path.expanduser('~/repos/sage'),
os.path.expanduser('~/sage'), os.path.expanduser('~/sage'),
] ]
for cand in candidates: for cand in candidates:
if os.path.isdir(os.path.join(cand, 'wwwroot')) and os.path.isdir(os.path.join(cand, 'py3', 'bin')): if os.path.isdir(os.path.join(cand, 'wwwroot')) and os.path.isdir(os.path.join(cand, 'py3', 'bin')):
return os.path.abspath(cand) return cand
return None return None
def main(): def main():
sage_root = find_sage_root() sage_root = _find_sage_root()
if not sage_root: if not sage_root:
print('[script_engine] 未找到 Sage 根目录，跳过 RBAC 登记（由宿主 load_path.py 兜底）') print('[%s] 未找到宿主 Sage/wwwroot，跳过 RBAC 注册' % MODULE)
return 0 sys.exit(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) sys.path.insert(0, sage_root)
import importlib.util sys.path.insert(0, os.path.join(sage_root, 'py3', 'bin'))
spec = importlib.util.spec_from_file_location('set_role_perm', set_role_perm) try:
mod = importlib.util.module_from_spec(spec) from set_role_perm import set_role_perm
spec.loader.exec_module(mod) except Exception as e:
for p in PATHS_ANY: print('[%s] 无法导入 set_role_perm: %s' % (MODULE, e))
mod.set_role_perm(p, 'any') sys.exit(1)
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('[%s] 已注册 %s -> logined' % (MODULE, path))
return 0 print('[%s] RBAC 注册完成（%d 条）' % (MODULE, len(PATHS_LOGINED)))
if __name__ == '__main__': if __name__ == '__main__':
sys.exit(main()) main()

2
wwwroot/api/script_create.dspy
View File

@ -1,2 +1,4 @@
# script_create.dspy -- create a script record
# @api: POST /script_engine/api/script_create.dspy
result = await create_script(request, params_kw) result = await create_script(request, params_kw)
return result return result

2
wwwroot/api/script_delete.dspy
View File

@ -1,2 +1,4 @@
# script_delete.dspy -- delete a script record
# @api: POST /script_engine/api/script_delete.dspy
result = await delete_script(request, params_kw) result = await delete_script(request, params_kw)
return result return result

2
wwwroot/api/script_execute.dspy
View File

@ -1,2 +1,4 @@
# script_execute.dspy -- execute a script by id or inline content
# @api: POST /script_engine/api/script_execute.dspy
result = await execute_script(request, params_kw) result = await execute_script(request, params_kw)
return result return result

2
wwwroot/api/script_get.dspy
View File

@ -1,2 +1,4 @@
# script_get.dspy -- get one script record detail
# @api: GET/POST /script_engine/api/script_get.dspy
result = await get_script(request, params_kw) result = await get_script(request, params_kw)
return result return result

2
wwwroot/api/script_list.dspy
View File

@ -1,2 +1,4 @@
# script_list.dspy -- paged script list
# @api: GET/POST /script_engine/api/script_list.dspy
result = await list_scripts(request, params_kw) result = await list_scripts(request, params_kw)
return result return result

2
wwwroot/api/script_update.dspy
View File

@ -1,2 +1,4 @@
# script_update.dspy -- update a script record
# @api: POST /script_engine/api/script_update.dspy
result = await update_script(request, params_kw) result = await update_script(request, params_kw)
return result return result

2
wwwroot/api/script_validate.dspy
View File

@ -1,2 +1,4 @@
# script_validate.dspy -- validate script content only (no persist, no execute)
# @api: POST /script_engine/api/script_validate.dspy
result = await validate_script_api(request, params_kw) result = await validate_script_api(request, params_kw)
return result return result

76
wwwroot/index.ui
View File

@ -1 +1,75 @@
{"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",
"backgroundColor": "#F5F6FA"
},
"subwidgets": [
{
"widgettype": "Text",
"options": {
"label": "脚本引擎（W-06）",
"fontSize": "24px",
"fontWeight": "bold"
}
},
{
"widgettype": "ResponsableBox",
"options": {
"gap": "16px",
"minWidth": "250px"
},
"subwidgets": [
{
"widgettype": "VBox",
"options": {
"backgroundColor": "#FFFFFF",
"padding": "20px",
"cursor": "pointer",
"borderRadius": "8px"
},
"binds": [
{
"wid": "self",
"event": "click",
"actiontype": "urlwidget",
"target": "app.script_engine_content",
"options": {
"url": "{{entire_url('/script_engine/script_engine_list')}}"
},
"mode": "replace"
}
],
"subwidgets": [
{
"widgettype": "Text",
"options": {
"label": "脚本管理",
"fontSize": "16px"
}
},
{
"widgettype": "Text",
"options": {
"label": "脚本 CRUD / 执行 / 校验",
"fontSize": "13px",
"color": "#888888"
}
}
]
}
]
},
{
"widgettype": "VBox",
"id": "script_engine_content",
"options": {
"width": "100%",
"flex": "1",
"marginTop": "20px"
}
}
]
}