admin/udmin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(flow): 改进 Rhai 脚本执行错误处理和前后端代码节点映射

Browse Source 
- 修改 eval_rhai_expr_json 返回 Result 以提供错误信息
- 统一使用 unwrap_or_else 处理 Rhai 表达式执行错误
- 前后端代码节点类型映射支持 JavaScript 和 Rhai 语言
- 前端代码编辑器添加语言选择器
- 优化 WebSocket 错误处理和关闭逻辑
This commit is contained in:
ayou
2025-09-22 06:52:22 +08:00
parent 067c6829f0
commit 3362268575
7 changed files with 95 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
backend/src/flow/engine.rs
View File

@ -99,7 +99,7 @@ fn eval_rhai_expr_bool(expr: &str, ctx: &serde_json::Value) -> bool {
// 构造作用域并注入 ctx
let mut scope = rhai::Scope::new();
let dyn_ctx = match rhai::serde::to_dynamic(ctx.clone()) { Ok(d) => d, Err(_) => rhai::Dynamic::UNIT };
scope.push("ctx", dyn_ctx);
scope.push_dynamic("ctx", dyn_ctx);
// 先从缓存读取 AST；未命中则编译并写入缓存，然后执行
let cached = AST_CACHE.with(|c| c.borrow().get(expr).cloned());
@ -107,7 +107,7 @@ fn eval_rhai_expr_bool(expr: &str, ctx: &serde_json::Value) -> bool {
return RHIA_ENGINE.with(|eng| eng.borrow().eval_ast_with_scope::<bool>(&mut scope, &ast).unwrap_or(false));
}
let compiled = RHIA_ENGINE.with(|eng| eng.borrow().compile(expr));
let compiled = RHIA_ENGINE.with(|eng| eng.borrow().compile_with_scope(&mut scope, expr));
match compiled {
Ok(ast) => {
// 简单容量控制：超过 1024 条时清空，避免无限增长
@ -123,20 +123,20 @@ fn eval_rhai_expr_bool(expr: &str, ctx: &serde_json::Value) -> bool {
}
// 通用：评估 Rhai 表达式并转换为 serde_json::Value，失败返回 None
pub(crate) fn eval_rhai_expr_json(expr: &str, ctx: &serde_json::Value) -> Option<serde_json::Value> {
pub(crate) fn eval_rhai_expr_json(expr: &str, ctx: &serde_json::Value) -> Result<serde_json::Value, String> {
// 构造作用域并注入 ctx
let mut scope = rhai::Scope::new();
let dyn_ctx = match rhai::serde::to_dynamic(ctx.clone()) { Ok(d) => d, Err(_) => rhai::Dynamic::UNIT };
scope.push("ctx", dyn_ctx);
scope.push_dynamic("ctx", dyn_ctx);
// 先从缓存读取 AST；未命中则编译并写入缓存，然后执行
let cached = AST_CACHE.with(|c| c.borrow().get(expr).cloned());
let eval = |ast: &AST, scope: &mut rhai::Scope| -> Option<serde_json::Value> {
let eval = |ast: &AST, scope: &mut rhai::Scope| -> Result<serde_json::Value, String> {
RHIA_ENGINE.with(|eng| {
eng.borrow()
.eval_ast_with_scope::<rhai::Dynamic>(scope, ast)
.ok()
.and_then(|d| rhai::serde::from_dynamic(&d).ok())
.map_err(|e| e.to_string())
.and_then(|d| rhai::serde::from_dynamic(&d).map_err(|e| e.to_string()))
})
};
@ -144,7 +144,7 @@ pub(crate) fn eval_rhai_expr_json(expr: &str, ctx: &serde_json::Value) -> Option
return eval(&ast, &mut scope);
}
let compiled = RHIA_ENGINE.with(|eng| eng.borrow().compile(expr));
let compiled = RHIA_ENGINE.with(|eng| eng.borrow().compile_with_scope(&mut scope, expr));
match compiled {
Ok(ast) => {
AST_CACHE.with(|c| {
@ -154,7 +154,7 @@ pub(crate) fn eval_rhai_expr_json(expr: &str, ctx: &serde_json::Value) -> Option
});
eval(&ast, &mut scope)
}
Err(_) => None,
Err(e) => Err(e.to_string()),
}
}

2
backend/src/flow/executors/condition.rs
View File

@ -166,7 +166,7 @@ pub(crate) fn resolve_value(ctx: &serde_json::Value, v: &serde_json::Value) -> R
"expression" => {
let expr = v.get("content").and_then(|x| x.as_str()).unwrap_or("");
if expr.trim().is_empty() { return Ok(V::Null); }
Ok(crate::flow::engine::eval_rhai_expr_json(expr, ctx).unwrap_or(V::Null))
Ok(crate::flow::engine::eval_rhai_expr_json(expr, ctx).unwrap_or_else(|_| V::Null))
}
_ => Ok(V::Null),
}

14
backend/src/flow/executors/script_rhai.rs
View File

@ -62,7 +62,7 @@ pub fn exec_rhai_file(node_id: &NodeId, path: &str, ctx: &mut Value) -> anyhow::
let res = eval_rhai_expr_json(&wrapped, ctx);
let dur_ms = start.elapsed().as_millis();
match res {
Some(new_ctx) => {
Ok(new_ctx) => {
let (added, removed, modified) = shallow_diff(&before_ctx, &new_ctx);
*ctx = new_ctx;
info!(target = "udmin.flow", node=%node_id.0, ms=%dur_ms, added=%added.len(), removed=%removed.len(), modified=%modified.len(), "script task: Rhai file executed and ctx updated");
@ -70,8 +70,8 @@ pub fn exec_rhai_file(node_id: &NodeId, path: &str, ctx: &mut Value) -> anyhow::
debug!(target = "udmin.flow", node=%node_id.0, ?added, ?removed, ?modified, "script task: ctx shallow diff");
}
}
None => {
info!(target = "udmin.flow", node=%node_id.0, ms=%dur_ms, preview=%preview, "script task: Rhai file execution failed, ctx unchanged");
Err(err) => {
info!(target = "udmin.flow", node=%node_id.0, ms=%dur_ms, preview=%preview, err=%err, "script task: Rhai file execution failed, ctx unchanged");
}
}
Ok(())
@ -114,10 +114,10 @@ impl Executor for ScriptRhaiTask {
let before_ctx = ctx.clone();
let wrapped = format!("{{ {} ; ctx }}", script);
let res = super::super::engine::eval_rhai_expr_json(&wrapped, ctx);
let res = eval_rhai_expr_json(&wrapped, ctx);
let dur_ms = start.elapsed().as_millis();
match res {
Some(new_ctx) => {
Ok(new_ctx) => {
let (added, removed, modified) = shallow_diff(&before_ctx, &new_ctx);
*ctx = new_ctx;
info!(target = "udmin.flow", node=%node_id.0, ms=%dur_ms, added=%added.len(), removed=%removed.len(), modified=%modified.len(), "script_rhai task: inline executed and ctx updated");
@ -125,8 +125,8 @@ impl Executor for ScriptRhaiTask {
debug!(target = "udmin.flow", node=%node_id.0, ?added, ?removed, ?modified, "script_rhai task: ctx shallow diff");
}
}
None => {
info!(target = "udmin.flow", node=%node_id.0, ms=%dur_ms, preview=%script_preview, "script_rhai task: inline execution failed, ctx unchanged");
Err(err) => {
info!(target = "udmin.flow", node=%node_id.0, ms=%dur_ms, preview=%script_preview, err=%err, "script_rhai task: inline execution failed, ctx unchanged");
}
}
return Ok(());

4
backend/src/flow/executors/variable.rs
View File

@ -56,7 +56,7 @@ fn resolve_assign_value(ctx: &Value, v: &Value) -> Value {
}
// ctx[...] / ctx. 前缀 -> 表达式求值
if s_trim.starts_with("ctx[") || s_trim.starts_with("ctx.") {
return eval_rhai_expr_json(s_trim, ctx).unwrap_or(V::Null);
return eval_rhai_expr_json(s_trim, ctx).unwrap_or_else(|_| V::Null);
}
}
v.get("content").cloned().unwrap_or(V::Null)
@ -97,7 +97,7 @@ fn resolve_assign_value(ctx: &Value, v: &Value) -> Value {
"expression" => {
let expr = v.get("content").and_then(|x| x.as_str()).unwrap_or("");
if expr.trim().is_empty() { return V::Null; }
eval_rhai_expr_json(expr, ctx).unwrap_or(V::Null)
eval_rhai_expr_json(expr, ctx).unwrap_or_else(|_| V::Null)
}
_ => {
// fallback: if content exists, treat as constant

18
frontend/src/flows/editor.tsx
View File

@ -113,14 +113,26 @@ export function Editor() {
export default Editor;
// 新增：将后端存储的 javascript 类型还原为前端 UI 的 code 类型
// 新增：将后端存储的 javascript/script/script_rhai 类型还原为前端 UI 的 code 类型，并设置语言
function transformDesignJsonFromBackend(json: any): any {
try {
const clone = JSON.parse(JSON.stringify(json));
if (Array.isArray(clone?.nodes)) {
clone.nodes = clone.nodes.map((n: any) => {
if (n && n.type === 'javascript') {
return { ...n, type: 'code' };
if (!n) return n;
if (n.type === 'javascript') {
// JS 节点 -> 前端 code，保留/设置 language=javascript
const data = n.data || {};
const script = data.script || {};
if (!script.language) script.language = 'javascript';
return { ...n, type: 'code', data: { ...data, script } };
}
if (n.type === 'script' || n.type === 'script_rhai') {
// Rhai 节点 -> 前端 code，设置 language=rhai
const data = n.data || {};
const script = data.script || {};
script.language = 'rhai';
return { ...n, type: 'code', data: { ...data, script } };
}
return n;
});

43
frontend/src/flows/nodes/code/components/code.tsx
View File

@ -5,7 +5,7 @@
import { Field } from '@flowgram.ai/free-layout-editor';
import { CodeEditor } from '@flowgram.ai/form-materials';
import { Divider } from '@douyinfe/semi-ui';
import { Divider, Select, Typography } from '@douyinfe/semi-ui';
import { useIsSidebar, useNodeRenderContext } from '../../../hooks';
import { FormItem } from '../../../form-components';
@ -21,14 +21,39 @@ export function Code() {
return (
<>
<Divider />
<Field<string> name="script.content">
{({ field }) => (
<CodeEditor
languageId="typescript"
value={field.value}
onChange={(value) => field.onChange(value)}
readonly={readonly}
/>
{/* Language selector + editor */}
<Field<string> name="script.language">
{({ field: langField }) => (
<>
<FormItem name={<Typography.Text strong>Language</Typography.Text> as any} vertical>
<Select
value={langField.value || 'javascript'}
disabled={readonly}
onChange={(value) => langField.onChange(value as string)}
style={{ width: '100%' }}
optionList={[
{ label: 'JavaScript', value: 'javascript' },
{ label: 'Rhai', value: 'rhai' },
]}
/>
</FormItem>
<Field<string> name="script.content">
{({ field }) => {
const lang = langField.value || 'javascript';
const languageId: 'typescript' | 'python' | 'shell' | 'json' =
lang === 'javascript' ? 'typescript' : 'shell';
return (
<CodeEditor
languageId={languageId}
value={field.value}
onChange={(value) => field.onChange(value)}
readonly={readonly}
/>
);
}}
</Field>
</>
)}
</Field>
</>

83
frontend/src/flows/services/custom-service.ts
View File

@ -40,13 +40,18 @@ function getFlowIdFromUrl(): string {
}
// 新增：针对后端的 design_json 兼容性转换
// - 将前端 UI 的 type: 'code' 映射为后端可识别并映射到 script_js 执行器的 'javascript'
// - 将前端 UI 的 type: 'code' 按语言映射：javascript -> 'javascript'，rhai -> 'script'（后端映射到 script_rhai）
// - 其余字段保持不变
function transformDesignJsonForBackend(json: any): any {
try {
const clone = JSON.parse(JSON.stringify(json));
clone.nodes = (clone.nodes || []).map((n: any) => {
if (n && n.type === 'code') {
const lang = n?.data?.script?.language;
if (lang === 'rhai') {
return { ...n, type: 'script' };
}
// 默认或显式 javascript
return { ...n, type: 'javascript' };
}
return n;
@ -78,7 +83,7 @@ export class CustomService {
}
const json = this.document.toJSON() as any;
const yaml = stringifyFlowDoc(json);
// 使用转换后的 design_json，以便后端将 code 节点识别为 javascript 并选择 script_js 执行器
// 使用转换后的 design_json，以便后端根据语言选择正确的执行器
const designForBackend = transformDesignJsonForBackend(json);
const design_json = JSON.stringify(designForBackend);
const { data } = await api.put<ApiResp<{ saved: boolean }>>(`/flows/${id}`, { yaml, design_json });
@ -219,23 +224,24 @@ export class CustomService {
ws.onerror = (ev: Event) => {
if (!finished) {
handlers?.onFatal?.(new Error('WebSocket error'));
rejectDone(new Error('WebSocket error'));
}
};
ws.onclose = () => {
if (!finished) {
resolveDone(null);
handlers?.onFatal?.(new Error('WebSocket closed'));
rejectDone(new Error('WebSocket closed'));
}
};
const cancel = () => {
try { finished = true; ws?.close(); } catch {}
};
return { cancel, done } as const;
return {
cancel: () => { try { ws?.close(); } catch {} },
done,
} as const;
}
// 现有：SSE 流式运行
// SSE 方案保留
runStream(input: any = {}, handlers?: { onNode?: (e: StreamEvent & { type: 'node' }) => void; onDone?: (e: StreamEvent & { type: 'done' }) => void; onError?: (e: StreamEvent & { type: 'error' }) => void; onFatal?: (err: Error) => void; }) {
const id = getFlowIdFromUrl();
if (!id) {
@ -244,55 +250,20 @@ export class CustomService {
return { cancel: () => {}, done: Promise.resolve<RunResult | null>(null) } as const;
}
// 在开发环境通过 Vite 代理前缀 /sse 转发到 8866（vite.config.ts 已配置 rewrite 到 /api）
const useSseProxy = !!((import.meta as any).env?.DEV);
let url: string;
if (useSseProxy) {
url = `/sse/flows/${id}/run/stream`;
} else {
const base = (api.defaults.baseURL || '') as string;
// 参照 WS：SSE 使用独立端口，默认 8866，可通过 VITE_SSE_PORT 覆盖
const ssePort = (import.meta as any).env?.VITE_SSE_PORT || '8866';
let sseBase: string;
if (base.startsWith('http://') || base.startsWith('https://')) {
const res = postSSE(`/flows/${id}/run/stream`, { input }, {
onMessage: (evt: StreamEvent) => {
try {
const u = new URL(base);
// 协议保持与 base 一致，仅替换端口
u.port = ssePort;
sseBase = `${u.protocol}//${u.host}${u.pathname.replace(/\/$/, '')}`;
} catch {
const loc = window.location;
sseBase = `${loc.protocol}//${loc.hostname}:${ssePort}${base.startsWith('/') ? base : '/' + base}`;
const e = evt;
if (e.type === 'node') { handlers?.onNode?.(e as any); return; }
if (e.type === 'error') { handlers?.onError?.(e as any); return; }
if (e.type === 'done') { handlers?.onDone?.(e as any); return; }
} catch (err: unknown) {
// ignore
}
} else {
const loc = window.location;
sseBase = `${loc.protocol}//${loc.hostname}:${ssePort}${base.startsWith('/') ? base : '/' + base}`;
}
url = sseBase + `/flows/${id}/run/stream`;
}
const { cancel, done } = postSSE<RunResult | null>(url, { input }, {
onMessage: (json: any) => {
try {
const evt = json as StreamEvent
if (evt.type === 'node') {
handlers?.onNode?.(evt as any)
return undefined
}
if (evt.type === 'error') {
handlers?.onError?.(evt as any)
return undefined
}
if (evt.type === 'done') {
handlers?.onDone?.(evt as any)
return { ok: (evt as any).ok, ctx: (evt as any).ctx, logs: (evt as any).logs }
}
} catch (_) {}
return undefined
},
onFatal: (e: any) => handlers?.onFatal?.(e),
})
return { cancel, done } as const;
onFatal: (err: Error) => handlers?.onFatal?.(err),
});
return res;
}
}