boos/BoosAPI
2
0
Fork 2
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: polish workflow studio and workspace assets

Browse Source
This commit is contained in:
boos
2026-05-25 20:42:38 +08:00
parent cb35d5fc6e
commit 1ba293c262
29 changed files with 2299 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/workflow/executor.rs
+148 -8
View File
@@ -9,7 +9,7 @@ use crate::{
AppState,
};
use serde_json::{json, Map, Value};
use std::{collections::HashMap, sync::Arc};
use std::{collections::HashMap, sync::Arc, time::Duration};
#[derive(Debug, Clone)]
pub enum WorkflowProgressEvent {
@@ -20,16 +20,26 @@ pub enum WorkflowProgressEvent {
NodeStarted {
prompt_id: String,
node_id: String,
node_type: String,
index: usize,
total: usize,
},
NodeCompleted {
prompt_id: String,
node_id: String,
node_type: String,
index: usize,
total: usize,
output: Value,
},
NodeProgress {
prompt_id: String,
node_id: String,
node_type: String,
value: usize,
max: usize,
message: String,
},
Finished {
prompt_id: String,
success: bool,
@@ -86,6 +96,7 @@ pub async fn run_workflow_with_progress(
WorkflowProgressEvent::NodeStarted {
prompt_id: run_public_id.clone(),
node_id: node.id.clone(),
node_type: node.node_type.clone(),
index,
total,
},
@@ -94,7 +105,18 @@ pub async fn run_workflow_with_progress(
let node_run_id =
storage::insert_node_run_started(state, run_db_id, &node.id, &node.node_type, &inputs)
.await?;
match execute_node(state, &ctx, node, inputs.clone()).await {
match execute_node(
state,
&ctx,
node,
inputs.clone(),
progress.clone(),
run_public_id.clone(),
index,
total,
)
.await
{
Ok(result) => {
storage::complete_node_run(state, node_run_id, &result.outputs).await?;
emit_progress(
@@ -102,6 +124,7 @@ pub async fn run_workflow_with_progress(
WorkflowProgressEvent::NodeCompleted {
prompt_id: run_public_id.clone(),
node_id: node.id.clone(),
node_type: node.node_type.clone(),
index: index + 1,
total,
output: result.outputs.clone(),
@@ -183,6 +206,10 @@ async fn execute_node(
ctx: &RunContext,
node: &WorkflowNode,
inputs: Value,
progress: Option<WorkflowProgressCallback>,
prompt_id: String,
index: usize,
total: usize,
) -> AppResult<NodeExecution> {
match node.node_type.as_str() {
"input.text" => {
@@ -202,8 +229,12 @@ async fn execute_node(
"comfyui.submit_workflow" => submit_comfyui_workflow(state, node, inputs).await,
"boosapi.prompt_rewrite" => chat_completion(state, ctx, node, inputs, true).await,
"boosapi.text_to_image" => text_to_image(state, ctx, node, inputs).await,
"boosapi.text_to_video" => text_to_video(ctx, node, inputs).await,
"boosapi.image_to_video" => image_to_video(ctx, node, inputs).await,
"boosapi.text_to_video" => {
text_to_video(ctx, node, inputs, progress, prompt_id, index, total).await
}
"boosapi.image_to_video" => {
image_to_video(ctx, node, inputs, progress, prompt_id, index, total).await
}
"boosapi.video_status" => video_status(ctx, node, inputs).await,
"CheckpointLoaderSimple" => {
let model = param_str(node, "ckpt_name", "gpt-image-2");
@@ -542,6 +573,10 @@ async fn text_to_video(
ctx: &RunContext,
node: &WorkflowNode,
inputs: Value,
progress: Option<WorkflowProgressCallback>,
prompt_id: String,
index: usize,
total: usize,
) -> AppResult<NodeExecution> {
let prompt = value_to_string(inputs.get("prompt"));
if prompt.trim().is_empty() {
@@ -551,13 +586,17 @@ async fn text_to_video(
)));
}
let body = video_request_body(node, prompt, None);
execute_video_generation(ctx, inputs, body).await
execute_video_generation(ctx, node, inputs, body, progress, prompt_id, index, total).await
}
async fn image_to_video(
ctx: &RunContext,
node: &WorkflowNode,
inputs: Value,
progress: Option<WorkflowProgressCallback>,
prompt_id: String,
index: usize,
total: usize,
) -> AppResult<NodeExecution> {
let prompt = value_to_string(inputs.get("prompt"));
if prompt.trim().is_empty() {
@@ -573,7 +612,7 @@ async fn image_to_video(
Some(image_url)
};
let body = video_request_body(node, prompt, image_url);
execute_video_generation(ctx, inputs, body).await
execute_video_generation(ctx, node, inputs, body, progress, prompt_id, index, total).await
}
async fn video_status(
@@ -627,22 +666,94 @@ fn video_request_body(node: &WorkflowNode, prompt: String, image_url: Option<Str
async fn execute_video_generation(
ctx: &RunContext,
node: &WorkflowNode,
inputs: Value,
body: Value,
progress: Option<WorkflowProgressCallback>,
prompt_id: String,
index: usize,
total: usize,
) -> AppResult<NodeExecution> {
emit_node_progress(
&progress,
&prompt_id,
node,
index,
total,
"正在创建视频任务...",
);
let raw = post_json(ctx, "/videos", &body).await?;
let task_id = extract_task_id(&raw);
let video = video_output_from_response(&raw);
let mut final_raw = raw.clone();
let mut video = video_output_from_response(&raw);
if video.is_null() && !task_id.is_null() && param_bool(node, "wait_for_result", true) {
let model = body
.get("model")
.and_then(Value::as_str)
.unwrap_or("seedance-1-0-lite-t2v");
let interval = param_i64(node, "poll_interval_secs", 5).clamp(1, 30) as u64;
let timeout = param_i64(node, "timeout_secs", 180).clamp(10, 1800) as u64;
let task = task_id.as_str().unwrap_or_default();
let started = std::time::Instant::now();
let mut attempt = 0_usize;
let max_attempts = ((timeout + interval - 1) / interval).max(1) as usize;
while started.elapsed() < Duration::from_secs(timeout) {
attempt += 1;
emit_node_progress(
&progress,
&prompt_id,
node,
index,
total,
&format!("视频任务已创建，正在查询结果 {attempt}/{max_attempts}..."),
);
tokio::time::sleep(Duration::from_secs(interval)).await;
let status_path = format!(
"/videos/{}?model={}",
simple_urlencode(task),
simple_urlencode(model)
);
final_raw = get_json(ctx, &status_path).await?;
video = video_output_from_response(&final_raw);
if !video.is_null() || video_task_is_terminal_failure(&final_raw) {
break;
}
}
}
Ok(NodeExecution {
inputs,
outputs: json!({
"task_id": task_id,
"video": video,
"raw": raw
"raw": final_raw,
"create_response": raw
}),
})
}
fn emit_node_progress(
progress: &Option<WorkflowProgressCallback>,
prompt_id: &str,
node: &WorkflowNode,
index: usize,
total: usize,
message: &str,
) {
let max = total.saturating_mul(10).max(1);
let value = index.saturating_mul(10).saturating_add(4).min(max);
emit_progress(
progress,
WorkflowProgressEvent::NodeProgress {
prompt_id: prompt_id.to_string(),
node_id: node.id.clone(),
node_type: node.node_type.clone(),
value,
max,
message: message.to_string(),
},
);
}
async fn vae_decode_to_image(
state: &AppState,
ctx: &RunContext,
@@ -804,11 +915,40 @@ fn video_output_from_response(value: &Value) -> Value {
}
}
fn video_task_is_terminal_failure(value: &Value) -> bool {
let status = find_string_key(value, &["status", "state", "task_status", "taskStatus"])
.unwrap_or_default()
.to_ascii_lowercase();
matches!(
status.as_str(),
"failed" | "fail" | "error" | "cancelled" | "canceled"
)
}
fn extract_media_url(value: Option<&Value>) -> Option<String> {
let value = value?;
find_url_key(value, &["url", "remote_url", "remoteUrl", "image_url", "imageUrl"])
}
fn find_string_key(value: &Value, keys: &[&str]) -> Option<String> {
match value {
Value::Object(map) => {
for key in keys {
if let Some(text) = map
.get(*key)
.and_then(Value::as_str)
.filter(|text| !text.trim().is_empty())
{
return Some(text.trim().to_string());
}
}
map.values().find_map(|child| find_string_key(child, keys))
}
Value::Array(items) => items.iter().find_map(|child| find_string_key(child, keys)),
_ => None,
}
}
fn find_url_key(value: &Value, keys: &[&str]) -> Option<String> {
match value {
Value::Object(map) => {
src/workflow/models.rs
+1
View File
@@ -53,6 +53,7 @@ pub struct WorkflowRunSummary {
pub public_id: String,
pub workflow_id: i64,
pub workflow_public_id: Option<String>,
pub workflow_name: Option<String>,
pub user_id: Option<i64>,
pub status: String,
pub input_overrides: Value,
src/workflow/registry.rs
+8 -2
View File
@@ -186,7 +186,10 @@ pub fn node_types() -> Vec<NodeTypeDefinition> {
"ratio": { "ui": "text", "type": "STRING", "default": "16:9" },
"duration": { "ui": "number", "type": "NUMBER", "default": 5, "min": 1, "max": 30 },
"resolution": { "ui": "text", "type": "STRING", "default": "720p" },
"fps": { "ui": "number", "type": "NUMBER", "default": 24, "min": 1, "max": 60 }
"fps": { "ui": "number", "type": "NUMBER", "default": 24, "min": 1, "max": 60 },
"wait_for_result": { "ui": "boolean", "type": "BOOLEAN", "default": true },
"poll_interval_secs": { "ui": "number", "type": "NUMBER", "default": 5, "min": 1, "max": 30 },
"timeout_secs": { "ui": "number", "type": "NUMBER", "default": 180, "min": 10, "max": 1800 }
}),
outputs: json!({
"task_id": { "type": "STRING" },
@@ -209,7 +212,10 @@ pub fn node_types() -> Vec<NodeTypeDefinition> {
"ratio": { "ui": "text", "type": "STRING", "default": "16:9" },
"duration": { "ui": "number", "type": "NUMBER", "default": 5, "min": 1, "max": 30 },
"resolution": { "ui": "text", "type": "STRING", "default": "720p" },
"fps": { "ui": "number", "type": "NUMBER", "default": 24, "min": 1, "max": 60 }
"fps": { "ui": "number", "type": "NUMBER", "default": 24, "min": 1, "max": 60 },
"wait_for_result": { "ui": "boolean", "type": "BOOLEAN", "default": true },
"poll_interval_secs": { "ui": "number", "type": "NUMBER", "default": 5, "min": 1, "max": 30 },
"timeout_secs": { "ui": "number", "type": "NUMBER", "default": 180, "min": 10, "max": 1800 }
}),
outputs: json!({
"task_id": { "type": "STRING" },
src/workflow/storage.rs
+3 -2
View File
@@ -69,6 +69,7 @@ fn row_to_run(
public_id: row.try_get("public_id").unwrap_or_default(),
workflow_id: row.try_get("workflow_id").unwrap_or_default(),
workflow_public_id: row.try_get("workflow_public_id").ok(),
workflow_name: row.try_get("workflow_name").ok(),
user_id: row.try_get::<Option<i64>, _>("user_id").ok().flatten(),
status: row.try_get("status").unwrap_or_default(),
input_overrides: parse_json_text(row.try_get("input_overrides").ok(), json!({})),
@@ -149,7 +150,7 @@ pub async fn get_run(
allow_admin: bool,
) -> AppResult<WorkflowRunSummary> {
let row = sqlx::query(
"SELECT workflow_runs.*, workflows.public_id AS workflow_public_id
"SELECT workflow_runs.*, workflows.public_id AS workflow_public_id, workflows.name AS workflow_name
FROM workflow_runs
JOIN workflows ON workflows.id = workflow_runs.workflow_id
WHERE workflow_runs.public_id = ?
@@ -183,7 +184,7 @@ pub async fn list_runs(
limit: i64,
) -> AppResult<Vec<WorkflowRunSummary>> {
let rows = sqlx::query(
"SELECT workflow_runs.*, workflows.public_id AS workflow_public_id
"SELECT workflow_runs.*, workflows.public_id AS workflow_public_id, workflows.name AS workflow_name
FROM workflow_runs
JOIN workflows ON workflows.id = workflow_runs.workflow_id
WHERE (? = 1 OR workflow_runs.user_id = ?)