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

fix: 回填目标机现场好代码——远端main残缺/空导致部署后模块不可用

Browse Source
This commit is contained in:
pipeline-agent 2026-08-30 13:26:14 +08:00
parent b325354acf
commit 4785665c46
31 changed files with 300 additions and 1050 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
.gitignore vendored
View File

@ -1,17 +1,9 @@
# Python
__pycache__/ __pycache__/
*.py[cod] *.py[cod]
*.egg-info/ *.egg-info/
build/ build/
dist/ dist/
.venv/
# generated artifacts
mysql.ddl.sql
wwwroot/script_engine_list/
# editors
*.swp *.swp
*.swo *.swo
.idea/ models/mysql.ddl.sql
.vscode/ wwwroot/script_engine/

36
README.md
View File

@ -1,30 +1,24 @@
# script_engine 逻辑编程（脚本/规则引擎）模块 # script_engine 模块
脚本/规则引擎：`script_engine` 脚本表 CRUD + 执行/校验接口。独立模块，无业务依赖。 逻辑编程（脚本/规则引擎）模块：管理逻辑脚本/规则表 `script_engine`，提供脚本执行与语法校验接口。
## 功能 ## 功能
- 脚本表 CRUD：`create_script / update_script / delete_script / get_script / list_scripts`（含复数别名） - `script_engine` 表 CRUD：脚本名称、编码、所属世界/场景/绑定实体、脚本类型、触发事件、脚本内容、状态。
- 执行接口：`execute_script`（按 id 或 content+script_type，校验后执行） - `api/execute_script.dspy`：按 `script_type` 选择执行器执行脚本内容，返回 `{success, result}`。
- 校验接口：`validate_script_api`（仅校验，不落库不执行） - `api/validate_script.dspy`：按 `script_type` 做语法/格式校验，返回 `{valid, message}`。
- 脚本类型：`0`=Python，`1`=SQL
- Python 受限执行：AST 校验（禁 import/类/λ/async/await/对象方法调用）+ builtins 白名单
- SQL 仅允许只读单语句（SELECT/SHOW/DESCRIBE/EXPLAIN），禁写语句/多语句/危险关键字
## 数据表 ## 数据表
- `script_engine`：id(str32 PK)、script_name(str100)、script_type(str32 default '0')、content(text)、description(str255)、status(str32 default '1')、created_at、updated_at - `script_engine`（逻辑脚本表），脚本内容 `content` 使用 `text` 字段。
- 索引：idx_script_name、idx_script_type
- 编码：script_type→appcodes_kv parentid='script_type'（0=Python，1=SQL）；status→parentid='script_status'（0=停用，1=启用）
## 集成 ## 脚本类型（appcodes script_type）
```python - `0` = Python 脚本（受限命名空间执行，通过 `result` 变量返回结果）
from script_engine.init import load_script_engine - `1` = 规则(JSON)（解析 JSON 作为结果）
load_script_engine() # 注册全部函数到 ServerEnv - `2` = 表达式（eval 求值）
```
- 取库名：`ServerEnv().get_module_dbname('script_engine')`（.py）/ `get_module_dbname('script_engine')`（.dspy 全局）
- REST：`wwwroot/api/script_create.dspy` 等 11 个端点
- RBAC：`scripts/load_path.py` 显式注册（无通配符）
## 构建 ## 挂载
宿主应用入口 `init()` 中调用 `load_script_engine()`。
## 安装
```bash ```bash
./build.sh # 1) 装 xls2ddl 2) models→mysql.ddl.sql 3) json→CRUD UI 4) wwwroot 软链 pip install .
``` ```

11
__init__.py
View File

