Route Comfy workflows through video API

src/proxy/executors/mod.rs

@@ -15,7 +15,7 @@ use crate::transformers::{
 };
 use crate::AppState;
 use reqwest::Client;
-use serde_json::Value;
+use serde_json::{json, Value};
 use std::time::Duration;
 
 /// Resolved upstream info for a given model request.
@@ -518,6 +518,23 @@ pub async fn execute_proxy_get(route: UpstreamRoute, path_suffix: &str) -> AppRe
     parse_upstream_json(&text, &upstream_url, status)
 }
 
+pub async fn execute_video_task_status(
+    state: AppState,
+    route: UpstreamRoute,
+    id: &str,
+) -> AppResult<Value> {
+    if comfy_video_template_id(&route).is_some() {
+        return comfy_history(&state, id).await;
+    }
+
+    let suffix = if is_volcengine_ark_base(&route.base_url) {
+        format!("/contents/generations/tasks/{id}")
+    } else {
+        format!("/videos/{id}")
+    };
+    execute_proxy_get(route, &suffix).await
+}
+
 /// Execute an OpenAI-shaped video task request, with Volcengine Ark native
 /// task endpoints as a first-class compatibility target.
 pub async fn execute_video_generation(
@@ -525,6 +542,10 @@ pub async fn execute_video_generation(
     route: UpstreamRoute,
     body: Value,
 ) -> AppResult<Value> {
+    if let Some(template_id) = comfy_video_template_id(&route) {
+        return execute_comfy_video_generation(state, route, body, &template_id).await;
+    }
+
     if is_volcengine_ark_base(&route.base_url) {
         let body = volcengine_video_create_body(body, &route.model)?;
         return execute_proxy_request(state, route, "/contents/generations/tasks", body).await;
@@ -533,6 +554,76 @@ pub async fn execute_video_generation(
     execute_proxy_request(state, route, "/videos", body).await
 }
 
+fn comfy_video_template_id(route: &UpstreamRoute) -> Option<String> {
+    route.base_url
+        .strip_prefix("comfy://")
+        .or_else(|| route.base_url.strip_prefix("comfy-template://"))
+        .or_else(|| route.model.strip_prefix("comfy://"))
+        .or_else(|| route.model.strip_prefix("comfy:"))
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+        .map(str::to_string)
+}
+
+async fn execute_comfy_video_generation(
+    state: AppState,
+    route: UpstreamRoute,
+    body: Value,
+    template_id: &str,
+) -> AppResult<Value> {
+    if !state.config.comfyui.uses_native_backend() {
+        return Err(AppError::BadRequest(
+            "Comfy 视频模型需要配置 comfyui.backend = cloud、external 或 runninghub".into(),
+        ));
+    }
+
+    let workflow = crate::workflow::storage::get_workflow(&state.db, template_id, None, true)
+        .await
+        .map_err(|err| AppError::BadRequest(format!("Comfy 视频模型模板不可用: {err}")))?;
+    let mut prompt = workflow_prompt_from_graph(&workflow.graph)
+        .ok_or_else(|| AppError::BadRequest("Comfy 视频模型模板缺少 prompt/workflow JSON".into()))?;
+    if prompt_contains_class_type(&prompt, "BoosAPITextToVideo") {
+        return Err(AppError::BadRequest(
+            "Comfy 视频模型模板不能包含 BoosAPITextToVideo 自定义节点，请选择真实 Comfy 视频节点模板".into(),
+        ));
+    }
+    let mut mapping = comfy_mapping_from_graph(&workflow.graph).unwrap_or_else(default_comfy_mapping);
+    if !mapping_has_node(&mapping, &["prompt", "positive", "text", "promptNode"]) {
+        mapping = infer_comfy_mapping(&prompt);
+    }
+    let video_prompt = inject_comfy_video_inputs(&mut prompt, &mapping, &body)?;
+    let client_id = format!("boosapi-videos-{}", uuid::Uuid::new_v4().simple());
+    let response = comfy_json(
+        &state,
+        reqwest::Method::POST,
+        "/prompt",
+        Some(json!({
+            "client_id": client_id,
+            "prompt": prompt,
+        })),
+    )
+    .await?;
+    let prompt_id = scalar_string(
+        response
+            .get("prompt_id")
+            .or_else(|| response.get("promptId"))
+            .or_else(|| response.pointer("/data/prompt_id"))
+            .or_else(|| response.pointer("/data/promptId")),
+    )
+    .ok_or_else(|| AppError::UpstreamError(format!("Comfy response missing prompt_id: {response}")))?;
+
+    Ok(json!({
+        "id": prompt_id,
+        "object": "video.generation.task",
+        "model": route.model,
+        "status": "queued",
+        "backend": "comfy",
+        "template_id": template_id,
+        "prompt": video_prompt,
+        "response": response,
+    }))
+}
+
 fn is_volcengine_ark_base(base_url: &str) -> bool {
     base_url.contains("ark.cn-beijing.volces.com") || base_url.contains("volces.com/api/v3")
 }
@@ -602,6 +693,299 @@ fn scalar_to_string(value: &Value) -> Option<String> {
     }
 }
 
+fn scalar_string(value: Option<&Value>) -> Option<String> {
+    scalar_to_string(value?)
+}
+
+fn workflow_prompt_from_graph(graph: &Value) -> Option<Value> {
+    graph.get("prompt")
+        .cloned()
+        .or_else(|| graph.get("workflow").cloned())
+        .or_else(|| graph.get("comfyPrompt").cloned())
+        .or_else(|| graph_to_comfy_prompt(graph))
+        .or_else(|| graph.is_object().then(|| graph.clone()))
+}
+
+fn comfy_mapping_from_graph(graph: &Value) -> Option<Value> {
+    graph.pointer("/metadata/moneyPrinter/nodeMapping")
+        .or_else(|| graph.pointer("/moneyPrinter/nodeMapping"))
+        .or_else(|| graph.pointer("/metadata/video/nodeMapping"))
+        .or_else(|| graph.pointer("/video/nodeMapping"))
+        .or_else(|| graph.pointer("/metadata/digitalHuman/nodeMapping"))
+        .or_else(|| graph.pointer("/digitalHuman/nodeMapping"))
+        .cloned()
+}
+
+fn graph_to_comfy_prompt(graph: &Value) -> Option<Value> {
+    let nodes = graph.get("nodes")?.as_array()?;
+    let mut prompt = serde_json::Map::new();
+    for node in nodes {
+        let id = scalar_string(node.get("id"))?;
+        let class_type = node
+            .get("type")
+            .or_else(|| node.get("class_type"))
+            .and_then(Value::as_str)
+            .unwrap_or_default()
+            .to_string();
+        if class_type.trim().is_empty() {
+            continue;
+        }
+        let inputs = node
+            .get("params")
+            .or_else(|| node.get("inputs"))
+            .and_then(Value::as_object)
+            .cloned()
+            .unwrap_or_default();
+        prompt.insert(id, json!({ "class_type": class_type, "inputs": inputs }));
+    }
+    for edge in graph.get("edges").and_then(Value::as_array).into_iter().flatten() {
+        let Some(from) = edge.get("from").and_then(Value::as_str) else { continue; };
+        let Some(to) = edge.get("to").and_then(Value::as_str) else { continue; };
+        let Some((from_id, _from_port)) = from.split_once('.') else { continue; };
+        let Some((to_id, to_field)) = to.split_once('.') else { continue; };
+        if let Some(inputs) = prompt
+            .get_mut(to_id)
+            .and_then(Value::as_object_mut)
+            .and_then(|node| node.get_mut("inputs"))
+            .and_then(Value::as_object_mut)
+        {
+            inputs.insert(to_field.to_string(), json!([from_id, 0]));
+        }
+    }
+    Some(Value::Object(prompt))
+}
+
+fn default_comfy_mapping() -> Value {
+    json!({
+        "prompt": { "nodeId": "", "field": "prompt" },
+        "negativePrompt": { "nodeId": "", "field": "text" },
+        "ratio": { "nodeId": "", "field": "ratio" },
+        "resolution": { "nodeId": "", "field": "resolution" },
+        "duration": { "nodeId": "", "field": "duration" }
+    })
+}
+
+fn mapping_value<'a>(mapping: &'a Value, keys: &[&str]) -> Option<&'a Value> {
+    keys.iter().find_map(|key| mapping.get(*key))
+}
+
+fn mapping_node_id(mapping: &Value, keys: &[&str]) -> Option<String> {
+    mapping_value(mapping, keys).and_then(|value| {
+        if value.is_object() {
+            scalar_string(value.get("nodeId").or_else(|| value.get("node_id")).or(Some(value)))
+        } else {
+            scalar_string(Some(value))
+        }
+    })
+}
+
+fn mapping_field(mapping: &Value, keys: &[&str], default_field: &str) -> String {
+    mapping_value(mapping, keys)
+        .and_then(|value| {
+            if value.is_object() {
+                scalar_string(value.get("field").or_else(|| value.get("fieldName")))
+            } else {
+                None
+            }
+        })
+        .unwrap_or_else(|| default_field.to_string())
+}
+
+fn mapping_has_node(mapping: &Value, keys: &[&str]) -> bool {
+    mapping_node_id(mapping, keys).is_some_and(|value| !value.trim().is_empty())
+}
+
+fn first_node_by_class(prompt: &serde_json::Map<String, Value>, class_type: &str) -> Option<String> {
+    prompt
+        .iter()
+        .find(|(_, node)| node.get("class_type").and_then(Value::as_str) == Some(class_type))
+        .map(|(id, _)| id.clone())
+}
+
+fn infer_comfy_mapping(prompt: &Value) -> Value {
+    let Some(object) = prompt.as_object() else {
+        return default_comfy_mapping();
+    };
+    let prompt_node = object
+        .iter()
+        .find(|(_, node)| {
+            let class_type = node.get("class_type").and_then(Value::as_str).unwrap_or_default().to_ascii_lowercase();
+            let inputs = node
+                .get("inputs")
+                .and_then(Value::as_object)
+                .map(|value| value.keys().cloned().collect::<Vec<_>>().join(" ").to_ascii_lowercase())
+                .unwrap_or_default();
+            class_type.contains("text")
+                || class_type.contains("prompt")
+                || class_type.contains("wan")
+                || class_type.contains("ltx")
+                || class_type.contains("video")
+                || inputs.contains("prompt")
+                || inputs.contains("positive")
+                || inputs.contains("text")
+        })
+        .map(|(id, _)| id.clone())
+        .unwrap_or_default();
+    let video_node = first_node_by_class(object, "Wan2ImageToVideoApi")
+        .or_else(|| first_node_by_class(object, "Wan2TextToVideoApi"))
+        .or_else(|| first_node_by_class(object, "TextGenerateLTX2Prompt"))
+        .or_else(|| first_node_by_class(object, "KSampler"))
+        .unwrap_or_else(|| prompt_node.clone());
+    json!({
+        "prompt": { "nodeId": prompt_node, "field": "text" },
+        "ratio": { "nodeId": video_node, "field": "ratio" },
+        "resolution": { "nodeId": video_node, "field": "resolution" },
+        "duration": { "nodeId": video_node, "field": "duration" }
+    })
+}
+
+fn apply_node_input(prompt: &mut Value, node_id: &str, field: &str, value: Value) -> AppResult<()> {
+    if node_id.trim().is_empty() {
+        return Err(AppError::BadRequest("Comfy 视频模型节点映射包含空 nodeId".into()));
+    }
+    let Some(node) = prompt.get_mut(node_id) else {
+        return Err(AppError::BadRequest(format!("Comfy prompt missing node {node_id}")));
+    };
+    let Some(inputs) = node.get_mut("inputs").and_then(Value::as_object_mut) else {
+        return Err(AppError::BadRequest(format!("Comfy node {node_id} has no inputs")));
+    };
+    inputs.insert(field.to_string(), value);
+    Ok(())
+}
+
+fn inject_comfy_video_inputs(prompt: &mut Value, mapping: &Value, body: &Value) -> AppResult<String> {
+    let video_prompt = scalar_string(body.get("prompt"))
+        .ok_or_else(|| AppError::BadRequest("prompt is required for Comfy video models".into()))?;
+    let prompt_node = mapping_node_id(mapping, &["prompt", "positive", "text", "promptNode"])
+        .ok_or_else(|| AppError::BadRequest("Comfy 视频模型缺少提示词节点映射".into()))?;
+    apply_node_input(
+        prompt,
+        &prompt_node,
+        &mapping_field(mapping, &["prompt", "positive", "text", "promptNode"], "prompt"),
+        json!(video_prompt),
+    )?;
+    if let Some(negative_node) = mapping_node_id(mapping, &["negativePrompt", "negative", "negativeNode"]) {
+        let negative = scalar_string(body.get("negative_prompt").or_else(|| body.get("negative")))
+            .unwrap_or_else(|| "low quality, blurry, distorted, watermark, text artifacts".to_string());
+        apply_node_input(
+            prompt,
+            &negative_node,
+            &mapping_field(mapping, &["negativePrompt", "negative", "negativeNode"], "text"),
+            json!(negative),
+        )?;
+    }
+    for (keys, request_keys, default_field) in [
+        (&["ratio", "ratioNode"][..], &["ratio", "aspectRatio", "aspect_ratio"][..], "ratio"),
+        (&["resolution", "resolutionNode"][..], &["resolution", "size"][..], "resolution"),
+        (&["duration", "durationNode"][..], &["duration"][..], "duration"),
+    ] {
+        if let Some(node_id) = mapping_node_id(mapping, keys) {
+            if let Some(value) = request_keys.iter().find_map(|key| body.get(*key).cloned()) {
+                apply_node_input(prompt, &node_id, &mapping_field(mapping, keys, default_field), value)?;
+            }
+        }
+    }
+    Ok(video_prompt)
+}
+
+fn prompt_contains_class_type(prompt: &Value, class_type: &str) -> bool {
+    match prompt {
+        Value::Object(object) => {
+            object
+                .get("class_type")
+                .and_then(Value::as_str)
+                .is_some_and(|value| value == class_type)
+                || object.values().any(|value| prompt_contains_class_type(value, class_type))
+        }
+        Value::Array(items) => items.iter().any(|value| prompt_contains_class_type(value, class_type)),
+        _ => false,
+    }
+}
+
+fn comfy_native_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())
+        )
+    } else {
+        path.to_string()
+    }
+}
+
+fn comfy_native_url(state: &AppState, path: &str) -> String {
+    format!(
+        "{}{}",
+        state.config.comfyui.target_base_url(),
+        comfy_native_path(state, path)
+    )
+}
+
+async fn comfy_json(
+    state: &AppState,
+    method: reqwest::Method,
+    path: &str,
+    body: Option<Value>,
+) -> AppResult<Value> {
+    let client = http_client();
+    let mut request = client.request(method, comfy_native_url(state, path));
+    if let Some(api_key) = state
+        .config
+        .comfyui
+        .api_key
+        .as_deref()
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+    {
+        request = request
+            .header("X-API-Key", api_key)
+            .header("Authorization", format!("Bearer {api_key}"));
+    }
+    if let Some(body) = body {
+        request = request.json(&body);
+    }
+    let response = request
+        .send()
+        .await
+        .map_err(|err| AppError::UpstreamError(format!("Comfy backend unreachable: {err}")))?;
+    let status = response.status();
+    let text = response
+        .text()
+        .await
+        .map_err(|err| AppError::UpstreamError(err.to_string()))?;
+    let value = serde_json::from_str::<Value>(&text).unwrap_or_else(|_| json!({ "body": text }));
+    if !status.is_success() {
+        return Err(AppError::UpstreamError(value.to_string()));
+    }
+    Ok(value)
+}
+
+async fn comfy_history(state: &AppState, prompt_id: &str) -> AppResult<Value> {
+    let history = comfy_json(state, reqwest::Method::GET, &format!("/history/{prompt_id}"), None).await?;
+    Ok(json!({
+        "id": prompt_id,
+        "object": "video.generation.task",
+        "backend": "comfy",
+        "status": if comfy_output_objects(&history) { "succeeded" } else { "running" },
+        "history": history,
+    }))
+}
+
+fn comfy_output_objects(value: &Value) -> bool {
+    match value {
+        Value::Object(object) => {
+            ["videos", "gifs", "images"]
+                .iter()
+                .any(|key| object.get(*key).and_then(Value::as_array).is_some_and(|items| !items.is_empty()))
+                || object.values().any(comfy_output_objects)
+        }
+        Value::Array(items) => items.iter().any(comfy_output_objects),
+        _ => false,
+    }
+}
+
 /// Execute an image generation request.
 pub async fn execute_image_generation(
     _state: AppState,

src/routes/proxy/mod.rs

@@ -347,12 +347,7 @@ async fn video_task_handler(
         .unwrap_or("default")
         .to_string();
     let route = resolve_route(&state, &model).await?;
-    let suffix = if route.base_url.contains("ark.cn-beijing.volces.com") {
-        format!("/contents/generations/tasks/{id}")
-    } else {
-        format!("/videos/{id}")
-    };
-    let result = executors::execute_proxy_get(route, &suffix).await?;
+    let result = executors::execute_video_task_status(state.clone(), route, &id).await?;
     record_billing(&state, key_id, &model, 0, 0, 0.0).await;
     Ok(Json(result))
 }