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

refactor(workflow_ui): 抽出 run_convert 子模块

Browse Source 
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
boos
2026-06-18 22:32:51 +08:00
parent a37546cb2d
commit f7f9a9ca09
2 changed files with 295 additions and 285 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/routes/workflow_ui/mod.rs
+8 -285
View File
@@ -2,7 +2,7 @@ use crate::workflow::executor::{WorkflowProgressCallback, WorkflowProgressEvent}
use crate::{
error::{AppError, AppResult},
workflow::{
models::{RunContext, WorkflowGraph},
models::RunContext,
},
AppState,
};
@@ -19,7 +19,7 @@ use axum::{
use include_dir::{include_dir, Dir};
use once_cell::sync::Lazy;
use serde::{Deserialize, Serialize};
use serde_json::{json, Map, Value};
use serde_json::{json, Value};
use std::{
collections::{HashMap, HashSet},
path::{Path as StdPath, PathBuf},
@@ -161,6 +161,11 @@ use native_proxy::{
native_workflow_response, native_workflow_status_ttl, native_workflow_video_response,
native_workflow_view_response, timestamp_ms, upload_image, upload_mask, workflow_cache_status,
};
mod run_convert;
use run_convert::{
node_output_for_comfy, run_outputs_for_comfy, workflow_graph_to_comfy_workflow,
workflow_run_to_history, workflow_run_to_history_task, workflow_run_to_queue_task,
};
pub fn routes() -> Router<AppState> {
Router::new()
@@ -592,48 +597,6 @@ pub(super) fn workflow_validation_hints(errors: usize, warnings: usize) -> Vec<S
hints
}
pub(super) fn workflow_run_to_history(run: crate::workflow::models::WorkflowRunSummary) -> Value {
let outputs = run_outputs_for_comfy(&run);
let status_str = if run.status == "failed" {
"error"
} else {
"success"
};
json!({
"prompt": workflow_run_to_prompt_tuple(&run),
"status": {
"status_str": status_str,
"completed": run.status != "running",
"messages": run_status_messages(&run)
},
"outputs": outputs,
"meta": {},
})
}
pub(super) fn workflow_run_to_history_task(run: crate::workflow::models::WorkflowRunSummary) -> Value {
let history = workflow_run_to_history(run);
json!({
"taskType": "History",
"prompt": history.get("prompt").cloned().unwrap_or_else(|| json!([])),
"status": history.get("status").cloned().unwrap_or_else(|| json!({})),
"outputs": history.get("outputs").cloned().unwrap_or_else(|| json!({})),
})
}
pub(super) fn workflow_run_to_queue_task(run: crate::workflow::models::WorkflowRunSummary) -> Value {
json!({
"taskType": "Running",
"prompt": workflow_run_to_prompt_tuple(&run),
"status": {
"status_str": "running",
"completed": false,
"messages": run_status_messages(&run)
},
"outputs": run_outputs_for_comfy(&run),
})
}
pub(super) fn prompt_error_response(message: impl Into<String>, details: impl Into<String>) -> Response {
let message = message.into();
let details = details.into();
@@ -697,246 +660,6 @@ pub(super) fn simple_urlencode(value: &str) -> String {
.replace('/', "%2F")
}
fn workflow_timestamp_ms(value: &str) -> i64 {
let trimmed = value.trim();
if trimmed.is_empty() {
return timestamp_ms();
}
chrono::DateTime::parse_from_rfc3339(trimmed)
.map(|dt| dt.timestamp_millis())
.or_else(|_| {
chrono::NaiveDateTime::parse_from_str(trimmed, "%Y-%m-%d %H:%M:%S")
.map(|dt| dt.and_utc().timestamp_millis())
})
.or_else(|_| {
chrono::NaiveDateTime::parse_from_str(trimmed, "%Y-%m-%dT%H:%M:%S")
.map(|dt| dt.and_utc().timestamp_millis())
})
.unwrap_or_else(|_| timestamp_ms())
}
pub(super) fn workflow_graph_to_comfy_workflow(graph: &WorkflowGraph) -> Value {
let nodes = graph
.nodes
.iter()
.map(|node| {
json!({
"id": node.id,
"type": node.node_type,
"pos": [
node.position.get("x").and_then(Value::as_f64).unwrap_or(80.0),
node.position.get("y").and_then(Value::as_f64).unwrap_or(120.0)
],
"widgets_values": node.params,
"properties": { "Node name for S&R": node.node_type },
})
})
.collect::<Vec<_>>();
json!({
"id": graph.name.clone().unwrap_or_else(|| "boosapi-workflow".to_string()),
"version": 0.4,
"nodes": nodes,
"links": [],
"groups": [],
"config": {},
"extra": {},
})
}
fn run_status_messages(run: &crate::workflow::models::WorkflowRunSummary) -> Vec<Value> {
let mut messages = vec![json!(["execution_start", {
"prompt_id": run.public_id,
"timestamp": workflow_timestamp_ms(run.started_at.as_deref().unwrap_or(&run.created_at))
}])];
if run.status == "running" {
let completed = run
.node_runs
.iter()
.filter(|node| node.status == "completed")
.count();
let total = run.node_runs.len().max(1);
messages.push(json!(["progress", {
"value": completed,
"max": total
}]));
if let Some(current) = run
.node_runs
.iter()
.find(|node| node.status == "running")
.or_else(|| run.node_runs.last())
{
let label = workflow_node_label(&current.node_type, &current.node_id);
messages.push(json!(["executing", {
"prompt_id": run.public_id,
"node": current.node_id,
"node_type": current.node_type,
"display_node": label,
"label": label,
}]));
}
} else if run.status == "failed" {
let failed_node = run
.node_runs
.iter()
.find(|node| node.status == "failed")
.or_else(|| run.node_runs.last());
messages.push(json!(["execution_error", {
"prompt_id": run.public_id,
"timestamp": workflow_timestamp_ms(run.completed_at.as_deref().unwrap_or(&run.created_at)),
"node_id": failed_node.map(|node| node.node_id.as_str()).unwrap_or("workflow"),
"node_type": failed_node.map(|node| node.node_type.as_str()).unwrap_or("Workflow"),
"executed": run.node_runs.iter().filter(|node| node.status == "completed").map(|node| node.node_id.clone()).collect::<Vec<_>>(),
"exception_message": run.error_message.clone().unwrap_or_else(|| "Workflow execution failed".to_string()),
"exception_type": "BoosAPIWorkflowError",
"traceback": [],
"current_inputs": failed_node.map(|node| node.inputs.clone()).unwrap_or_else(|| json!({})),
"current_outputs": failed_node.and_then(|node| node.outputs.clone()).unwrap_or_else(|| json!({})),
}]));
} else {
messages.push(json!(["execution_success", {
"prompt_id": run.public_id,
"timestamp": workflow_timestamp_ms(run.completed_at.as_deref().unwrap_or(&run.created_at))
}]));
}
messages
}
pub(super) fn run_outputs_for_comfy(run: &crate::workflow::models::WorkflowRunSummary) -> Value {
let mut outputs = Map::new();
for node in &run.node_runs {
if let Some(node_outputs) = &node.outputs {
let media = node_output_for_comfy(node_outputs)
.unwrap_or_else(|| json!({ "text": [workflow_text_output(node_outputs)] }));
outputs.insert(node.node_id.clone(), media);
}
}
Value::Object(outputs)
}
fn node_output_for_comfy(node_outputs: &Value) -> Option<Value> {
let mut media = Map::new();
if let Some(videos) = comfy_media_items(node_outputs, "comfy_videos")
.or_else(|| comfy_media_items(node_outputs, "videos"))
.or_else(|| comfy_media_items(node_outputs, "video"))
{
media.insert("videos".to_string(), videos);
}
if let Some(images) = comfy_media_items(node_outputs, "comfy_images")
.or_else(|| comfy_media_items(node_outputs, "image"))
.or_else(|| comfy_media_items(node_outputs, "images"))
.or_else(|| comfy_media_items(node_outputs, "IMAGE"))
{
media.insert("images".to_string(), images);
}
if media.is_empty() {
None
} else {
Some(Value::Object(media))
}
}
fn comfy_media_items(node_outputs: &Value, key: &str) -> Option<Value> {
let value = node_outputs.get(key)?;
let items = if let Some(array) = value.as_array() {
array.clone()
} else {
vec![value.clone()]
};
let normalized = items
.into_iter()
.filter_map(normalize_comfy_media_item)
.collect::<Vec<_>>();
if normalized.is_empty() {
None
} else {
Some(Value::Array(normalized))
}
}
fn normalize_comfy_media_item(value: Value) -> Option<Value> {
let object = value.as_object()?;
let filename = object
.get("filename")
.and_then(Value::as_str)
.or_else(|| object.get("public_id").and_then(Value::as_str))
.or_else(|| object.get("publicId").and_then(Value::as_str))
.or_else(|| object.get("file_name").and_then(Value::as_str))
.or_else(|| object.get("fileName").and_then(Value::as_str))
.filter(|value| !value.trim().is_empty())?;
let file_type = object
.get("type")
.and_then(Value::as_str)
.filter(|value| !value.trim().is_empty())
.unwrap_or("output");
let subfolder = object
.get("subfolder")
.and_then(Value::as_str)
.unwrap_or("");
let mut normalized = Map::new();
normalized.insert("filename".to_string(), json!(filename));
normalized.insert("subfolder".to_string(), json!(subfolder));
normalized.insert("type".to_string(), json!(file_type));
if let Some(format) = object.get("format").and_then(Value::as_str) {
normalized.insert("format".to_string(), json!(format));
}
if let Some(frame_rate) = object.get("frame_rate").or_else(|| object.get("frameRate")) {
normalized.insert("frame_rate".to_string(), frame_rate.clone());
}
Some(Value::Object(normalized))
}
fn workflow_text_output(value: &Value) -> Value {
for key in ["text", "prompt", "task_id", "file"] {
if let Some(text) = value.get(key).and_then(Value::as_str) {
if !text.trim().is_empty() {
return json!({
"filename": text.chars().take(80).collect::<String>(),
"subfolder": "",
"type": "text",
});
}
}
}
json!({
"filename": "workflow-result",
"subfolder": "",
"type": "text",
})
}
fn workflow_run_to_prompt_tuple(run: &crate::workflow::models::WorkflowRunSummary) -> Value {
let workflow_name = run
.workflow_name
.as_deref()
.or(run.workflow_public_id.as_deref())
.unwrap_or("BoosAPI workflow");
let workflow = json!({ "name": workflow_name, "nodes": [], "links": [] });
let prompt = run
.node_runs
.iter()
.map(|node| {
(
node.node_id.clone(),
json!({
"inputs": node.inputs,
"class_type": node.node_type,
}),
)
})
.collect::<Map<_, _>>();
json!([
run.id,
run.public_id,
prompt,
{
"extra_pnginfo": { "workflow": workflow },
"client_id": "boosapi-workflow-studio"
},
run.node_runs.iter().map(|node| node.node_id.clone()).collect::<Vec<_>>()
])
}
fn workflow_session_sender(session_id: &str) -> broadcast::Sender<Value> {
let mut sessions = WORKFLOW_WS_SESSIONS.lock();
if let Some(sender) = sessions.get(session_id) {
@@ -958,7 +681,7 @@ fn publish_workflow_ws_event(session_id: Option<&str>, event_type: &str, data: V
}));
}
fn workflow_node_label(node_type: &str, node_id: &str) -> String {
pub(super) fn workflow_node_label(node_type: &str, node_id: &str) -> String {
if node_type.trim().is_empty() {
node_id.to_string()
} else {
src/routes/workflow_ui/run_convert.rs
+287
View File
@@ -0,0 +1,287 @@
use crate::workflow::models::WorkflowGraph;
use serde_json::{json, Map, Value};
use super::native_proxy::timestamp_ms;
use super::workflow_node_label;
pub(super) fn workflow_run_to_history(run: crate::workflow::models::WorkflowRunSummary) -> Value {
let outputs = run_outputs_for_comfy(&run);
let status_str = if run.status == "failed" {
"error"
} else {
"success"
};
json!({
"prompt": workflow_run_to_prompt_tuple(&run),
"status": {
"status_str": status_str,
"completed": run.status != "running",
"messages": run_status_messages(&run)
},
"outputs": outputs,
"meta": {},
})
}
pub(super) fn workflow_run_to_history_task(run: crate::workflow::models::WorkflowRunSummary) -> Value {
let history = workflow_run_to_history(run);
json!({
"taskType": "History",
"prompt": history.get("prompt").cloned().unwrap_or_else(|| json!([])),
"status": history.get("status").cloned().unwrap_or_else(|| json!({})),
"outputs": history.get("outputs").cloned().unwrap_or_else(|| json!({})),
})
}
pub(super) fn workflow_run_to_queue_task(run: crate::workflow::models::WorkflowRunSummary) -> Value {
json!({
"taskType": "Running",
"prompt": workflow_run_to_prompt_tuple(&run),
"status": {
"status_str": "running",
"completed": false,
"messages": run_status_messages(&run)
},
"outputs": run_outputs_for_comfy(&run),
})
}
fn workflow_timestamp_ms(value: &str) -> i64 {
let trimmed = value.trim();
if trimmed.is_empty() {
return timestamp_ms();
}
chrono::DateTime::parse_from_rfc3339(trimmed)
.map(|dt| dt.timestamp_millis())
.or_else(|_| {
chrono::NaiveDateTime::parse_from_str(trimmed, "%Y-%m-%d %H:%M:%S")
.map(|dt| dt.and_utc().timestamp_millis())
})
.or_else(|_| {
chrono::NaiveDateTime::parse_from_str(trimmed, "%Y-%m-%dT%H:%M:%S")
.map(|dt| dt.and_utc().timestamp_millis())
})
.unwrap_or_else(|_| timestamp_ms())
}
pub(super) fn workflow_graph_to_comfy_workflow(graph: &WorkflowGraph) -> Value {
let nodes = graph
.nodes
.iter()
.map(|node| {
json!({
"id": node.id,
"type": node.node_type,
"pos": [
node.position.get("x").and_then(Value::as_f64).unwrap_or(80.0),
node.position.get("y").and_then(Value::as_f64).unwrap_or(120.0)
],
"widgets_values": node.params,
"properties": { "Node name for S&R": node.node_type },
})
})
.collect::<Vec<_>>();
json!({
"id": graph.name.clone().unwrap_or_else(|| "boosapi-workflow".to_string()),
"version": 0.4,
"nodes": nodes,
"links": [],
"groups": [],
"config": {},
"extra": {},
})
}
fn run_status_messages(run: &crate::workflow::models::WorkflowRunSummary) -> Vec<Value> {
let mut messages = vec![json!(["execution_start", {
"prompt_id": run.public_id,
"timestamp": workflow_timestamp_ms(run.started_at.as_deref().unwrap_or(&run.created_at))
}])];
if run.status == "running" {
let completed = run
.node_runs
.iter()
.filter(|node| node.status == "completed")
.count();
let total = run.node_runs.len().max(1);
messages.push(json!(["progress", {
"value": completed,
"max": total
}]));
if let Some(current) = run
.node_runs
.iter()
.find(|node| node.status == "running")
.or_else(|| run.node_runs.last())
{
let label = workflow_node_label(&current.node_type, &current.node_id);
messages.push(json!(["executing", {
"prompt_id": run.public_id,
"node": current.node_id,
"node_type": current.node_type,
"display_node": label,
"label": label,
}]));
}
} else if run.status == "failed" {
let failed_node = run
.node_runs
.iter()
.find(|node| node.status == "failed")
.or_else(|| run.node_runs.last());
messages.push(json!(["execution_error", {
"prompt_id": run.public_id,
"timestamp": workflow_timestamp_ms(run.completed_at.as_deref().unwrap_or(&run.created_at)),
"node_id": failed_node.map(|node| node.node_id.as_str()).unwrap_or("workflow"),
"node_type": failed_node.map(|node| node.node_type.as_str()).unwrap_or("Workflow"),
"executed": run.node_runs.iter().filter(|node| node.status == "completed").map(|node| node.node_id.clone()).collect::<Vec<_>>(),
"exception_message": run.error_message.clone().unwrap_or_else(|| "Workflow execution failed".to_string()),
"exception_type": "BoosAPIWorkflowError",
"traceback": [],
"current_inputs": failed_node.map(|node| node.inputs.clone()).unwrap_or_else(|| json!({})),
"current_outputs": failed_node.and_then(|node| node.outputs.clone()).unwrap_or_else(|| json!({})),
}]));
} else {
messages.push(json!(["execution_success", {
"prompt_id": run.public_id,
"timestamp": workflow_timestamp_ms(run.completed_at.as_deref().unwrap_or(&run.created_at))
}]));
}
messages
}
pub(super) fn run_outputs_for_comfy(run: &crate::workflow::models::WorkflowRunSummary) -> Value {
let mut outputs = Map::new();
for node in &run.node_runs {
if let Some(node_outputs) = &node.outputs {
let media = node_output_for_comfy(node_outputs)
.unwrap_or_else(|| json!({ "text": [workflow_text_output(node_outputs)] }));
outputs.insert(node.node_id.clone(), media);
}
}
Value::Object(outputs)
}
pub(super) fn node_output_for_comfy(node_outputs: &Value) -> Option<Value> {
let mut media = Map::new();
if let Some(videos) = comfy_media_items(node_outputs, "comfy_videos")
.or_else(|| comfy_media_items(node_outputs, "videos"))
.or_else(|| comfy_media_items(node_outputs, "video"))
{
media.insert("videos".to_string(), videos);
}
if let Some(images) = comfy_media_items(node_outputs, "comfy_images")
.or_else(|| comfy_media_items(node_outputs, "image"))
.or_else(|| comfy_media_items(node_outputs, "images"))
.or_else(|| comfy_media_items(node_outputs, "IMAGE"))
{
media.insert("images".to_string(), images);
}
if media.is_empty() {
None
} else {
Some(Value::Object(media))
}
}
fn comfy_media_items(node_outputs: &Value, key: &str) -> Option<Value> {
let value = node_outputs.get(key)?;
let items = if let Some(array) = value.as_array() {
array.clone()
} else {
vec![value.clone()]
};
let normalized = items
.into_iter()
.filter_map(normalize_comfy_media_item)
.collect::<Vec<_>>();
if normalized.is_empty() {
None
} else {
Some(Value::Array(normalized))
}
}
fn normalize_comfy_media_item(value: Value) -> Option<Value> {
let object = value.as_object()?;
let filename = object
.get("filename")
.and_then(Value::as_str)
.or_else(|| object.get("public_id").and_then(Value::as_str))
.or_else(|| object.get("publicId").and_then(Value::as_str))
.or_else(|| object.get("file_name").and_then(Value::as_str))
.or_else(|| object.get("fileName").and_then(Value::as_str))
.filter(|value| !value.trim().is_empty())?;
let file_type = object
.get("type")
.and_then(Value::as_str)
.filter(|value| !value.trim().is_empty())
.unwrap_or("output");
let subfolder = object
.get("subfolder")
.and_then(Value::as_str)
.unwrap_or("");
let mut normalized = Map::new();
normalized.insert("filename".to_string(), json!(filename));
normalized.insert("subfolder".to_string(), json!(subfolder));
normalized.insert("type".to_string(), json!(file_type));
if let Some(format) = object.get("format").and_then(Value::as_str) {
normalized.insert("format".to_string(), json!(format));
}
if let Some(frame_rate) = object.get("frame_rate").or_else(|| object.get("frameRate")) {
normalized.insert("frame_rate".to_string(), frame_rate.clone());
}
Some(Value::Object(normalized))
}
fn workflow_text_output(value: &Value) -> Value {
for key in ["text", "prompt", "task_id", "file"] {
if let Some(text) = value.get(key).and_then(Value::as_str) {
if !text.trim().is_empty() {
return json!({
"filename": text.chars().take(80).collect::<String>(),
"subfolder": "",
"type": "text",
});
}
}
}
json!({
"filename": "workflow-result",
"subfolder": "",
"type": "text",
})
}
fn workflow_run_to_prompt_tuple(run: &crate::workflow::models::WorkflowRunSummary) -> Value {
let workflow_name = run
.workflow_name
.as_deref()
.or(run.workflow_public_id.as_deref())
.unwrap_or("BoosAPI workflow");
let workflow = json!({ "name": workflow_name, "nodes": [], "links": [] });
let prompt = run
.node_runs
.iter()
.map(|node| {
(
node.node_id.clone(),
json!({
"inputs": node.inputs,
"class_type": node.node_type,
}),
)
})
.collect::<Map<_, _>>();
json!([
run.id,
run.public_id,
prompt,
{
"extra_pnginfo": { "workflow": workflow },
"client_id": "boosapi-workflow-studio"
},
run.node_runs.iter().map(|node| node.node_id.clone()).collect::<Vec<_>>()
])
}