10 Commits

Author SHA1 Message Date
boos d548525ef2 refactor(workflow_ui): 抽出 bootstrap 子模块
Co-Authored-By: Claude <noreply@anthropic.com>
2026-06-18 22:52:59 +08:00
boos 3aa1468fbb refactor(workflow_ui): 抽出 ws 子模块
Co-Authored-By: Claude <noreply@anthropic.com>
2026-06-18 22:44:53 +08:00
boos f7f9a9ca09 refactor(workflow_ui): 抽出 run_convert 子模块
Co-Authored-By: Claude <noreply@anthropic.com>
2026-06-18 22:32:51 +08:00
boos a37546cb2d refactor(workflow_ui): 抽出 native_proxy 子模块
Co-Authored-By: Claude <noreply@anthropic.com>
2026-06-18 22:22:19 +08:00
boos f43daed3ee refactor(workflow_ui): 抽出 repair 子模块
Co-Authored-By: Claude <noreply@anthropic.com>
2026-06-18 22:05:30 +08:00
boos 35f63c46f1 refactor(workflow_ui): 抽出 model_inject 子模块
Co-Authored-By: Claude <noreply@anthropic.com>
2026-06-18 21:55:27 +08:00
boos e296c01b4b docs: workflow_ui/mod.rs 续拆(6 子模块)设计
把 mod.rs 残留 4095 行按职责再拆 6 组(model_inject/repair/native_proxy/
run_convert/ws/bootstrap),mod.rs 瘦到 ~1000-1200。纯机械移动,编译器+5测试验证。

