LaPaoPao/BoosAPI
1
0
Fork 0
forked from boos/BoosAPI
Code Pull Requests 1 Activity

chore: prune legacy node backend and mix workflow models

Browse Source
This commit is contained in:
boos
2026-05-26 15:18:33 +08:00
parent 6869504119
commit 0f9fe4a81d
32 changed files with 525 additions and 5019 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/routes/comfyui/mod.rs
+3 -1
View File
@@ -200,7 +200,9 @@ async fn proxy_ws_bridge(mut client_socket: axum::extract::ws::WebSocket, target
Ok(request) => request,
Err(err) => {
let _ = client_socket
.send(Message::Text(format!("invalid ComfyUI websocket URL: {err}").into()))
.send(Message::Text(
format!("invalid ComfyUI websocket URL: {err}").into(),
))
.await;
return;
}
src/routes/comfyui_agent.rs
+2 -1
View File
@@ -701,7 +701,8 @@ fn attachment_summary(attachment: &AgentAttachment) -> String {
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or("file");
let header = format!("\n\n[素材引用: {reference}, 名称: {name}, 类型: {mime}, 分类: {media_kind}]");
let header =
format!("\n\n[素材引用: {reference}, 名称: {name}, 类型: {mime}, 分类: {media_kind}]");
if let Some(text) = attachment
.text
src/routes/digital_human.rs
+7 -1
View File
@@ -118,7 +118,13 @@ async fn generate(
Json(mut body): Json<Value>,
) -> AppResult<Json<Value>> {
let auth = resolve_auth(&headers, &state).await?;
if body.get("model").and_then(Value::as_str).unwrap_or("").trim().is_empty() {
if body
.get("model")
.and_then(Value::as_str)
.unwrap_or("")
.trim()
.is_empty()
{
return Err(AppError::BadRequest("model is required".into()));
}
if body
src/routes/workflow_ui.rs
+165 -50
View File
@@ -1,3 +1,4 @@
use crate::workflow::executor::{WorkflowProgressCallback, WorkflowProgressEvent};
use crate::{
auth::jwt::validate_jwt,
error::{AppError, AppResult},
@@ -9,7 +10,6 @@ use crate::{
},
AppState,
};
use crate::workflow::executor::{WorkflowProgressCallback, WorkflowProgressEvent};
use axum::{
body::Bytes,
extract::{
@@ -27,7 +27,7 @@ use serde::{Deserialize, Serialize};
use serde_json::{json, Map, Value};
use sqlx::Row;
use std::{
collections::HashMap,
collections::{HashMap, HashSet},
path::PathBuf,
sync::{mpsc, Arc},
time::{SystemTime, UNIX_EPOCH},
@@ -138,7 +138,10 @@ pub fn routes() -> Router<AppState> {
)
.route("/workflow-ui/api/{*path}", any(native_workflow_api_proxy))
.route("/workflow-ui/api/queue", get(queue).post(queue_mutate))
.route("/workflow-ui/api/history", get(history).post(history_mutate))
.route(
"/workflow-ui/api/history",
get(history).post(history_mutate),
)
.route("/workflow-ui/api/history/{id}", get(history_by_id))
.route("/workflow-ui/api/interrupt", post(interrupt))
.route("/workflow-ui/api/view", get(view))
@@ -1039,6 +1042,84 @@ fn inject_model_choices(defs: &mut Map<String, Value>, models: &[String]) {
}
}
fn boosapi_node_defs_with_models(models: &[String]) -> Map<String, Value> {
let mut defs = core_node_definitions()
.into_iter()
.chain(registry::node_types())
.map(|def| (def.node_type.to_string(), node_type_def_to_comfy(&def)))
.collect::<Map<_, _>>();
inject_model_choices(&mut defs, models);
defs
}
fn inject_boosapi_object_info(value: &mut Value, models: &[String]) {
let Some(object) = value.as_object_mut() else {
return;
};
for (key, def) in boosapi_node_defs_with_models(models) {
object.insert(key, def);
}
}
fn builtin_model_folders() -> Vec<&'static str> {
vec![
"boosapi_models",
"checkpoints",
"chat_models",
"video_models",
"audio_tts_models",
"embeddings",
"vision_models",
"3d_models",
"loras",
"vae",
]
}
fn inject_boosapi_model_folders(value: &mut Value) {
let Some(items) = value.as_array_mut() else {
return;
};
let mut seen = items
.iter()
.filter_map(|item| {
item.get("name")
.and_then(Value::as_str)
.map(|name| name.to_ascii_lowercase())
})
.collect::<HashSet<_>>();
for name in builtin_model_folders() {
if seen.insert(name.to_ascii_lowercase()) {
items.push(json!({ "name": name }));
}
}
}
fn prompt_contains_boosapi_nodes(prompt: &Value) -> bool {
prompt
.as_object()
.map(|nodes| {
nodes.values().any(|node| {
node.get("class_type")
.and_then(Value::as_str)
.map(|node_type| {
node_type.starts_with("boosapi.")
|| matches!(
node_type,
"input.text"
| "output.preview_image"
| "output.preview_video"
| "output.save_file"
| "util.text_join"
| "util.json_extract"
)
})
.unwrap_or(false)
})
})
.unwrap_or(false)
}
async fn workflow_ui_index() -> Html<String> {
let dir = workflow_ui_root();
let index = dir.join("index.html");
@@ -1364,18 +1445,17 @@ async fn embeddings(State(state): State<AppState>, headers: HeaderMap) -> AppRes
async fn object_info(State(state): State<AppState>, headers: HeaderMap) -> AppResult<Json<Value>> {
let _ = auth_caller_optional(&state, &headers);
let route_models = workflow_route_models(&state).await.unwrap_or_default();
if state.config.comfyui.uses_native_backend() {
return native_workflow_json(&state, Method::GET, "/object_info", None).await;
let mut value = native_workflow_json(&state, Method::GET, "/object_info", None)
.await?
.0;
inject_boosapi_object_info(&mut value, &route_models);
return Ok(Json(value));
}
let mut defs = core_node_definitions()
.into_iter()
.chain(registry::node_types())
.into_iter()
.map(|def| (def.node_type.to_string(), node_type_def_to_comfy(&def)))
.collect::<serde_json::Map<_, _>>();
let models = workflow_route_models(&state).await.unwrap_or_default();
inject_model_choices(&mut defs, &models);
Ok(Json(Value::Object(defs)))
Ok(Json(Value::Object(boosapi_node_defs_with_models(
&route_models,
))))
}
async fn system_stats(State(state): State<AppState>, headers: HeaderMap) -> AppResult<Json<Value>> {
@@ -1420,8 +1500,13 @@ async fn queue(State(state): State<AppState>, headers: HeaderMap) -> AppResult<J
}
let caller =
auth_caller_optional(&state, &headers).unwrap_or(WorkflowCaller::User { user_id: 0 });
let rows = storage::list_runs(&state.db, caller_user_id(&caller), caller_is_admin(&caller), 25)
.await?;
let rows = storage::list_runs(
&state.db,
caller_user_id(&caller),
caller_is_admin(&caller),
25,
)
.await?;
let running = rows
.into_iter()
.filter(|run| run.status == "running")
@@ -1487,7 +1572,8 @@ async fn history_mutate(
return native_workflow_json(&state, Method::POST, "/history", Some(value)).await;
}
let affected = if payload.clear {
storage::clear_runs(&state, caller_user_id(&caller), caller_is_admin(&caller)).await? as usize
storage::clear_runs(&state, caller_user_id(&caller), caller_is_admin(&caller)).await?
as usize
} else {
storage::delete_runs(
&state,
@@ -1598,6 +1684,9 @@ async fn queue_prompt(
) -> AppResult<Response> {
let caller = auth_caller(&state, &headers).await?;
if state.config.comfyui.uses_native_backend() {
if prompt_contains_boosapi_nodes(&payload.prompt) {
return queue_boosapi_prompt(state, caller, payload).await;
}
return native_workflow_response(
&state,
Method::POST,
@@ -1606,6 +1695,14 @@ async fn queue_prompt(
)
.await;
}
queue_boosapi_prompt(state, caller, payload).await
}
async fn queue_boosapi_prompt(
state: AppState,
caller: WorkflowCaller,
payload: PromptPayload,
) -> AppResult<Response> {
let prompt = payload.prompt.clone();
let graph = match normalize_prompt_payload(&prompt) {
Ok(graph) => graph,
@@ -1667,7 +1764,9 @@ async fn queue_prompt(
.await;
match result {
Ok(run_id) => {
if let Ok(run) = storage::get_run(&state_for_task.db, &run_id, user_id, allow_admin).await {
if let Ok(run) =
storage::get_run(&state_for_task.db, &run_id, user_id, allow_admin).await
{
let outputs = run_outputs_for_comfy(&run);
publish_run_events(Some(session_for_task.as_str()), &run, &outputs);
}
@@ -1781,7 +1880,10 @@ async fn view(
}
}
let remote_url: Option<String> = row.try_get("remote_url").ok();
if let Some(url) = remote_url.as_deref().filter(|value| !value.trim().is_empty()) {
if let Some(url) = remote_url
.as_deref()
.filter(|value| !value.trim().is_empty())
{
return Ok((
StatusCode::TEMPORARY_REDIRECT,
[(axum::http::header::LOCATION, url.to_string())],
@@ -1906,7 +2008,12 @@ fn native_workflow_base(state: &AppState) -> String {
fn native_workflow_http_path(state: &AppState, path: &str) -> String {
if state.config.comfyui.uses_cloud_backend() {
format!("/api{}", path.strip_prefix('/').map(|value| format!("/{value}")).unwrap_or_else(|| path.to_string()))
format!(
"/api{}",
path.strip_prefix('/')
.map(|value| format!("/{value}"))
.unwrap_or_else(|| path.to_string())
)
} else {
path.to_string()
}
@@ -1919,9 +2026,13 @@ fn native_workflow_request(
body: Option<Value>,
) -> reqwest::RequestBuilder {
let client = reqwest::Client::new();
let url = format!("{}{}", native_workflow_base(state), native_workflow_http_path(state, path));
let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())
.unwrap_or(reqwest::Method::GET);
let url = format!(
"{}{}",
native_workflow_base(state),
native_workflow_http_path(state, path)
);
let reqwest_method =
reqwest::Method::from_bytes(method.as_str().as_bytes()).unwrap_or(reqwest::Method::GET);
let mut request = client.request(reqwest_method, url);
if let Some(api_key) = state
.config
@@ -2014,14 +2125,7 @@ async fn native_workflow_view_response(
.await
.map_err(|err| AppError::UpstreamError(err.to_string()))?;
if status.is_success() {
archive_native_workflow_asset(
state,
caller,
params,
&content_type,
bytes.as_ref(),
)
.await;
archive_native_workflow_asset(state, caller, params, &content_type, bytes.as_ref()).await;
}
Ok((
StatusCode::from_u16(status.as_u16()).unwrap_or(StatusCode::BAD_GATEWAY),
@@ -2114,9 +2218,10 @@ async fn native_workflow_api_proxy(
let body = if body.is_empty() {
None
} else {
Some(serde_json::from_slice::<Value>(&body).unwrap_or_else(|_| {
json!({ "body": String::from_utf8_lossy(&body).to_string() })
}))
Some(
serde_json::from_slice::<Value>(&body)
.unwrap_or_else(|_| json!({ "body": String::from_utf8_lossy(&body).to_string() })),
)
};
native_workflow_response(&state, method, &target, body).await
}
@@ -2673,24 +2778,16 @@ async fn experiment_models(
) -> AppResult<Json<Value>> {
if state.config.comfyui.uses_native_backend() {
let _ = auth_caller_optional(&state, &headers);
return native_workflow_json(&state, Method::GET, "/experiment/models", None).await;
let mut value = native_workflow_json(&state, Method::GET, "/experiment/models", None)
.await?
.0;
inject_boosapi_model_folders(&mut value);
return Ok(Json(value));
}
let caller =
auth_caller_optional(&state, &headers).unwrap_or(WorkflowCaller::User { user_id: 0 });
let _ = caller;
let folders = vec![
"boosapi_models",
"checkpoints",
"chat_models",
"video_models",
"audio_tts_models",
"embeddings",
"vision_models",
"3d_models",
"loras",
"vae",
];
let items = folders
let items = builtin_model_folders()
.into_iter()
.map(|name| json!({ "name": name }))
.collect::<Vec<_>>();
@@ -2704,6 +2801,16 @@ async fn experiment_models_in_folder(
) -> AppResult<Json<Value>> {
if state.config.comfyui.uses_native_backend() {
let _ = auth_caller_optional(&state, &headers);
let route_models = workflow_route_models(&state).await.unwrap_or_default();
let api_models = workflow_folder_models(&folder, &route_models);
if !api_models.is_empty() {
return Ok(Json(Value::Array(
api_models
.into_iter()
.map(|name| json!({ "name": name, "pathIndex": 0 }))
.collect(),
)));
}
let path = format!("/experiment/models/{}", simple_urlencode(&folder));
return native_workflow_json(&state, Method::GET, &path, None).await;
}
@@ -2743,7 +2850,11 @@ async fn view_metadata(
) -> AppResult<Json<Value>> {
if state.config.comfyui.uses_native_backend() {
let _ = auth_caller_optional(&state, &headers);
let path = format!("/view_metadata/{}{}", simple_urlencode(&folder), build_query_string(&query));
let path = format!(
"/view_metadata/{}{}",
simple_urlencode(&folder),
build_query_string(&query)
);
return native_workflow_json(&state, Method::GET, &path, None).await;
}
let caller = auth_caller(&state, &headers).await?;
@@ -2815,7 +2926,9 @@ async fn workflow_ws(
.cloned()
.unwrap_or_else(|| "boosapi-workflow".to_string());
if native_workflow_enabled(&state) {
return Ok(ws.on_upgrade(move |socket| native_workflow_ws_bridge(state, socket, query, sid)));
return Ok(
ws.on_upgrade(move |socket| native_workflow_ws_bridge(state, socket, query, sid))
);
}
Ok(ws.on_upgrade(move |socket| async move {
use tokio::time::{interval, Duration};
@@ -2883,7 +2996,9 @@ async fn native_workflow_ws_bridge(
.map(str::trim)
.filter(|value| !value.is_empty())
{
query.entry("token".to_string()).or_insert_with(|| api_key.to_string());
query
.entry("token".to_string())
.or_insert_with(|| api_key.to_string());
}
ws_url.push_str("/ws");
ws_url.push_str(&build_query_string(&query));
src/routes/workspace.rs
+9 -2
View File
@@ -13,6 +13,7 @@ use axum::{
use serde::Deserialize;
use serde_json::{json, Value};
use std::path::PathBuf;
use tokio_util::io::ReaderStream;
#[derive(Debug, Clone)]
enum WorkspaceCaller {
@@ -318,7 +319,11 @@ async fn get_asset_file(
.filter(|value| !value.trim().is_empty())
.ok_or_else(|| AppError::NotFound("Workspace asset file is not archived locally".into()))?;
let path = PathBuf::from(file_path);
let bytes = tokio::fs::read(&path)
let file = tokio::fs::File::open(&path)
.await
.map_err(|_| AppError::NotFound("Workspace asset file not found".into()))?;
let metadata = file
.metadata()
.await
.map_err(|_| AppError::NotFound("Workspace asset file not found".into()))?;
let mime_type = row
@@ -334,11 +339,13 @@ async fn get_asset_file(
Response::builder()
.status(StatusCode::OK)
.header(header::CONTENT_TYPE, mime_type)
.header(header::CONTENT_LENGTH, metadata.len().to_string())
.header(header::ACCEPT_RANGES, "bytes")
.header(
header::CONTENT_DISPOSITION,
format!("inline; filename=\"{}\"", file_name.replace('"', "")),
)
.body(Body::from(bytes))
.body(Body::from_stream(ReaderStream::new(file)))
.map_err(|err| AppError::Internal(anyhow::anyhow!(err)))
}