@ -1,8 +1,7 @@
# -*- coding: utf-8 -*- """script_engine 模块 —— 逻辑编程（脚本/规则引擎）。
"""script_engine module root package.
Delegates to the real package script_engine.script_engine so that 通过 load_script_engine() 挂载到宿主应用。
`import script_engine` exposes the full public API.
""" """
from script_engine.script_engine import * # noqa: F401,F403 from .script_engine.init import load_script_engine
from script_engine.script_engine import __all__ # noqa: F401
__all__ = ['load_script_engine']

56
build.sh
View File

@ -1,56 +0,0 @@
#!/usr/bin/env bash
# script_engine module build - four steps:
# 1) install xls2ddl
# 2) models/*.json -> mysql.ddl.sql (json2ddl)
# 3) json/*.json -> generated CRUD UI (xls2ui)
# 4) symlink module wwwroot into host app wwwroot
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
cd "$SCRIPT_DIR"
PYTHON="${PYTHON:-python3}"
echo "[1/4] install xls2ddl..."
pip install xls2ddl
echo "[2/4] generate DDL from models/*.json..."
if [ -d models ]; then
$PYTHON -m xls2ddl.json2ddl mysql . > mysql.ddl.sql
fi
echo "[3/4] generate CRUD UI from json/*.json..."
if [ -d json ]; then
$PYTHON -m xls2ddl.xls2ui -m models -o wwwroot script_engine json/*.json
fi
echo "[4/4] symlink module wwwroot into host app wwwroot..."
# locate host app wwwroot: prefer sibling app dir under the same workspace
HOST_WWWROOT=""
for candidate in \
"$SCRIPT_DIR/../../apps/yuanjing/wwwroot" \
"$SCRIPT_DIR/../../yuanjing/wwwroot" \
"$SCRIPT_DIR/../yuanjing/wwwroot" \
"$HOME/repos/yuanjing/wwwroot"; do
if [ -d "$candidate" ]; then
HOST_WWWROOT="$(cd "$candidate" && pwd)"
break
fi
done
if [ -n "$HOST_WWWROOT" ]; then
mkdir -p "$HOST_WWWROOT/script_engine"
ln -sfn "$SCRIPT_DIR/wwwroot" "$HOST_WWWROOT/script_engine"
# link generated CRUD subdirs (script_engine_list) individually
for sub in "$SCRIPT_DIR/wwwroot"/*/; do
name="$(basename "$sub")"
case "$name" in
api|styles|scripts) continue ;;
esac
ln -sfn "$sub" "$HOST_WWWROOT/script_engine/$name"
done
echo "wwwroot linked to $HOST_WWWROOT/script_engine"
else
echo "WARN: host app wwwroot not found - skip symlink (deploy step will link)"
fi
echo "script_engine build done"

7
init/data.json
View File

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

14
json/script.json
View File

@ -1,14 +0,0 @@
{
"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')}}"
}
}
}

38
json/script_engine.json
View File

@ -1,38 +1,30 @@
{ {
"tblname": "script_engine", "tblname": "script_engine",
"title": "脚本管理", "title": "逻辑脚本",
"params": { "params": {
"sortby": ["created_at desc"], "sortby": ["updated_at desc"],
"editable": {
"new_data_url": "{{entire_url('../api/script_engine_create.dspy')}}",
"update_data_url": "{{entire_url('../api/script_engine_update.dspy')}}",
"delete_data_url": "{{entire_url('../api/script_engine_delete.dspy')}}"
},
"browserfields": { "browserfields": {
"exclouded": ["id", "content"], "exclouded": ["id", "content"],
"alters": { "alters": {
"script_type": { "world_id": {"uitype": "code", "dataurl": "{{entire_url('../api/get_search_world_id.dspy')}}"},
"uitype": "code", "scene_id": {"uitype": "code", "dataurl": "{{entire_url('../api/get_search_scene_id.dspy')}}"},
"dataurl": "{{entire_url('../api/get_search_script_type.dspy')}}" "entity_id": {"uitype": "code", "dataurl": "{{entire_url('../api/get_search_entity_id.dspy')}}"},
}, "script_type": {"uitype": "code", "dataurl": "{{entire_url('../api/get_search_script_type.dspy')}}"},
"status": { "status": {"uitype": "code", "dataurl": "{{entire_url('../api/get_search_status.dspy')}}"}
"uitype": "code",
"dataurl": "{{entire_url('../api/get_search_status.dspy')}}"
}
} }
}, },
"editexclouded": ["id", "created_at", "updated_at"], "editexclouded": ["id", "created_at", "updated_at"],
"data_filter": { "data_filter": {
"AND": [ "AND": [
{"field": "script_name", "op": "LIKE", "var": "script_name"}, {"field": "name", "op": "LIKE", "var": "name"},
{"field": "script_type", "op": "=", "var": "script_type"}, {"field": "world_id", "op": "=", "var": "world_id"},
{"field": "status", "op": "=", "var": "status"} {"field": "entity_id", "op": "=", "var": "entity_id"}
] ]
},
"filter_labels": {
"script_name": "脚本名称",
"script_type": "脚本类型",
"status": "状态"
},
"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')}}"
} }
} }
} }

54
models/script.json
View File

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

100
models/script_engine.json
View File

@ -2,93 +2,35 @@
"summary": [ "summary": [
{ {
"name": "script_engine", "name": "script_engine",
"title": "脚本表", "title": "逻辑脚本表",
"primary": ["id"], "primary": ["id"],
"catelog": "entity" "catelog": "entity"
} }
], ],
"fields": [ "fields": [
{ {"name": "id", "title": "主键ID", "type": "str", "length": 32, "nullable": "no"},
"name": "id", {"name": "world_id", "title": "所属世界", "type": "str", "length": 32, "nullable": "no"},
"title": "主键ID", {"name": "scene_id", "title": "所属场景", "type": "str", "length": 32, "nullable": "yes"},
"type": "str", {"name": "entity_id", "title": "绑定实体", "type": "str", "length": 32, "nullable": "yes"},
"length": 32, {"name": "name", "title": "脚本名称", "type": "str", "length": 255, "nullable": "no"},
"nullable": "no" {"name": "code", "title": "脚本编码", "type": "str", "length": 64, "nullable": "no"},
}, {"name": "script_type", "title": "脚本类型", "type": "str", "length": 16, "nullable": "no", "default": "0"},
{ {"name": "trigger_event", "title": "触发事件", "type": "str", "length": 64, "nullable": "yes"},
"name": "script_name", {"name": "content", "title": "脚本内容", "type": "text", "nullable": "no"},
"title": "脚本名称", {"name": "status", "title": "状态", "type": "str", "length": 16, "nullable": "no", "default": "0"},
"type": "str", {"name": "created_at", "title": "创建时间", "type": "timestamp", "nullable": "no"},
"length": 100, {"name": "updated_at", "title": "更新时间", "type": "timestamp", "nullable": "yes"}
"nullable": "no"
},
{
"name": "script_type",
"title": "脚本类型",
"type": "str",
"length": 32,
"nullable": "no",
"default": "0"
},
{
"name": "content",
"title": "脚本内容",
"type": "text",
"nullable": "no"
},
{
"name": "description",
"title": "描述",
"type": "str",
"length": 255
},
{
"name": "status",
"title": "状态",
"type": "str",
"length": 32,
"nullable": "no",
"default": "1"
},
{
"name": "created_at",
"title": "创建时间",
"type": "timestamp",
"nullable": "no"
},
{
"name": "updated_at",
"title": "更新时间",
"type": "timestamp",
"nullable": "no"
}
], ],
"indexes": [ "indexes": [
{ {"name": "idx_script_code", "idxtype": "unique", "idxfields": ["code"]},
"name": "idx_script_name", {"name": "idx_script_world", "idxtype": "index", "idxfields": ["world_id"]},
"idxtype": "index", {"name": "idx_script_entity", "idxtype": "index", "idxfields": ["entity_id"]}
"idxfields": ["script_name"]
},
{
"name": "idx_script_type",
"idxtype": "index",
"idxfields": ["script_type"]
}
], ],
"codes": [ "codes": [
{ {"field": "world_id", "table": "world", "valuefield": "id", "textfield": "name"},
"field": "script_type", {"field": "scene_id", "table": "scene", "valuefield": "id", "textfield": "name"},
"table": "appcodes_kv", {"field": "entity_id", "table": "entity", "valuefield": "id", "textfield": "name"},
"valuefield": "k", {"field": "script_type", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='script_type'"},
"textfield": "v", {"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='script_status'"}
"cond": "parentid='script_type'"
},
{
"field": "status",
"table": "appcodes_kv",
"valuefield": "k",
"textfield": "v",
"cond": "parentid='script_status'"
}
] ]
} }

1
pyproject.toml
View File

@ -5,7 +5,6 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "script_engine" name = "script_engine"
version = "1.0.0" version = "1.0.0"
description = "Logic programming (script/rule engine) module: script table CRUD + execute/validate interfaces (0=Python, 1=SQL)"
requires-python = ">=3.8" requires-python = ">=3.8"
dependencies = ["sqlor", "bricks_for_python"] dependencies = ["sqlor", "bricks_for_python"]

36
script_engine/__init__.py
View File

@ -1,36 +1,18 @@
"""script_engine package: logic programming (script/rule engine) module.""" """script_engine 包 —— 逻辑编程（脚本/规则引擎）模块实现。"""
from .engine import (
execute_python,
validate_python_source,
validate_sql,
)
from .init import ( from .init import (
create_script,
create_scripts,
delete_script,
delete_scripts,
execute_script,
get_script,
list_scripts,
load_script_engine, load_script_engine,
execute_script,
validate_script,
create_script,
update_script, update_script,
update_scripts, delete_script,
validate_script_api,
) )
__all__ = [ __all__ = [
'create_script',
'create_scripts',
'delete_script',
'delete_scripts',
'execute_python',
'execute_script',
'get_script',
'list_scripts',
'load_script_engine', 'load_script_engine',
'execute_script',
'validate_script',
'create_script',
'update_script', 'update_script',
'update_scripts', 'delete_script',
'validate_python_source',
'validate_script_api',
'validate_sql',
] ]

146
script_engine/engine.py
View File

@ -1,146 +0,0 @@
# -*- coding: utf-8 -*-
"""script_engine execution engine.
Python scripts run in a restricted namespace: only whitelisted builtin
functions are callable; import / class / lambda / async / await and
object method calls are forbidden (AST validation).
SQL scripts are read-only single statements (SELECT/SHOW/DESCRIBE/EXPLAIN);
write statements, multi-statement and dangerous keywords are rejected.
"""
import ast
import builtins
import contextlib
import io
import re
import traceback
# whitelisted builtin names callable inside python scripts
PY_WHITELIST = {
'abs', 'all', 'any', 'bin', 'bool', 'bytearray', 'bytes', 'callable',
'chr', 'complex', 'dict', 'divmod', 'enumerate', 'filter', 'float',
'format', 'frozenset', 'hash', 'hex', 'int', 'isinstance', 'issubclass',
'iter', 'len', 'list', 'map', 'max', 'min', 'next', 'oct', 'ord', 'pow',
'print', 'range', 'repr', 'reversed', 'round', 'set', 'slice', 'sorted',
'str', 'sum', 'tuple', 'zip',
'True', 'False', 'None',
}
# builtins that are never allowed even if listed elsewhere
PY_FORBIDDEN_BUILTINS = {
'__import__', 'eval', 'exec', 'compile', 'open', 'input', 'globals',
'locals', 'vars', 'dir', 'getattr', 'setattr', 'delattr', 'hasattr',
'memoryview', 'breakpoint', 'exit', 'quit', 'help', 'type', 'object',
}
# AST node types that are forbidden in python scripts
PY_FORBIDDEN_NODE_TYPES = (
ast.Import,
ast.ImportFrom,
ast.ClassDef,
ast.Lambda,
ast.AsyncFunctionDef,
ast.AsyncFor,
ast.AsyncWith,
ast.Await,
ast.Global,
ast.Nonlocal,
ast.Delete,
ast.Yield,
ast.YieldFrom,
)
# SQL statement starters that are read-only
SQL_READONLY_FIRST = ('SELECT', 'SHOW', 'DESCRIBE', 'DESC', 'EXPLAIN')
# SQL keywords that indicate write / execution / privilege operations
SQL_FORBIDDEN_KEYWORDS = (
'INSERT', 'UPDATE', 'DELETE', 'DROP', 'ALTER', 'CREATE', 'TRUNCATE',
'GRANT', 'REVOKE', 'MERGE', 'REPLACE', 'RENAME', 'CALL', 'LOAD',
'LOCK', 'UNLOCK', 'SET', 'USE', 'BEGIN', 'COMMIT', 'ROLLBACK',
'SAVEPOINT', 'INTO', 'OUTFILE', 'INFILE', 'PREPARE', 'EXECUTE',
'DEALLOCATE', 'PROCEDURE', 'TRIGGER', 'EVENT', 'TEMPORARY', 'PARTITION',
)
def _has_keyword(text, kw):
return re.search(r'\b' + kw + r'\b', text, re.IGNORECASE) is not None
def validate_python_source(content):
"""AST-validate python source. Returns (ok, message)."""
if not content or not str(content).strip():
return False, 'python content is empty'
try:
tree = ast.parse(str(content))
except SyntaxError as e:
return False, 'syntax error: %s' % str(e)
for node in ast.walk(tree):
if isinstance(node, PY_FORBIDDEN_NODE_TYPES):
return False, 'forbidden python syntax: ' + type(node).__name__
if isinstance(node, ast.Call):
fn = node.func
if isinstance(fn, ast.Name):
if fn.id in PY_FORBIDDEN_BUILTINS:
return False, 'forbidden builtin: ' + fn.id
if fn.id not in PY_WHITELIST:
return False, 'unknown function: ' + fn.id
elif isinstance(fn, ast.Attribute):
return False, 'object method call is not allowed'
return True, ''
def execute_python(content, params=None):
"""Execute python source in a restricted namespace.
Script may read `params` dict and set `result`; stdout is captured.
Returns {code, message, field, detail, data}.
"""
ok, msg = validate_python_source(content)
if not ok:
return {'code': 1, 'message': msg, 'field': 'content', 'detail': ''}
safe_builtins = {}
for name in PY_WHITELIST:
if name in ('True', 'False', 'None'):
continue
if hasattr(builtins, name):
safe_builtins[name] = getattr(builtins, name)
safe_builtins['__builtins__'] = safe_builtins
ns = {'params': dict(params or {})}
ns.update(safe_builtins)
buf = io.StringIO()
try:
with contextlib.redirect_stdout(buf):
exec(compile(str(content), '<script_engine>', 'exec'), ns)
return {
'code': 0,
'message': 'ok',
'field': '',
'detail': '',
'data': {'result': ns.get('result'), 'output': buf.getvalue()},
}
except Exception as e:
return {
'code': 1,
'message': str(e),
'field': '',
'detail': traceback.format_exc(),
}
def validate_sql(sql):
"""Validate SQL source. Returns (ok, message)."""
if not sql or not str(sql).strip():
return False, 'sql content is empty'
text = str(sql).strip().rstrip(';').strip()
if not text:
return False, 'sql content is empty'
parts = [p.strip() for p in text.split(';') if p.strip()]
if len(parts) > 1:
return False, 'multi-statement sql is not allowed'
first = text.split(None, 1)[0].upper()
if first not in SQL_READONLY_FIRST:
return False, 'only SELECT/SHOW/DESCRIBE/EXPLAIN statements are allowed'
for kw in SQL_FORBIDDEN_KEYWORDS:
if _has_keyword(text, kw):
return False, 'forbidden sql keyword: ' + kw
return True, ''

413
script_engine/init.py
View File

@ -1,327 +1,134 @@
# -*- coding: utf-8 -*- """script_engine 模块初始化。
"""script_engine module init.
Register script table CRUD + execute/validate interfaces to ServerEnv. 逻辑编程（脚本/规则引擎）领域模块，通过 load_script_engine() 挂载到宿主应用，
Functions: 向 ServerEnv 注册业务函数：
create_script / update_script / delete_script / get_script / list_scripts - execute_script(script_id, context_json) 执行脚本
execute_script / validate_script_api - validate_script(content, script_type) 校验脚本语法
plural aliases: create_scripts / update_scripts / delete_scripts - create_script / update_script / delete_script CRUD 业务逻辑
Entry point: load_script_engine()
""" """
try: import json as _json
from ahserver.serverenv import ServerEnv
except ImportError: # pragma: no cover - alternate package layout
try:
from ahserver.serverEnv import ServerEnv
except ImportError: # pragma: no cover - alternate package layout
from sqlor.dbpools import ServerEnv
from appPublic.log import debug
from appPublic.timeUtils import curDateString
from appPublic.uniqueID import getID from appPublic.uniqueID import getID
from appPublic.timeUtils import curDateString
from ahserver.serverenv import ServerEnv
from sqlor.dbpools import DBPools from sqlor.dbpools import DBPools
from .engine import ( _MODULE = 'script_engine'
execute_python,
validate_python_source,
validate_sql,
)
MODULE = 'script_engine'
TABLE = 'script_engine'
SCRIPT_TYPE_PYTHON = '0'
SCRIPT_TYPE_SQL = '1'
STATUS_ENABLED = '1'
def _dbname(): def _dbname():
return ServerEnv().get_module_dbname(MODULE) """取模块库名（禁止硬编码，由宿主应用 get_module_dbname 决定）。"""
return ServerEnv().get_module_dbname(_MODULE)
def _clean_params(params_kw): async def _get_script(sor, script_id):
if hasattr(params_kw, 'copy'): rows = await sor.sqlExe(
return params_kw.copy() 'SELECT * FROM script_engine WHERE id = ${id}$', {'id': script_id})
return dict(params_kw or {}) return rows[0] if rows else None
def _check_name(name): async def execute_script(script_id, context_json=None):
if not name: """执行指定脚本（逻辑编程运行时）。"""
return 'script_name is required' if not script_id:
if len(name) > 100: return {'success': False, 'error': '缺少 script_id 参数'}
return 'script_name is too long (max 100)'
return ''
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
def _check_type(stype): db = DBPools()
if stype not in (SCRIPT_TYPE_PYTHON, SCRIPT_TYPE_SQL): async with db.sqlorContext(_dbname()) as sor:
return 'invalid script_type (0=python, 1=sql)' script = await _get_script(sor, script_id)
return '' if not script:
return {'success': False, 'error': '脚本不存在: %s' % script_id}
script_type = script.get('script_type') or '0'
def _validate_content(stype, content): content = script.get('content') or ''
if stype == SCRIPT_TYPE_PYTHON:
return validate_python_source(content)
return validate_sql(content)
async def create_script(request, params_kw):
ns = _clean_params(params_kw)
name = str(ns.get('script_name') or '').strip()
err = _check_name(name)
if err:
return {'code': 1, 'message': err, 'field': 'script_name', 'detail': ''}
stype = str(ns.get('script_type') or SCRIPT_TYPE_PYTHON)
err = _check_type(stype)
if err:
return {'code': 1, 'message': err, 'field': 'script_type', 'detail': ''}
content = str(ns.get('content') or '')
if not content.strip():
return {'code': 1, 'message': 'content is required', 'field': 'content', 'detail': ''}
ok, msg = _validate_content(stype, content)
if not ok:
return {'code': 1, 'message': msg, 'field': 'content', 'detail': ''}
now = curDateString()
rec = {
'id': getID(),
'script_name': name,
'script_type': stype,
'content': content,
'description': str(ns.get('description') or '')[:255],
'status': str(ns.get('status') or STATUS_ENABLED),
'created_at': now,
'updated_at': now,
}
dbname = _dbname()
try:
async with DBPools().sqlorContext(dbname) as sor:
await sor.C(TABLE, rec)
except Exception as e:
debug('script_engine create_script error: %s' % str(e))
return {'code': 1, 'message': str(e), 'field': '', 'detail': ''}
return {'code': 0, 'message': 'ok', 'data': {'id': rec['id']}}
async def update_script(request, params_kw):
ns = _clean_params(params_kw)
sid = str(ns.get('id') or '').strip()
if not sid:
return {'code': 1, 'message': 'id is required', 'field': 'id', 'detail': ''}
upd = {'id': sid, 'updated_at': curDateString()}
if ns.get('script_name') is not None:
name = str(ns.get('script_name') or '').strip()
err = _check_name(name)
if err:
return {'code': 1, 'message': err, 'field': 'script_name', 'detail': ''}
upd['script_name'] = name
if ns.get('script_type') is not None:
stype = str(ns.get('script_type'))
err = _check_type(stype)
if err:
return {'code': 1, 'message': err, 'field': 'script_type', 'detail': ''}
upd['script_type'] = stype
if ns.get('content') is not None:
stype = str(ns.get('script_type') or upd.get('script_type') or SCRIPT_TYPE_PYTHON)
content = str(ns.get('content') or '')
if not content.strip():
return {'code': 1, 'message': 'content is required', 'field': 'content', 'detail': ''}
ok, msg = _validate_content(stype, content)
if not ok:
return {'code': 1, 'message': msg, 'field': 'content', 'detail': ''}
upd['content'] = content
if ns.get('description') is not None:
upd['description'] = str(ns.get('description'))[:255]
if ns.get('status') is not None:
upd['status'] = str(ns.get('status'))
dbname = _dbname()
try:
async with DBPools().sqlorContext(dbname) as sor:
await sor.U(TABLE, upd)
except Exception as e:
debug('script_engine update_script error: %s' % str(e))
return {'code': 1, 'message': str(e), 'field': '', 'detail': ''}
return {'code': 0, 'message': 'ok', 'data': {'id': sid}}
async def delete_script(request, params_kw):
ns = _clean_params(params_kw)
sid = str(ns.get('id') or '').strip()
if not sid:
return {'code': 1, 'message': 'id is required', 'field': 'id', 'detail': ''}
dbname = _dbname()
try:
async with DBPools().sqlorContext(dbname) as sor:
await sor.D(TABLE, {'id': sid})
except Exception as e:
debug('script_engine delete_script error: %s' % str(e))
return {'code': 1, 'message': str(e), 'field': '', 'detail': ''}
return {'code': 0, 'message': 'ok', 'data': {'id': sid}}
async def get_script(request, params_kw):
ns = _clean_params(params_kw)
sid = str(ns.get('id') or '').strip()
if not sid:
return {'code': 1, 'message': 'id is required', 'field': 'id', 'detail': ''}
dbname = _dbname()
try:
async with DBPools().sqlorContext(dbname) as sor:
recs = await sor.R(TABLE, {'id': sid})
except Exception as e:
debug('script_engine get_script error: %s' % str(e))
return {'code': 1, 'message': str(e), 'field': '', 'detail': ''}
if not recs:
return {'code': 1, 'message': 'script not found', 'field': 'id', 'detail': ''}
r = recs[0]
return {'code': 0, 'message': 'ok', 'data': {
'id': r.id,
'script_name': r.script_name,
'script_type': r.script_type,
'content': r.content,
'description': getattr(r, 'description', ''),
'status': r.status,
'created_at': r.created_at,
'updated_at': r.updated_at,
}}
async def list_scripts(request, params_kw):
ns = _clean_params(params_kw)
try:
page = int(ns.get('page') or 1)
except (TypeError, ValueError):
page = 1
try:
rows = int(ns.get('rows') or ns.get('pagerows') or 20)
except (TypeError, ValueError):
rows = 20
page = max(page, 1)
rows = min(max(rows, 1), 500)
offset = (page - 1) * rows
conds = []
vals = {}
name = str(ns.get('script_name') or '').strip()
if name:
conds.append('script_name like ${script_name}$')
vals['script_name'] = '%' + name + '%'
stype = str(ns.get('script_type') or '').strip()
if stype:
conds.append('script_type = ${script_type}$')
vals['script_type'] = stype
status = str(ns.get('status') or '').strip()
if status:
conds.append('status = ${status}$')
vals['status'] = status
where = ''
if conds:
where = ' where ' + ' and '.join(conds)
dbname = _dbname()
try:
async with DBPools().sqlorContext(dbname) as sor:
cnts = await sor.sqlExe('select count(*) as cnt from ' + TABLE + where, vals)
total = int(cnts[0].cnt) if cnts else 0
recs = await sor.sqlExe(
'select id, script_name, script_type, description, status, '
'created_at, updated_at from ' + TABLE + where +
' order by created_at desc limit %d offset %d' % (rows, offset),
vals)
items = [{
'id': r.id,
'script_name': r.script_name,
'script_type': r.script_type,
'description': getattr(r, 'description', ''),
'status': r.status,
'created_at': r.created_at,
'updated_at': r.updated_at,
} for r in recs]
except Exception as e:
debug('script_engine list_scripts error: %s' % str(e))
return {'code': 1, 'message': str(e), 'field': '', 'detail': ''}
return {'code': 0, 'message': 'ok', 'data': {'list': items, 'total': total}}
async def _execute_sql(sql, params):
dbname = _dbname()
try:
async with DBPools().sqlorContext(dbname) as sor:
recs = await sor.sqlExe(sql, dict(params or {}))
rows = list(recs)
except Exception as e:
debug('script_engine execute_sql error: %s' % str(e))
return {'code': 1, 'message': str(e), 'field': '', 'detail': ''}
return {'code': 0, 'message': 'ok', 'data': {'rows': rows, 'total': len(rows)}}
async def execute_script(request, params_kw):
"""Execute a script by id, or by content + script_type (validated first)."""
ns = _clean_params(params_kw)
sid = str(ns.get('id') or '').strip()
content = ns.get('content')
stype = str(ns.get('script_type') or SCRIPT_TYPE_PYTHON)
params = ns.get('params') or {}
if sid:
dbname = _dbname()
try: try:
async with DBPools().sqlorContext(dbname) as sor: if script_type == '0':
recs = await sor.R(TABLE, {'id': sid}) 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: except Exception as e:
debug('script_engine execute_script load error: %s' % str(e)) return {'success': False, 'error': '%s: %s' % (type(e).__name__, e)}
return {'code': 1, 'message': str(e), 'field': '', 'detail': ''}
if not recs:
return {'code': 1, 'message': 'script not found', 'field': 'id', 'detail': ''}
rec = recs[0]
if str(rec.status) != STATUS_ENABLED:
return {'code': 1, 'message': 'script is disabled', 'field': 'status', 'detail': ''}
content = rec.content
stype = str(rec.script_type)
err = _check_type(stype)
if err:
return {'code': 1, 'message': err, 'field': 'script_type', 'detail': ''}
if content is None or not str(content).strip():
return {'code': 1, 'message': 'content is required', 'field': 'content', 'detail': ''}
ok, msg = _validate_content(stype, str(content))
if not ok:
return {'code': 1, 'message': msg, 'field': 'content', 'detail': ''}
if stype == SCRIPT_TYPE_PYTHON:
return execute_python(str(content), params)
return await _execute_sql(str(content), params)
async def validate_script_api(request, params_kw): async def validate_script(content, script_type='0'):
"""Validate only - does not save to db and does not execute.""" """校验脚本语法。"""
ns = _clean_params(params_kw) script_type = script_type or '0'
content = str(ns.get('content') or '') content = content or ''
if not content.strip(): try:
return {'code': 1, 'message': 'content is required', 'field': 'content', 'detail': ''} if script_type == '0':
stype = str(ns.get('script_type') or SCRIPT_TYPE_PYTHON) compile(content, '<script>', 'exec')
err = _check_type(stype) return {'valid': True, 'message': 'Python 脚本语法正确'}
if err: if script_type == '1':
return {'code': 1, 'message': err, 'field': 'script_type', 'detail': ''} _json.loads(content or '{}')
ok, msg = _validate_content(stype, content) return {'valid': True, 'message': 'JSON 规则格式正确'}
if not ok: compile(content or '0', '<expr>', 'eval')
return {'code': 1, 'message': msg, 'field': 'content', 'detail': ''} return {'valid': True, 'message': '表达式语法正确'}
return {'code': 0, 'message': 'ok', 'data': {'valid': True}} except Exception as e:
return {'valid': False, 'message': '%s: %s' % (type(e).__name__, e)}
# plural aliases (CRUD framework convention: dspy wrappers use plural names) async def create_script(ns):
create_scripts = create_script """新建脚本。"""
update_scripts = update_script code = ns.get('code') or ''
delete_scripts = delete_script 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
ns['updated_at'] = now
ns.setdefault('status', '0')
ns.setdefault('script_type', '0')
await sor.C('script_engine', ns)
return {'success': True, 'id': ns['id']}
def load_script_engine(env=None): async def update_script(ns):
"""Register all script_engine functions to ServerEnv.""" """更新脚本。"""
if env is None: rid = ns.get('id')
env = ServerEnv() if not rid:
env.create_script = create_script return {'success': False, 'error': '缺少 id 参数'}
env.create_scripts = create_scripts upd = {k: v for k, v in ns.items() if k != 'id'}
env.update_script = update_script upd['id'] = rid
env.update_scripts = update_scripts upd['updated_at'] = curDateString()
env.delete_script = delete_script db = DBPools()
env.delete_scripts = delete_scripts async with db.sqlorContext(_dbname()) as sor:
env.get_script = get_script await sor.U('script_engine', upd)
env.list_scripts = list_scripts return {'success': True, 'id': rid}
async def delete_script(ns):
"""删除脚本。"""
rid = ns.get('id')
if not rid:
return {'success': False, 'error': '缺少 id 参数'}
db = DBPools()
async with db.sqlorContext(_dbname()) as sor:
await sor.D('script_engine', {'id': rid})
return {'success': True, 'id': rid}
def load_script_engine():
"""挂载 script_engine 模块到 ServerEnv。"""
env = ServerEnv()
env.execute_script = execute_script env.execute_script = execute_script
env.validate_script_api = validate_script_api env.validate_script = validate_script
debug('script_engine module loaded') env.create_script = create_script
return env env.update_script = update_script
env.delete_script = delete_script

113
script_engine/tests/test_validate_python.py
View File

@ -1,113 +0,0 @@
# -*- coding: utf-8 -*-
"""Regression tests for script_engine.validate_python (Bug kGZMpApWK3I1ed4OHpKQ0).
Validates that:
* params.get() and other whitelisted method calls now PASS validation
* every non-whitelisted method call is rejected (100% interception)
* dunder attribute access (sandbox escapes) is rejected
* attribute assignment / deletion is rejected
* forbidden constructs (import/class/lambda/async/...) are still rejected
Run (from modules/script_engine):
python script_engine/tests/test_validate_python.py
or:
python -m pytest script_engine/tests/test_validate_python.py -v
The engine module is loaded by file path so this test does not require the
ahserver/sqlor runtime environment.
"""
import importlib.util
import os
_HERE = os.path.dirname(os.path.abspath(__file__))
_ENGINE_PATH = os.path.join(os.path.dirname(_HERE), 'engine.py')
_spec = importlib.util.spec_from_file_location('_se_engine', _ENGINE_PATH)
_se_engine = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_se_engine)
validate_python = _se_engine.validate_python
execute_python = _se_engine.execute_python
_PASS = 0
_FAIL = 0
def check(content, expect_ok, why):
"""Assert validate_python(content) == expect_ok and print the outcome."""
global _PASS, _FAIL
r = validate_python(content)
ok = r.get('code') == 0
status = 'PASS' if ok == expect_ok else 'FAIL'
if ok == expect_ok:
_PASS += 1
else:
_FAIL += 1
print('%-4s expect_ok=%-5s %-52s -> %s' % (
status, expect_ok, why, r.get('message')))
def test_positive():
# the reported bug: params.get() must pass validation
check("result = params.get('age', 0) + 1", True, 'params.get')
check("result = params.get('name', '').strip().upper()", True, 'str chain on params.get')
check("result = sum(params.get('nums', []))", True, 'sum + params.get')
check("r = 0\nfor i in range(5):\n r += i\nresult = r", True, 'loop + range')
check("d = {'a': 1}\nd['b'] = 2\nresult = d.get('b')", True, 'dict subscript assign')
check("result = [x * 2 for x in range(3)]", True, 'list comprehension')
check("result = params.get('a', '').replace('x', 'y')", True, 'str.replace whitelisted')
check("result = params.get('l', []).append(1) or params.get('l')", True, 'list.append whitelisted')
def test_negative():
# forbidden constructs (must STILL be rejected)
check("import os\nresult = 1", False, 'import')
check("from os import system\nresult = 1", False, 'from-import')
check("class X:\n pass", False, 'class def')
check("f = lambda x: x", False, 'lambda')
check("async def f():\n pass", False, 'async def')
check("def g():\n yield 1", False, 'yield')
# non-whitelisted builtin calls
check("result = open('/etc/passwd')", False, 'open not whitelisted')
check("result = os.system('id')", False, 'os.system (name call)')
check("result = eval('1+1')", False, 'eval not whitelisted')
check("result = exec('x=1')", False, 'exec not whitelisted')
check("result = __import__('os')", False, '__import__ not whitelisted')
# dunder attribute access (sandbox escapes)
check("result = params.__class__", False, 'dunder read __class__')
check("result = ().__class__.__mro__", False, 'dunder escape chain')
check("result = params.get.__self__", False, 'dunder __self__')
check("result = (lambda: 1).__globals__", False, 'dunder __globals__')
# non-whitelisted object methods (100% interception)
check("result = params.popitem()", False, 'popitem not whitelisted')
check("result = params.foo()", False, 'arbitrary method foo')
check("result = params.get('a').isidentifier()", False, 'isidentifier not whitelisted')
check("result = [].index.__call__()", False, '__call__ dunder')
# attribute assignment / deletion
check("obj = params\nobj.x = 1", False, 'attribute assign')
check("obj = params\nobj.x += 1", False, 'attribute aug-assign')
check("obj = params\ndel obj.x", False, 'attribute delete')
# other call expressions
check("result = params['__class__']()", False, 'subscript call expression')
def test_execute():
out = execute_python("result = params.get('a', 0) + 10", {'params': {'a': 5}})
assert out == 15, 'execute_python params.get -> %r' % out
print('PASS execute_python params.get({a:5})+10 ->', out)
global _PASS
_PASS += 1
def main():
test_positive()
test_negative()
test_execute()
print('-' * 78)
print('TOTAL: %d passed, %d failed' % (_PASS, _FAIL))
if _FAIL:
raise SystemExit(1)
print('ALL TESTS PASSED')
if __name__ == '__main__':
main()

102
scripts/load_path.py
View File

@ -1,88 +1,64 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
"""script_engine RBAC path registration. """script_engine 模块 RBAC 权限注册（显式路径，禁用通配符）。"""
Run: python3 scripts/load_path.py (from module repo root)
Explicit paths only - NO wildcards.
"""
import os import os
import sys import sys
MODULE = 'script_engine' MODULE = 'script_engine'
# paths accessible to any (logged-out) visitor
PATHS_ANY = [
'/script_engine/menu.ui',
]
# paths accessible to any authenticated user
PATHS_LOGINED = [ PATHS_LOGINED = [
'/script_engine', '/{0}'.format(MODULE),
'/script_engine/index.ui', '/{0}/index.ui'.format(MODULE),
'/script_engine/script_execute_page.ui', '/{0}/script_engine'.format(MODULE),
'/script_engine/script_engine_list', '/{0}/script_engine/index.ui'.format(MODULE),
'/script_engine/script_engine_list/index.ui', '/{0}/script_engine/get_script_engine.dspy'.format(MODULE),
'/script_engine/script_engine_list/get_script_engine_list.dspy', '/{0}/script_engine/add_script_engine.dspy'.format(MODULE),
'/script_engine/script_engine_list/add_script_engine.dspy', '/{0}/script_engine/update_script_engine.dspy'.format(MODULE),
'/script_engine/script_engine_list/update_script_engine.dspy', '/{0}/script_engine/delete_script_engine.dspy'.format(MODULE),
'/script_engine/script_engine_list/delete_script_engine.dspy', '/{0}/api/execute_script.dspy'.format(MODULE),
'/script_engine/api/script_create.dspy', '/{0}/api/validate_script.dspy'.format(MODULE),
'/script_engine/api/script_update.dspy', '/{0}/api/script_engine_create.dspy'.format(MODULE),
'/script_engine/api/script_delete.dspy', '/{0}/api/script_engine_update.dspy'.format(MODULE),
'/script_engine/api/script_get.dspy', '/{0}/api/script_engine_delete.dspy'.format(MODULE),
'/script_engine/api/script_list.dspy', '/{0}/api/get_search_world_id.dspy'.format(MODULE),
'/script_engine/api/script_execute.dspy', '/{0}/api/get_search_scene_id.dspy'.format(MODULE),
'/script_engine/api/execute_script.dspy', '/{0}/api/get_search_entity_id.dspy'.format(MODULE),
'/script_engine/api/script_validate.dspy', '/{0}/api/get_search_script_type.dspy'.format(MODULE),
'/script_engine/api/validate_script.dspy', '/{0}/api/get_search_status.dspy'.format(MODULE),
'/script_engine/api/get_search_script_type.dspy',
'/script_engine/api/get_search_status.dspy',
] ]
# role-restricted paths PATHS_ANY = []
PATHS_ROLE = [
('/script_engine/api/script_create.dspy', 'owner.operator'),
('/script_engine/api/script_update.dspy', 'owner.operator'),
('/script_engine/api/script_delete.dspy', 'owner.operator'),
]
def find_sage_root(): def find_sage_root():
script_dir = os.path.dirname(os.path.abspath(__file__)) cur = os.path.dirname(os.path.abspath(__file__))
candidates = [ for _ in range(6):
os.path.join(script_dir, '..', '..'), if os.path.isdir(os.path.join(cur, 'wwwroot')) and os.path.isdir(os.path.join(cur, 'py3', 'bin')):
os.path.join(os.path.expanduser('~'), 'repos', 'sage'), return cur
os.path.join(os.path.expanduser('~'), 'sage'), parent = os.path.dirname(cur)
] if parent == cur:
for cand in candidates: break
cand = os.path.abspath(cand) cur = parent
if os.path.isdir(os.path.join(cand, 'wwwroot')) and os.path.isdir(os.path.join(cand, 'py3', 'bin')):
return cand
return None return None
def register(sage_root, path, role):
set_role_perm = os.path.join(sage_root, 'py3', 'bin', 'set_role_perm.py')
if not os.path.exists(set_role_perm):
print('set_role_perm.py not found: %s' % set_role_perm)
return
cmd = '%s %s %s %s' % (sys.executable, set_role_perm, role, path)
print('RUN: %s' % cmd)
os.system(cmd)
def main(): def main():
sage_root = find_sage_root() sage_root = find_sage_root()
if not sage_root: if not sage_root:
print('ERROR: sage root not found; skip RBAC registration') print('[{0}] Sage root not found, skip RBAC registration'.format(MODULE))
return
sys.path.insert(0, sage_root)
try:
from set_role_perm import set_role_perm
except ImportError:
print('[{0}] set_role_perm not found, skip RBAC registration'.format(MODULE))
return return
for path in PATHS_ANY: for path in PATHS_ANY:
register(sage_root, path, 'any') set_role_perm(path, 'any')
for path in PATHS_LOGINED: for path in PATHS_LOGINED:
register(sage_root, path, 'logined') set_role_perm(path, 'logined')
for path, role in PATHS_ROLE: print('[{0}] registered {1} any + {2} logined paths'.format(
register(sage_root, path, role) MODULE, len(PATHS_ANY), len(PATHS_LOGINED)))
print('script_engine RBAC registration done')
if __name__ == '__main__': if __name__ == '__main__':

50
skill/SKILL.md
View File

@ -1,45 +1,33 @@
--- ---
name: script_engine name: script_engine
description: 逻辑编程（脚本/规则引擎）模块——脚本表 CRUD + execute_script/validate_script 接口，通过 load_script_engine() 挂载。脚本类型 0=Python / 1=SQL。 description: 逻辑编程（脚本/规则引擎）模块——脚本表 CRUD + execute_script/validate_script 接口，通过 load_script_engine() 挂载。
--- ---
# script_engine 模块 # script_engine 模块
脚本/规则引擎：脚本表（`script_engine`）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_engine`：`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` - `api/execute_script.dspy`：输入 `script_id` / `context_json`（可选），按 `script_type` 选择执行器执行 `content`，返回 `{success, result}`。
- 编码：`script_type` → appcodes_kv parentid='script_type'（0=Python，1=SQL）；`status` → appcodes_kv parentid='script_status'（0=停用，1=启用） - `api/validate_script.dspy`：输入 `content` / `script_type`，语法/格式校验，返回 `{valid, message}`。
## 关键接口（ServerEnv 注册函数，.dspy 直接调用） ## 脚本类型编码（appcodes script_type）
- `0` = Python脚本（受限命名空间，通过 `result` 变量返回结果）
- `1` = 规则(JSON)（解析 JSON 作为结果）
- `2` = 表达式（eval 求值）
- `create_script / update_script / delete_script / get_script / list_scripts` ## 状态编码（appcodes script_status）
- `execute_script`（id 或 content+script_type，校验后执行） - `0` = 禁用、`1` = 启用
- `validate_script_api`（仅校验，不落库不执行）
- 复数别名 `create_scripts/update_scripts/delete_scripts` 与单数同实现（CRUD 框架约定）
## 错误结构与分页 ## 挂载
宿主应用入口 `init()` 调用 `load_script_engine()`，向 ServerEnv 注册 `execute_script` / `validate_script` / `create_script` / `update_script` / `delete_script`。
- 统一 `{code, message, field, detail}`；code=0 成功
- 分页 `{list, total}`（list_scripts 返回 data.list / data.total）
## REST 接口（wwwroot/api/*.dspy）
- script_create.dspy / script_update.dspy / script_delete.dspy / script_get.dspy / script_list.dspy
- script_execute.dspy / execute_script.dspy / script_validate.dspy / validate_script.dspy
- get_search_script_type.dspy / get_search_status.dspy（下拉 [{value,text}]）
## 陷阱 ## 陷阱
- 脚本执行使用受限命名空间（`__builtins__={}`），Python 脚本须通过 `result` 变量返回结果。
- 取库名用 `ServerEnv().get_module_dbname('script_engine')`，禁止硬编码 DBNAME - `content` 为 `text` 字段：列表页隐藏（`browserfields.exclouded`），编辑页显示。
- 非法输入（空名/超长/非法类型/语法错误/受限语法/危险调用）100% 拦截，不落库不执行 - `code` 唯一索引，create 时校验重复。
- Python 执行受限命名空间：仅内置白名单函数，禁 import/类/λ/async/await/对象方法调用（engine.py ast 校验 + 白名单 builtins） - 取库名统一 `ServerEnv().get_module_dbname('script_engine')`，禁止硬编码 DBNAME。
- SQL 仅允许只读单语句（SELECT/SHOW/DESCRIBE/EXPLAIN），禁写语句/多语句/危险关键字
- .dspy 无 import/print/uuid/f-string，显式 return；helper 在 init.py
- init/data.json 为 Format B appcodes（parentid + items），幂等落库
## 依赖
- 无业务依赖；依赖基础包 sqlor / ahserver / appPublic

6
wwwroot/api/execute_script.dspy
View File

@ -1,4 +1,4 @@
# execute_script.dspy - alias endpoint for script execution script_id = params_kw.get('script_id')
debug('execute_script.dspy: START params_kw=%s' % dict(params_kw)) context_json = params_kw.get('context_json')
result = await execute_script(request, params_kw) result = await execute_script(script_id, context_json)
return result return result

12
wwwroot/api/get_search_script_type.dspy
View File

@ -1,9 +1,9 @@
# get_search_script_type.dspy - dropdown for script_type (appcodes_kv parentid=script_type) result = [{'value': '', 'text': '全部'}]
debug('get_search_script_type.dspy: START')
try: try:
async with get_sor_context(request._run_ns, 'appbase') as sor: async with get_sor_context(request._run_ns, 'appbase') as sor:
recs = await sor.sqlExe("select k as value, v as text from appcodes_kv where parentid='script_type' order by k", {}) rows = await sor.sqlExe(
return json.dumps([{'value': '', 'text': '全部'}] + list(recs), ensure_ascii=False) "select k as value, v as text from appcodes_kv where parentid='script_type' order by k", {})
result = [{'value': '', 'text': '全部'}] + list(rows)
except Exception as e: except Exception as e:
debug('get_search_script_type.dspy error: %s' % str(e)) debug('get_search_script_type error: %s' % e)
return json.dumps([{'value': '', 'text': '全部'}], ensure_ascii=False) return result

12
wwwroot/api/get_search_status.dspy
View File

@ -1,9 +1,9 @@
# get_search_status.dspy - dropdown for status (appcodes_kv parentid=script_status) result = [{'value': '', 'text': '全部'}]
debug('get_search_status.dspy: START')
try: try:
async with get_sor_context(request._run_ns, 'appbase') as sor: async with get_sor_context(request._run_ns, 'appbase') as sor:
recs = await sor.sqlExe("select k as value, v as text from appcodes_kv where parentid='script_status' order by k", {}) rows = await sor.sqlExe(
return json.dumps([{'value': '', 'text': '全部'}] + list(recs), ensure_ascii=False) "select k as value, v as text from appcodes_kv where parentid='script_status' order by k", {})
result = [{'value': '', 'text': '全部'}] + list(rows)
except Exception as e: except Exception as e:
debug('get_search_status.dspy error: %s' % str(e)) debug('get_search_status error: %s' % e)
return json.dumps([{'value': '', 'text': '全部'}], ensure_ascii=False) return result

6
wwwroot/api/script_create.dspy
View File

@ -1,6 +0,0 @@
# script_create.dspy - create script (validates syntax before save)
debug('script_create.dspy: START params_kw=%s' % dict(params_kw))
result = await create_script(request, params_kw)
if result.get('code') == 0:
return {'status': 'ok', 'message': 'ok', 'data': result.get('data', {})}
return {'status': 'error', 'message': result.get('message', 'create failed'), 'field': result.get('field', ''), 'data': result}

6
wwwroot/api/script_delete.dspy
View File

@ -1,6 +0,0 @@
# script_delete.dspy - delete script by id
debug('script_delete.dspy: START params_kw=%s' % dict(params_kw))
result = await delete_script(request, params_kw)
if result.get('code') == 0:
return {'status': 'ok', 'message': 'ok', 'data': result.get('data', {})}
return {'status': 'error', 'message': result.get('message', 'delete failed'), 'field': result.get('field', ''), 'data': result}

4
wwwroot/api/script_execute.dspy
View File

@ -1,4 +0,0 @@
# script_execute.dspy - execute script by id or content+script_type
debug('script_execute.dspy: START params_kw=%s' % dict(params_kw))
result = await execute_script(request, params_kw)
return result

4
wwwroot/api/script_get.dspy
View File

@ -1,4 +0,0 @@
# script_get.dspy - get script detail by id
debug('script_get.dspy: START params_kw=%s' % dict(params_kw))
result = await get_script(request, params_kw)
return result

7
wwwroot/api/script_list.dspy
View File

@ -1,7 +0,0 @@
# script_list.dspy - paginated script list (returns DataViewer rows/total)
debug('script_list.dspy: START params_kw=%s' % dict(params_kw))
result = await list_scripts(request, params_kw)
if result.get('code') == 0:
data = result.get('data', {})
return {'status': 'ok', 'rows': data.get('list', []), 'total': data.get('total', 0)}
return {'status': 'error', 'message': result.get('message', 'list failed'), 'rows': [], 'total': 0}

6
wwwroot/api/script_update.dspy
View File

@ -1,6 +0,0 @@
# script_update.dspy - update script (validates syntax before save)
debug('script_update.dspy: START params_kw=%s' % dict(params_kw))
result = await update_script(request, params_kw)
if result.get('code') == 0:
return {'status': 'ok', 'message': 'ok', 'data': result.get('data', {})}
return {'status': 'error', 'message': result.get('message', 'update failed'), 'field': result.get('field', ''), 'data': result}

4
wwwroot/api/script_validate.dspy
View File

@ -1,4 +0,0 @@
# script_validate.dspy - validate script content only (no save, no execute)
debug('script_validate.dspy: START params_kw=%s' % dict(params_kw))
result = await validate_script_api(request, params_kw)
return result

6
wwwroot/api/validate_script.dspy
View File

@ -1,4 +1,4 @@
# validate_script.dspy - alias endpoint for script validation content = params_kw.get('content')
debug('validate_script.dspy: START params_kw=%s' % dict(params_kw)) script_type = params_kw.get('script_type', '0')
result = await validate_script_api(request, params_kw) result = await validate_script(content, script_type)
return result return result

15
wwwroot/i18n/script_engine.json
View File

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

32
wwwroot/index.ui
View File

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

31
wwwroot/script_execute_page.ui
View File

@ -1,31 +0,0 @@
{
"widgettype": "VBox",
"options": {"width": "100%", "height": "100%", "padding": "20px"},
"subwidgets": [
{"widgettype": "Text", "options": {"label": "脚本执行 / 校验", "fontSize": "20px"}},
{"widgettype": "HBox", "options": {"width": "100%", "gap": "16px"}, "subwidgets": [
{"widgettype": "VBox", "options": {"backgroundColor": "#FFFFFF", "padding": "16px", "flex": "1"}, "subwidgets": [
{"widgettype": "Text", "options": {"label": "脚本类型"}},
{"widgettype": "Select", "id": "script_type", "options": {
"data": [{"value": "0", "text": "Python"}, {"value": "1", "text": "SQL"}],
"value": "0"}},
{"widgettype": "Text", "options": {"label": "脚本内容"}},
{"widgettype": "Textarea", "id": "script_content", "options": {"width": "100%", "height": "220px", "placeholder": "Python: 设置 result 变量输出；SQL: 只读单条 SELECT/SHOW/DESCRIBE/EXPLAIN"}},
{"widgettype": "HBox", "options": {"gap": "12px"}, "subwidgets": [
{"widgettype": "Button", "id": "btn_validate", "options": {"label": "校验"},
"binds": [{"wid": "btn_validate", "event": "click", "actiontype": "urlwidget", "target": "script_result",
"options": {"method": "POST", "url": "{{entire_url('api/script_validate.dspy')}}",
"params": {"script_type": "{{params_kw.script_type}}", "content": "{{params_kw.script_content}}"}}}]},
{"widgettype": "Button", "id": "btn_execute", "options": {"label": "执行"},
"binds": [{"wid": "btn_execute", "event": "click", "actiontype": "urlwidget", "target": "script_result",
"options": {"method": "POST", "url": "{{entire_url('api/script_execute.dspy')}}",
"params": {"script_type": "{{params_kw.script_type}}", "content": "{{params_kw.script_content}}"}}}]}
]}
]},
{"widgettype": "VBox", "options": {"backgroundColor": "#FFFFFF", "padding": "16px", "flex": "1"}, "subwidgets": [
{"widgettype": "Text", "options": {"label": "执行结果"}},
{"widgettype": "VBox", "id": "script_result", "options": {"width": "100%", "minHeight": "260px", "backgroundColor": "#F7F7F7"}}
]}
]}
]
}

10
wwwroot/script_list.ui
View File

@ -1,10 +0,0 @@
{"widgettype":"VBox","options":{"width":"100%","height":"100%"},"subwidgets":[
{"widgettype":"Text","options":{"label":"脚本管理","fontSize":"18px","marginBottom":"10px"}},
{"widgettype":"DataViewer","options":{"data_url":"{{entire_url('api/script_list.dspy')}}",
"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')}}",
"browserfields":{"exclouded":["content"],
"alters":{"script_type":{"uitype":"code","dataurl":"{{entire_url('api/get_search_script_type.dspy')}}"},
"status":{"uitype":"code","dataurl":"{{entire_url('api/get_search_status.dspy')}}"}}},
"editexclouded":["id","created_at","updated_at"]}}]}