Co-Authored-By: Claude <noreply@anthropic.com>
2026-06-18 21:33:55 +08:00
boos a6c4db64c4 refactor(workflow_ui): 抽出 queue 子模块
Co-Authored-By: Claude <noreply@anthropic.com>
2026-06-18 15:15:17 +08:00
boos 7a02545fb8 refactor(workflow_ui): 抽出 validate 子模块
Co-Authored-By: Claude <noreply@anthropic.com>
2026-06-18 14:58:58 +08:00
boos 9cad69b88c refactor(workflow_ui): 抽出 comfy_proxy 子模块
Co-Authored-By: Claude <noreply@anthropic.com>
2026-06-18 14:47:24 +08:00
11 changed files with 4363 additions and 4108 deletions
@@ -0,0 +1,58 @@
# workflow_ui/mod.rs 续拆(6 子模块)设计
日期:2026-06-18
分支:`feat/comfy-cloud-adapter`
状态:已通过 brainstorming,待用户复审
## 背景
上一轮已把 `workflow_ui.rs`(6181 行)拆成 8 模块,但 `mod.rs` 仍 4095 行——一批
未在上轮清单内的函数(native proxy / repair 链 / bootstrap 注入 / run 转换 / ws / 模型注入)
留在了 mod.rs。本轮继续把这 6 组抽出,让 mod.rs 瘦到 ~1000-1200 行(只剩 routes()
+ 共享 struct + 小工具)。
延续上轮纪律:**纯机械移动,函数体一字不改**,只调可见性 + use,编译器 + 5 测试验证。
## 已确认决策
1. 续拆 workflow_ui/mod.rs 残留。
2. 拆成 6 子模块(见下表)。
3. 纯机械移动,零逻辑改动。
4. 安全:每抽一组即 cargo build -j 2 + 5 测试保持;末尾各文件行数核对。
## 子模块划分
| 子模块 | 函数组 |
|--------|--------|
| `model_inject.rs` | comfy_combo_input, inject_model_choices, is_boosapi_extension_node_type, workflow_node_defs_with_models, inject_boosapi_object_info, builtin_model_folders, inject_boosapi_model_folders, workflow_model_folder_payload, native_model_folder_payload_or_fallback, model_payload_items_from_names, normalize_model_item_name, workflow_model_display_aliases, model_payload_item_with_name, merge_model_payload_items, normalize_model_folder_payload |
| `repair.rs` | inject_native_prompt_cloud_api_key, repair_resize_image_mask_nodes(+_without_workflow), repair_text_generate_ltx2_prompt_nodes, repair_dynamic_combo_nodes(+ dynamic_combo_* 全部 helper), repair_wan2_api_nodes(+ wan2_* / is_wan2_* helper), collect_workflow_nodes, prompt_subgraph_node_id, id_to_prompt_key |
| `native_proxy.rs` | native_workflow_enabled/base/http_path/request/json/json_cached/response/view_response/video_response/file_response/raw_response/api_proxy, archive_native_workflow_asset, repair_native_prompt_body_value, workflow_cache_kind_from_key, workflow_cache_status, upload_image(+_to_local_backend/_to_native_backend), upload_mask, timestamp_ms |
| `run_convert.rs` | workflow_run_to_history/_task/_queue_task/_prompt_tuple, run_status_messages, run_outputs_for_comfy, node_output_for_comfy, comfy_media_items, normalize_comfy_media_item, workflow_text_output, workflow_graph_to_comfy_workflow, workflow_timestamp_ms |
| `ws.rs` | workflow_session_sender, publish_workflow_ws_event, workflow_node_label, workflow_progress_callback_with_run_sender, publish_run_events, workflow_ws, native_workflow_ws_bridge |
| `bootstrap.rs` | inject_bootstrap, workflow_secret_cleanup_bootstrap, inject_external_bootstrap, inject_enhanced_bootstrap |
> 留 mod.rs:routes()、共享 struct(WorkflowCaller/CachedJson/各 Query/Payload)、跨多模块小工具
> (safe_view_path/user_home/guess_mime_from_path/parse_boolish/auth_caller_optional/
> workflow_context/parse_workflow_cookie/build_query_string/simple_urlencode/各 workflow_widget_*
> validation helper 等)、experiment_models/view_metadata/custom_nodes_i18n/internal_* 等零散 handler。
> 实现时按实际依赖判定:被多组共用的留 mod.rs `pub(super)`。
## 安全策略
同上轮:每抽一组 → `cargo build -j 2`(编译器是主安全网,纯移动)→ `cargo test -p boosapi
--lib routes::workflow_ui`(5 测试基线不减)→ 提交。跨模块依赖的私有符号按编译报错逐个
改 `pub(super)`。子模块内部 helper 也 `pub(super)` 或保持模块私有(看是否跨文件用)。
## 测试
编译器逐组验证;5 个现有测试全程保持;末尾核对 8+6 个子文件行数,mod.rs 应 <1500。
## 非目标
- 不改逻辑、不改路由、不动 handler 行为。
- 不拆其他巨文件(oauth/payment/digital_human 各自后续循环)。
- 不顺手优化函数实现。
## 提交约定
分支 `feat/comfy-cloud-adapter`。每抽 1 组 + 编译通过即提交。中文 commit。
+799
View File
@@ -0,0 +1,799 @@
pub(super) fn inject_bootstrap(html: String, config: &crate::config::ComfyUIConfig) -> String {
if config.uses_native_backend() && !config.uses_cloud_backend() {
return inject_external_bootstrap(html);
}
inject_enhanced_bootstrap(html)
}
fn workflow_secret_cleanup_bootstrap() -> &'static str {
r#"<script>
(function () {
const clearLeakedComfyApiKey = () => {
try {
localStorage.removeItem('comfy_api_key');
sessionStorage.removeItem('comfy_api_key');
document.cookie = 'comfy_api_key=; Path=/workflow-ui; Max-Age=0; SameSite=Lax';
} catch (err) {}
};
clearLeakedComfyApiKey();
window.addEventListener('DOMContentLoaded', clearLeakedComfyApiKey);
})();
</script>"#
}
fn inject_external_bootstrap(html: String) -> String {
let secret_cleanup_bootstrap = workflow_secret_cleanup_bootstrap();
let shell = r#"<script>
(function () {
const renameComfyUiBrand = () => {
document.title = document.title.replace(/ComfyUI/g, 'BoosAI');
const candidates = Array.from(document.querySelectorAll('header *, nav *, .comfy-menu-header *, .comfyui-menu *, .p-menubar *')).slice(0, 80);
for (const el of candidates) {
if (el.childNodes.length === 1 && el.firstChild && el.firstChild.nodeType === Node.TEXT_NODE) {
const value = String(el.textContent || '').trim();
if (value === 'ComfyUI' || value === 'Comfy UI') el.textContent = 'BoosAI';
}
}
};
try {
const userJwt = localStorage.getItem('user_jwt_token');
const adminToken = localStorage.getItem('auth_token');
const token = adminToken || userJwt;
if (token) {
localStorage.setItem('auth_token', token);
document.cookie = 'boosapi_workflow_token=' + encodeURIComponent(token) + '; Path=/workflow-ui; SameSite=Lax';
}
} catch (err) {}
const getCurrentWorkflowJson = async () => {
try {
if (window.app?.graphToPrompt) {
const result = await window.app.graphToPrompt();
return result?.workflow || result?.output || result?.prompt || result;
}
} catch (err) {}
try {
const graph = window.app?.graph || window.LGraphCanvas?.active_canvas?.graph;
if (graph?.serialize) return graph.serialize();
} catch (err) {}
return null;
};
const installWorkflowValidationButton = () => {
if (document.getElementById('boosapi-workflow-validate-button')) return;
const button = document.createElement('button');
button.id = 'boosapi-workflow-validate-button';
button.type = 'button';
button.textContent = '检查模板';
button.title = '检查当前工作流的节点、模型和输入素材兼容性';
button.style.cssText = 'position:fixed;right:112px;top:10px;z-index:10000;height:32px;padding:0 12px;border:1px solid rgba(120,130,150,.45);border-radius:6px;background:rgba(20,24,32,.88);color:#fff;font-size:13px;cursor:pointer;';
button.addEventListener('click', async () => {
button.disabled = true;
const oldText = button.textContent;
button.textContent = '检查中...';
try {
const workflow = await getCurrentWorkflowJson();
if (!workflow) throw new Error('没有读取到当前工作流');
const response = await fetch('/workflow-ui/api/workflow/validate?check_files=true', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(workflow),
});
const report = await response.json();
const summary = report?.summary || {};
const lines = [`节点 ${summary.nodes || 0},连线 ${summary.links || 0},错误 ${summary.errors || 0},警告 ${summary.warnings || 0}`];
for (const item of report?.missingNodeTypes || []) lines.push(`缺失节点: ${item.nodeType} #${item.nodeId}${item.connected ? '(已连接)' : '(未连接)'}`);
for (const item of report?.emptyModelFields || []) lines.push(`空模型列表: ${item.nodeType} #${item.nodeId} ${item.field} = ${item.selected}`);
for (const item of report?.missingModelChoices || []) lines.push(`缺失模型: ${item.nodeType} #${item.nodeId} ${item.field} = ${item.selected}`);
for (const item of report?.inputFiles || []) if (item.checked) lines.push(`素材: ${item.filename} ${item.reachable ? '可访问' : '不可访问'}`);
if (report?.hints?.length) lines.push('', ...report.hints);
window.alert(lines.join('\n'));
} catch (err) {
window.alert('检查失败:' + (err?.message || err));
} finally {
button.disabled = false;
button.textContent = oldText;
}
});
document.body.appendChild(button);
};
window.setTimeout(renameComfyUiBrand, 250);
window.setTimeout(renameComfyUiBrand, 1200);
window.setTimeout(installWorkflowValidationButton, 800);
window.setTimeout(installWorkflowValidationButton, 1800);
})();
</script>"#;
let bootstrap = format!("{secret_cleanup_bootstrap}{shell}");
html.replace("</head>", &format!("{bootstrap}\n</head>"))
}
pub(super) fn inject_enhanced_bootstrap(html: String) -> String {
let secret_cleanup_bootstrap = workflow_secret_cleanup_bootstrap();
let shell = r#"<style id="boosapi-workflow-shell-style">
</style>
<script>
(function () {
const renameComfyUiBrand = () => {
document.title = document.title.replace(/ComfyUI/g, 'BoosAI');
const candidates = Array.from(document.querySelectorAll('header *, nav *, .comfy-menu-header *, .comfyui-menu *, .p-menubar *')).slice(0, 120);
for (const el of candidates) {
if (el.childNodes.length === 1 && el.firstChild && el.firstChild.nodeType === Node.TEXT_NODE) {
const value = String(el.textContent || '').trim();
if (value === 'ComfyUI' || value === 'Comfy UI') {
el.textContent = 'BoosAI';
}
}
for (const attr of ['aria-label', 'title']) {
const value = el.getAttribute(attr);
if (value && /Comfy\s*UI|ComfyUI/.test(value)) {
el.setAttribute(attr, value.replace(/Comfy\s*UI|ComfyUI/g, 'BoosAI'));
}
}
}
};
const installBoosApiFetchAuth = () => {
try {
const api = window.app?.api || window.comfyAPI?.api || window.api;
if (!api || api.__boosapiFetchAuthPatch || typeof api.fetchApi !== 'function') return;
const originalFetchApi = api.fetchApi.bind(api);
api.fetchApi = function(route, options) {
options ||= {};
const headers = new Headers(options.headers || {});
const token = localStorage.getItem('auth_token') || localStorage.getItem('user_jwt_token');
if (token && !headers.has('Authorization')) headers.set('Authorization', 'Bearer ' + token);
options.headers = headers;
options.credentials ||= 'same-origin';
return originalFetchApi(route, options);
};
api.__boosapiFetchAuthPatch = true;
} catch (err) {}
};
const workflowTextMap = new Map(Object.entries({
'Add Node': '添加节点',
'Search': '搜索',
'Reroute': '转接点',
'PreviewImage': '预览图像',
'SaveImage': '保存图像',
'Load Image': '加载图像',
'CheckpointLoaderSimple': 'Checkpoint 加载器(简易)',
'CLIPTextEncode': 'CLIP 文本编码',
'KSampler': 'K 采样器',
'VAEDecode': 'VAE 解码',
'VAEEncode': 'VAE 编码',
'EmptyLatentImage': '空潜空间图像',
'Latent': '潜空间',
'Input': '输入',
'Output': '输出',
'Util': '工具',
'utils': '工具',
'ComfyUI': 'BoosAI',
'IMAGE': '图像',
'LATENT': '潜空间',
'MODEL': '模型',
'CLIP': 'CLIP',
'VAE': 'VAE',
'CONDITIONING': '条件',
'BoosAPI': 'BoosAPI',
'boosapi.image_to_video': 'BoosAPI 图生视频',
'boosapi.text_to_video': 'BoosAPI 文生视频',
'boosapi.video_status': '查询视频任务',
'output.preview_video': '输出:预览视频',
'output.preview_image': '输出:预览图像',
'input.text': '输入:文本',
'boosapi.chat_completion': 'BoosAPI 对话',
'boosapi.prompt_rewrite': 'BoosAPI 提示词改写',
'boosapi.text_to_image': 'BoosAPI 文生图',
'output.save_file': '输出:保存文件'
}));
const translateWorkflowText = (text) => {
const raw = String(text || '');
const trimmed = raw.replace(/\s+/g, ' ').trim();
if (!trimmed) return null;
if (workflowTextMap.has(trimmed)) return workflowTextMap.get(trimmed);
return null;
};
const localizeWorkflowMenus = () => {
const selector = [
'.litecontextmenu .litemenu-entry',
'.litecontextmenu .litemenu-title',
'.litecontextmenu .litemenu-submenu',
'.litecontextmenu li',
'.litecontextmenu div',
'.p-contextmenu [role="menuitem"]',
'.p-contextmenu .p-menuitem-text',
'.p-tieredmenu [role="menuitem"]',
'.p-tieredmenu .p-menuitem-text',
'[role="menu"] [role="menuitem"]'
].join(',');
for (const el of document.querySelectorAll(selector)) {
if (!el || el.children.length > 1) continue;
const textNode = Array.from(el.childNodes).find((node) => node.nodeType === Node.TEXT_NODE && String(node.textContent || '').trim());
const target = textNode || (el.children.length === 0 ? el : null);
if (!target) continue;
const translated = translateWorkflowText(target.textContent);
if (translated && target.textContent !== translated) target.textContent = translated;
}
for (const el of document.querySelectorAll('[title],[aria-label]')) {
for (const attr of ['title', 'aria-label']) {
const translated = translateWorkflowText(el.getAttribute(attr));
if (translated) el.setAttribute(attr, translated);
}
}
};
const observeWorkflowMenuLocalization = () => {
if (window.__boosapiWorkflowMenuI18nObserver) return;
window.__boosapiWorkflowMenuI18nObserver = true;
const observer = new MutationObserver((mutations) => {
if (mutations.some((item) => item.addedNodes && item.addedNodes.length)) {
window.requestAnimationFrame(localizeWorkflowMenus);
}
});
observer.observe(document.body, { childList: true, subtree: true });
};
const normalizeWorkflowPortType = (value) => {
const normalized = String(value || '').trim().toLowerCase();
if (normalized === 'conditioning' || normalized === 'condition' || normalized === 'cond' || normalized === '条件' || normalized === '条件化' || normalized === '正面条件' || normalized === '负面条件') return 'conditioning';
if (normalized === 'latent' || normalized === '潜空间') return 'latent';
if (normalized === 'model' || normalized === '模型') return 'model';
if (normalized === 'clip') return 'clip';
if (normalized === 'vae') return 'vae';
if (normalized === 'image' || normalized === '图像' || normalized === '图片') return 'image';
if (normalized === 'video' || normalized === '视频') return 'video';
return normalized;
};
const normalizeWorkflowSlotType = (slot) => {
if (!slot) return;
const normalized = normalizeWorkflowPortType(slot.type || slot.name || slot.localized_name);
const canonical = {
conditioning: 'CONDITIONING',
latent: 'LATENT',
model: 'MODEL',
clip: 'CLIP',
vae: 'VAE',
image: 'IMAGE',
video: 'VIDEO'
}[normalized];
if (canonical) slot.type = canonical;
};
const slotName = (slot) => String(slot?.name || slot?.localized_name || slot?.label || '').trim().toLowerCase();
const normalizeWorkflowNodeSlots = (node) => {
try {
if (!node) return;
if (node.type === 'CLIPTextEncode' && Array.isArray(node.outputs)) {
for (const output of node.outputs) {
const name = slotName(output);
if (name.includes('conditioning') || name.includes('condition') || name.includes('条件') || name === 'conditioning') {
output.type = 'CONDITIONING';
output.name = output.name || 'CONDITIONING';
output.localized_name = output.localized_name || '条件';
}
}
}
if (node.type === 'KSampler' && Array.isArray(node.inputs)) {
for (const input of node.inputs) {
const name = slotName(input);
if (name === 'positive' || name === 'negative' || name.includes('positive') || name.includes('negative') || name.includes('正面条件') || name.includes('负面条件')) {
input.type = 'CONDITIONING';
}
}
}
node.inputs?.forEach(normalizeWorkflowSlotType);
node.outputs?.forEach(normalizeWorkflowSlotType);
} catch (err) {}
};
const normalizeWorkflowGraphSlots = () => {
try {
const graph = window.app?.graph || window.LGraphCanvas?.active_canvas?.graph;
if (!graph?.nodes) return;
for (const node of graph.nodes) normalizeWorkflowNodeSlots(node);
graph.setDirtyCanvas?.(true, true);
window.app?.canvas?.setDirty?.(true, true);
} catch (err) {}
};
const patchWorkflowConnectionTypes = () => {
try {
const liteGraph = window.LiteGraph;
if (!liteGraph || liteGraph.__boosapiConnectionPatch) return;
const original = liteGraph.isValidConnection;
if (typeof original !== 'function') return;
liteGraph.isValidConnection = function (typeA, typeB) {
const a = normalizeWorkflowPortType(typeA);
const b = normalizeWorkflowPortType(typeB);
if (a && b && a === b) return true;
return original.call(this, typeA, typeB);
};
const proto = window.LGraphNode?.prototype;
if (proto?.connect && !proto.__boosapiConnectPatch) {
const originalConnect = proto.connect;
proto.connect = function (slot, targetNode, targetSlot, afterRerouteId) {
normalizeWorkflowNodeSlots(this);
normalizeWorkflowNodeSlots(targetNode);
return originalConnect.call(this, slot, targetNode, targetSlot, afterRerouteId);
};
proto.__boosapiConnectPatch = true;
}
const originalCreateNode = liteGraph.createNode;
if (typeof originalCreateNode === 'function' && !liteGraph.__boosapiCreateNodePatch) {
liteGraph.createNode = function () {
const node = originalCreateNode.apply(this, arguments);
normalizeWorkflowNodeSlots(node);
return node;
};
liteGraph.__boosapiCreateNodePatch = true;
}
liteGraph.__boosapiConnectionPatch = true;
normalizeWorkflowGraphSlots();
} catch (err) {}
};
const startWorkflowSlotNormalizer = () => {
if (window.__boosapiWorkflowSlotNormalizer) return;
window.__boosapiWorkflowSlotNormalizer = true;
window.setInterval(normalizeWorkflowGraphSlots, 1000);
window.setTimeout(normalizeWorkflowGraphSlots, 100);
window.setTimeout(normalizeWorkflowGraphSlots, 600);
window.setTimeout(normalizeWorkflowGraphSlots, 1500);
};
const workflowConnectionQuickNodes = {
model: [
{ title: 'K 采样器', types: ['KSampler'], hint: '把模型接入采样流程' },
{ title: 'LoRA 加载器', types: ['LoraLoader', 'LoraLoaderModelOnly'], hint: '给模型叠加 LoRA' },
{ title: '模型采样设置', types: ['ModelSamplingDiscrete', 'ModelSamplingContinuousEDM'], hint: '调整模型采样方式' },
{ title: '保存 Checkpoint', types: ['CheckpointSave'], hint: '保存模型/CLIP/VAE 组合' }
],
clip: [
{ title: 'CLIP 文本编码', types: ['CLIPTextEncode'], hint: '输入提示词生成条件' },
{ title: 'CLIP 跳层', types: ['CLIPSetLastLayer'], hint: '设置 CLIP skip' },
{ title: 'LoRA 加载器', types: ['LoraLoader'], hint: '同时接入模型与 CLIP' },
{ title: '保存 Checkpoint', types: ['CheckpointSave'], hint: '保存模型/CLIP/VAE 组合' }
],
vae: [
{ title: 'VAE 解码', types: ['VAEDecode', 'VAEDecodeTiled'], hint: '潜空间转图像' },
{ title: 'VAE 编码', types: ['VAEEncode', 'VAEEncodeTiled'], hint: '图像转潜空间' },
{ title: '保存 Checkpoint', types: ['CheckpointSave'], hint: '保存模型/CLIP/VAE 组合' }
],
image: [
{ title: '预览图像', types: ['PreviewImage'], hint: '查看这一条图像输出' },
{ title: '保存图像', types: ['SaveImage'], hint: '保存到输出目录' },
{ title: 'VAE 编码', types: ['VAEEncode', 'VAEEncodeTiled'], hint: '图像转潜空间' },
{ title: '放大图像', types: ['ImageScale', 'ImageScaleBy', 'ImageUpscaleWithModel'], hint: '调整尺寸/放大' },
{ title: '图像对比', types: ['Image Comparer (rgthree)', 'Image Comparer', 'ImageCompare'], hint: '对比输入图像' },
{ title: '图像转遮罩', types: ['ImageToMask'], hint: '从图像提取遮罩' }
],
latent: [
{ title: 'VAE 解码', types: ['VAEDecode', 'VAEDecodeTiled'], hint: '潜空间转图像' },
{ title: 'K 采样器', types: ['KSampler'], hint: '继续采样' },
{ title: '潜空间放大', types: ['LatentUpscale', 'LatentUpscaleBy'], hint: '调整潜空间尺寸' },
{ title: '潜空间合成', types: ['LatentComposite'], hint: '合成两个潜空间' }
],
conditioning: [
{ title: 'K 采样器', types: ['KSampler'], hint: '接入正/负条件' },
{ title: '合并条件', types: ['ConditioningCombine', 'ConditioningConcat'], hint: '组合提示条件' },
{ title: '条件平均', types: ['ConditioningAverage'], hint: '混合两组条件' },
{ title: '条件区域', types: ['ConditioningSetArea'], hint: '限制条件作用区域' }
],
mask: [
{ title: '遮罩转图像', types: ['MaskToImage'], hint: '查看/保存遮罩' },
{ title: '图像转遮罩', types: ['ImageToMask'], hint: '从图像提取遮罩' },
{ title: '遮罩合成', types: ['MaskComposite'], hint: '组合遮罩' },
{ title: '遮罩扩张', types: ['GrowMask'], hint: '扩大或收缩遮罩' }
],
video: [
{ title: '预览视频', types: ['VHS_VideoCombine', 'Video Combine 🎥🅥🅗🅢', 'VideoCombine'], hint: '合成为可预览视频' },
{ title: '保存动画', types: ['SaveAnimatedPNG', 'SaveAnimatedWEBP'], hint: '保存序列动画' },
{ title: '加载视频', types: ['VHS_LoadVideo', 'LoadVideo'], hint: '读取视频素材' }
],
audio: [
{ title: '预览音频', types: ['PreviewAudio', 'SaveAudio'], hint: '试听或保存音频' },
{ title: '加载音频', types: ['LoadAudio'], hint: '读取音频素材' }
],
string: [
{ title: '文本输入', types: ['PrimitiveNode', 'StringConstant', 'Text Input'], hint: '创建文本值' },
{ title: '显示文本', types: ['ShowText|pysssss', 'ShowText', 'PreviewText'], hint: '查看文本输出' }
],
int: [
{ title: '数值输入', types: ['PrimitiveNode', 'Integer Constant'], hint: '创建整数值' }
],
float: [
{ title: '数值输入', types: ['PrimitiveNode', 'Float Constant'], hint: '创建小数值' }
],
boolean: [
{ title: '开关输入', types: ['PrimitiveNode', 'Boolean Constant'], hint: '创建布尔值' }
]
};
const findWorkflowNodeType = (candidates) => {
const registry = window.LiteGraph?.registered_node_types || {};
for (const name of candidates) {
if (registry[name]) return name;
}
const entries = Object.keys(registry);
for (const name of candidates) {
const normalized = String(name).toLowerCase().replace(/[\s_-]+/g, '');
const found = entries.find((item) => String(item).toLowerCase().replace(/[\s_-]+/g, '') === normalized);
if (found) return found;
}
return '';
};
const getWorkflowConnectionType = (options) => {
try {
const link = options || window.__boosapiLastConnectionMenuOptions || {};
if (link.nodeFrom && Number.isInteger(link.slotFrom)) {
return normalizeWorkflowPortType(link.nodeFrom.outputs?.[link.slotFrom]?.type || link.nodeFrom.outputs?.[link.slotFrom]?.name);
}
if (link.nodeTo && Number.isInteger(link.slotTo)) {
return normalizeWorkflowPortType(link.nodeTo.inputs?.[link.slotTo]?.type || link.nodeTo.inputs?.[link.slotTo]?.name);
}
const title = String(document.querySelector('.litecontextmenu .litemenu-title')?.textContent || '');
if (title.includes('IMAGE') || title.includes('图像')) return 'image';
if (title.includes('LATENT') || title.includes('潜空间')) return 'latent';
if (title.includes('CONDITIONING') || title.includes('条件')) return 'conditioning';
if (title.includes('MODEL') || title.includes('模型')) return 'model';
if (title.includes('CLIP')) return 'clip';
if (title.includes('VAE')) return 'vae';
if (title.includes('MASK') || title.includes('遮罩')) return 'mask';
if (title.includes('VIDEO') || title.includes('视频')) return 'video';
if (title.includes('AUDIO') || title.includes('音频')) return 'audio';
if (title.includes('STRING') || title.includes('TEXT') || title.includes('文本')) return 'string';
if (title.includes('INT') || title.includes('整数')) return 'int';
if (title.includes('FLOAT') || title.includes('NUMBER') || title.includes('数字')) return 'float';
if (title.includes('BOOLEAN') || title.includes('BOOL') || title.includes('布尔')) return 'boolean';
} catch (err) {}
return '';
};
const workflowNodeDisplayTitle = (nodeType) => {
const registry = window.LiteGraph?.registered_node_types || {};
const ctor = registry[nodeType];
return String(ctor?.title || ctor?.nodeData?.display_name || ctor?.comfyClass || nodeType || '').trim();
};
const fallbackWorkflowConnectionItems = (connectionType) => {
try {
const options = window.__boosapiLastConnectionMenuOptions || {};
const liteGraph = window.LiteGraph || {};
const canonical = {
image: 'IMAGE',
latent: 'LATENT',
conditioning: 'CONDITIONING',
model: 'MODEL',
clip: 'CLIP',
vae: 'VAE',
mask: 'MASK',
video: 'VIDEO',
audio: 'AUDIO',
string: 'STRING',
int: 'INT',
float: 'FLOAT',
boolean: 'BOOLEAN'
}[connectionType] || String(connectionType || '').toUpperCase();
const defaults = options.nodeFrom
? liteGraph.slot_types_default_out?.[canonical]
: liteGraph.slot_types_default_in?.[canonical];
const registered = options.nodeFrom
? liteGraph.registered_slot_in_types?.[String(canonical).toLowerCase()]?.nodes
: liteGraph.registered_slot_out_types?.[canonical]?.nodes;
const names = [...(Array.isArray(defaults) ? defaults : []), ...(Array.isArray(registered) ? registered : [])]
.map((value) => typeof value === 'string' ? value : value?.comfyClass || value?.type || value?.title)
.filter(Boolean);
return [...new Set(names)].slice(0, 8).map((nodeType) => ({
title: workflowNodeDisplayTitle(nodeType),
types: [nodeType],
hint: `${canonical} 兼容节点`
}));
} catch (err) {
return [];
}
};
const findCompatibleSlotIndex = (slots, type) => {
const wanted = normalizeWorkflowPortType(type);
if (!Array.isArray(slots)) return 0;
const exact = slots.findIndex((slot) => normalizeWorkflowPortType(slot?.type || slot?.name || slot?.localized_name) === wanted);
return exact >= 0 ? exact : 0;
};
const connectedNodePosition = (options) => {
const event = options?.e || {};
if (typeof event.canvasX === 'number' && typeof event.canvasY === 'number') {
return [event.canvasX + 24, event.canvasY - 28];
}
try {
const canvas = window.app?.canvas || window.LGraphCanvas?.active_canvas;
const area = canvas?.ds?.visible_area;
if (Array.isArray(area)) return [area[0] + area[2] / 2 - 120, area[1] + area[3] / 2 - 80];
} catch (err) {}
return [180, 180];
};
const addConnectedWorkflowQuickNode = (item) => {
try {
const options = window.__boosapiLastConnectionMenuOptions || {};
const nodeType = findWorkflowNodeType(item.types || []);
const graph = window.app?.graph || window.LGraphCanvas?.active_canvas?.graph;
const canvas = window.app?.canvas || window.LGraphCanvas?.active_canvas;
if (!nodeType || !graph || !window.LiteGraph?.createNode) return false;
const node = window.LiteGraph.createNode(nodeType, item.title || nodeType);
if (!node) return false;
node.pos = connectedNodePosition(options);
graph.add(node);
normalizeWorkflowNodeSlots(node);
if (options.nodeFrom && Number.isInteger(options.slotFrom)) {
const output = options.nodeFrom.outputs?.[options.slotFrom];
const inputIndex = findCompatibleSlotIndex(node.inputs, output?.type || output?.name);
options.nodeFrom.connect(options.slotFrom, node, inputIndex, options.afterRerouteId);
} else if (options.nodeTo && Number.isInteger(options.slotTo)) {
const input = options.nodeTo.inputs?.[options.slotTo];
const outputIndex = findCompatibleSlotIndex(node.outputs, input?.type || input?.name);
node.connect(outputIndex, options.nodeTo, options.slotTo, options.afterRerouteId);
}
graph.setDirtyCanvas?.(true, true);
canvas?.selectNode?.(node);
if (window.LiteGraph?.closeAllContextMenus) window.LiteGraph.closeAllContextMenus(window);
return true;
} catch (err) {
return false;
}
};
const installWorkflowConnectionMenuMemory = () => {
try {
const proto = window.LGraphCanvas?.prototype;
if (!proto?.showConnectionMenu || proto.__boosapiConnectionMenuMemory) return;
const original = proto.showConnectionMenu;
proto.showConnectionMenu = function (options) {
window.__boosapiLastConnectionMenuOptions = options || {};
const result = original.apply(this, arguments);
window.setTimeout(enhanceWorkflowConnectionMenu, 0);
window.setTimeout(enhanceWorkflowConnectionMenu, 80);
return result;
};
proto.__boosapiConnectionMenuMemory = true;
} catch (err) {}
};
const enhanceWorkflowConnectionMenu = () => {
try {
const menu = document.querySelector('.litecontextmenu');
if (!menu || menu.__boosapiConnectionSuggestions) return;
const connectionType = getWorkflowConnectionType();
const items = [
...(workflowConnectionQuickNodes[connectionType] || []),
...fallbackWorkflowConnectionItems(connectionType)
];
if (!items?.length) return;
const available = items
.map((item) => ({ ...item, nodeType: findWorkflowNodeType(item.types || []) }))
.filter((item) => item.nodeType);
if (!available.length) return;
menu.__boosapiConnectionSuggestions = true;
const panel = document.createElement('div');
panel.className = 'boosapi-connection-suggestions';
panel.style.cssText = 'padding:8px 8px 6px;border-bottom:1px solid rgba(255,255,255,.12);background:rgba(20,24,30,.96);min-width:260px;';
const title = document.createElement('div');
title.textContent = '最相关';
title.style.cssText = 'font-size:12px;font-weight:700;color:#e9f7f5;margin:0 0 6px 2px;';
panel.appendChild(title);
for (const item of available.slice(0, 6)) {
const button = document.createElement('button');
button.type = 'button';
button.textContent = item.title;
button.title = item.hint || item.nodeType;
button.style.cssText = 'display:flex;width:100%;align-items:center;justify-content:space-between;gap:12px;margin:2px 0;padding:8px 10px;border:0;border-radius:6px;background:transparent;color:#fff;text-align:left;font-size:13px;cursor:pointer;';
button.addEventListener('mouseenter', () => { button.style.background = 'rgba(20,184,166,.22)'; });
button.addEventListener('mouseleave', () => { button.style.background = 'transparent'; });
button.addEventListener('click', (event) => {
event.preventDefault();
event.stopPropagation();
addConnectedWorkflowQuickNode(item);
}, true);
panel.appendChild(button);
}
menu.insertBefore(panel, menu.firstChild);
} catch (err) {}
};
const installWorkflowConnectionMenuObserver = () => {
if (window.__boosapiWorkflowConnectionMenuObserver) return;
window.__boosapiWorkflowConnectionMenuObserver = true;
const observer = new MutationObserver((mutations) => {
if (mutations.some((item) => item.addedNodes && item.addedNodes.length)) {
window.requestAnimationFrame(enhanceWorkflowConnectionMenu);
}
});
observer.observe(document.body, { childList: true, subtree: true });
};
const boostWorkflowSlotDefaults = () => {
try {
const liteGraph = window.LiteGraph;
if (!liteGraph) return;
const prepend = (target, type, values) => {
const current = Array.isArray(target[type]) ? target[type] : [];
target[type] = [...values.filter((value) => findWorkflowNodeType([value])), ...current.filter((value) => !values.includes(value))].slice(0, 24);
};
liteGraph.slot_types_default_out ||= {};
liteGraph.slot_types_default_in ||= {};
prepend(liteGraph.slot_types_default_out, 'MODEL', ['KSampler', 'LoraLoader', 'ModelSamplingDiscrete']);
prepend(liteGraph.slot_types_default_out, 'CLIP', ['CLIPTextEncode', 'CLIPSetLastLayer', 'LoraLoader']);
prepend(liteGraph.slot_types_default_out, 'VAE', ['VAEDecode', 'VAEEncode']);
prepend(liteGraph.slot_types_default_out, 'IMAGE', ['PreviewImage', 'SaveImage', 'VAEEncode', 'ImageScale', 'ImageScaleBy']);
prepend(liteGraph.slot_types_default_in, 'IMAGE', ['LoadImage', 'EmptyImage', 'PreviewImage']);
prepend(liteGraph.slot_types_default_out, 'LATENT', ['VAEDecode', 'KSampler']);
prepend(liteGraph.slot_types_default_out, 'CONDITIONING', ['KSampler', 'ConditioningCombine']);
prepend(liteGraph.slot_types_default_out, 'MASK', ['MaskToImage', 'ImageToMask', 'MaskComposite', 'GrowMask']);
prepend(liteGraph.slot_types_default_out, 'VIDEO', ['VHS_VideoCombine', 'VideoCombine', 'SaveAnimatedPNG']);
prepend(liteGraph.slot_types_default_out, 'AUDIO', ['PreviewAudio', 'SaveAudio', 'LoadAudio']);
prepend(liteGraph.slot_types_default_out, 'STRING', ['PrimitiveNode', 'ShowText', 'PreviewText']);
prepend(liteGraph.slot_types_default_out, 'INT', ['PrimitiveNode']);
prepend(liteGraph.slot_types_default_out, 'FLOAT', ['PrimitiveNode']);
prepend(liteGraph.slot_types_default_out, 'BOOLEAN', ['PrimitiveNode']);
} catch (err) {}
};
const isWorkflowVideoModelName = (modelName) => {
const value = String(modelName || '').toLowerCase();
return ['seedance', 'wan2', 't2v', 'i2v', 'flf2v', 'video', 'seaweed'].some((keyword) => value.includes(keyword));
};
const addTextToImageNodeFromModel = (modelName) => {
try {
const name = String(modelName || '').trim();
if (!name || !window.LiteGraph) return false;
const app = window.app;
const graph = app?.graph || window.LGraphCanvas?.active_canvas?.graph;
const canvas = app?.canvas || window.LGraphCanvas?.active_canvas;
if (!graph) return false;
const node = window.LiteGraph.createNode('boosapi.text_to_image', 'BoosAPI 文生图');
if (!node) return false;
const area = canvas?.ds?.visible_area;
node.pos = Array.isArray(area) ? [area[0] + area[2] / 2 - 190, area[1] + area[3] / 2 - 120] : [160, 160];
graph.add(node);
normalizeWorkflowNodeSlots(node);
const widget = node.widgets?.find((item) => item.name === 'model');
if (widget) widget.value = name;
node.properties ||= {};
node.properties['Node name for S&R'] = 'boosapi.text_to_image';
graph.setDirtyCanvas?.(true, true);
canvas?.selectNode?.(node);
return true;
} catch (err) {
return false;
}
};
const addTextToVideoNodeFromModel = (modelName) => {
try {
const name = String(modelName || '').trim();
if (!name || !window.LiteGraph) return false;
const app = window.app;
const graph = app?.graph || window.LGraphCanvas?.active_canvas?.graph;
const canvas = app?.canvas || window.LGraphCanvas?.active_canvas;
if (!graph) return false;
const node = window.LiteGraph.createNode('boosapi.text_to_video', 'BoosAPI 文生视频');
if (!node) return false;
const area = canvas?.ds?.visible_area;
node.pos = Array.isArray(area) ? [area[0] + area[2] / 2 - 190, area[1] + area[3] / 2 - 120] : [160, 160];
graph.add(node);
normalizeWorkflowNodeSlots(node);
const widget = node.widgets?.find((item) => item.name === 'model');
if (widget) widget.value = name;
node.properties ||= {};
node.properties['Node name for S&R'] = 'boosapi.text_to_video';
graph.setDirtyCanvas?.(true, true);
canvas?.selectNode?.(node);
return true;
} catch (err) {
return false;
}
};
const patchModelLibraryClicks = () => {
document.querySelectorAll('#model-library .p-tree-node-content, #model-library [role="treeitem"], .model-lib-tree-explorer .p-tree-node-content').forEach((el) => {
if (el.__boosapiModelClickPatch) return;
el.__boosapiModelClickPatch = true;
el.addEventListener('dblclick', (event) => {
const text = String(el.textContent || '').replace(/\s+/g, ' ').trim();
if (!text || text.includes('checkpoints')) return;
const added = isWorkflowVideoModelName(text)
? addTextToVideoNodeFromModel(text)
: addTextToImageNodeFromModel(text);
if (added) {
event.preventDefault();
event.stopPropagation();
}
}, true);
});
};
const getCurrentWorkflowJson = async () => {
try {
if (window.app?.graphToPrompt) {
const result = await window.app.graphToPrompt();
return result?.workflow || result?.output || result?.prompt || result;
}
} catch (err) {}
try {
const graph = window.app?.graph || window.LGraphCanvas?.active_canvas?.graph;
if (graph?.serialize) return graph.serialize();
} catch (err) {}
return null;
};
const showWorkflowValidationReport = (report) => {
const lines = [];
const summary = report?.summary || {};
lines.push(`节点 ${summary.nodes || 0},连线 ${summary.links || 0},错误 ${summary.errors || 0},警告 ${summary.warnings || 0}`);
for (const item of report?.missingNodeTypes || []) {
lines.push(`缺失节点: ${item.nodeType} #${item.nodeId}${item.connected ? '(已连接)' : '(未连接)'}`);
}
for (const item of report?.emptyModelFields || []) {
lines.push(`空模型列表: ${item.nodeType} #${item.nodeId} ${item.field} = ${item.selected}`);
}
for (const item of report?.missingModelChoices || []) {
lines.push(`缺失模型: ${item.nodeType} #${item.nodeId} ${item.field} = ${item.selected}`);
if (item.near?.length) lines.push(` 相近: ${item.near.join(', ')}`);
}
for (const item of report?.inputFiles || []) {
if (item.checked) lines.push(`素材: ${item.filename} ${item.reachable ? '可访问' : '不可访问'}`);
}
if (report?.hints?.length) lines.push('', ...report.hints);
window.alert(lines.join('\n') || '没有发现明显问题');
};
const runWorkflowValidation = async () => {
const workflow = await getCurrentWorkflowJson();
if (!workflow) {
window.alert('没有读取到当前工作流');
return;
}
const response = await fetch('/workflow-ui/api/workflow/validate?check_files=true', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(workflow),
});
const report = await response.json();
showWorkflowValidationReport(report);
};
const installWorkflowValidationButton = () => {
if (document.getElementById('boosapi-workflow-validate-button')) return;
const host = document.querySelector('header, nav, .p-menubar, .comfy-menu, .comfyui-menu') || document.body;
const button = document.createElement('button');
button.id = 'boosapi-workflow-validate-button';
button.type = 'button';
button.textContent = '检查模板';
button.title = '检查当前工作流的节点、模型和输入素材兼容性';
button.style.cssText = 'position:fixed;right:112px;top:10px;z-index:10000;height:32px;padding:0 12px;border:1px solid rgba(120,130,150,.45);border-radius:6px;background:rgba(20,24,32,.88);color:#fff;font-size:13px;cursor:pointer;';
button.addEventListener('click', async () => {
button.disabled = true;
const oldText = button.textContent;
button.textContent = '检查中...';
try {
await runWorkflowValidation();
} catch (err) {
window.alert('检查失败:' + (err?.message || err));
} finally {
button.disabled = false;
button.textContent = oldText;
}
});
host.appendChild(button);
};
try {
const userJwt = localStorage.getItem('user_jwt_token');
const adminToken = localStorage.getItem('auth_token');
const token = adminToken || userJwt;
if (token) {
localStorage.setItem('auth_token', token);
document.cookie = 'boosapi_workflow_token=' + encodeURIComponent(token) + '; Path=/workflow-ui; SameSite=Lax';
}
} catch (err) {}
let shellPatchQueued = false;
const patchWorkflowShell = () => {
shellPatchQueued = false;
installBoosApiFetchAuth();
patchWorkflowConnectionTypes();
startWorkflowSlotNormalizer();
installWorkflowConnectionMenuMemory();
installWorkflowConnectionMenuObserver();
boostWorkflowSlotDefaults();
enhanceWorkflowConnectionMenu();
patchModelLibraryClicks();
installWorkflowValidationButton();
localizeWorkflowMenus();
observeWorkflowMenuLocalization();
renameComfyUiBrand();
};
const queueWorkflowShellPatch = () => {
if (shellPatchQueued) return;
shellPatchQueued = true;
window.setTimeout(patchWorkflowShell, 120);
};
document.addEventListener('click', queueWorkflowShellPatch, true);
window.setTimeout(patchWorkflowShell, 250);
window.setTimeout(patchWorkflowShell, 1200);
})();
</script>"#;
let bootstrap = format!("{secret_cleanup_bootstrap}{shell}");
html.replace("</head>", &format!("{bootstrap}\n</head>"))
}
+321
View File
@@ -0,0 +1,321 @@
use super::auth::{auth_caller, caller_is_admin, caller_user_id};
use super::models::workflow_route_models;
use super::node_defs::{inject_boosapi_workflow_template_payload, workflow_template_payload};
use super::{
auth_caller_optional, inject_boosapi_object_info, native_workflow_json_cached,
native_workflow_metadata_ttl, native_workflow_status_ttl, parse_boolish, safe_view_path,
user_home, workflow_node_defs_with_models, UserDataQuery, WorkflowCaller,
};
use crate::{
error::{AppError, AppResult},
AppState,
};
use axum::{
extract::{Path, Query, State},
http::{header, HeaderMap, Method, StatusCode},
response::{IntoResponse, Response},
Json,
};
use serde_json::{json, Value};
use std::collections::HashMap;
pub(super) async fn get_user_config(
State(state): State<AppState>,
headers: HeaderMap,
) -> AppResult<Json<Value>> {
let caller = auth_caller(&state, &headers).await?;
let audience = match caller {
WorkflowCaller::Admin => "admin",
WorkflowCaller::User { .. } => "user",
};
Ok(Json(json!({
"username": audience,
"role": audience,
"id": caller_user_id(&caller),
})))
}
pub(super) async fn create_user(
State(state): State<AppState>,
headers: HeaderMap,
) -> AppResult<Response> {
let _ = auth_caller(&state, &headers).await?;
Ok((
StatusCode::METHOD_NOT_ALLOWED,
Json(json!({ "error": "read-only" })),
)
.into_response())
}
pub(super) async fn get_settings(
State(state): State<AppState>,
headers: HeaderMap,
) -> AppResult<Json<Value>> {
let _ = auth_caller(&state, &headers).await?;
if state.config.comfyui.uses_cloud_backend() {
return Ok(Json(json!({
"Comfy.Workflow.ShowMissingModelsWarning": false
})));
}
Ok(Json(json!({})))
}
pub(super) async fn get_setting(
State(state): State<AppState>,
headers: HeaderMap,
Path(id): Path<String>,
) -> AppResult<Json<Value>> {
let _ = auth_caller(&state, &headers).await?;
if state.config.comfyui.uses_cloud_backend() && id == "Comfy.Workflow.ShowMissingModelsWarning"
{
return Ok(Json(Value::Bool(false)));
}
Ok(Json(Value::Null))
}
pub(super) async fn store_settings(
State(state): State<AppState>,
headers: HeaderMap,
Json(_body): Json<Value>,
) -> AppResult<Json<Value>> {
let _ = auth_caller(&state, &headers).await?;
Ok(Json(json!({ "success": true })))
}
pub(super) async fn store_setting(
State(state): State<AppState>,
headers: HeaderMap,
Path(_id): Path<String>,
Json(_body): Json<Value>,
) -> AppResult<Json<Value>> {
let _ = auth_caller(&state, &headers).await?;
Ok(Json(json!({ "success": true })))
}
pub(super) async fn userdata_list(
State(state): State<AppState>,
headers: HeaderMap,
Query(query): Query<UserDataQuery>,
) -> AppResult<Response> {
let caller =
auth_caller_optional(&state, &headers).unwrap_or(WorkflowCaller::User { user_id: 0 });
let user_id = caller_user_id(&caller);
let dir = query.dir.unwrap_or_else(|| "workflow".to_string());
let base = user_home(user_id).join("userdata").join(&dir);
if tokio::fs::metadata(&base).await.is_err() {
return Ok(Json(json!([])).into_response());
}
if query.full_info {
let mut entries = Vec::new();
let mut reader = match tokio::fs::read_dir(&base).await {
Ok(value) => value,
Err(_) => return Ok(Json(json!([])).into_response()),
};
while let Ok(Some(entry)) = reader.next_entry().await {
let path = entry.path();
let meta = match entry.metadata().await {
Ok(meta) => meta,
Err(_) => continue,
};
entries.push(json!({
"path": path.to_string_lossy().to_string(),
"name": entry.file_name().to_string_lossy().to_string(),
"type": if meta.is_dir() { "directory" } else { "file" },
"size": meta.len(),
}));
}
return Ok(Json(Value::Array(entries)).into_response());
}
let mut items = Vec::new();
let mut reader = match tokio::fs::read_dir(&base).await {
Ok(value) => value,
Err(_) => return Ok(Json(json!([])).into_response()),
};
while let Ok(Some(entry)) = reader.next_entry().await {
let name = entry.file_name().to_string_lossy().to_string();
if query.split {
items.push(json!([name.clone()]));
} else {
items.push(Value::String(name));
}
}
Ok(Json(Value::Array(items)).into_response())
}
pub(super) async fn userdata_get(
State(state): State<AppState>,
headers: HeaderMap,
Path(file): Path<String>,
) -> AppResult<Response> {
let caller = auth_caller(&state, &headers).await?;
let root = user_home(caller_user_id(&caller)).join("userdata");
let path = root.join(safe_view_path(&file));
let bytes = tokio::fs::read(&path)
.await
.map_err(|_| AppError::NotFound("User data file not found".into()))?;
Ok((
StatusCode::OK,
[(header::CONTENT_TYPE, "application/octet-stream")],
bytes,
)
.into_response())
}
pub(super) async fn userdata_store(
State(state): State<AppState>,
headers: HeaderMap,
Path(file): Path<String>,
Query(query): Query<HashMap<String, String>>,
body: axum::body::Bytes,
) -> AppResult<Response> {
let caller = auth_caller(&state, &headers).await?;
let root = user_home(caller_user_id(&caller)).join("userdata");
let path = root.join(safe_view_path(&file));
if let Some(parent) = path.parent() {
tokio::fs::create_dir_all(parent)
.await
.map_err(|err| AppError::Internal(anyhow::anyhow!(err)))?;
}
tokio::fs::write(&path, body)
.await
.map_err(|err| AppError::Internal(anyhow::anyhow!(err)))?;
Ok(Json(json!({
"name": file,
"subfolder": query.get("subfolder").cloned().unwrap_or_default(),
"type": query.get("type").cloned().unwrap_or_else(|| "input".to_string())
}))
.into_response())
}
pub(super) async fn userdata_delete(
State(state): State<AppState>,
headers: HeaderMap,
Path(file): Path<String>,
) -> AppResult<Response> {
let caller = auth_caller(&state, &headers).await?;
let root = user_home(caller_user_id(&caller)).join("userdata");
let path = root.join(safe_view_path(&file));
let _ = tokio::fs::remove_file(&path).await;
Ok(StatusCode::NO_CONTENT.into_response())
}
pub(super) async fn move_userdata(
State(state): State<AppState>,
headers: HeaderMap,
Path((source, dest)): Path<(String, String)>,
Query(query): Query<HashMap<String, String>>,
) -> AppResult<Response> {
let caller = auth_caller(&state, &headers).await?;
let root = user_home(caller_user_id(&caller)).join("userdata");
let src = root.join(safe_view_path(&source));
let dst = root.join(safe_view_path(&dest));
if let Some(parent) = dst.parent() {
tokio::fs::create_dir_all(parent)
.await
.map_err(|err| AppError::Internal(anyhow::anyhow!(err)))?;
}
let overwrite = parse_boolish(query.get("overwrite").map(String::as_str));
if !overwrite && tokio::fs::metadata(&dst).await.is_ok() {
return Err(AppError::BadRequest("destination already exists".into()));
}
tokio::fs::rename(&src, &dst)
.await
.map_err(|err| AppError::Internal(anyhow::anyhow!(err)))?;
Ok(Json(json!({ "success": true })).into_response())
}
pub(super) async fn extensions(
State(state): State<AppState>,
headers: HeaderMap,
) -> AppResult<Json<Value>> {
let _ = auth_caller_optional(&state, &headers);
Ok(Json(json!(["extensions/core"])))
}
pub(super) async fn workflow_templates(
State(state): State<AppState>,
headers: HeaderMap,
) -> AppResult<Json<Value>> {
let _ = auth_caller_optional(&state, &headers);
if state.config.comfyui.uses_cloud_backend() {
let mut value = native_workflow_json_cached(
&state,
Method::GET,
"/workflow_templates",
None,
native_workflow_metadata_ttl(&state),
)
.await
.map(|value| value.0)
.unwrap_or_else(|_| json!({}));
inject_boosapi_workflow_template_payload(&mut value);
return Ok(Json(value));
}
Ok(Json(workflow_template_payload()))
}
pub(super) async fn embeddings(
State(state): State<AppState>,
headers: HeaderMap,
) -> AppResult<Json<Value>> {
let _ = auth_caller_optional(&state, &headers);
Ok(Json(json!([])))
}
pub(super) 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() {
let mut value = native_workflow_json_cached(
&state,
Method::GET,
"/object_info",
None,
native_workflow_metadata_ttl(&state),
)
.await?
.0;
inject_boosapi_object_info(&mut value, &route_models);
return Ok(Json(value));
}
Ok(Json(Value::Object(workflow_node_defs_with_models(
&route_models,
true,
))))
}
pub(super) async fn system_stats(
State(state): State<AppState>,
headers: HeaderMap,
) -> AppResult<Json<Value>> {
if state.config.comfyui.uses_native_backend() {
let _ = auth_caller_optional(&state, &headers);
return native_workflow_json_cached(
&state,
Method::GET,
"/system_stats",
None,
native_workflow_status_ttl(&state),
)
.await;
}
let caller =
auth_caller_optional(&state, &headers).unwrap_or(WorkflowCaller::User { user_id: 0 });
let host = if state.config.server.host == "0.0.0.0" || state.config.server.host == "::" {
"127.0.0.1"
} else {
state.config.server.host.as_str()
};
Ok(Json(json!({
"system": {
"host": host,
"port": state.config.server.port,
"role": if caller_is_admin(&caller) { "admin" } else { "user" }
},
"devices": [],
"queue_remaining": 0,
})))
}
File diff suppressed because it is too large Load Diff
+280
View File
@@ -0,0 +1,280 @@
use super::models::{image_model_choices, is_video_workflow_model};
use super::node_defs::{core_node_definitions, node_type_def_to_comfy};
use super::DEFAULT_TEXT_TO_VIDEO_MODEL;
use crate::workflow::registry;
use serde_json::{json, Map, Value};
use std::collections::HashSet;
pub(super) fn comfy_combo_input(options: &[String], fallback: &str) -> Value {
let choices = if options.is_empty() {
vec![fallback.to_string()]
} else {
options.to_vec()
};
let default = choices
.iter()
.find(|model| model.as_str() == fallback)
.cloned()
.unwrap_or_else(|| choices[0].clone());
json!([choices, { "default": default }])
}
pub(super) fn inject_model_choices(defs: &mut Map<String, Value>, models: &[String]) {
let image_models = image_model_choices(models);
let video_models = models
.iter()
.filter(|model| is_video_workflow_model(model))
.cloned()
.collect::<Vec<_>>();
if let Some(node) = defs.get_mut("boosapi.text_to_image") {
if let Some(optional) = node
.pointer_mut("/input/optional")
.and_then(Value::as_object_mut)
{
optional.insert(
"model".to_string(),
comfy_combo_input(&image_models, "gpt-image-2"),
);
}
}
for node_type in [
"boosapi.text_to_video",
"boosapi.image_to_video",
"boosapi.video_status",
] {
if let Some(node) = defs.get_mut(node_type) {
if let Some(optional) = node
.pointer_mut("/input/optional")
.and_then(Value::as_object_mut)
{
optional.insert(
"model".to_string(),
comfy_combo_input(&video_models, DEFAULT_TEXT_TO_VIDEO_MODEL),
);
}
}
}
for node_type in ["boosapi.chat_completion", "boosapi.comfyui_workflow"] {
if let Some(node) = defs.get_mut(node_type) {
if let Some(optional) = node
.pointer_mut("/input/optional")
.and_then(Value::as_object_mut)
{
optional.insert("model".to_string(), comfy_combo_input(models, "gpt-5.4"));
}
}
}
}
pub(super) fn is_boosapi_extension_node_type(node_type: &str) -> bool {
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"
| "comfyui.submit_workflow"
)
}
pub(super) fn workflow_node_defs_with_models(
models: &[String],
include_compat_core: bool,
) -> Map<String, Value> {
let core_defs = include_compat_core
.then(core_node_definitions)
.unwrap_or_default();
let mut defs = core_defs
.into_iter()
.chain(
registry::node_types()
.into_iter()
.filter(|def| include_compat_core || is_boosapi_extension_node_type(def.node_type)),
)
.map(|def| (def.node_type.to_string(), node_type_def_to_comfy(&def)))
.collect::<Map<_, _>>();
inject_model_choices(&mut defs, models);
defs
}
pub(super) fn inject_boosapi_object_info(value: &mut Value, models: &[String]) {
let Some(object) = value.as_object_mut() else {
return;
};
for (key, def) in workflow_node_defs_with_models(models, false) {
object.entry(key).or_insert(def);
}
}
pub(super) 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",
]
}
pub(super) fn inject_boosapi_model_folders(value: &mut Value) {
let Some(items) = value.as_array_mut() else {
*value = workflow_model_folder_payload();
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 }));
}
}
}
pub(super) fn workflow_model_folder_payload() -> Value {
Value::Array(
builtin_model_folders()
.into_iter()
.map(|name| json!({ "name": name }))
.collect(),
)
}
pub(super) fn native_model_folder_payload_or_fallback(value: Value) -> Value {
match value {
Value::Array(_) => {
let mut value = value;
normalize_model_folder_payload(&mut value);
inject_boosapi_model_folders(&mut value);
value
}
Value::Object(object)
if object
.get("code")
.and_then(Value::as_i64)
.is_some_and(|code| code == 404) =>
{
workflow_model_folder_payload()
}
_ => workflow_model_folder_payload(),
}
}
pub(super) fn model_payload_items_from_names(names: Vec<String>, path_index: i64) -> Vec<Value> {
names
.into_iter()
.map(|name| json!({ "name": name, "pathIndex": path_index }))
.collect()
}
pub(super) fn normalize_model_item_name(item: &Value) -> Option<String> {
item.get("name")
.and_then(Value::as_str)
.or_else(|| item.as_str())
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
}
pub(super) fn workflow_model_display_aliases(name: &str) -> Vec<String> {
let trimmed = name.trim();
if trimmed.is_empty() {
return Vec::new();
}
let mut aliases = Vec::new();
let lower = trimmed.to_ascii_lowercase();
if lower == "ltx-2.3-22b-distilled-1.1_lora-dynamic_fro09_avg_rank_111_bf16.safetensors" {
aliases.push(
"ltx_2.3_22b_distilled_1.1_lora_dynamic_fro09_avg_rank_111_bf16.safetensors"
.to_string(),
);
aliases.push(
"ltx_2.3_22b_distilled_1.1_lora_dynamic_fr009_avg_rank_111_bf16.safetensors"
.to_string(),
);
}
if lower.contains("ltx") && lower.contains("lora") {
aliases.push(trimmed.replace('-', "_"));
}
aliases.sort();
aliases.dedup_by(|a, b| a.eq_ignore_ascii_case(b));
aliases
}
pub(super) fn model_payload_item_with_name(source: &Value, name: String) -> Value {
match source {
Value::Object(object) => {
let mut object = object.clone();
object.insert("name".to_string(), Value::String(name));
object
.entry("pathIndex".to_string())
.or_insert_with(|| Value::Number(0.into()));
Value::Object(object)
}
_ => json!({ "name": name, "pathIndex": 0 }),
}
}
pub(super) fn merge_model_payload_items(primary: Vec<Value>, extra: Vec<Value>) -> Value {
let mut seen = HashSet::<String>::new();
let mut items = Vec::with_capacity(primary.len() + extra.len());
for item in primary.into_iter().chain(extra) {
let Some(name) = normalize_model_item_name(&item) else {
continue;
};
if seen.insert(name.to_ascii_lowercase()) {
let aliases = workflow_model_display_aliases(&name);
items.push(match item {
Value::String(_) => json!({ "name": name.clone(), "pathIndex": 0 }),
Value::Object(mut object) => {
object
.entry("name".to_string())
.or_insert_with(|| Value::String(name.clone()));
object
.entry("pathIndex".to_string())
.or_insert_with(|| Value::Number(0.into()));
Value::Object(object)
}
_ => json!({ "name": name.clone(), "pathIndex": 0 }),
});
let source = items.last().cloned().unwrap_or_else(|| json!({}));
for alias in aliases {
if seen.insert(alias.to_ascii_lowercase()) {
items.push(model_payload_item_with_name(&source, alias));
}
}
}
}
Value::Array(items)
}
pub(super) fn normalize_model_folder_payload(value: &mut Value) {
let Some(items) = value.as_array_mut() else {
return;
};
for item in items {
if let Some(name) = item.as_str().map(str::to_string) {
*item = json!({ "name": name });
}
}
}
+646
View File
@@ -0,0 +1,646 @@
use super::auth::{auth_caller, caller_user_id};
use super::queue::repair_native_prompt_payload;
use super::{
auth_caller_optional, build_query_string, safe_view_path, user_home, workspace_kind_from_mime,
CachedJson, PromptPayload, UploadQuery, WorkflowCaller, NATIVE_WORKFLOW_CACHE_MAX_AGE,
NATIVE_WORKFLOW_JSON_CACHE,
};
use crate::{
error::{AppError, AppResult},
services::workspace::{self, WorkspaceAssetInput},
AppState,
};
use axum::{
body::Bytes,
extract::{multipart::Multipart, FromRequest, Path, Query, State},
http::{header, HeaderMap, HeaderValue, Method, StatusCode, Uri},
response::{IntoResponse, Response},
Json,
};
use serde_json::{json, Value};
use std::{
collections::HashMap,
time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};
pub(super) async fn workflow_cache_status(
State(state): State<AppState>,
headers: HeaderMap,
) -> AppResult<Json<Value>> {
let _ = auth_caller_optional(&state, &headers);
let base = native_workflow_base(&state);
let now = Instant::now();
let mut items = NATIVE_WORKFLOW_JSON_CACHE
.lock()
.iter()
.filter(|(key, _)| key.starts_with(&base))
.map(|(key, item)| {
json!({
"key": key,
"ageSecs": now.duration_since(item.inserted_at).as_secs(),
"kind": workflow_cache_kind_from_key(key),
})
})
.collect::<Vec<_>>();
items.sort_by(|a, b| {
a.get("key")
.and_then(Value::as_str)
.cmp(&b.get("key").and_then(Value::as_str))
});
Ok(Json(json!({
"success": true,
"backend": if state.config.comfyui.uses_runninghub_backend() {
"runninghub"
} else if state.config.comfyui.uses_cloud_backend() {
"cloud"
} else if state.config.comfyui.uses_native_backend() {
"external"
} else {
"boosapi"
},
"baseUrl": base,
"items": items,
})))
}
pub(super) async fn upload_image(
State(state): State<AppState>,
headers: HeaderMap,
Query(query): Query<UploadQuery>,
body: Bytes,
) -> AppResult<Response> {
let caller = auth_caller(&state, &headers).await?;
if state.config.comfyui.uses_native_backend() {
return upload_image_to_native_backend(&state, &headers, body).await;
}
let user_id = caller_user_id(&caller);
let multipart = Multipart::from_request(
axum::http::Request::builder()
.method(Method::POST)
.header(
header::CONTENT_TYPE,
headers
.get(header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.unwrap_or_default(),
)
.body(axum::body::Body::from(body))
.map_err(|err| AppError::BadRequest(err.to_string()))?,
&state,
)
.await
.map_err(|err| AppError::BadRequest(err.to_string()))?;
upload_image_to_local_backend(user_id, query, multipart).await
}
async fn upload_image_to_local_backend(
user_id: Option<i64>,
query: UploadQuery,
mut multipart: Multipart,
) -> AppResult<Response> {
let mut image_name: Option<String> = None;
let mut image_bytes: Option<Vec<u8>> = None;
let mut subfolder = query.subfolder.unwrap_or_default();
let mut file_type = query.file_type.unwrap_or_else(|| "input".to_string());
while let Some(field) = multipart
.next_field()
.await
.map_err(|err| AppError::BadRequest(err.to_string()))?
{
let name = field.name().unwrap_or_default().to_string();
if name == "image" {
image_name = field.file_name().map(str::to_string);
image_bytes = Some(
field
.bytes()
.await
.map_err(|err| AppError::BadRequest(err.to_string()))?
.to_vec(),
);
} else if name == "subfolder" {
subfolder = field.text().await.unwrap_or_default();
} else if name == "type" {
file_type = field.text().await.unwrap_or_else(|_| "input".to_string());
}
}
let Some(bytes) = image_bytes else {
return Err(AppError::BadRequest("image is required".into()));
};
let file_name =
image_name.unwrap_or_else(|| format!("upload-{}.png", uuid::Uuid::new_v4().simple()));
let root = user_home(user_id).join("userdata").join(&file_type);
let dir = if subfolder.trim().is_empty() {
root.clone()
} else {
root.join(safe_view_path(&subfolder))
};
tokio::fs::create_dir_all(&dir)
.await
.map_err(|err| AppError::Internal(anyhow::anyhow!(err)))?;
let path = dir.join(&file_name);
tokio::fs::write(&path, bytes)
.await
.map_err(|err| AppError::Internal(anyhow::anyhow!(err)))?;
Ok(Json(json!({
"name": file_name,
"subfolder": subfolder,
"type": file_type,
}))
.into_response())
}
async fn upload_image_to_native_backend(
state: &AppState,
headers: &HeaderMap,
body: Bytes,
) -> AppResult<Response> {
if body.is_empty() {
return Err(AppError::BadRequest("empty upload request body".into()));
}
let content_type = headers
.get(header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.filter(|value| {
value
.to_ascii_lowercase()
.starts_with("multipart/form-data")
})
.ok_or_else(|| AppError::BadRequest("multipart/form-data content-type is required".into()))?
.to_string();
let response = native_workflow_request(state, Method::POST, "/upload/image", None)
.header(reqwest::header::CONTENT_TYPE, content_type)
.body(body)
.send()
.await
.map_err(|err| AppError::UpstreamError(format!("ComfyUI upload failed: {err}")))?;
let status = response.status();
let content_type = response
.headers()
.get(reqwest::header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.unwrap_or("application/json")
.to_string();
let bytes = response
.bytes()
.await
.map_err(|err| AppError::UpstreamError(err.to_string()))?;
if !status.is_success() {
let body = String::from_utf8_lossy(&bytes);
tracing::warn!(status = status.as_u16(), body = %body, "Workflow native upload failed");
}
Ok((
StatusCode::from_u16(status.as_u16()).unwrap_or(StatusCode::BAD_GATEWAY),
[(header::CONTENT_TYPE, content_type)],
bytes,
)
.into_response())
}
pub(super) fn timestamp_ms() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_millis() as i64)
.unwrap_or_default()
}
pub(super) fn native_workflow_enabled(state: &AppState) -> bool {
state.config.comfyui.uses_native_backend()
}
pub(super) fn native_workflow_metadata_ttl(state: &AppState) -> Duration {
if state.config.comfyui.uses_runninghub_backend() {
Duration::from_secs(21600)
} else {
Duration::from_secs(300)
}
}
pub(super) fn native_workflow_status_ttl(state: &AppState) -> Duration {
if state.config.comfyui.uses_runninghub_backend() {
Duration::from_secs(120)
} else {
Duration::from_secs(30)
}
}
pub(super) fn native_workflow_base(state: &AppState) -> String {
state.config.comfyui.target_base_url()
}
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())
)
} else {
path.to_string()
}
}
pub(super) fn native_workflow_request(
state: &AppState,
method: Method,
path: &str,
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 mut request = client.request(reqwest_method, url);
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);
}
request
}
pub(super) async fn native_workflow_json(
state: &AppState,
method: Method,
path: &str,
body: Option<Value>,
) -> AppResult<Json<Value>> {
let response = native_workflow_request(state, method, path, body)
.send()
.await
.map_err(|err| AppError::UpstreamError(format!("ComfyUI 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(Json(value))
}
pub(super) async fn native_workflow_json_cached(
state: &AppState,
method: Method,
path: &str,
body: Option<Value>,
ttl: Duration,
) -> AppResult<Json<Value>> {
if method != Method::GET || body.is_some() || ttl.is_zero() {
return native_workflow_json(state, method, path, body).await;
}
let cache_key = format!(
"{}|{}|{}",
native_workflow_base(state),
method.as_str(),
native_workflow_http_path(state, path)
);
let now = Instant::now();
if let Some(cached) = NATIVE_WORKFLOW_JSON_CACHE.lock().get(&cache_key).cloned() {
if now.duration_since(cached.inserted_at) <= ttl {
return Ok(Json(cached.value));
}
}
let value = match native_workflow_json(state, method, path, None).await {
Ok(value) => value.0,
Err(err) => {
if let Some(cached) = NATIVE_WORKFLOW_JSON_CACHE.lock().get(&cache_key).cloned() {
tracing::warn!(
path,
error = %err,
"using stale ComfyUI workflow metadata cache after refresh failure"
);
return Ok(Json(cached.value));
}
return Err(err);
}
};
let mut cache = NATIVE_WORKFLOW_JSON_CACHE.lock();
if cache.len() > 64 {
cache.retain(|_, item| {
now.duration_since(item.inserted_at) <= NATIVE_WORKFLOW_CACHE_MAX_AGE
});
}
cache.insert(
cache_key,
CachedJson {
inserted_at: now,
value: value.clone(),
},
);
Ok(Json(value))
}
fn workflow_cache_kind_from_key(key: &str) -> &'static str {
if key.contains("/object_info") {
"object_info"
} else if key.contains("/experiment/models") {
"models"
} else if key.contains("/system_stats") {
"status"
} else {
"json"
}
}
pub(super) async fn native_workflow_response(
state: &AppState,
method: Method,
path: &str,
body: Option<Value>,
) -> AppResult<Response> {
let response = native_workflow_request(state, method, path, body)
.send()
.await
.map_err(|err| AppError::UpstreamError(format!("ComfyUI backend unreachable: {err}")))?;
let status = response.status();
let content_type = response
.headers()
.get(reqwest::header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.unwrap_or("application/json")
.to_string();
let bytes = response
.bytes()
.await
.map_err(|err| AppError::UpstreamError(err.to_string()))?;
Ok((
StatusCode::from_u16(status.as_u16()).unwrap_or(StatusCode::BAD_GATEWAY),
[(header::CONTENT_TYPE, content_type)],
bytes,
)
.into_response())
}
pub(super) async fn native_workflow_view_response(
state: &AppState,
caller: &WorkflowCaller,
params: &HashMap<String, String>,
) -> AppResult<Response> {
native_workflow_file_response(state, caller, "/view", params).await
}
pub(super) async fn native_workflow_video_response(
state: &AppState,
caller: &WorkflowCaller,
params: &HashMap<String, String>,
) -> AppResult<Response> {
native_workflow_file_response(state, caller, "/viewvideo", params).await
}
async fn native_workflow_file_response(
state: &AppState,
caller: &WorkflowCaller,
path: &str,
params: &HashMap<String, String>,
) -> AppResult<Response> {
let query = build_query_string(params);
let response = native_workflow_request(state, Method::GET, &format!("{path}{query}"), None)
.send()
.await
.map_err(|err| AppError::UpstreamError(format!("ComfyUI backend unreachable: {err}")))?;
let status = response.status();
let content_type = response
.headers()
.get(reqwest::header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.unwrap_or("application/octet-stream")
.to_string();
let bytes = response
.bytes()
.await
.map_err(|err| AppError::UpstreamError(err.to_string()))?;
if status.is_success() && matches!(path, "/view" | "/viewvideo") {
archive_native_workflow_asset(state, caller, path, params, &content_type, bytes.as_ref())
.await;
}
Ok((
StatusCode::from_u16(status.as_u16()).unwrap_or(StatusCode::BAD_GATEWAY),
[(header::CONTENT_TYPE, content_type)],
bytes,
)
.into_response())
}
async fn archive_native_workflow_asset(
state: &AppState,
caller: &WorkflowCaller,
path: &str,
params: &HashMap<String, String>,
content_type: &str,
bytes: &[u8],
) {
let file_type = params.get("type").map(String::as_str).unwrap_or("").trim();
if file_type != "output" {
return;
}
let Some(kind) = workspace_kind_from_mime(content_type) else {
return;
};
let filename = params
.get("filename")
.map(String::as_str)
.unwrap_or("")
.trim();
if filename.is_empty() || bytes.is_empty() {
return;
}
let user_id = caller_user_id(caller);
let source = "workflow-studio";
let already_exists = sqlx::query_scalar::<_, i64>(
"SELECT COUNT(*)
FROM workspace_assets
WHERE deleted_at IS NULL
AND (
(? = 1 AND user_id IS NULL)
OR (? = 0 AND user_id = ?)
)
AND source = ?
AND json_extract(metadata, '$.workflowUi.filename') = ?
AND json_extract(metadata, '$.workflowUi.type') = ?
AND json_extract(metadata, '$.workflowUi.subfolder') = ?",
)
.bind(user_id.is_none())
.bind(user_id.is_none())
.bind(user_id)
.bind(source)
.bind(filename)
.bind(file_type)
.bind(params.get("subfolder").map(String::as_str).unwrap_or(""))
.fetch_one(&*state.db)
.await
.unwrap_or_default();
if already_exists > 0 {
return;
}
let input = WorkspaceAssetInput {
user_id,
source: source.to_string(),
kind: kind.to_string(),
title: Some(filename.chars().take(80).collect()),
prompt: None,
model: None,
remote_url: None,
metadata: json!({
"workspaceAsset": { "role": "output" },
"workflowUi": {
"filename": filename,
"type": file_type,
"subfolder": params.get("subfolder").map(String::as_str).unwrap_or(""),
"endpoint": path,
}
}),
};
if let Err(err) = workspace::record_bytes_asset(state, input, bytes, content_type).await {
tracing::warn!(%err, filename, "failed to archive workflow-ui asset into workspace");
}
}
pub(super) async fn native_workflow_api_proxy(
State(state): State<AppState>,
headers: HeaderMap,
Path(path): Path<String>,
method: Method,
uri: Uri,
body: Bytes,
) -> AppResult<Response> {
let _ = auth_caller_optional(&state, &headers);
if !native_workflow_enabled(&state) {
return Err(AppError::NotFound("Workflow API route not found".into()));
}
let target = match uri.query() {
Some(query) => format!("/{path}?{query}"),
None => format!("/{path}"),
};
if state.config.comfyui.uses_cloud_backend() {
return native_workflow_raw_response(&state, method, &target, &headers, body).await;
}
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() })),
)
};
native_workflow_response(&state, method, &target, body).await
}
async fn native_workflow_raw_response(
state: &AppState,
method: Method,
path: &str,
headers: &HeaderMap,
body: Bytes,
) -> AppResult<Response> {
let is_prompt_post = method == Method::POST && path.split('?').next() == Some("/prompt");
let mut request = native_workflow_request(state, method, path, None);
if let Some(content_type) = headers
.get(header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
{
request = request.header(reqwest::header::CONTENT_TYPE, content_type);
}
if let Some(accept) = headers
.get(header::ACCEPT)
.and_then(|value| value.to_str().ok())
{
request = request.header(reqwest::header::ACCEPT, accept);
}
if !body.is_empty() {
if is_prompt_post {
if let Ok(mut value) = serde_json::from_slice::<Value>(&body) {
repair_native_prompt_body_value(state, &mut value).await;
request = request.json(&value);
} else {
request = request.body(body);
}
} else {
request = request.body(body);
}
}
let response = request
.send()
.await
.map_err(|err| AppError::UpstreamError(format!("ComfyUI backend unreachable: {err}")))?;
let status = response.status();
let response_headers = response.headers().clone();
let content_type = response_headers
.get(reqwest::header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.unwrap_or("application/json")
.to_string();
let bytes = response
.bytes()
.await
.map_err(|err| AppError::UpstreamError(err.to_string()))?;
let mut builder = Response::builder()
.status(StatusCode::from_u16(status.as_u16()).unwrap_or(StatusCode::BAD_GATEWAY))
.header(header::CONTENT_TYPE, content_type);
for name in [
header::CACHE_CONTROL,
header::ETAG,
header::LAST_MODIFIED,
header::LOCATION,
header::SET_COOKIE,
header::WWW_AUTHENTICATE,
] {
if let Some(value) = response_headers
.get(name.as_str())
.and_then(|value| value.to_str().ok())
.and_then(|value| HeaderValue::from_str(value).ok())
{
builder = builder.header(name, value);
}
}
builder
.body(axum::body::Body::from(bytes))
.map_err(|err| AppError::Internal(anyhow::anyhow!(err)))
}
async fn repair_native_prompt_body_value(state: &AppState, value: &mut Value) {
if let Ok(mut payload) = serde_json::from_value::<PromptPayload>(value.clone()) {
repair_native_prompt_payload(state, &mut payload).await;
if let Ok(repaired) = serde_json::to_value(payload) {
*value = repaired;
}
}
}
pub(super) async fn upload_mask(
State(state): State<AppState>,
headers: HeaderMap,
body: Bytes,
) -> AppResult<Response> {
upload_image(
State(state),
headers,
Query(UploadQuery {
subfolder: None,
file_type: Some("input".into()),
}),
body,
)
.await
}
+435
View File
@@ -0,0 +1,435 @@
use crate::{
error::{AppError, AppResult},
workflow::storage,
AppState,
};
use axum::{
extract::{Path, Query, State},
http::{HeaderMap, Method, StatusCode},
response::{Html, IntoResponse, Response},
Json,
};
use serde_json::{json, Value};
use sqlx::Row;
use std::{collections::HashMap, path::PathBuf, sync::mpsc};
use super::auth::{auth_caller, caller_is_admin, caller_user_id};
use super::node_defs::normalize_prompt_payload;
use super::{
auth_caller_optional, guess_mime_from_path, inject_native_prompt_cloud_api_key,
native_workflow_json, native_workflow_json_cached, native_workflow_metadata_ttl,
native_workflow_response, native_workflow_view_response, prompt_contains_boosapi_nodes,
prompt_error_response, publish_run_events, repair_dynamic_combo_nodes,
repair_resize_image_mask_nodes, repair_text_generate_ltx2_prompt_nodes, repair_wan2_api_nodes,
run_outputs_for_comfy, safe_view_path, simple_urlencode, user_home, workflow_context,
workflow_graph_to_comfy_workflow, workflow_progress_callback_with_run_sender,
workflow_run_to_history, workflow_run_to_history_task, workflow_run_to_queue_task,
HistoryQuery, PromptPayload, QueueMutation, WorkflowCaller,
};
pub(super) async fn queue_status(
State(state): State<AppState>,
headers: HeaderMap,
) -> AppResult<Json<Value>> {
let _ = auth_caller_optional(&state, &headers);
if state.config.comfyui.uses_native_backend() {
return native_workflow_json(&state, Method::GET, "/prompt", None).await;
}
Ok(Json(json!({
"status_str": "success",
"completed": true,
"messages": []
})))
}
pub(super) async fn queue(
State(state): State<AppState>,
headers: HeaderMap,
) -> AppResult<Json<Value>> {
if state.config.comfyui.uses_native_backend() {
let _ = auth_caller_optional(&state, &headers);
return native_workflow_json(&state, Method::GET, "/queue", None).await;
}
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 running = rows
.into_iter()
.filter(|run| run.status == "running")
.map(workflow_run_to_queue_task)
.collect::<Vec<_>>();
Ok(Json(json!({
"Running": running,
"Pending": []
})))
}
pub(super) async fn queue_mutate(
State(state): State<AppState>,
headers: HeaderMap,
Json(_payload): Json<QueueMutation>,
) -> AppResult<Json<Value>> {
let _ = auth_caller(&state, &headers).await?;
if state.config.comfyui.uses_native_backend() {
let value = serde_json::to_value(&_payload).unwrap_or_else(|_| json!({}));
return native_workflow_json(&state, Method::POST, "/queue", Some(value)).await;
}
Ok(Json(json!({
"success": true,
"Running": [],
"Pending": []
})))
}
pub(super) async fn history(
State(state): State<AppState>,
headers: HeaderMap,
Query(query): Query<HistoryQuery>,
) -> AppResult<Json<Value>> {
if state.config.comfyui.uses_native_backend() {
let _ = auth_caller_optional(&state, &headers);
let path = format!("/history?max_items={}", query.max_items);
return native_workflow_json(&state, Method::GET, &path, None).await;
}
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),
query.max_items,
)
.await?;
let history = rows
.into_iter()
.map(workflow_run_to_history_task)
.collect::<Vec<_>>();
Ok(Json(json!({ "History": history })))
}
pub(super) async fn history_mutate(
State(state): State<AppState>,
headers: HeaderMap,
Json(payload): Json<QueueMutation>,
) -> AppResult<Json<Value>> {
let caller = auth_caller(&state, &headers).await?;
if state.config.comfyui.uses_native_backend() {
let value = serde_json::to_value(&payload).unwrap_or_else(|_| json!({}));
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
} else {
storage::delete_runs(
&state,
&payload.delete,
caller_user_id(&caller),
caller_is_admin(&caller),
)
.await?
};
Ok(Json(json!({
"success": true,
"deleted": affected
})))
}
pub(super) async fn history_by_id(
State(state): State<AppState>,
headers: HeaderMap,
Path(id): Path<String>,
) -> AppResult<Json<Value>> {
if state.config.comfyui.uses_native_backend() {
let _ = auth_caller_optional(&state, &headers);
let path = format!("/history/{}", simple_urlencode(&id));
return native_workflow_json(&state, Method::GET, &path, None).await;
}
let caller =
auth_caller_optional(&state, &headers).unwrap_or(WorkflowCaller::User { user_id: 0 });
let run = storage::get_run(
&state.db,
&id,
caller_user_id(&caller),
caller_is_admin(&caller),
)
.await?;
Ok(Json(workflow_run_to_history(run)))
}
pub(super) async fn queue_prompt(
State(state): State<AppState>,
headers: HeaderMap,
Json(mut payload): Json<PromptPayload>,
) -> AppResult<Response> {
let caller = auth_caller(&state, &headers).await?;
if state.config.comfyui.uses_native_backend() {
if state.config.comfyui.uses_cloud_backend() {
repair_native_prompt_payload(&state, &mut payload).await;
return native_workflow_response(
&state,
Method::POST,
"/prompt",
Some(serde_json::to_value(payload).unwrap_or_else(|_| json!({}))),
)
.await;
}
if prompt_contains_boosapi_nodes(&payload.prompt) {
return queue_boosapi_prompt(state, caller, payload).await;
}
repair_native_prompt_payload(&state, &mut payload).await;
return native_workflow_response(
&state,
Method::POST,
"/prompt",
Some(serde_json::to_value(payload).unwrap_or_else(|_| json!({}))),
)
.await;
}
queue_boosapi_prompt(state, caller, payload).await
}
pub(super) async fn repair_native_prompt_payload(state: &AppState, payload: &mut PromptPayload) {
inject_native_prompt_cloud_api_key(state, payload);
let object_info = native_workflow_json_cached(
state,
Method::GET,
"/object_info",
None,
native_workflow_metadata_ttl(state),
)
.await
.ok()
.map(|value| value.0);
if let Some(object_info) = object_info.as_ref() {
repair_dynamic_combo_nodes(&mut payload.prompt, object_info);
}
repair_resize_image_mask_nodes(&mut payload.prompt, &payload.extra_data);
repair_text_generate_ltx2_prompt_nodes(&mut payload.prompt);
repair_wan2_api_nodes(&mut payload.prompt);
}
pub(super) 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,
Err(err) => {
return Ok(prompt_error_response("工作流无法执行", err.to_string()));
}
};
if graph.nodes.is_empty() {
return Ok(prompt_error_response(
"画布里还没有可执行节点",
"请先从左侧节点库添加 BoosAPI 节点并连接后再执行。",
));
}
let name = payload
.extra_data
.pointer("/extra_pnginfo/workflow/name")
.and_then(Value::as_str)
.unwrap_or("ComfyUI workflow");
let comfy_workflow = payload
.extra_data
.pointer("/extra_pnginfo/workflow")
.cloned()
.unwrap_or_else(|| workflow_graph_to_comfy_workflow(&graph));
let workflow = storage::insert_workflow(
&state,
caller_user_id(&caller),
name,
Some("Imported from ComfyUI frontend"),
&serde_json::to_value(&graph).unwrap_or_else(|_| json!({})),
"private",
false,
)
.await?;
let ctx = workflow_context(&state, &caller).await?;
let progress_session_id = payload
.client_id
.clone()
.filter(|value| !value.trim().is_empty())
.unwrap_or_else(|| "boosapi-workflow".to_string());
let (run_tx, run_rx) = mpsc::sync_channel::<String>(1);
let progress =
workflow_progress_callback_with_run_sender(Some(progress_session_id.clone()), Some(run_tx));
let state_for_task = state.clone();
let graph_value = serde_json::to_value(&graph).unwrap_or_else(|_| json!({}));
let task_graph_value = graph_value.clone();
let user_id = caller_user_id(&caller);
let allow_admin = caller_is_admin(&caller);
let session_for_task = progress_session_id.clone();
tokio::spawn(async move {
let result = crate::workflow::executor::run_workflow_with_progress(
&state_for_task,
workflow.id,
task_graph_value,
json!({}),
ctx,
progress,
)
.await;
match result {
Ok(run_id) => {
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);
}
}
Err(err) => {
tracing::warn!("workflow background execution failed: {err}");
}
}
});
let run_id = match tokio::task::spawn_blocking(move || run_rx.recv()).await {
Ok(Ok(run_id)) => run_id,
Ok(Err(_)) | Err(_) => {
return Ok(prompt_error_response(
"工作流执行失败",
"后台任务没有成功进入队列,请稍后重试。",
));
}
};
Ok(Json(json!({
"prompt_id": run_id,
"number": payload.number.unwrap_or(0),
"node_errors": [],
"client_id": payload.client_id,
"front": payload.front.unwrap_or(false),
"outputs": {},
"workflow": comfy_workflow
}))
.into_response())
}
pub(super) async fn interrupt(
State(state): State<AppState>,
headers: HeaderMap,
) -> AppResult<Json<Value>> {
let _ = auth_caller_optional(&state, &headers);
if state.config.comfyui.uses_native_backend() {
return native_workflow_json(&state, Method::POST, "/interrupt", Some(json!({}))).await;
}
Ok(Json(json!({ "success": true })))
}
pub(super) async fn view(
State(state): State<AppState>,
headers: HeaderMap,
Query(params): Query<HashMap<String, String>>,
) -> AppResult<Response> {
let caller = auth_caller(&state, &headers).await?;
if state.config.comfyui.uses_native_backend() {
return native_workflow_view_response(&state, &caller, &params).await;
}
let file = params.get("filename").cloned().unwrap_or_default();
if file.is_empty() {
return Err(AppError::BadRequest("filename is required".into()));
}
let file_type = params
.get("type")
.cloned()
.unwrap_or_else(|| "input".to_string());
let subfolder = params.get("subfolder").cloned().unwrap_or_default();
let user_root = user_home(caller_user_id(&caller))
.join("userdata")
.join(&file_type);
let local_path = if subfolder.trim().is_empty() {
user_root.join(&file)
} else {
user_root.join(safe_view_path(&subfolder)).join(&file)
};
if tokio::fs::metadata(&local_path).await.is_ok() {
let bytes = tokio::fs::read(&local_path)
.await
.map_err(|err| AppError::UpstreamError(err.to_string()))?;
return Ok((
StatusCode::OK,
[(
axum::http::header::CONTENT_TYPE,
guess_mime_from_path(&local_path),
)],
bytes,
)
.into_response());
}
let row = sqlx::query(
"SELECT file_path, remote_url, mime_type
FROM workspace_assets
WHERE file_name = ? OR public_id = ?
ORDER BY id DESC
LIMIT 1",
)
.bind(&file)
.bind(&file)
.fetch_optional(&*state.db)
.await?;
let Some(row) = row else {
return Err(AppError::NotFound("Workspace asset not found".into()));
};
let file_path: Option<String> = row.try_get("file_path").ok();
if let Some(path) = file_path {
let path_buf = PathBuf::from(path);
match tokio::fs::read(&path_buf).await {
Ok(bytes) => {
return Ok((
StatusCode::OK,
[(
axum::http::header::CONTENT_TYPE,
guess_mime_from_path(&path_buf),
)],
bytes,
)
.into_response())
}
Err(err) => return Err(AppError::UpstreamError(err.to_string())),
}
}
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())
{
return Ok((
StatusCode::TEMPORARY_REDIRECT,
[(axum::http::header::LOCATION, url.to_string())],
)
.into_response());
}
Err(AppError::NotFound(
remote_url.unwrap_or_else(|| "asset unavailable".into()),
))
}
pub(super) async fn internal_logs(
State(state): State<AppState>,
headers: HeaderMap,
) -> AppResult<Html<String>> {
let _ = auth_caller(&state, &headers).await?;
Ok(Html("".to_string()))
}
pub(super) async fn internal_logs_raw(
State(state): State<AppState>,
headers: HeaderMap,
) -> AppResult<Json<Value>> {
let _ = auth_caller(&state, &headers).await?;
Ok(Json(
json!({ "size": { "cols": 0, "row": 0 }, "entries": [] }),
))
}
+644
View File
@@ -0,0 +1,644 @@
use super::workflow_cloud_api_key;
use super::PromptPayload;
use crate::AppState;
use serde_json::{json, Map, Value};
use std::collections::HashMap;
pub(super) fn inject_native_prompt_cloud_api_key(state: &AppState, payload: &mut PromptPayload) {
if !state.config.comfyui.uses_cloud_backend() {
return;
}
let Some(api_key) = workflow_cloud_api_key(&state.config.comfyui) else {
return;
};
let Value::Object(extra_data) = &mut payload.extra_data else {
payload.extra_data = json!({ "api_key_comfy_org": api_key });
return;
};
extra_data.insert(
"api_key_comfy_org".to_string(),
Value::String(api_key.to_string()),
);
}
pub(super) fn repair_resize_image_mask_nodes(prompt: &mut Value, extra_data: &Value) {
let Some(prompt_nodes) = prompt.as_object_mut() else {
return;
};
let workflow = extra_data
.pointer("/extra_pnginfo/workflow")
.cloned()
.unwrap_or(Value::Null);
let mut workflow_nodes = Vec::<Value>::new();
collect_workflow_nodes(&workflow, &mut workflow_nodes);
if workflow_nodes.is_empty() {
return;
}
let workflow_by_id = workflow_nodes
.iter()
.filter_map(|node| {
let id = node.get("id")?;
Some((id_to_prompt_key(id), node))
})
.collect::<HashMap<_, _>>();
for (node_id, node) in prompt_nodes.iter_mut() {
let Some(object) = node.as_object_mut() else {
continue;
};
if object
.get("class_type")
.and_then(Value::as_str)
.is_none_or(|class_type| class_type != "ResizeImageMaskNode")
{
continue;
}
let Some(inputs) = object.get_mut("inputs").and_then(Value::as_object_mut) else {
continue;
};
let crop_is_invalid = inputs
.get("resize_type.crop")
.is_some_and(|value| !matches!(value.as_str(), Some("disabled" | "center")));
if !crop_is_invalid {
continue;
}
let Some(workflow_node) = workflow_by_id
.get(node_id)
.or_else(|| prompt_subgraph_node_id(node_id).and_then(|id| workflow_by_id.get(id)))
else {
repair_resize_image_mask_node_inputs_without_workflow(node_id, inputs);
continue;
};
let Some(values) = workflow_node
.get("widgets_values")
.and_then(Value::as_array)
else {
repair_resize_image_mask_node_inputs_without_workflow(node_id, inputs);
continue;
};
if values
.first()
.and_then(Value::as_str)
.is_some_and(|value| value == "scale dimensions")
&& values.len() >= 4
{
inputs.insert(
"resize_type".to_string(),
Value::String("scale dimensions".to_string()),
);
inputs.insert("resize_type.width".to_string(), values[1].clone());
inputs.insert("resize_type.height".to_string(), values[2].clone());
inputs.insert("resize_type.crop".to_string(), values[3].clone());
if let Some(method) = values.get(4) {
inputs.insert("resize_type.method".to_string(), method.clone());
}
tracing::warn!(node_id = %node_id, "repaired ResizeImageMaskNode dynamic widget prompt inputs");
}
}
}
fn repair_resize_image_mask_node_inputs_without_workflow(
node_id: &str,
inputs: &mut Map<String, Value>,
) {
let crop_is_invalid = inputs
.get("resize_type.crop")
.is_some_and(|value| !matches!(value.as_str(), Some("disabled" | "center")));
if !crop_is_invalid {
return;
}
inputs
.entry("resize_type".to_string())
.or_insert_with(|| Value::String("scale dimensions".to_string()));
inputs.insert(
"resize_type.crop".to_string(),
Value::String("center".to_string()),
);
tracing::warn!(
node_id = %node_id,
"repaired ResizeImageMaskNode crop prompt input without workflow metadata"
);
}
pub(super) fn repair_text_generate_ltx2_prompt_nodes(prompt: &mut Value) {
let Some(prompt_nodes) = prompt.as_object_mut() else {
return;
};
for (node_id, node) in prompt_nodes.iter_mut() {
let Some(object) = node.as_object_mut() else {
continue;
};
if object
.get("class_type")
.and_then(Value::as_str)
.is_none_or(|class_type| class_type != "TextGenerateLTX2Prompt")
{
continue;
}
let Some(inputs) = object.get_mut("inputs").and_then(Value::as_object_mut) else {
continue;
};
let repetition_penalty_invalid = inputs
.get("sampling_mode.repetition_penalty")
.and_then(Value::as_f64)
.is_some_and(|value| !(0.0..=5.0).contains(&value));
if !repetition_penalty_invalid {
continue;
}
inputs.insert("sampling_mode".to_string(), Value::String("on".to_string()));
inputs.insert("sampling_mode.temperature".to_string(), json!(0.7));
inputs.insert("sampling_mode.top_k".to_string(), json!(64));
inputs.insert("sampling_mode.top_p".to_string(), json!(0.95));
inputs.insert("sampling_mode.min_p".to_string(), json!(0.05));
inputs.insert("sampling_mode.repetition_penalty".to_string(), json!(1.05));
inputs
.entry("sampling_mode.seed".to_string())
.or_insert_with(|| json!(0));
inputs.insert("sampling_mode.presence_penalty".to_string(), json!(0.0));
tracing::warn!(
node_id = %node_id,
"repaired TextGenerateLTX2Prompt dynamic sampling prompt inputs"
);
}
}
#[derive(Debug, Clone)]
struct DynamicComboField {
name: String,
value_type: String,
options: Vec<String>,
default: Option<Value>,
min: Option<f64>,
max: Option<f64>,
}
pub(super) fn repair_dynamic_combo_nodes(prompt: &mut Value, object_info: &Value) {
let Some(prompt_nodes) = prompt.as_object_mut() else {
return;
};
for (node_id, node) in prompt_nodes.iter_mut() {
let Some(object) = node.as_object_mut() else {
continue;
};
let class_type = object
.get("class_type")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
if class_type.is_empty() {
continue;
}
let dynamic_groups = dynamic_combo_groups_for_node(object_info, &class_type);
if dynamic_groups.is_empty() {
continue;
}
let Some(inputs) = object.get_mut("inputs").and_then(Value::as_object_mut) else {
continue;
};
let mut repaired_groups = Vec::<String>::new();
for (group_name, option_key, fields) in dynamic_groups {
if repair_dynamic_combo_group(inputs, &group_name, &option_key, &fields) {
repaired_groups.push(group_name);
}
}
if !repaired_groups.is_empty() {
tracing::warn!(
node_id = %node_id,
class_type = %class_type,
groups = ?repaired_groups,
"repaired ComfyUI dynamic combo prompt inputs from object_info schema"
);
}
}
}
fn dynamic_combo_groups_for_node(
object_info: &Value,
class_type: &str,
) -> Vec<(String, String, Vec<DynamicComboField>)> {
let Some(node) = object_info.get(class_type) else {
return Vec::new();
};
let mut groups = Vec::new();
for section in ["required", "optional"] {
let Some(inputs) = node
.pointer(&format!("/input/{section}"))
.and_then(Value::as_object)
else {
continue;
};
for (name, input) in inputs {
let Some(input_type) = input.get(0).and_then(Value::as_str) else {
continue;
};
if !input_type.starts_with("COMFY_DYNAMICCOMBO") {
continue;
}
let Some(option) = input
.get(1)
.and_then(|config| config.get("options"))
.and_then(Value::as_array)
.and_then(|options| options.first())
else {
continue;
};
let option_key = option
.get("key")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
let fields = dynamic_combo_fields_from_option(option);
if !fields.is_empty() {
groups.push((name.clone(), option_key, fields));
}
}
}
groups
}
fn dynamic_combo_fields_from_option(option: &Value) -> Vec<DynamicComboField> {
let Some(required) = option
.pointer("/inputs/required")
.and_then(Value::as_object)
else {
return Vec::new();
};
required
.iter()
.filter_map(|(name, spec)| {
let value_type = spec.get(0).and_then(Value::as_str)?.to_string();
let config = spec.get(1).and_then(Value::as_object);
let options = config
.and_then(|config| config.get("options"))
.and_then(Value::as_array)
.map(|items| {
items
.iter()
.filter_map(Value::as_str)
.map(ToOwned::to_owned)
.collect::<Vec<_>>()
})
.unwrap_or_default();
Some(DynamicComboField {
name: name.clone(),
value_type,
options,
default: config.and_then(|config| config.get("default")).cloned(),
min: config
.and_then(|config| config.get("min"))
.and_then(Value::as_f64),
max: config
.and_then(|config| config.get("max"))
.and_then(Value::as_f64),
})
})
.collect()
}
fn repair_dynamic_combo_group(
inputs: &mut Map<String, Value>,
group_name: &str,
option_key: &str,
fields: &[DynamicComboField],
) -> bool {
if fields.is_empty() {
return false;
}
let mut values = Vec::<Value>::new();
if let Some(value) = inputs.get(group_name).cloned() {
values.push(value);
}
for field in fields {
if let Some(value) = inputs.get(&format!("{group_name}.{}", field.name)).cloned() {
values.push(value);
}
}
if values.is_empty() {
return false;
}
let invalid = fields.iter().any(|field| {
let key = format!("{group_name}.{}", field.name);
inputs
.get(&key)
.is_some_and(|value| !dynamic_combo_value_matches_field(value, field))
});
if !invalid {
return false;
}
let mut used = vec![false; values.len()];
if !option_key.is_empty() {
inputs.insert(
group_name.to_string(),
Value::String(option_key.to_string()),
);
for (idx, value) in values.iter().enumerate() {
if value.as_str() == Some(option_key) {
used[idx] = true;
break;
}
}
}
for field in fields {
let key = format!("{group_name}.{}", field.name);
let selected_idx = values.iter().enumerate().find_map(|(idx, value)| {
(!used[idx] && dynamic_combo_value_matches_field(value, field)).then_some(idx)
});
let repaired = selected_idx
.map(|idx| {
used[idx] = true;
normalize_dynamic_combo_value(&values[idx], field)
})
.unwrap_or_else(|| default_dynamic_combo_value(field));
inputs.insert(key, repaired);
}
true
}
fn dynamic_combo_value_matches_field(value: &Value, field: &DynamicComboField) -> bool {
match field.value_type.as_str() {
"STRING" => value.as_str().is_some(),
"COMBO" => value
.as_str()
.is_some_and(|text| field.options.iter().any(|option| option == text)),
"INT" => dynamic_combo_value_as_i64(value).is_some_and(|number| {
field.min.is_none_or(|min| (number as f64) >= min)
&& field.max.is_none_or(|max| (number as f64) <= max)
}),
"FLOAT" => dynamic_combo_value_as_f64(value).is_some_and(|number| {
field.min.is_none_or(|min| number >= min) && field.max.is_none_or(|max| number <= max)
}),
"BOOLEAN" => dynamic_combo_value_as_bool(value).is_some(),
_ => true,
}
}
fn normalize_dynamic_combo_value(value: &Value, field: &DynamicComboField) -> Value {
match field.value_type.as_str() {
"INT" => dynamic_combo_value_as_i64(value)
.map(|number| json!(clamp_i64_to_field(number, field)))
.unwrap_or_else(|| default_dynamic_combo_value(field)),
"FLOAT" => dynamic_combo_value_as_f64(value)
.map(|number| json!(clamp_f64_to_field(number, field)))
.unwrap_or_else(|| default_dynamic_combo_value(field)),
"BOOLEAN" => dynamic_combo_value_as_bool(value)
.map(Value::Bool)
.unwrap_or_else(|| default_dynamic_combo_value(field)),
_ => value.clone(),
}
}
fn default_dynamic_combo_value(field: &DynamicComboField) -> Value {
if let Some(default) = &field.default {
return normalize_dynamic_combo_value(default, field);
}
match field.value_type.as_str() {
"COMBO" => field
.options
.first()
.cloned()
.map(Value::String)
.unwrap_or(Value::Null),
"STRING" => Value::String(String::new()),
"INT" => json!(field.min.unwrap_or(0.0).ceil() as i64),
"FLOAT" => json!(field.min.unwrap_or(0.0)),
"BOOLEAN" => Value::Bool(false),
_ => Value::Null,
}
}
fn dynamic_combo_value_as_i64(value: &Value) -> Option<i64> {
value
.as_i64()
.or_else(|| value.as_u64().and_then(|number| i64::try_from(number).ok()))
.or_else(|| value.as_f64().map(|number| number.round() as i64))
.or_else(|| {
value
.as_str()
.and_then(|text| text.trim().parse::<i64>().ok())
})
}
fn dynamic_combo_value_as_f64(value: &Value) -> Option<f64> {
value.as_f64().or_else(|| {
value
.as_str()
.and_then(|text| text.trim().parse::<f64>().ok())
})
}
fn dynamic_combo_value_as_bool(value: &Value) -> Option<bool> {
value.as_bool().or_else(|| {
value
.as_str()
.and_then(|text| match text.trim().to_ascii_lowercase().as_str() {
"true" | "1" | "yes" | "on" => Some(true),
"false" | "0" | "no" | "off" => Some(false),
_ => None,
})
})
}
fn clamp_i64_to_field(value: i64, field: &DynamicComboField) -> i64 {
let min = field
.min
.map(|value| value.ceil() as i64)
.unwrap_or(i64::MIN);
let max = field
.max
.map(|value| value.floor() as i64)
.unwrap_or(i64::MAX);
value.clamp(min, max)
}
fn clamp_f64_to_field(value: f64, field: &DynamicComboField) -> f64 {
let min = field.min.unwrap_or(f64::NEG_INFINITY);
let max = field.max.unwrap_or(f64::INFINITY);
value.clamp(min, max)
}
pub(super) fn repair_wan2_api_nodes(prompt: &mut Value) {
let Some(prompt_nodes) = prompt.as_object_mut() else {
return;
};
for (node_id, node) in prompt_nodes.iter_mut() {
let Some(object) = node.as_object_mut() else {
continue;
};
let class_type = object
.get("class_type")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
let field_names = match class_type.as_str() {
"Wan2ImageToVideoApi" => &["prompt", "negative_prompt", "resolution", "duration"][..],
"Wan2TextToVideoApi" | "Wan2ReferenceVideoApi" => &[
"prompt",
"negative_prompt",
"resolution",
"ratio",
"duration",
][..],
_ => continue,
};
let Some(inputs) = object.get_mut("inputs").and_then(Value::as_object_mut) else {
continue;
};
if !wan2_model_inputs_look_shifted(inputs) {
continue;
}
repair_wan2_model_inputs(inputs, field_names);
tracing::warn!(
node_id = %node_id,
class_type = %class_type,
"repaired Wan2 dynamic model prompt inputs"
);
}
}
fn wan2_model_inputs_look_shifted(inputs: &Map<String, Value>) -> bool {
inputs
.get("model.resolution")
.is_some_and(|value| !matches!(value.as_str(), Some("720P" | "1080P")))
|| inputs.get("model.ratio").is_some_and(|value| {
!matches!(
value.as_str(),
Some("16:9" | "9:16" | "1:1" | "4:3" | "3:4")
)
})
|| inputs
.get("model.duration")
.is_some_and(|value| value.as_i64().is_none())
}
fn repair_wan2_model_inputs(inputs: &mut Map<String, Value>, field_names: &[&str]) {
let mut values = field_names
.iter()
.filter_map(|name| inputs.remove(&format!("model.{name}")))
.collect::<Vec<_>>();
if let Some(model) = inputs.remove("model") {
values.insert(0, model);
}
let prompt = values
.iter()
.find_map(|value| {
value.as_str().and_then(|text| {
let trimmed = text.trim();
if trimmed.len() > 20 && !is_wan2_resolution(trimmed) && !is_wan2_ratio(trimmed) {
Some(Value::String(text.to_string()))
} else {
None
}
})
})
.unwrap_or_else(|| Value::String(String::new()));
let negative_prompt = values
.iter()
.filter_map(Value::as_str)
.find(|text| {
let trimmed = text.trim();
!trimmed.is_empty()
&& trimmed.len() <= 20
&& !is_wan2_model_key(trimmed)
&& !is_wan2_resolution(trimmed)
&& !is_wan2_ratio(trimmed)
})
.map(|text| Value::String(text.to_string()))
.unwrap_or_else(|| Value::String(String::new()));
let resolution = values
.iter()
.filter_map(Value::as_str)
.find(|text| is_wan2_resolution(text.trim()))
.map(|text| Value::String(text.trim().to_string()))
.unwrap_or_else(|| Value::String("720P".to_string()));
let ratio = values
.iter()
.filter_map(Value::as_str)
.find(|text| is_wan2_ratio(text.trim()))
.map(|text| Value::String(text.trim().to_string()))
.unwrap_or_else(|| Value::String("16:9".to_string()));
let duration = values
.iter()
.find_map(|value| {
value.as_i64().or_else(|| {
value
.as_str()
.and_then(|text| text.trim().parse::<i64>().ok())
})
})
.map(|value| value.clamp(2, 15))
.unwrap_or(5);
inputs.insert(
"model".to_string(),
infer_wan2_model_key(field_names, &values),
);
inputs.insert("model.prompt".to_string(), prompt);
inputs.insert("model.negative_prompt".to_string(), negative_prompt);
inputs.insert("model.resolution".to_string(), resolution);
if field_names.contains(&"ratio") {
inputs.insert("model.ratio".to_string(), ratio);
}
inputs.insert("model.duration".to_string(), json!(duration));
}
fn infer_wan2_model_key(field_names: &[&str], values: &[Value]) -> Value {
if let Some(model) = values
.iter()
.filter_map(Value::as_str)
.find(|value| is_wan2_model_key(value.trim()))
{
return Value::String(model.trim().to_string());
}
if field_names.contains(&"ratio") {
Value::String("wan2.7-t2v".to_string())
} else {
Value::String("wan2.7-i2v".to_string())
}
}
fn is_wan2_resolution(value: &str) -> bool {
matches!(value, "720P" | "1080P")
}
fn is_wan2_ratio(value: &str) -> bool {
matches!(value, "16:9" | "9:16" | "1:1" | "4:3" | "3:4")
}
fn is_wan2_model_key(value: &str) -> bool {
matches!(value, "wan2.7-i2v" | "wan2.7-t2v" | "wan2.7-r2v")
}
fn collect_workflow_nodes(value: &Value, nodes: &mut Vec<Value>) {
if let Some(items) = value.get("nodes").and_then(Value::as_array) {
nodes.extend(items.iter().cloned());
}
match value {
Value::Array(items) => {
for item in items {
collect_workflow_nodes(item, nodes);
}
}
Value::Object(object) => {
for item in object.values() {
collect_workflow_nodes(item, nodes);
}
}
_ => {}
}
}
fn prompt_subgraph_node_id(node_id: &str) -> Option<&str> {
node_id
.rsplit_once(':')
.map(|(_, child_id)| child_id.trim())
.filter(|value| !value.is_empty())
}
fn id_to_prompt_key(id: &Value) -> String {
if let Some(value) = id.as_i64() {
return value.to_string();
}
id.as_str().unwrap_or_default().to_string()
}
+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<_>>()
])
}
+305
View File
@@ -0,0 +1,305 @@
use crate::{error::AppResult, AppState};
use axum::{
extract::{Query, State},
http::{HeaderMap, Method},
Json,
};
use serde_json::{json, Value};
use std::collections::HashMap;
use super::auth::{auth_caller, caller_user_id};
use super::models::{combo_fields_from_node, workflow_route_models};
use super::{
build_query_string, field_names_are_equivalent, inject_boosapi_object_info,
is_input_file_field, is_modelish_field, native_workflow_json_cached,
native_workflow_metadata_ttl, native_workflow_request, safe_view_path, user_home,
workflow_linked_node_ids, workflow_node_defs_with_models, workflow_validation_hints,
workflow_widget_inputs, workflow_widget_string, WorkflowCaller, WorkflowValidateQuery,
};
pub(super) async fn validate_workflow(
State(state): State<AppState>,
headers: HeaderMap,
Query(query): Query<WorkflowValidateQuery>,
Json(workflow): Json<Value>,
) -> AppResult<Json<Value>> {
let caller = auth_caller(&state, &headers).await?;
let object_info = if state.config.comfyui.uses_native_backend() {
let mut value = native_workflow_json_cached(
&state,
Method::GET,
"/object_info",
None,
native_workflow_metadata_ttl(&state),
)
.await?
.0;
let route_models = workflow_route_models(&state).await.unwrap_or_default();
inject_boosapi_object_info(&mut value, &route_models);
value
} else {
let route_models = workflow_route_models(&state).await.unwrap_or_default();
Value::Object(workflow_node_defs_with_models(&route_models, true))
};
let report = validate_workflow_against_object_info(
&state,
&caller,
&workflow,
&object_info,
query.check_files,
)
.await;
Ok(Json(report))
}
async fn validate_workflow_against_object_info(
state: &AppState,
caller: &WorkflowCaller,
workflow: &Value,
object_info: &Value,
check_files: bool,
) -> Value {
let nodes = workflow
.get("nodes")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
let links = workflow
.get("links")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
let linked_node_ids = workflow_linked_node_ids(&links);
let mut missing_node_types = Vec::<Value>::new();
let mut empty_model_fields = Vec::<Value>::new();
let mut missing_model_choices = Vec::<Value>::new();
let mut unknown_model_fields = Vec::<Value>::new();
let mut input_files = Vec::<Value>::new();
let mut node_type_counts = HashMap::<String, usize>::new();
for node in &nodes {
let node_id = node.get("id").and_then(Value::as_i64).unwrap_or_default();
let node_type = node
.get("type")
.and_then(Value::as_str)
.unwrap_or("")
.trim()
.to_string();
if node_type.is_empty() {
continue;
}
*node_type_counts.entry(node_type.clone()).or_default() += 1;
let connected = linked_node_ids.contains(&node_id);
let Some(node_def) = object_info.get(&node_type) else {
missing_node_types.push(json!({
"nodeId": node_id,
"nodeType": node_type,
"connected": connected,
"severity": if connected { "error" } else { "warning" },
}));
continue;
};
let widgets = node
.get("widgets_values")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
let combo_fields = combo_fields_from_node(object_info, &node_type);
let widget_inputs = node
.get("inputs")
.and_then(Value::as_array)
.map(|inputs| workflow_widget_inputs(inputs))
.unwrap_or_default();
let mut widget_index = 0usize;
for input in widget_inputs {
let field = input
.get("name")
.and_then(Value::as_str)
.unwrap_or("")
.to_string();
let input_type = input
.get("type")
.and_then(Value::as_str)
.unwrap_or("")
.to_string();
let Some(widget_value) = widgets.get(widget_index) else {
break;
};
widget_index += 1;
if !is_modelish_field(&field, &input_type, widget_value) {
continue;
}
let Some(selected) = workflow_widget_string(widget_value) else {
continue;
};
if selected.trim().is_empty() {
continue;
}
if is_input_file_field(&field, &input_type, &selected) {
if check_files {
input_files.push(
validate_workflow_input_file(state, caller, node_id, &node_type, &selected)
.await,
);
} else {
input_files.push(json!({
"nodeId": node_id,
"nodeType": node_type,
"filename": selected,
"checked": false,
}));
}
continue;
}
let choices = combo_fields
.iter()
.find(|(name, _)| name == &field)
.map(|(_, choices)| choices.clone())
.or_else(|| {
combo_fields
.iter()
.find(|(name, _)| field_names_are_equivalent(name, &field))
.map(|(_, choices)| choices.clone())
});
match choices {
Some(choices) if choices.is_empty() => empty_model_fields.push(json!({
"nodeId": node_id,
"nodeType": node_type,
"field": field,
"selected": selected,
"severity": if connected { "error" } else { "warning" },
"message": "后端这个模型/LoRA字段返回空列表,当前选择无法确认可用。",
})),
Some(choices) if !choices.iter().any(|choice| choice == &selected) => {
let near = choices
.iter()
.filter(|choice| {
let a = choice.to_ascii_lowercase();
let b = selected.to_ascii_lowercase();
a.contains(&b) || b.contains(&a)
})
.take(8)
.cloned()
.collect::<Vec<_>>();
missing_model_choices.push(json!({
"nodeId": node_id,
"nodeType": node_type,
"field": field,
"selected": selected,
"near": near,
"availableCount": choices.len(),
"severity": if connected { "error" } else { "warning" },
}));
}
Some(_) => {}
None => {
if node_def.get("input").is_some() {
unknown_model_fields.push(json!({
"nodeId": node_id,
"nodeType": node_type,
"field": field,
"selected": selected,
"severity": "info",
}));
}
}
}
}
}
let errors = missing_node_types
.iter()
.chain(empty_model_fields.iter())
.chain(missing_model_choices.iter())
.filter(|item| {
item.get("severity")
.and_then(Value::as_str)
.is_some_and(|severity| severity == "error")
})
.count();
let warnings = missing_node_types
.iter()
.chain(empty_model_fields.iter())
.chain(missing_model_choices.iter())
.filter(|item| {
item.get("severity")
.and_then(Value::as_str)
.is_some_and(|severity| severity == "warning")
})
.count();
json!({
"success": true,
"compatible": errors == 0,
"summary": {
"nodes": nodes.len(),
"links": links.len(),
"nodeTypes": node_type_counts.len(),
"errors": errors,
"warnings": warnings,
},
"missingNodeTypes": missing_node_types,
"emptyModelFields": empty_model_fields,
"missingModelChoices": missing_model_choices,
"unknownModelFields": unknown_model_fields,
"inputFiles": input_files,
"hints": workflow_validation_hints(errors, warnings),
})
}
async fn validate_workflow_input_file(
state: &AppState,
caller: &WorkflowCaller,
node_id: i64,
node_type: &str,
filename: &str,
) -> Value {
if state.config.comfyui.uses_native_backend() {
let mut params = HashMap::new();
params.insert("filename".to_string(), filename.to_string());
params.insert("type".to_string(), "input".to_string());
let path = format!("/view{}", build_query_string(&params));
let result = native_workflow_request(state, Method::GET, &path, None)
.header(reqwest::header::RANGE, "bytes=0-0")
.send()
.await;
return match result {
Ok(response) => json!({
"nodeId": node_id,
"nodeType": node_type,
"filename": filename,
"checked": true,
"reachable": response.status().is_success(),
"status": response.status().as_u16(),
"contentType": response
.headers()
.get(reqwest::header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.unwrap_or(""),
}),
Err(err) => json!({
"nodeId": node_id,
"nodeType": node_type,
"filename": filename,
"checked": true,
"reachable": false,
"error": err.to_string(),
}),
};
}
let user_id = caller_user_id(caller);
let path = user_home(user_id)
.join("userdata")
.join("input")
.join(safe_view_path(filename));
json!({
"nodeId": node_id,
"nodeType": node_type,
"filename": filename,
"checked": true,
"reachable": tokio::fs::metadata(&path).await.is_ok(),
"localPath": path,
})
}
+494
View File
@@ -0,0 +1,494 @@
use crate::workflow::executor::{WorkflowProgressCallback, WorkflowProgressEvent};
use crate::{error::AppResult, AppState};
use axum::{
extract::{
ws::{Message, WebSocketUpgrade},
Query, State,
},
http::HeaderMap,
response::Response,
};
use once_cell::sync::Lazy;
use serde_json::{json, Value};
use std::{
collections::HashMap,
sync::{mpsc, Arc},
};
use tokio::sync::broadcast;
use tokio_tungstenite::{connect_async, tungstenite};
use tungstenite::client::IntoClientRequest;
use super::workflow_node_label;
use super::{
build_query_string, native_workflow_base, native_workflow_enabled, node_output_for_comfy,
timestamp_ms,
};
use super::auth::auth_caller;
static WORKFLOW_WS_SESSIONS: Lazy<parking_lot::Mutex<HashMap<String, broadcast::Sender<Value>>>> =
Lazy::new(|| parking_lot::Mutex::new(HashMap::new()));
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) {
return sender.clone();
}
let (sender, _) = broadcast::channel(128);
sessions.insert(session_id.to_string(), sender.clone());
sender
}
fn publish_workflow_ws_event(session_id: Option<&str>, event_type: &str, data: Value) {
let Some(session_id) = session_id.map(str::trim).filter(|value| !value.is_empty()) else {
return;
};
let sender = workflow_session_sender(session_id);
let _ = sender.send(json!({
"type": event_type,
"data": data,
}));
}
pub(super) fn workflow_progress_callback_with_run_sender(
session_id: Option<String>,
run_sender: Option<mpsc::SyncSender<String>>,
) -> Option<WorkflowProgressCallback> {
let session_id = session_id.map(|value| value.trim().to_string())?;
if session_id.is_empty() {
return None;
}
let run_sender = Arc::new(parking_lot::Mutex::new(run_sender));
Some(Arc::new(move |event| match event {
WorkflowProgressEvent::Started { prompt_id, total } => {
if let Some(sender) = run_sender.lock().take() {
let _ = sender.send(prompt_id.clone());
}
publish_workflow_ws_event(
Some(&session_id),
"status",
json!({ "status": { "exec_info": { "queue_remaining": 1 } } }),
);
publish_workflow_ws_event(
Some(&session_id),
"execution_start",
json!({
"prompt_id": prompt_id,
"timestamp": timestamp_ms(),
}),
);
publish_workflow_ws_event(
Some(&session_id),
"progress",
json!({ "value": 0, "max": total.max(1) }),
);
}
WorkflowProgressEvent::NodeStarted {
prompt_id,
node_id,
node_type,
index,
total,
} => {
let label = workflow_node_label(&node_type, &node_id);
publish_workflow_ws_event(
Some(&session_id),
"executing",
json!({
"prompt_id": prompt_id,
"node": node_id,
"node_type": node_type,
"display_node": label,
"label": label,
}),
);
publish_workflow_ws_event(
Some(&session_id),
"progress",
json!({ "value": index, "max": total.max(1) }),
);
}
WorkflowProgressEvent::NodeCompleted {
prompt_id,
node_id,
node_type,
index,
total,
output,
} => {
let label = workflow_node_label(&node_type, &node_id);
let output = node_output_for_comfy(&output).unwrap_or(output);
publish_workflow_ws_event(
Some(&session_id),
"executed",
json!({
"prompt_id": prompt_id,
"node": node_id,
"node_type": node_type,
"display_node": label,
"label": label,
"output": output,
}),
);
publish_workflow_ws_event(
Some(&session_id),
"progress",
json!({ "value": index, "max": total.max(1) }),
);
}
WorkflowProgressEvent::NodeProgress {
prompt_id,
node_id,
node_type,
value,
max,
message,
} => {
let label = workflow_node_label(&node_type, &node_id);
publish_workflow_ws_event(
Some(&session_id),
"executing",
json!({
"prompt_id": prompt_id,
"node": node_id,
"node_type": node_type,
"display_node": label,
"label": label,
"message": message,
}),
);
publish_workflow_ws_event(
Some(&session_id),
"progress",
json!({ "value": value, "max": max.max(1), "message": message }),
);
}
WorkflowProgressEvent::Finished {
prompt_id,
success,
error,
} => {
publish_workflow_ws_event(
Some(&session_id),
"executing",
json!({
"prompt_id": prompt_id,
"node": Value::Null,
"display_node": Value::Null,
}),
);
publish_workflow_ws_event(
Some(&session_id),
"status",
json!({ "status": { "exec_info": { "queue_remaining": 0 } } }),
);
publish_workflow_ws_event(
Some(&session_id),
if success {
"execution_success"
} else {
"execution_error"
},
if success {
json!({ "prompt_id": prompt_id, "timestamp": timestamp_ms() })
} else {
json!({
"prompt_id": prompt_id,
"timestamp": timestamp_ms(),
"node_id": "workflow",
"node_type": "Workflow",
"exception_message": error.unwrap_or_else(|| "Workflow execution failed".to_string()),
"exception_type": "BoosAPIWorkflowError",
"traceback": [],
})
},
);
}
}))
}
// APPEND_MARKER_PUBLISH_RUN_EVENTS
pub(super) fn publish_run_events(
session_id: Option<&str>,
run: &crate::workflow::models::WorkflowRunSummary,
outputs: &Value,
) {
let Some(session_id) = session_id.map(str::trim).filter(|value| !value.is_empty()) else {
return;
};
publish_workflow_ws_event(
Some(session_id),
"execution_start",
json!({
"prompt_id": run.public_id,
"timestamp": timestamp_ms(),
}),
);
for node in &run.node_runs {
let label = workflow_node_label(&node.node_type, &node.node_id);
publish_workflow_ws_event(
Some(session_id),
"executing",
json!({
"prompt_id": run.public_id,
"node": node.node_id,
"node_type": node.node_type,
"display_node": label,
"label": label,
}),
);
if let Some(output) = outputs.get(&node.node_id) {
publish_workflow_ws_event(
Some(session_id),
"executed",
json!({
"prompt_id": run.public_id,
"node": node.node_id,
"node_type": node.node_type,
"display_node": label,
"label": label,
"output": output,
}),
);
}
}
publish_workflow_ws_event(
Some(session_id),
"executing",
json!({
"prompt_id": run.public_id,
"node": Value::Null,
"display_node": Value::Null,
}),
);
publish_workflow_ws_event(
Some(session_id),
if run.status == "failed" {
"execution_error"
} else {
"execution_success"
},
if run.status == "failed" {
json!({
"prompt_id": run.public_id,
"timestamp": timestamp_ms(),
"node_id": run.node_runs.last().map(|node| node.node_id.as_str()).unwrap_or("workflow"),
"node_type": run.node_runs.last().map(|node| node.node_type.as_str()).unwrap_or("Workflow"),
"exception_message": run.error_message.clone().unwrap_or_else(|| "Workflow execution failed".to_string()),
"exception_type": "BoosAPIWorkflowError",
"traceback": [],
})
} else {
json!({
"prompt_id": run.public_id,
"timestamp": timestamp_ms(),
})
},
);
}
// APPEND_MARKER_WS_HANDLERS
pub(super) async fn workflow_ws(
State(state): State<AppState>,
headers: HeaderMap,
ws: WebSocketUpgrade,
Query(query): Query<HashMap<String, String>>,
) -> AppResult<Response> {
let _ = auth_caller(&state, &headers).await?;
let sid = query
.get("clientId")
.or_else(|| query.get("client_id"))
.cloned()
.unwrap_or_else(|| "boosapi-workflow".to_string());
if native_workflow_enabled(&state) && !state.config.comfyui.uses_runninghub_backend() {
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};
let mut socket = socket;
let mut receiver = workflow_session_sender(&sid).subscribe();
let mut ticker = interval(Duration::from_secs(20));
let _ = socket
.send(Message::Text(
json!({
"type": "status",
"data": {
"status": { "exec_info": { "queue_remaining": 0 } },
"sid": sid
}
})
.to_string()
.into(),
))
.await;
loop {
tokio::select! {
event = receiver.recv() => {
match event {
Ok(event) => {
if socket.send(Message::Text(event.to_string().into())).await.is_err() {
break;
}
}
Err(broadcast::error::RecvError::Lagged(_)) => {}
Err(broadcast::error::RecvError::Closed) => break,
}
}
_ = ticker.tick() => {
if socket.send(Message::Text(json!({"type":"status","data":{"status":{"exec_info":{"queue_remaining":0}},"sid":sid}}).to_string().into())).await.is_err() {
break;
}
}
next = socket.recv() => {
match next {
Some(Ok(Message::Close(_))) | None => break,
Some(Ok(_)) => {}
Some(Err(_)) => break,
}
}
}
}
}))
}
// APPEND_MARKER_BRIDGE
async fn native_workflow_ws_bridge(
state: AppState,
mut client_socket: axum::extract::ws::WebSocket,
mut query: HashMap<String, String>,
sid: String,
) {
let mut ws_url = native_workflow_base(&state)
.replacen("https://", "wss://", 1)
.replacen("http://", "ws://", 1);
query.entry("clientId".to_string()).or_insert(sid.clone());
if let Some(api_key) = state
.config
.comfyui
.api_key
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
{
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));
let request = match ws_url.as_str().into_client_request() {
Ok(mut request) => {
if let Some(api_key) = state
.config
.comfyui
.api_key
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
{
if let Ok(value) = tungstenite::http::HeaderValue::from_str(api_key) {
request.headers_mut().insert("X-API-Key", value);
}
if let Ok(value) =
tungstenite::http::HeaderValue::from_str(&format!("Bearer {api_key}"))
{
request
.headers_mut()
.insert(tungstenite::http::header::AUTHORIZATION, value);
}
}
request
}
Err(err) => {
let _ = client_socket
.send(Message::Text(
json!({"type":"execution_error","data":{"exception_message":format!("invalid ComfyUI websocket URL: {err}")}})
.to_string()
.into(),
))
.await;
return;
}
};
let upstream_target = ws_url
.split('?')
.next()
.unwrap_or(ws_url.as_str())
.to_string();
let (upstream_socket, _) = match connect_async(request).await {
Ok(value) => value,
Err(err) => {
tracing::warn!(target = %upstream_target, error = %err, "ComfyUI native websocket connect failed");
let _ = client_socket
.send(Message::Text(
json!({"type":"execution_error","data":{"exception_message":format!("ComfyUI websocket connect failed: {err}")}})
.to_string()
.into(),
))
.await;
return;
}
};
use futures::{SinkExt, StreamExt};
let (mut upstream_write, mut upstream_read) = upstream_socket.split();
loop {
tokio::select! {
from_upstream = upstream_read.next() => {
match from_upstream {
Some(Ok(message)) => {
let outgoing = match message {
tungstenite::Message::Text(text) => Message::Text(text.to_string().into()),
tungstenite::Message::Binary(bytes) => Message::Binary(bytes.to_vec().into()),
tungstenite::Message::Ping(bytes) => Message::Ping(bytes.to_vec().into()),
tungstenite::Message::Pong(bytes) => Message::Pong(bytes.to_vec().into()),
tungstenite::Message::Close(frame) => {
let _ = client_socket.send(Message::Close(frame.map(|frame| axum::extract::ws::CloseFrame {
code: frame.code.into(),
reason: frame.reason.to_string().into(),
}))).await;
break;
}
tungstenite::Message::Frame(_) => continue,
};
if client_socket.send(outgoing).await.is_err() {
break;
}
}
Some(Err(_)) | None => break,
}
}
from_client = client_socket.recv() => {
match from_client {
Some(Ok(message)) => {
let outgoing = match message {
Message::Text(text) => tungstenite::Message::Text(text.to_string().into()),
Message::Binary(bytes) => tungstenite::Message::Binary(bytes.to_vec().into()),
Message::Ping(bytes) => tungstenite::Message::Ping(bytes.to_vec().into()),
Message::Pong(bytes) => tungstenite::Message::Pong(bytes.to_vec().into()),
Message::Close(frame) => {
let _ = upstream_write.send(tungstenite::Message::Close(frame.map(|frame| tungstenite::protocol::CloseFrame {
code: frame.code.into(),
reason: frame.reason.to_string().into(),
}))).await;
break;
}
};
if upstream_write.send(outgoing).await.is_err() {
break;
}
}
Some(Err(_)) | None => break,
}
}
}
}
}