refactor: workflow_service改名app

app/comfy/comfy_queue.py

@@ -0,0 +1,234 @@
+import asyncio
+import logging
+from typing import Any, Optional
+
+import aiohttp
+
+from app.comfy.comfy_workflow import ComfyWorkflow
+from app.comfy.comfy_run import ComfyRun
+from app.config import Settings
+from app.comfy.comfy_server import server_manager, ComfyUIServerInfo
+
+settings = Settings()
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+# 全局任务队列管理器
+class WorkflowQueueManager:
+    def __init__(self, monitor_interval: int = 5):
+        self.running_tasks = {}  # server_url -> task_info
+        self.pending_tasks = []  # 等待队列
+        self.lock = asyncio.Lock()
+        self._queue_monitor_task: Optional[asyncio.Task] = None
+        self._monitor_interval = monitor_interval  # 监控间隔（秒）
+        self._last_processing_time = None  # 上次处理队列的时间
+
+    async def start_queue_monitor(self):
+        """启动队列监控任务"""
+        if self._queue_monitor_task is None or self._queue_monitor_task.done():
+            self._queue_monitor_task = asyncio.create_task(self._monitor_queue())
+            logger.info(f"队列监控任务已启动，监控间隔: {self._monitor_interval}秒")
+
+    async def stop_queue_monitor(self):
+        """停止队列监控任务"""
+        if self._queue_monitor_task and not self._queue_monitor_task.done():
+            self._queue_monitor_task.cancel()
+            try:
+                await self._queue_monitor_task
+            except asyncio.CancelledError:
+                pass
+            logger.info("队列监控任务已停止")
+
+    async def add_task(
+        self,
+        workflow: ComfyWorkflow,
+        request_data: dict,
+    ):
+        """添加新任务到队列"""
+        async with self.lock:
+            # 创建ComfyRun实例并保存到数据库
+            comfy_run = await ComfyRun.create(workflow, request_data)
+
+            # 添加到待处理队列
+            self.pending_tasks.append(comfy_run.run_id)
+            logger.info(
+                f"任务 {comfy_run.run_id} 已添加到队列，当前队列长度: {len(self.pending_tasks)}"
+            )
+
+            return comfy_run.run_id
+
+    async def get_server_status(
+        self, server: ComfyUIServerInfo, session: aiohttp.ClientSession
+    ) -> dict[str, Any]:
+        """
+        检查单个ComfyUI服务器的详细状态。
+        返回一个包含可达性、队列状态和详细队列内容的字典。
+        """
+        # [BUG修复] 确保初始字典结构与成功时的结构一致，以满足Pydantic模型
+        status_info = {
+            "is_reachable": False,
+            "is_free": False,
+            "queue_details": {"running_count": 0, "pending_count": 0},
+        }
+        try:
+            queue_url = f"{server.http_url}/queue"
+            async with session.get(queue_url, timeout=60) as response:
+                response.raise_for_status()
+                queue_data = await response.json()
+
+                status_info["is_reachable"] = True
+
+                running_count = len(queue_data.get("queue_running", []))
+                pending_count = len(queue_data.get("queue_pending", []))
+
+                status_info["queue_details"] = {
+                    "running_count": running_count,
+                    "pending_count": pending_count,
+                }
+
+                status_info["is_free"] = running_count == 0 and pending_count == 0
+
+        except Exception as e:
+            # 当请求失败时，将返回上面定义的、结构正确的初始 status_info
+            logger.warning(f"无法检查服务器 {server.http_url} 的队列状态: {e}")
+
+        return status_info
+
+    async def _monitor_queue(self):
+        """定期监控队列，当有可用服务器时自动处理"""
+        while True:
+            try:
+                await asyncio.sleep(self._monitor_interval)
+
+                # 每次定时检测触发时，打印当前状态
+                async with self.lock:
+                    from datetime import datetime
+
+                    pending_count = len(self.pending_tasks)
+                    running_count = len(self.running_tasks)
+                    current_time = datetime.now()
+                    timestamp = current_time.strftime("%Y-%m-%d %H:%M:%S")
+
+                    # 计算距离上次处理队列的时间
+                    time_since_last_processing = ""
+                    if self._last_processing_time:
+                        time_diff = current_time - self._last_processing_time
+                        minutes = int(time_diff.total_seconds() // 60)
+                        seconds = int(time_diff.total_seconds() % 60)
+                        time_since_last_processing = (
+                            f"上次处理: {minutes}分{seconds}秒前"
+                        )
+                    else:
+                        time_since_last_processing = "上次处理: 从未"
+
+                    # 格式化状态信息，提高可读性
+                    status_info = (
+                        f"⏰ 定时检测触发 [{timestamp}]\t"
+                        f"待处理/运行中: {pending_count}/{running_count}\t"
+                        f"监控间隔: {self._monitor_interval} 秒\t"
+                        f"🕐 {time_since_last_processing}"
+                    )
+                    logger.info(status_info)
+
+                    if self.pending_tasks:
+                        available_servers = await self._get_available_servers()
+                        if available_servers:
+                            logger.info(
+                                f"监控发现 {len(available_servers)} 个可用服务器，开始处理队列"
+                            )
+                            # 使用create_task避免阻塞监控循环
+                            asyncio.create_task(self._process_queue())
+                        else:
+                            # 只在有任务等待时记录日志，避免日志噪音
+                            if len(self.pending_tasks) > 0:
+                                logger.debug(
+                                    f"队列中有 {len(self.pending_tasks)} 个待处理任务，但无可用服务器"
+                                )
+                    else:
+                        # 队列为空时，减少日志输出
+                        logger.debug("队列为空，继续监控中...")
+
+            except asyncio.CancelledError:
+                logger.info("队列监控任务被取消")
+                break
+            except Exception as e:
+                logger.error(f"队列监控任务出错: {e}")
+                # 出错时稍微延长等待时间，避免频繁重试
+                await asyncio.sleep(self._monitor_interval * 2)
+
+    async def _process_queue(self):
+        """处理队列中的任务"""
+        async with self.lock:
+            # 更新上次处理队列的时间
+            from datetime import datetime
+
+            self._last_processing_time = datetime.now()
+
+            if not self.pending_tasks:
+                logger.debug("队列为空，无需处理")
+                return
+
+            # 检查是否有空闲的服务器
+            available_servers = await self._get_available_servers()
+            if not available_servers:
+                logger.info(
+                    f"队列中有 {len(self.pending_tasks)} 个待处理任务，但没有可用的服务器"
+                )
+                return
+
+            # 获取一个待处理任务
+            workflow_run_id = self.pending_tasks.pop(0)
+            server = available_servers[0]
+
+            logger.info(
+                f"开始处理任务 {workflow_run_id}，使用服务器 {server.name} ({server.http_url})"
+            )
+
+            # 记录运行中任务
+            self.running_tasks[server.http_url] = {
+                "workflow_run_id": workflow_run_id,
+                "started_at": datetime.now(),
+                "server_name": server.name,
+            }
+
+            # 启动任务执行
+            asyncio.create_task(self._execute_task(workflow_run_id, server))
+
+            logger.info(
+                f"任务 {workflow_run_id} 已分配到服务器 {server.name}，当前运行中任务数: {len(self.running_tasks)}"
+            )
+
+    async def _get_available_servers(self) -> list[ComfyUIServerInfo]:
+        """获取可用的服务器"""
+        # 使用新的服务器管理器获取可用服务器
+        available_server = await server_manager.get_available_server()
+        if available_server:
+            return [available_server]
+        return []
+
+    async def _execute_task(self, workflow_run_id: str, server: ComfyUIServerInfo):
+        """执行任务"""
+        try:
+            # 从run_id创建ComfyRun实例并执行
+            comfy_run = await ComfyRun.from_run_id(workflow_run_id)
+            if not comfy_run:
+                raise Exception(f"找不到工作流运行记录: {workflow_run_id}")
+
+            # 执行工作流（ComfyRun内部处理所有逻辑包括资源管理）
+            await comfy_run.execute(server)
+
+        except Exception as e:
+            logger.error(f"执行任务 {workflow_run_id} 时出错: {e}")
+
+        finally:
+            # 清理运行状态
+            async with self.lock:
+                if server.http_url in self.running_tasks:
+                    del self.running_tasks[server.http_url]
+
+
+# 全局队列管理器实例
+# 从配置中读取监控间隔，默认为5秒
+queue_manager = WorkflowQueueManager(monitor_interval=5)

app/comfy/comfy_run.py

@@ -0,0 +1,345 @@
+import json
+import logging
+import random
+import uuid
+from typing import Optional
+
+import aiohttp
+import websockets
+from aiohttp import ClientTimeout
+
+from app.comfy.comfy_workflow import ComfyWorkflow, API_OUTPUT_PREFIX
+from app.comfy.comfy_server import ComfyUIServerInfo, server_manager
+from app.database.api import (
+    create_workflow_run,
+    create_workflow_run_nodes,
+    get_workflow_run,
+    update_workflow_run_status,
+    update_workflow_run_node_status,
+)
+
+
+class ComfyUIExecutionError(Exception):
+    """ComfyUI执行错误"""
+
+    def __init__(self, error_data):
+        self.error_data = error_data
+        super().__init__(f"ComfyUI execution error: {error_data}")
+
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class ComfyRun:
+    """
+    ComfyUI工作流运行实例，封装单个工作流运行的所有操作
+    """
+
+    def __init__(self, workflow: ComfyWorkflow, run_id: str, request_data: dict):
+        """
+        初始化工作流运行实例
+
+        Args:
+            workflow: ComfyWorkflow实例
+            run_id: 运行ID
+            request_data: 请求数据
+        """
+        self.workflow = workflow
+        self.run_id = run_id
+        self.request_data = request_data
+        self.client_id = str(uuid.uuid4())
+        self.prompt_id: Optional[str] = None
+        self.server: Optional[ComfyUIServerInfo] = None
+
+    @classmethod
+    async def create(cls, workflow: ComfyWorkflow, request_data: dict) -> "ComfyRun":
+        """
+        创建新的ComfyRun实例并保存到数据库
+
+        Args:
+            workflow: ComfyWorkflow实例
+            request_data: 请求数据
+
+        Returns:
+            ComfyRun实例
+        """
+        run_id = str(uuid.uuid4())
+
+        # 创建任务记录
+        await create_workflow_run(
+            workflow_run_id=run_id,
+            workflow_name=workflow.workflow_name,
+            workflow_json=json.dumps(workflow.workflow_data.model_dump()),
+            api_spec=json.dumps(workflow.get_api_spec().model_dump()),
+            request_data=json.dumps(request_data),
+        )
+
+        # 创建工作流节点记录
+        nodes_data = []
+        for node in workflow.workflow_data.nodes:
+            nodes_data.append({"id": str(node.id), "type": node.type})
+        await create_workflow_run_nodes(run_id, nodes_data)
+
+        return cls(workflow, run_id, request_data)
+
+    @classmethod
+    async def from_run_id(cls, run_id: str) -> Optional["ComfyRun"]:
+        """
+        从run_id创建ComfyRun实例
+
+        Args:
+            run_id: 运行ID
+
+        Returns:
+            ComfyRun实例，如果找不到则返回None
+        """
+        workflow_run = await get_workflow_run(run_id)
+        if not workflow_run:
+            return None
+
+        workflow_data = json.loads(workflow_run.workflow_json)
+        workflow = ComfyWorkflow(workflow_run.workflow_name, workflow_data)
+        request_data = json.loads(workflow_run.request_data)
+
+        return cls(workflow, run_id, request_data)
+
+    async def execute(self, server: ComfyUIServerInfo) -> dict:
+        """
+        在指定的服务器上执行工作流
+
+        Args:
+            server: ComfyUI服务器信息
+
+        Returns:
+            执行结果
+        """
+        self.server = server
+
+        try:
+            # 分配服务器资源
+            await server_manager.allocate_server(server.name)
+
+            # 构建prompt
+            prompt = await self.workflow.build_prompt(server, self.request_data)
+
+            # 更新运行状态为running
+            await self._update_status(
+                "running", server.http_url, client_id=self.client_id
+            )
+
+            # 提交到ComfyUI
+            self.prompt_id = await self._queue_prompt(prompt, server.http_url)
+
+            # 更新prompt_id
+            await self._update_status(
+                "running", server.http_url, self.prompt_id, self.client_id
+            )
+
+            logger.info(
+                f"工作流 {self.run_id} 已在 {server.http_url} 上入队，Prompt ID: {self.prompt_id}"
+            )
+
+            # 获取执行结果
+            results = await self._get_execution_results(server.ws_url)
+
+            # 标记完成
+            await self._update_status(
+                "completed", result=json.dumps(results, ensure_ascii=False)
+            )
+
+            return results
+
+        except Exception as e:
+            # 标记失败
+            await self._update_status("failed", error_message=str(e))
+            raise
+
+        finally:
+            # 释放服务器资源
+            try:
+                await server_manager.release_server(server.name)
+            except Exception as e:
+                logger.error(f"释放服务器资源时出错: {e}")
+
+    async def _update_status(
+        self,
+        status: str,
+        server_url: Optional[str] = None,
+        prompt_id: Optional[str] = None,
+        client_id: Optional[str] = None,
+        result: Optional[str] = None,
+        error_message: Optional[str] = None,
+    ):
+        """更新工作流运行状态"""
+        await update_workflow_run_status(
+            self.run_id, status, server_url, prompt_id, client_id, error_message, result
+        )
+
+    async def _queue_prompt(self, prompt: dict, http_url: str) -> str:
+        """提交工作流到ComfyUI服务器"""
+        # 添加随机缓存破坏器避免缓存
+        for node_id in prompt:
+            prompt[node_id]["inputs"][
+                f"cache_buster_{uuid.uuid4().hex}"
+            ] = random.random()
+
+        payload = {
+            "prompt": prompt,
+            "client_id": self.client_id,
+            "extra_data": {
+                "api_key_comfy_org": "",
+                "extra_pnginfo": {"workflow": self.workflow.workflow_data.model_dump()},
+            },
+        }
+
+        async with aiohttp.ClientSession(timeout=ClientTimeout(total=90)) as session:
+            prompt_url = f"{http_url}/prompt"
+            try:
+                async with session.post(prompt_url, json=payload) as response:
+                    logger.info(f"ComfyUI /prompt 端点返回的响应: {response}")
+                    response.raise_for_status()
+                    result = await response.json()
+                    logger.info(f"ComfyUI /prompt 端点返回的响应: {result}")
+
+                    if "prompt_id" not in result:
+                        raise Exception(
+                            f"从 ComfyUI /prompt 端点返回的响应无效: {result}"
+                        )
+
+                    return result["prompt_id"]
+
+            except Exception as e:
+                logger.error(f"提交到 ComfyUI /prompt 端点时发生错误: {e}")
+                raise
+
+    async def _get_execution_results(self, ws_url: str) -> dict:
+        """
+        通过WebSocket连接获取执行结果，支持节点级别的状态跟踪
+        """
+        aggregated_outputs = {}
+        full_ws_url = f"{ws_url}?clientId={self.client_id}"
+
+        try:
+            async with websockets.connect(full_ws_url) as websocket:
+                logger.info(f"已连接到WebSocket: {full_ws_url}")
+
+                while True:
+                    try:
+                        out = await websocket.recv()
+                        if not isinstance(out, str):
+                            continue
+
+                        message = json.loads(out)
+                        msg_type = message.get("type")
+                        data = message.get("data")
+
+                        if not (data and data.get("prompt_id") == self.prompt_id):
+                            continue
+
+                        # 捕获并处理执行错误
+                        if msg_type == "execution_error":
+                            error_data = data
+                            logger.error(
+                                f"ComfyUI执行错误 (Prompt ID: {self.prompt_id}): {error_data}"
+                            )
+
+                            # 更新节点状态为失败
+                            node_id = error_data.get("node_id")
+                            if node_id:
+                                await update_workflow_run_node_status(
+                                    self.run_id,
+                                    node_id,
+                                    "failed",
+                                    error_message=error_data.get(
+                                        "exception_message", "Unknown error"
+                                    ),
+                                )
+
+                            # 抛出自定义异常，将错误详情传递出去
+                            raise ComfyUIExecutionError(error_data)
+
+                        # 处理节点开始执行
+                        if msg_type == "executing" and data.get("node"):
+                            node_id = data.get("node")
+                            logger.info(
+                                f"节点 {node_id} 开始执行 (Prompt ID: {self.prompt_id})"
+                            )
+
+                            # 更新节点状态为运行中
+                            await update_workflow_run_node_status(
+                                self.run_id, node_id, "running"
+                            )
+
+                        # 处理节点执行完成
+                        elif msg_type == "executed":
+                            await self._handle_node_executed(data, aggregated_outputs)
+
+                        # 处理整个工作流执行完成
+                        elif msg_type == "executing" and data.get("node") is None:
+                            logger.info(f"Prompt ID: {self.prompt_id} 执行完成。")
+                            return aggregated_outputs
+
+                    except websockets.exceptions.ConnectionClosed as e:
+                        logger.warning(
+                            f"WebSocket 连接已关闭 (Prompt ID: {self.prompt_id})。错误: {e}"
+                        )
+                        return aggregated_outputs
+                    except Exception as e:
+                        # 重新抛出我们自己的异常，或者处理其他意外错误
+                        if not isinstance(e, ComfyUIExecutionError):
+                            logger.error(
+                                f"处理 prompt {self.prompt_id} 时发生意外错误: {e}"
+                            )
+                        raise e
+
+        except websockets.exceptions.InvalidURI as e:
+            logger.error(
+                f"错误: 尝试连接的WebSocket URI无效: '{full_ws_url}'. 原始URL: '{ws_url}'. 错误: {e}"
+            )
+            raise e
+        except Exception as e:
+            logger.error(f"WebSocket连接出错: {e}")
+            raise
+
+    async def _handle_node_executed(self, data: dict, aggregated_outputs: dict):
+        """处理节点执行完成事件"""
+        node_id = data.get("node")
+        output_data = data.get("output")
+
+        if not node_id or not output_data:
+            return
+
+        # 查找对应的节点
+        node = next(
+            (x for x in self.workflow.workflow_data.nodes if str(x.id) == node_id),
+            None,
+        )
+
+        if not node:
+            return
+
+        # 如果是输出节点，收集结果
+        if node.title and node.title.startswith(API_OUTPUT_PREFIX):
+            title = node.title.replace(API_OUTPUT_PREFIX, "")
+            aggregated_outputs[title] = output_data
+
+        logger.info(f"收到节点 {node_id} 的输出 (Prompt ID: {self.prompt_id})")
+
+        # 更新节点状态为完成
+        await update_workflow_run_node_status(
+            self.run_id,
+            node_id,
+            "completed",
+            output_data=json.dumps(output_data),
+        )
+
+    def get_status_info(self) -> dict:
+        """获取运行状态信息"""
+        return {
+            "run_id": self.run_id,
+            "workflow_name": self.workflow.workflow_name,
+            "client_id": self.client_id,
+            "prompt_id": self.prompt_id,
+            "server_url": self.server.http_url if self.server else None,
+        }

app/comfy/comfy_server.py

@@ -0,0 +1,447 @@
+import asyncio
+import json
+import logging
+import time
+from datetime import datetime, timedelta
+from typing import Dict, List, Optional, Set
+from dataclasses import dataclass, asdict
+from enum import Enum
+
+import aiohttp
+from aiohttp import ClientTimeout
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, update, delete
+from sqlalchemy.orm import selectinload
+
+from app.database.connection import AsyncSessionLocal
+from app.database.models import ComfyUIServer as ComfyUIServerModel
+
+logger = logging.getLogger(__name__)
+
+
+class ServerStatus(Enum):
+    """服务器状态枚举"""
+
+    ONLINE = "online"
+    OFFLINE = "offline"
+    BUSY = "busy"
+    ERROR = "error"
+
+
+@dataclass
+class ComfyUIServerInfo:
+    """ComfyUI服务器信息"""
+
+    name: str  # 服务器名称，由注册方自行拟定
+    http_url: str
+    ws_url: str
+    status: ServerStatus = ServerStatus.OFFLINE
+    last_health_check: Optional[datetime] = None
+    current_tasks: int = 0
+    max_concurrent_tasks: int = 1
+    capabilities: Dict[str, any] = None  # 服务器能力信息，如支持的模型等
+    metadata: Dict[str, any] = None  # 其他元数据
+
+    def __post_init__(self):
+        if self.capabilities is None:
+            self.capabilities = {}
+        if self.metadata is None:
+            self.metadata = {}
+
+    @classmethod
+    def from_model(cls, model: ComfyUIServerModel) -> "ComfyUIServerInfo":
+        """从数据库模型创建实例"""
+        capabilities = json.loads(model.capabilities) if model.capabilities else {}
+        metadata = json.loads(model.server_metadata) if model.server_metadata else {}
+
+        return cls(
+            name=model.name,
+            http_url=model.http_url,
+            ws_url=model.ws_url,
+            status=ServerStatus(model.status),
+            last_health_check=model.last_health_check,
+            current_tasks=model.current_tasks,
+            max_concurrent_tasks=model.max_concurrent_tasks,
+            capabilities=capabilities,
+            metadata=metadata,
+        )
+
+    def to_model(self) -> ComfyUIServerModel:
+        """转换为数据库模型"""
+        return ComfyUIServerModel(
+            name=self.name,
+            http_url=self.http_url,
+            ws_url=self.ws_url,
+            status=self.status.value,
+            last_health_check=self.last_health_check,
+            current_tasks=self.current_tasks,
+            max_concurrent_tasks=self.max_concurrent_tasks,
+            capabilities=json.dumps(self.capabilities),
+            server_metadata=json.dumps(self.metadata),
+        )
+
+
+class ComfyUIServerManager:
+    """ComfyUI服务器管理器"""
+
+    def __init__(self):
+        self.servers: Dict[str, ComfyUIServerInfo] = {}  # name -> server_info
+        self.lock = asyncio.Lock()
+        self.health_check_interval = 30  # 健康检查间隔（秒）
+        self._health_check_task: Optional[asyncio.Task] = None
+        self._initialized = False
+
+    async def initialize(self):
+        """初始化管理器，从数据库加载服务器信息"""
+        if self._initialized:
+            return
+
+        async with self.lock:
+            try:
+                await self._load_servers_from_db()
+                self._initialized = True
+                logger.info(f"从数据库加载了 {len(self.servers)} 个服务器")
+            except Exception as e:
+                logger.error(f"从数据库加载服务器失败: {e}")
+                self._initialized = True  # 即使失败也标记为已初始化，避免重复尝试
+
+    async def _load_servers_from_db(self):
+        """从数据库加载服务器信息"""
+        async with AsyncSessionLocal() as session:
+            async with session.begin():
+                result = await session.execute(select(ComfyUIServerModel))
+                models = result.scalars().all()
+
+                for model in models:
+                    try:
+                        server_info = ComfyUIServerInfo.from_model(model)
+                        self.servers[server_info.name] = server_info
+                    except Exception as e:
+                        logger.error(f"加载服务器 {model.name} 失败: {e}")
+
+    async def _save_server_to_db(self, server: ComfyUIServerInfo):
+        """保存服务器信息到数据库"""
+        try:
+            async with AsyncSessionLocal() as session:
+                async with session.begin():
+                    # 检查是否已存在
+                    existing = await session.execute(
+                        select(ComfyUIServerModel).where(
+                            ComfyUIServerModel.name == server.name
+                        )
+                    )
+                    existing_model = existing.scalar_one_or_none()
+
+                    if existing_model:
+                        # 更新现有记录
+                        await session.execute(
+                            update(ComfyUIServerModel)
+                            .where(ComfyUIServerModel.name == server.name)
+                            .values(
+                                http_url=server.http_url,
+                                ws_url=server.ws_url,
+                                status=server.status.value,
+                                last_health_check=server.last_health_check,
+                                current_tasks=server.current_tasks,
+                                max_concurrent_tasks=server.max_concurrent_tasks,
+                                capabilities=json.dumps(server.capabilities),
+                                server_metadata=json.dumps(server.metadata),
+                                updated_at=datetime.utcnow(),
+                            )
+                        )
+                    else:
+                        # 创建新记录
+                        new_model = server.to_model()
+                        session.add(new_model)
+
+                    await session.commit()
+        except Exception as e:
+            logger.error(f"保存服务器 {server.name} 到数据库失败: {e}")
+
+    async def _delete_server_from_db(self, name: str):
+        """从数据库删除服务器"""
+        try:
+            async with AsyncSessionLocal() as session:
+                async with session.begin():
+                    await session.execute(
+                        delete(ComfyUIServerModel).where(
+                            ComfyUIServerModel.name == name
+                        )
+                    )
+                    await session.commit()
+        except Exception as e:
+            logger.error(f"从数据库删除服务器 {name} 失败: {e}")
+
+    async def register_server(
+        self,
+        name: str,
+        http_url: str,
+        ws_url: str,
+        max_concurrent_tasks: int = 1,
+        capabilities: Optional[Dict[str, any]] = None,
+        metadata: Optional[Dict[str, any]] = None,
+    ) -> bool:
+        """注册新的ComfyUI服务器"""
+        # 确保已初始化
+        await self.initialize()
+
+        # 确保健康检查任务已启动
+        await self._ensure_health_check_started()
+
+        async with self.lock:
+            # 解析URL获取IP和端口
+            from urllib.parse import urlparse
+            
+            parsed_url = urlparse(http_url)
+            host = parsed_url.hostname
+            port = parsed_url.port or (80 if parsed_url.scheme == 'http' else 443)
+            
+            # 检查是否已存在相同IP+端口的服务器
+            existing_server_name = None
+            for existing_server in self.servers.values():
+                existing_parsed = urlparse(existing_server.http_url)
+                existing_host = existing_parsed.hostname
+                existing_port = existing_parsed.port or (80 if existing_parsed.scheme == 'http' else 443)
+                
+                if existing_host == host and existing_port == port:
+                    existing_server_name = existing_server.name
+                    logger.info(f"检测到IP {host}:{port} 已被服务器 {existing_server_name} 使用，将自动更新配置")
+                    break
+            
+            # 如果IP+端口已存在，更新现有服务器
+            if existing_server_name:
+                # 删除旧的服务器记录
+                if existing_server_name in self.servers:
+                    del self.servers[existing_server_name]
+                logger.info(f"已删除旧服务器 {existing_server_name} 的记录")
+            
+            # 检查名称是否已存在
+            if name in self.servers:
+                logger.warning(f"服务器名称 {name} 已存在，将更新配置")
+
+            server_info = ComfyUIServerInfo(
+                name=name,
+                http_url=http_url.rstrip("/"),
+                ws_url=ws_url.rstrip("/"),
+                status=ServerStatus.OFFLINE,  # 注册时状态为离线，等待健康检查
+                max_concurrent_tasks=max_concurrent_tasks,
+                capabilities=capabilities or {},
+                metadata=metadata or {},
+            )
+
+            # 立即进行健康检查
+            if await self._check_server_health(server_info):
+                server_info.status = ServerStatus.ONLINE
+                server_info.last_health_check = datetime.now()
+
+            self.servers[name] = server_info
+
+            # 保存到数据库
+            await self._save_server_to_db(server_info)
+            
+            # 如果更新了现有服务器，从数据库删除旧记录
+            if existing_server_name:
+                await self._delete_server_from_db(existing_server_name)
+
+            logger.info(f"服务器 {name} 注册成功: {http_url} (IP: {host}:{port})")
+            return True
+
+    async def unregister_server(self, name: str) -> bool:
+        """注销ComfyUI服务器"""
+        async with self.lock:
+            if name in self.servers:
+                del self.servers[name]
+                # 从数据库删除
+                await self._delete_server_from_db(name)
+                logger.info(f"服务器 {name} 已注销")
+                return True
+            return False
+
+    async def get_available_server(
+        self, required_capabilities: Optional[Dict[str, any]] = None
+    ) -> Optional[ComfyUIServerInfo]:
+        """获取可用的服务器（负载均衡）"""
+        # 确保已初始化
+        await self.initialize()
+
+        # 确保健康检查任务已启动
+        await self._ensure_health_check_started()
+
+        async with self.lock:
+            available_servers = []
+
+            for server in self.servers.values():
+                if (
+                    server.status == ServerStatus.ONLINE
+                    and server.current_tasks < server.max_concurrent_tasks
+                ):
+
+                    # 检查能力要求
+                    if required_capabilities:
+                        if not self._check_capabilities(
+                            server.capabilities, required_capabilities
+                        ):
+                            continue
+
+                    available_servers.append(server)
+
+            if not available_servers:
+                return None
+
+            # 简单的负载均衡：选择当前任务数最少的服务器
+            return min(available_servers, key=lambda s: s.current_tasks)
+
+    async def allocate_server(self, name: str) -> bool:
+        """分配服务器资源"""
+        async with self.lock:
+            if name in self.servers:
+                server = self.servers[name]
+                if (
+                    server.status == ServerStatus.ONLINE
+                    and server.current_tasks < server.max_concurrent_tasks
+                ):
+                    server.current_tasks += 1
+                    if server.current_tasks >= server.max_concurrent_tasks:
+                        server.status = ServerStatus.BUSY
+                    # 保存到数据库
+                    await self._save_server_to_db(server)
+                    return True
+            return False
+
+    async def release_server(self, name: str) -> bool:
+        """释放服务器资源"""
+        async with self.lock:
+            if name in self.servers:
+                server = self.servers[name]
+                if server.current_tasks > 0:
+                    server.current_tasks -= 1
+                    if (
+                        server.status == ServerStatus.BUSY
+                        and server.current_tasks < server.max_concurrent_tasks
+                    ):
+                        server.status = ServerStatus.ONLINE
+                    # 保存到数据库
+                    await self._save_server_to_db(server)
+                    return True
+            return False
+
+    async def get_server_status(self, name: str) -> Optional[ComfyUIServerInfo]:
+        """获取服务器状态"""
+        # 确保已初始化
+        await self.initialize()
+
+        async with self.lock:
+            return self.servers.get(name)
+
+    async def get_all_servers(self) -> List[ComfyUIServerInfo]:
+        """获取所有服务器信息"""
+        # 确保已初始化
+        await self.initialize()
+
+        async with self.lock:
+            return list(self.servers.values())
+
+    def _check_capabilities(
+        self, server_capabilities: Dict[str, any], required_capabilities: Dict[str, any]
+    ) -> bool:
+        """检查服务器是否满足能力要求"""
+        for key, value in required_capabilities.items():
+            if key not in server_capabilities:
+                return False
+            if isinstance(value, dict) and isinstance(server_capabilities[key], dict):
+                if not self._check_capabilities(server_capabilities[key], value):
+                    return False
+            elif server_capabilities[key] != value:
+                return False
+        return True
+
+    async def _check_server_health(self, server: ComfyUIServerInfo) -> bool:
+        """检查服务器健康状态"""
+        try:
+            timeout = ClientTimeout(total=10)
+            async with aiohttp.ClientSession(timeout=timeout) as session:
+                async with session.get(f"{server.http_url}/system_stats") as response:
+                    if response.status == 200:
+                        return True
+        except Exception as e:
+            logger.debug(f"服务器 {server.name} 健康检查失败: {e}")
+
+        return False
+
+    async def _health_check_loop(self):
+        """健康检查循环"""
+        while True:
+            try:
+                await self._perform_health_checks()
+            except Exception as e:
+                logger.error(f"健康检查过程中发生错误: {e}")
+
+            await asyncio.sleep(self.health_check_interval)
+
+    async def _perform_health_checks(self):
+        """执行健康检查"""
+        current_time = datetime.now()
+
+        async with self.lock:
+            for server in self.servers.values():
+                # 定期健康检查
+                if (
+                    not server.last_health_check
+                    or current_time - server.last_health_check
+                    > timedelta(seconds=self.health_check_interval)
+                ):
+
+                    is_healthy = await self._check_server_health(server)
+                    server.last_health_check = current_time
+
+                    if is_healthy and server.status != ServerStatus.ONLINE:
+                        server.status = ServerStatus.ONLINE
+                        logger.info(f"服务器 {server.name} 恢复在线")
+                    elif not is_healthy and server.status == ServerStatus.ONLINE:
+                        server.status = ServerStatus.ERROR
+                        logger.warning(f"服务器 {server.name} 健康检查失败")
+
+                    # 保存到数据库
+                    await self._save_server_to_db(server)
+
+    def _start_health_check(self):
+        """启动健康检查任务"""
+        try:
+            # 检查是否有运行中的事件循环
+            loop = asyncio.get_running_loop()
+            if self._health_check_task is None or self._health_check_task.done():
+                self._health_check_task = asyncio.create_task(self._health_check_loop())
+        except RuntimeError:
+            # 没有运行中的事件循环，延迟启动
+            logger.debug("没有运行中的事件循环，健康检查将在事件循环启动后自动开始")
+
+    async def _ensure_health_check_started(self):
+        """确保健康检查任务已启动"""
+        if self._health_check_task is None or self._health_check_task.done():
+            self._start_health_check()
+
+    async def shutdown(self):
+        """关闭管理器"""
+        if self._health_check_task and not self._health_check_task.done():
+            self._health_check_task.cancel()
+            try:
+                await self._health_check_task
+            except asyncio.CancelledError:
+                pass
+
+
+# 全局服务器管理器实例
+server_manager = ComfyUIServerManager()
+
+
+# 兼容性函数，用于替代原有的静态配置
+async def get_available_servers() -> List[ComfyUIServerInfo]:
+    """获取可用的服务器列表（兼容性函数）"""
+    return await server_manager.get_all_servers()
+
+
+async def get_server_for_task(
+    required_capabilities: Optional[Dict[str, any]] = None
+) -> Optional[ComfyUIServerInfo]:
+    """为任务获取合适的服务器（兼容性函数）"""
+    return await server_manager.get_available_server(required_capabilities)

app/comfy/comfy_workflow.py

@@ -0,0 +1,760 @@
+import logging
+from typing import Any, Dict, List, Optional, Union
+
+import aiohttp
+from pydantic import BaseModel
+from app.comfy.comfy_server import ComfyUIServerInfo
+
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+API_INPUT_PREFIX = "INPUT_"
+API_OUTPUT_PREFIX = "OUTPUT_"
+
+
+class ComfyWorkflowNodeWidget(BaseModel):
+    """节点控件定义"""
+
+    name: str
+
+
+class ComfyWorkflowNodeInput(BaseModel):
+    """节点输入定义"""
+
+    label: Optional[str] = None
+    localized_name: Optional[str] = None
+    name: str
+    shape: Optional[int] = None
+    type: str
+    link: Optional[int] = None
+    widget: Optional[ComfyWorkflowNodeWidget] = None
+
+
+class ComfyWorkflowNodeOutput(BaseModel):
+    """节点输出定义"""
+
+    label: Optional[str] = None
+    localized_name: Optional[str] = None
+    name: str
+    type: str
+    links: Optional[List[int]] = None
+
+
+class ComfyWorkflowNodeProperties(BaseModel):
+    """节点属性定义"""
+
+    cnr_id: Optional[str] = None
+    aux_id: Optional[str] = None
+    ver: Optional[str] = None
+    Node_name_for_SR: Optional[str] = None
+
+
+class ComfyAPIFieldSpec(BaseModel):
+    """API字段规范"""
+
+    type: str
+    widget_name: str
+
+
+class ComfyAPIOutputSpec(BaseModel):
+    """API输出规范"""
+
+    output_name: str
+
+
+class ComfyAPINodeSpec(BaseModel):
+    """API节点规范"""
+
+    node_id: str
+    class_type: str
+    inputs: Optional[Dict[str, ComfyAPIFieldSpec]] = {}
+    outputs: Optional[Dict[str, ComfyAPIOutputSpec]] = {}
+
+
+class ComfyAPISpec(BaseModel):
+    """API规范定义"""
+
+    inputs: Dict[str, ComfyAPINodeSpec] = {}
+    outputs: Dict[str, ComfyAPINodeSpec] = {}
+
+
+class ComfyJSONSchemaProperty(BaseModel):
+    """JSON Schema属性定义"""
+
+    type: str
+    title: Optional[str] = None
+    description: Optional[str] = None
+    default: Optional[Any] = None
+    minimum: Optional[Union[int, float]] = None
+    maximum: Optional[Union[int, float]] = None
+    enum: Optional[List[str]] = None
+    format: Optional[str] = None
+    contentMediaType: Optional[str] = None
+    pattern: Optional[str] = None
+
+
+class ComfyJSONSchemaNode(BaseModel):
+    """JSON Schema节点定义"""
+
+    type: str = "object"
+    title: str
+    description: str
+    properties: Dict[str, ComfyJSONSchemaProperty] = {}
+    required: List[str] = []
+    additionalProperties: bool = False
+
+
+class ComfyJSONSchema(BaseModel):
+    """JSON Schema定义"""
+
+    schema: str = "http://json-schema.org/draft-07/schema#"
+    type: str = "object"
+    title: str = "ComfyUI Workflow Input Schema"
+    description: str = "工作流输入参数定义"
+    properties: Dict[str, ComfyJSONSchemaNode] = {}
+    required: List[str] = []
+
+    class Config:
+        fields = {"schema": "$schema"}
+
+
+class ComfyWorkflowNode(BaseModel):
+    """工作流节点定义"""
+
+    id: int
+    type: str
+    pos: List[float]
+    size: List[float]
+    flags: Dict[str, Any] = {}
+    order: int
+    mode: int
+    inputs: Optional[List[ComfyWorkflowNodeInput]] = []
+    outputs: Optional[List[ComfyWorkflowNodeOutput]] = []
+    properties: ComfyWorkflowNodeProperties = ComfyWorkflowNodeProperties()
+    widgets_values: Optional[List[Any]] = []
+    title: Optional[str] = None
+
+
+class ComfyWorkflowExtra(BaseModel):
+    """工作流额外配置"""
+
+    ds: Optional[Dict[str, Any]] = None
+    frontendVersion: Optional[str] = None
+    VHS_latentpreview: Optional[bool] = None
+    VHS_latentpreviewrate: Optional[int] = None
+    VHS_MetadataImage: Optional[bool] = None
+    VHS_KeepIntermediate: Optional[bool] = None
+
+
+class ComfyWorkflowDataSpec(BaseModel):
+    """ComfyUI工作流数据规范"""
+
+    id: str
+    revision: int = 0
+    last_node_id: int
+    last_link_id: int
+    nodes: List[ComfyWorkflowNode]
+    links: List[List[Union[int, str]]] = []
+    groups: List[Any] = []
+    config: Dict[str, Any] = {}
+    extra: ComfyWorkflowExtra = ComfyWorkflowExtra()
+    version: float
+
+
+class ComfyWorkflow:
+    """
+    ComfyUI工作流处理器类，提供面向对象的工作流管理和处理能力
+    """
+
+    def __init__(self, workflow_name: str, workflow_data: dict):
+        """
+        初始化工作流实例
+
+        Args:
+            workflow_name: 工作流名称
+            workflow_data: 工作流数据
+        """
+        self.workflow_data = ComfyWorkflowDataSpec.model_validate(workflow_data)
+        self.workflow_name = workflow_name
+        self._nodes_map = {str(node.id): node for node in self.workflow_data.nodes}
+        self._api_spec = self._parse_api_spec()
+        self._inputs_json_schema = self._parse_inputs_json_schema()
+
+    async def build_prompt(
+        self, server: ComfyUIServerInfo, request_data: dict[str, Any]
+    ) -> dict:
+        """
+        构建prompt
+
+        Args:
+            request_data: 请求数据
+
+        Returns:
+            dict: 构建好的prompt
+        """
+        patched_workflow = await self._patch_workflow(server, request_data)
+        if "nodes" not in patched_workflow:
+            raise ValueError("无效的工作流格式")
+
+        prompt_api_format = {}
+
+        # 建立从link_id到源节点的映射
+        link_map = {}
+        for link_data in patched_workflow.get("links", []):
+            link_id, origin_node_id, origin_slot_index = (
+                link_data[0],
+                link_data[1],
+                link_data[2],
+            )
+
+            # 键是目标节点的输入link_id
+            link_map[link_id] = [str(origin_node_id), origin_slot_index]
+
+        for node in patched_workflow["nodes"]:
+            node_id = str(node["id"])
+            inputs_dict = self._get_node_inputs(node, link_map)
+            prompt_api_format[node_id] = {
+                "inputs": inputs_dict,
+                "class_type": node["type"],
+                "_meta": {
+                    "title": node.get("title", ""),
+                },
+            }
+
+        # 过滤掉类型为 Note 的节点
+        prompt_api_format = {
+            k: v
+            for k, v in prompt_api_format.items()
+            if v["class_type"] not in ["Note"]
+        }
+
+        return prompt_api_format
+
+    def get_api_spec(self) -> ComfyAPISpec:
+        """
+        获取API规范
+
+        Returns:
+            ComfyAPISpec: API规范
+        """
+        return self._api_spec
+
+    def get_inputs_json_schema(self) -> ComfyJSONSchema:
+        """
+        获取输入参数的JSON Schema
+
+        Returns:
+            ComfyJSONSchema: JSON Schema
+        """
+        return self._inputs_json_schema
+
+    def _parse_api_spec(self) -> ComfyAPISpec:
+        """
+        解析工作流，并根据规范生成嵌套结构的API参数名。
+        结构: {'节点名': {'node_id': '节点ID', '字段名': {字段信息}}}
+        """
+        inputs = {}
+        outputs = {}
+
+        for node_id, node in self._nodes_map.items():
+            title: str = node.title
+            if not title:
+                continue
+
+            if title.startswith(API_INPUT_PREFIX):
+                # 提取节点名（去掉API_INPUT_PREFIX前缀）
+                node_name = title[len(API_INPUT_PREFIX) :]
+
+                # 如果节点名不在inputs中，创建新的节点条目
+                if node_name not in inputs:
+                    inputs[node_name] = ComfyAPINodeSpec(
+                        node_id=node_id, class_type=node.type, inputs={}, outputs={}
+                    )
+
+                if node.inputs:
+                    for a_input in node.inputs:
+                        if a_input.link is None and a_input.widget:
+                            widget_name = a_input.widget.name
+
+                            # 特殊处理：隐藏LoadImage节点的upload字段
+                            if node.type == "LoadImage" and widget_name == "upload":
+                                continue
+
+                            # 直接使用widget_name作为字段名
+                            inputs[node_name].inputs[widget_name] = ComfyAPIFieldSpec(
+                                type=self._get_param_type(a_input.type or "STRING"),
+                                widget_name=a_input.widget.name,
+                            )
+
+            elif title.startswith(API_OUTPUT_PREFIX):
+                # 提取节点名（去掉API_OUTPUT_PREFIX前缀）
+                node_name = title[len(API_OUTPUT_PREFIX) :]
+
+                # 如果节点名不在outputs中，创建新的节点条目
+                if node_name not in outputs:
+                    outputs[node_name] = ComfyAPINodeSpec(
+                        node_id=node_id, class_type=node.type, inputs={}, outputs={}
+                    )
+
+                if node.outputs:
+                    for an_output in node.outputs:
+                        output_name = an_output.name
+
+                        # 直接使用output_name作为字段名
+                        outputs[node_name].outputs[output_name] = ComfyAPIOutputSpec(
+                            output_name=an_output.name
+                        )
+
+        return ComfyAPISpec(inputs=inputs, outputs=outputs)
+
+    def _parse_inputs_json_schema(self) -> ComfyJSONSchema:
+        """
+        解析工作流，生成符合 JSON Schema 标准的 API 规范。
+        返回一个只包含输入参数的 JSON Schema 对象。
+        """
+        # 收集所有输入节点
+        properties = {}
+        required = []
+
+        for _, node in self._nodes_map.items():
+            title: str = node.title
+            if not title:
+                continue
+
+            if title.startswith(API_INPUT_PREFIX):
+                # 提取节点名（去掉API_INPUT_PREFIX前缀）
+                node_name = title[len(API_INPUT_PREFIX) :]
+
+                node_properties = {}
+                node_required = []
+
+                if node.inputs:
+                    for a_input in node.inputs:
+                        if a_input.link is None and a_input.widget:
+                            widget_name = a_input.widget.name
+
+                            # 特殊处理：隐藏LoadImage节点的upload字段
+                            if node.type == "LoadImage" and widget_name == "upload":
+                                continue
+
+                            # 获取节点的当前值作为默认值
+                            # 创建空的link_map，因为输入节点通常没有连接
+                            empty_link_map = {}
+                            node_current_inputs = self._get_node_inputs(
+                                node, empty_link_map
+                            )
+                            current_value = node_current_inputs.get(widget_name)
+
+                            # 生成字段的 JSON Schema 定义，包含默认值
+                            field_schema = self._generate_field_schema_model(
+                                a_input, current_value
+                            )
+                            node_properties[widget_name] = field_schema
+
+                            # 如果字段是必需的，添加到required列表
+                            # TODO: 从输入配置中获取是否必需的信息
+                            node_required.append(widget_name)
+
+                # 创建节点的JSON Schema
+                if node_properties:
+                    properties[node_name] = ComfyJSONSchemaNode(
+                        title=f"输入节点: {node_name}",
+                        description=f"节点类型: {node.type}",
+                        properties=node_properties,
+                        required=node_required,
+                    )
+
+                    # 如果节点有必需字段，将整个节点标记为必需
+                    if node_required:
+                        required.append(node_name)
+
+        return ComfyJSONSchema(properties=properties, required=required)
+
+    async def _patch_workflow(
+        self,
+        server: ComfyUIServerInfo,
+        request_data: dict[str, Any],
+    ) -> dict:
+        """
+        将request_data中的参数值，patch到workflow_data中。并返回修改后的workflow_data。
+
+        request_data结构: {"节点名称": {"字段名称": "字段值"}}
+        """
+
+        nodes_map = {k: v.model_copy(deep=True) for k, v in self._nodes_map.items()}
+
+        for node_name, node_fields in request_data.items():
+            if node_name not in self._api_spec.inputs:
+                continue
+
+            node_spec = self._api_spec.inputs[node_name]
+            node_id = node_spec.node_id
+            if node_id not in nodes_map:
+                continue
+
+            target_node = nodes_map[node_id]
+
+            # 处理该节点下的所有字段
+            for field_name, value in node_fields.items():
+                if field_name not in node_spec.inputs:
+                    continue
+
+                field_spec = node_spec.inputs[field_name]
+                widget_name_to_patch = field_spec.widget_name
+
+                # 找到所有属于widget的输入，并确定目标widget的索引
+                widget_inputs = [
+                    inp for inp in (target_node.inputs or []) if inp.widget is not None
+                ]
+
+                target_widget_index = -1
+                for i, widget_input in enumerate(widget_inputs):
+                    if (
+                        widget_input.widget
+                        and widget_input.widget.name == widget_name_to_patch
+                    ):
+                        target_widget_index = i
+                        break
+
+                if target_widget_index == -1:
+                    logger.warning(
+                        f"在节点 {node_id} 中未找到名为 '{widget_name_to_patch}' 的 widget。跳过此参数。"
+                    )
+                    continue
+
+                # 确保 `widgets_values` 存在且为列表
+                if not target_node.widgets_values or not isinstance(
+                    target_node.widgets_values, list
+                ):
+                    # 如果不存在或格式错误，根据widget数量创建一个占位符列表
+                    target_node.widgets_values = [None] * len(widget_inputs)
+
+                # 确保 `widgets_values` 列表足够长
+                while len(target_node.widgets_values) <= target_widget_index:
+                    target_node.widgets_values.append(None)
+
+                # 根据API规范转换数据类型
+                target_type = str
+                if field_spec.type == "int":
+                    target_type = int
+                elif field_spec.type == "float":
+                    target_type = float
+
+                # 在正确的位置上更新值
+                try:
+                    if target_node.type == "LoadImage":
+                        value = await self._upload_image_to_comfy(server, value)
+                    target_node.widgets_values[target_widget_index] = target_type(value)
+                except (ValueError, TypeError) as e:
+                    logger.warning(
+                        f"无法将节点 '{node_name}' 的字段 '{field_name}' 的值 '{value}' 转换为类型 '{field_spec.type}'。错误: {e}"
+                    )
+                    continue
+
+        # 创建新的workflow_data副本
+        patched_workflow_data = self.workflow_data.model_copy(deep=True)
+        patched_workflow_data.nodes = list(nodes_map.values())
+
+        # 对workflow_data中的nodes进行排序，按照id从小到大
+        patched_workflow_data.nodes = sorted(
+            patched_workflow_data.nodes, key=lambda x: x.id
+        )
+
+        # 转换为字典格式以保持向后兼容
+        return patched_workflow_data.model_dump()
+
+    def _convert_workflow_to_prompt_api_format(self, workflow_data: dict) -> dict:
+        """
+        将工作流（API格式）转换为提交到/prompt端点的格式。
+        此函数现在能正确处理已通过 `patch_workflow` 修改的 `widgets_values`。
+        """
+        if "nodes" not in workflow_data:
+            raise ValueError("无效的工作流格式")
+
+        prompt_api_format = {}
+
+        # 建立从link_id到源节点的映射
+        link_map = {}
+        for link_data in workflow_data.get("links", []):
+            link_id, origin_node_id, origin_slot_index = (
+                link_data[0],
+                link_data[1],
+                link_data[2],
+            )
+
+            # 键是目标节点的输入link_id
+            link_map[link_id] = [str(origin_node_id), origin_slot_index]
+
+        for node in workflow_data["nodes"]:
+            node_id = str(node["id"])
+            inputs_dict = self._get_node_inputs(node, link_map)
+            prompt_api_format[node_id] = {
+                "inputs": inputs_dict,
+                "class_type": node["type"],
+                "_meta": {
+                    "title": node.get("title", ""),
+                },
+            }
+
+        # 过滤掉类型为 Note 的节点
+        prompt_api_format = {
+            k: v
+            for k, v in prompt_api_format.items()
+            if v["class_type"] not in ["Note"]
+        }
+
+        return prompt_api_format
+
+    async def _upload_image_to_comfy(
+        self, server: ComfyUIServerInfo, file_path: str
+    ) -> str:
+        """
+        上传文件到服务器。
+        file 是一个http链接
+        返回一个名称
+        """
+        file_name = file_path.split("/")[-1]
+        file_data = await self._download_file(file_path)
+
+        form_data = aiohttp.FormData()
+        form_data.add_field(
+            "image", file_data, filename=file_name, content_type="image/*"
+        )
+        form_data.add_field("type", "input")
+
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                f"{server.http_url}/api/upload/image", data=form_data
+            ) as response:
+                response.raise_for_status()
+                result = await response.json()
+                return result["name"]
+
+    async def _download_file(self, src: str) -> bytes:
+        """
+        下载文件到本地。
+        """
+        async with aiohttp.ClientSession() as session:
+            async with session.get(src) as response:
+                response.raise_for_status()
+                return await response.read()
+
+    def _generate_field_schema(
+        self, input_field: dict, default_value: Any = None
+    ) -> dict:
+        """
+        根据输入字段信息生成 JSON Schema 字段定义（字典版本，向后兼容）
+
+        Args:
+            input_field: 输入字段配置
+            default_value: 字段的默认值
+        """
+        field_schema = {
+            "title": input_field["widget"]["name"],
+            "description": input_field.get("description", ""),
+        }
+
+        # 根据字段类型设置 schema 类型
+        field_type = self._get_param_type(input_field.get("type", "STRING"))
+
+        if field_type == "int":
+            field_schema.update(
+                {
+                    "type": "integer",
+                    "minimum": input_field.get("min", None),
+                    "maximum": input_field.get("max", None),
+                }
+            )
+        elif field_type == "float":
+            field_schema.update(
+                {
+                    "type": "number",
+                    "minimum": input_field.get("min", None),
+                    "maximum": input_field.get("max", None),
+                }
+            )
+        elif field_type == "UploadFile":
+            field_schema.update(
+                {
+                    "type": "string",
+                    "format": "binary",
+                    "contentMediaType": "image/*",
+                    "pattern": ".*\\.(jpg|jpeg|png)$",
+                    "description": "[ext:jpg,jpeg,png][file-size:1MB] 上传图片文件",
+                }
+            )
+        else:  # string
+            field_schema.update({"type": "string"})
+
+            # 处理枚举值
+            if "options" in input_field:
+                field_schema["enum"] = input_field["options"]
+                field_schema[
+                    "description"
+                ] += f" 可选值: {', '.join(input_field['options'])}"
+
+        # 设置默认值
+        if default_value is not None:
+            field_schema["default"] = default_value
+        elif "default" in input_field:
+            field_schema["default"] = input_field["default"]
+
+        return field_schema
+
+    def _generate_field_schema_model(
+        self, input_field: ComfyWorkflowNodeInput, default_value: Any = None
+    ) -> ComfyJSONSchemaProperty:
+        """
+        根据输入字段信息生成 JSON Schema 字段定义（模型版本）
+
+        Args:
+            input_field: 输入字段配置
+            default_value: 字段的默认值
+        """
+        title = input_field.widget.name if input_field.widget else input_field.name
+        description = input_field.localized_name or ""
+
+        # 根据字段类型设置 schema 类型
+        field_type = self._get_param_type(input_field.type or "STRING")
+
+        if field_type == "int":
+            return ComfyJSONSchemaProperty(
+                type="integer",
+                title=title,
+                description=description,
+                default=default_value,
+                # TODO: 从输入配置中获取min/max值
+            )
+        elif field_type == "float":
+            return ComfyJSONSchemaProperty(
+                type="number",
+                title=title,
+                description=description,
+                default=default_value,
+                # TODO: 从输入配置中获取min/max值
+            )
+        elif field_type == "UploadFile":
+            return ComfyJSONSchemaProperty(
+                type="string",
+                title=title,
+                format="binary",
+                contentMediaType="image/*",
+                pattern=".*\\.(jpg|jpeg|png)$",
+                description="[ext:jpg,jpeg,png][file-size:1MB] 上传图片文件",
+                default=default_value,
+            )
+        else:  # string
+            return ComfyJSONSchemaProperty(
+                type="string",
+                title=title,
+                description=description,
+                default=default_value,
+                # TODO: 从输入配置中获取枚举值
+            )
+
+    def _get_param_type(self, input_type_str: str) -> str:
+        """
+        根据输入类型字符串确定参数类型
+        """
+        input_type_str = input_type_str.upper()
+        if "COMBO" in input_type_str:
+            return "UploadFile"
+        elif "INT" in input_type_str:
+            return "int"
+        elif "FLOAT" in input_type_str:
+            return "float"
+        else:
+            return "string"
+
+    def _get_node_inputs(
+        self, node: Union[ComfyWorkflowNode, dict], link_map: dict
+    ) -> dict:
+        """
+        获取节点的输入字段值
+
+        Args:
+            node: 节点数据（可以是ComfyWorkflowNode模型或字典）
+            link_map: 从link_id到源节点的映射
+
+        Returns:
+            dict: 包含节点输入字段值的字典
+        """
+        inputs_dict = {}
+
+        # 兼容处理：如果是字典，转换为模型
+        if isinstance(node, dict):
+            try:
+                node_model = ComfyWorkflowNode.model_validate(node)
+            except Exception:
+                # 如果转换失败，使用原来的字典方式
+                return self._get_node_inputs_dict(node, link_map)
+        else:
+            node_model = node
+
+        # 1. 处理控件输入 (widgets)
+        widgets_values = node_model.widgets_values or []
+        widget_cursor = 0
+
+        for input_config in node_model.inputs or []:
+            # 如果是widget并且有对应的widgets_values
+            if input_config.widget and widget_cursor < len(widgets_values):
+                widget_name = input_config.widget.name
+                if widget_name:
+                    # 使用widgets_values中的值，因为这里已经包含了API传入的修改
+                    inputs_dict[widget_name] = widgets_values[widget_cursor]
+                widget_cursor += 1
+
+                # 特殊处理：如果节点是 EG_CYQ_JB ，则需要忽略widgets_values的第1个值，这个是用来控制control net的，它不参与赋值
+                if node_model.type == "EG_CYQ_JB" and widget_cursor == 1:
+                    widget_cursor += 1
+
+        # 2. 处理连接输入 (links)
+        for input_config in node_model.inputs or []:
+            if input_config.link is not None:
+                link_id = input_config.link
+                if link_id in link_map:
+                    # 输入名称是input_config中的'name'
+                    inputs_dict[input_config.name] = link_map[link_id]
+
+        return inputs_dict
+
+    def _get_node_inputs_dict(self, node: dict, link_map: dict) -> dict:
+        """
+        获取节点的输入字段值（字典版本，用于向后兼容）
+
+        Args:
+            node: 节点数据字典
+            link_map: 从link_id到源节点的映射
+
+        Returns:
+            dict: 包含节点输入字段值的字典
+        """
+        inputs_dict = {}
+
+        # 1. 处理控件输入 (widgets)
+        widgets_values = node.get("widgets_values", [])
+        widget_cursor = 0
+        for input_config in node.get("inputs", []):
+            # 如果是widget并且有对应的widgets_values
+            if "widget" in input_config and widget_cursor < len(widgets_values):
+                widget_name = input_config["widget"].get("name")
+                if widget_name:
+                    # 使用widgets_values中的值，因为这里已经包含了API传入的修改
+                    inputs_dict[widget_name] = widgets_values[widget_cursor]
+                widget_cursor += 1
+
+                # 特殊处理：如果节点是 EG_CYQ_JB ，则需要忽略widgets_values的第1个值，这个是用来控制control net的，它不参与赋值
+                if node["type"] == "EG_CYQ_JB" and widget_cursor == 1:
+                    widget_cursor += 1
+
+        # 2. 处理连接输入 (links)
+        for input_config in node.get("inputs", []):
+            if "link" in input_config and input_config["link"] is not None:
+                link_id = input_config["link"]
+                if link_id in link_map:
+                    # 输入名称是input_config中的'name'
+                    inputs_dict[input_config["name"]] = link_map[link_id]
+
+        return inputs_dict

app/config.py

@@ -0,0 +1,31 @@
+import json
+from pydantic import BaseModel, ValidationError
+from pydantic_settings import BaseSettings, SettingsConfigDict
+from typing import List
+
+
+class ComfyUIServer(BaseModel):
+    """定义单个ComfyUI服务器的配置模型。"""
+
+    http_url: str
+    ws_url: str
+    input_dir: str
+    output_dir: str
+
+
+class Settings(BaseSettings):
+    """
+    应用配置类，通过.env文件加载环境变量。
+    """
+
+    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
+
+    DB_FILE: str = "/db/workflows_service.sqlite"
+    # S3 和 AWS 配置保持不变
+    S3_BUCKET_NAME: str
+    AWS_ACCESS_KEY_ID: str
+    AWS_SECRET_ACCESS_KEY: str
+    AWS_REGION_NAME: str
+
+
+settings = Settings()

app/database/__init__.py

@@ -0,0 +1,48 @@
+"""
+数据库模块
+提供工作流和工作流运行的数据库操作接口
+"""
+
+from .models import Workflow, WorkflowRun, WorkflowRunNode, ComfyUIServer
+from .api import (
+    init_db,
+    save_workflow,
+    get_all_workflows,
+    get_latest_workflow_by_base_name,
+    get_workflow_by_version,
+    get_workflow,
+    delete_workflow,
+    create_workflow_run,
+    update_workflow_run_status,
+    create_workflow_run_nodes,
+    update_workflow_run_node_status,
+    get_workflow_run,
+    get_workflow_run_nodes,
+    get_pending_workflow_runs,
+    get_running_workflow_runs,
+)
+
+__all__ = [
+    # 模型
+    "Workflow",
+    "WorkflowRun", 
+    "WorkflowRunNode",
+    "ComfyUIServer",
+    
+    # API函数
+    "init_db",
+    "save_workflow",
+    "get_all_workflows",
+    "get_latest_workflow_by_base_name",
+    "get_workflow_by_version",
+    "get_workflow",
+    "delete_workflow",
+    "create_workflow_run",
+    "update_workflow_run_status",
+    "create_workflow_run_nodes",
+    "update_workflow_run_node_status",
+    "get_workflow_run",
+    "get_workflow_run_nodes",
+    "get_pending_workflow_runs",
+    "get_running_workflow_runs",
+]

app/database/api.py

@@ -0,0 +1,282 @@
+"""
+数据库操作API
+提供工作流相关的所有数据库操作函数
+"""
+
+from datetime import datetime
+from typing import Optional, List
+import json
+import uuid
+
+from sqlalchemy.future import select
+from sqlalchemy import delete
+
+from .models import Base, Workflow, WorkflowRun, WorkflowRunNode, ComfyUIServer
+from .connection import async_engine, AsyncSessionLocal
+
+
+async def init_db():
+    """初始化数据库表结构"""
+    async with async_engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    print(f"数据库表结构已创建完成。")
+
+
+async def save_workflow(name: str, workflow_json: str):
+    """保存工作流"""
+    version = datetime.now().strftime("%Y%m%d%H%M%S")
+    workflow = Workflow(
+        name=f"{name} [{version}]",
+        base_name=name,
+        version=version,
+        workflow_json=workflow_json,
+    )
+
+    async with AsyncSessionLocal() as session:
+        await session.merge(workflow)  # 使用merge实现INSERT OR REPLACE
+        await session.commit()
+
+
+async def get_all_workflows() -> List[dict]:
+    """获取所有工作流（最新版本）"""
+    async with AsyncSessionLocal() as session:
+        # 使用SQLAlchemy ORM语法，通过子查询获取每个base_name的最新版本
+        from sqlalchemy import func
+
+        # 子查询：获取每个base_name的最新版本
+        latest_versions = (
+            select(Workflow.base_name, func.max(Workflow.version).label("max_version"))
+            .group_by(Workflow.base_name)
+            .subquery()
+        )
+
+        # 主查询：关联获取完整的工作流信息
+        stmt = (
+            select(Workflow)
+            .join(
+                latest_versions,
+                (Workflow.base_name == latest_versions.c.base_name)
+                & (Workflow.version == latest_versions.c.max_version),
+            )
+            .order_by(Workflow.base_name)
+        )
+
+        result = await session.execute(stmt)
+        workflows = result.scalars().all()
+
+        return [
+            {"name": wf.name, "workflow": json.loads(wf.workflow_json)}
+            for wf in workflows
+        ]
+
+
+async def get_latest_workflow_by_base_name(base_name: str) -> Optional[dict]:
+    """根据基础名称获取最新版本的工作流"""
+    async with AsyncSessionLocal() as session:
+        stmt = (
+            select(Workflow)
+            .where(Workflow.base_name == base_name)
+            .order_by(Workflow.version.desc())
+            .limit(1)
+        )
+
+        result = await session.execute(stmt)
+        workflow = result.scalar_one_or_none()
+
+        return workflow.to_dict() if workflow else None
+
+
+async def get_workflow_by_version(base_name: str, version: str) -> Optional[dict]:
+    """根据版本获取工作流"""
+    name = f"{base_name} [{version}]"
+    async with AsyncSessionLocal() as session:
+        stmt = select(Workflow).where(Workflow.name == name)
+        result = await session.execute(stmt)
+        workflow = result.scalar_one_or_none()
+
+        return workflow.to_dict() if workflow else None
+
+
+async def get_workflow(name: str, version: Optional[str] = None) -> Optional[dict]:
+    """获取工作流"""
+    if version:
+        return await get_workflow_by_version(name, version)
+    else:
+        return await get_latest_workflow_by_base_name(name)
+
+
+async def delete_workflow(name: str) -> bool:
+    """删除工作流"""
+    async with AsyncSessionLocal() as session:
+        stmt = delete(Workflow).where(Workflow.name == name)
+        result = await session.execute(stmt)
+        await session.commit()
+        return result.rowcount > 0
+
+
+async def create_workflow_run(
+    workflow_run_id: str,
+    workflow_name: str,
+    workflow_json: str,
+    api_spec: str,
+    request_data: str,
+) -> str:
+    """创建新的工作流运行记录"""
+    workflow_run = WorkflowRun(
+        id=workflow_run_id,
+        workflow_name=workflow_name,
+        workflow_json=workflow_json,
+        api_spec=api_spec,
+        request_data=request_data,
+    )
+
+    async with AsyncSessionLocal() as session:
+        session.add(workflow_run)
+        await session.commit()
+
+    return workflow_run_id
+
+
+async def update_workflow_run_status(
+    workflow_run_id: str,
+    status: str,
+    server_url: str = None,
+    prompt_id: str = None,
+    client_id: str = None,
+    error_message: str = None,
+    result: str = None,
+):
+    """更新工作流运行状态"""
+    async with AsyncSessionLocal() as session:
+        workflow_run = await session.get(WorkflowRun, workflow_run_id)
+        if not workflow_run:
+            return
+
+        workflow_run.status = status
+
+        if status == "running":
+            workflow_run.server_url = server_url
+            workflow_run.prompt_id = prompt_id
+            workflow_run.client_id = client_id
+            workflow_run.started_at = datetime.utcnow()
+        elif status == "completed":
+            workflow_run.completed_at = datetime.utcnow()
+            workflow_run.result = result
+        elif status == "failed":
+            workflow_run.error_message = error_message
+            workflow_run.completed_at = datetime.utcnow()
+
+        await session.commit()
+
+
+async def create_workflow_run_nodes(workflow_run_id: str, nodes_data: List[dict]):
+    """创建工作流运行节点记录"""
+    nodes = []
+    for node in nodes_data:
+        node_obj = WorkflowRunNode(
+            id=str(uuid.uuid4()),
+            workflow_run_id=workflow_run_id,
+            node_id=node["id"],
+            node_type=node["type"],
+        )
+        nodes.append(node_obj)
+
+    async with AsyncSessionLocal() as session:
+        session.add_all(nodes)
+        await session.commit()
+
+
+async def update_workflow_run_node_status(
+    workflow_run_id: str,
+    node_id: str,
+    status: str,
+    output_data: str = None,
+    error_message: str = None,
+):
+    """更新工作流运行节点状态"""
+    async with AsyncSessionLocal() as session:
+        stmt = select(WorkflowRunNode).where(
+            WorkflowRunNode.workflow_run_id == workflow_run_id,
+            WorkflowRunNode.node_id == node_id,
+        )
+        result = await session.execute(stmt)
+        node = result.scalar_one_or_none()
+
+        if not node:
+            return
+
+        node.status = status
+
+        if status == "running":
+            node.started_at = datetime.utcnow()
+        elif status == "completed":
+            node.output_data = output_data
+            node.completed_at = datetime.utcnow()
+        elif status == "failed":
+            node.error_message = error_message
+            node.completed_at = datetime.utcnow()
+
+        await session.commit()
+
+
+async def get_workflow_run(workflow_run_id: str) -> Optional[WorkflowRun]:
+    """获取工作流运行记录"""
+    async with AsyncSessionLocal() as session:
+        workflow_run = await session.get(WorkflowRun, workflow_run_id)
+        return workflow_run
+
+
+async def get_workflow_run_nodes(workflow_run_id: str) -> List[dict]:
+    """获取工作流运行节点记录"""
+    async with AsyncSessionLocal() as session:
+        stmt = select(WorkflowRunNode).where(
+            WorkflowRunNode.workflow_run_id == workflow_run_id
+        )
+        result = await session.execute(stmt)
+        nodes = result.scalars().all()
+
+        return [node.to_dict() for node in nodes]
+
+
+async def get_pending_workflow_runs() -> List[dict]:
+    """获取所有待处理的工作流运行记录"""
+    async with AsyncSessionLocal() as session:
+        stmt = (
+            select(WorkflowRun)
+            .where(WorkflowRun.status == "pending")
+            .order_by(WorkflowRun.created_at.asc())
+        )
+
+        result = await session.execute(stmt)
+        runs = result.scalars().all()
+
+        return [run.to_dict() for run in runs]
+
+
+async def get_running_workflow_runs() -> List[dict]:
+    """获取所有正在运行的工作流运行记录"""
+    async with AsyncSessionLocal() as session:
+        stmt = select(WorkflowRun).where(WorkflowRun.status == "running")
+        result = await session.execute(stmt)
+        runs = result.scalars().all()
+
+        return [run.to_dict() for run in runs]
+
+
+async def get_workflow_runs_recent(
+    start_time: datetime, end_time: datetime
+) -> List[dict]:
+    """获取指定时间范围内的最近工作流运行记录"""
+    async with AsyncSessionLocal() as session:
+        stmt = (
+            select(WorkflowRun)
+            .where(
+                WorkflowRun.created_at >= start_time, WorkflowRun.created_at <= end_time
+            )
+            .order_by(WorkflowRun.created_at.desc())
+        )
+
+        result = await session.execute(stmt)
+        runs = result.scalars().all()
+
+        return [run.to_dict() for run in runs]

app/database/connection.py

@@ -0,0 +1,42 @@
+"""
+数据库连接配置
+管理数据库引擎、会话和连接池
+"""
+
+from sqlalchemy import create_engine
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
+from sqlalchemy.orm import sessionmaker
+
+from app.config import Settings
+
+# 数据库配置
+DATABASE_FILE = Settings().DB_FILE
+ASYNC_DATABASE_URL = f"sqlite+aiosqlite:///{DATABASE_FILE}"
+SYNC_DATABASE_URL = f"sqlite:///{DATABASE_FILE}"
+
+# 创建异步引擎
+async_engine = create_async_engine(
+    ASYNC_DATABASE_URL, echo=False, pool_pre_ping=True, pool_recycle=3600
+)
+
+# 创建异步会话工厂
+AsyncSessionLocal = async_sessionmaker(
+    async_engine, class_=AsyncSession, expire_on_commit=False
+)
+
+# 创建同步引擎和会话（用于初始化）
+sync_engine = create_engine(
+    SYNC_DATABASE_URL, echo=False, pool_pre_ping=True, pool_recycle=3600
+)
+SessionLocal = sessionmaker(bind=sync_engine)
+
+
+async def get_db_session() -> AsyncSession:
+    """获取数据库会话"""
+    async with AsyncSessionLocal() as session:
+        return session
+
+
+def get_sync_session():
+    """获取同步数据库会话（用于初始化等场景）"""
+    return SessionLocal()

app/database/models.py

@@ -0,0 +1,144 @@
+"""
+数据库模型定义
+定义工作流相关的所有数据表结构
+"""
+
+from datetime import datetime
+from sqlalchemy import Column, String, Text, DateTime, ForeignKey, Integer, Boolean
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import relationship
+
+# 创建基类
+Base = declarative_base()
+
+
+class Workflow(Base):
+    """工作流模型"""
+
+    __tablename__ = "workflows"
+
+    name = Column(String, primary_key=True)
+    base_name = Column(String, nullable=False)
+    version = Column(String, nullable=False)
+    workflow_json = Column(Text, nullable=False)
+
+    def to_dict(self) -> dict:
+        return {
+            "name": self.name,
+            "base_name": self.base_name,
+            "version": self.version,
+            "workflow_json": self.workflow_json,
+        }
+
+
+class WorkflowRun(Base):
+    """工作流运行记录模型"""
+
+    __tablename__ = "workflow_run"
+
+    id = Column(String, primary_key=True)
+    workflow_name = Column(String, nullable=False)
+    prompt_id = Column(String)
+    client_id = Column(String)
+    status = Column(String, nullable=False, default="pending")
+    server_url = Column(String)
+    created_at = Column(DateTime, default=datetime.utcnow)
+    started_at = Column(DateTime)
+    completed_at = Column(DateTime)
+    error_message = Column(Text)
+    workflow_json = Column(Text, nullable=False)
+    api_spec = Column(Text, nullable=False)
+    request_data = Column(Text, nullable=False)
+    result = Column(Text)
+
+    # 关联关系
+    nodes = relationship(
+        "WorkflowRunNode", back_populates="workflow_run", cascade="all, delete-orphan"
+    )
+
+    def to_dict(self) -> dict:
+        return {
+            "id": self.id,
+            "workflow_name": self.workflow_name,
+            "prompt_id": self.prompt_id,
+            "client_id": self.client_id,
+            "status": self.status,
+            "server_url": self.server_url,
+            "created_at": self.created_at.isoformat() if self.created_at else None,
+            "started_at": self.started_at.isoformat() if self.started_at else None,
+            "completed_at": (
+                self.completed_at.isoformat() if self.completed_at else None
+            ),
+            "error_message": self.error_message,
+            "workflow_json": self.workflow_json,
+            "api_spec": self.api_spec,
+            "request_data": self.request_data,
+            "result": self.result,
+        }
+
+
+class WorkflowRunNode(Base):
+    """工作流运行节点模型"""
+
+    __tablename__ = "workflow_run_nodes"
+
+    id = Column(String, primary_key=True)
+    workflow_run_id = Column(String, ForeignKey("workflow_run.id"), nullable=False)
+    node_id = Column(String, nullable=False)
+    node_type = Column(String, nullable=False)
+    status = Column(String, nullable=False, default="pending")
+    started_at = Column(DateTime)
+    completed_at = Column(DateTime)
+    output_data = Column(Text)
+    error_message = Column(Text)
+
+    # 关联关系
+    workflow_run = relationship("WorkflowRun", back_populates="nodes")
+
+    def to_dict(self) -> dict:
+        return {
+            "id": self.id,
+            "workflow_run_id": self.workflow_run_id,
+            "node_id": self.node_id,
+            "node_type": self.node_type,
+            "status": self.status,
+            "started_at": self.started_at.isoformat() if self.started_at else None,
+            "completed_at": (
+                self.completed_at.isoformat() if self.completed_at else None
+            ),
+            "output_data": self.output_data,
+            "error_message": self.error_message,
+        }
+
+
+class ComfyUIServer(Base):
+    """ComfyUI 服务器模型"""
+
+    __tablename__ = "comfyui_servers"
+
+    name = Column(String, primary_key=True)
+    http_url = Column(String, nullable=False)
+    ws_url = Column(String, nullable=False)
+    status = Column(String, nullable=False, default="offline")
+    last_health_check = Column(DateTime)
+    current_tasks = Column(Integer, default=0)
+    max_concurrent_tasks = Column(Integer, default=1)
+    capabilities = Column(Text)  # JSON 字符串
+    server_metadata = Column(Text)  # JSON 字符串
+    created_at = Column(DateTime, default=datetime.utcnow)
+    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
+
+    def to_dict(self) -> dict:
+        return {
+            "name": self.name,
+            "http_url": self.http_url,
+            "ws_url": self.ws_url,
+            "status": self.status,
+            "last_health_check": self.last_health_check.isoformat() if self.last_health_check else None,
+            "current_tasks": self.current_tasks,
+            "max_concurrent_tasks": self.max_concurrent_tasks,
+            "capabilities": self.capabilities,
+            "server_metadata": self.server_metadata,
+            "created_at": self.created_at.isoformat() if self.created_at else None,
+            "updated_at": self.updated_at.isoformat() if self.updated_at else None,
+        }

app/main.py

@@ -0,0 +1,75 @@
+import os
+from pathlib import Path
+
+from fastapi.responses import FileResponse
+import uvicorn
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.staticfiles import StaticFiles
+
+from app.comfy.comfy_server import server_manager
+from app.database.api import init_db
+from app.config import Settings
+from app.routes import service, workflow, run, runx, comfy_server
+from app.comfy.comfy_queue import queue_manager
+
+settings = Settings()
+
+web_app = FastAPI(title="ComfyUI Workflow Service & Management API")
+
+# 配置跨域支持
+web_app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # 允许所有来源，生产环境中建议限制具体域名
+    allow_credentials=True,
+    allow_methods=["*"],  # 允许所有HTTP方法
+    allow_headers=["*"],  # 允许所有请求头
+)
+
+# 挂载静态文件目录
+static_dir = Path(__file__).parent / "static"
+if static_dir.exists():
+    web_app.mount("/static", StaticFiles(directory=str(static_dir)), name="static")
+
+web_app.include_router(service.service_router)
+web_app.include_router(workflow.workflow_router)
+web_app.include_router(run.run_router)
+web_app.include_router(runx.runx_router)
+web_app.include_router(comfy_server.router)
+
+
+@web_app.get("/")
+async def dashboard():
+    """监控仪表板页面"""
+    return FileResponse(str(Path(__file__).parent / "static" / "index.html"))
+
+
+@web_app.on_event("startup")
+async def startup_event():
+    """服务启动时，初始化数据库。"""
+    await init_db()
+    try:
+        servers = await server_manager.get_all_servers()
+        print(f"检测到 {len(servers)} 个 ComfyUI 服务器配置。")
+        for server in servers:
+            print(f"  - 服务器: {server.http_url}")
+
+        # 启动队列监控任务
+        await queue_manager.start_queue_monitor()
+        print("队列监控任务已启动")
+
+    except ValueError as e:
+        print(f"错误: 无法在启动时获取服务器信息: {e}")
+
+
+@web_app.on_event("shutdown")
+async def shutdown_event():
+    """服务关闭时，停止队列监控任务。"""
+    await queue_manager.stop_queue_monitor()
+    print("队列监控任务已停止")
+
+
+if __name__ == "__main__":
+    uvicorn.run(
+        "workflow_service.main:web_app", host="0.0.0.0", port=18000, reload=True
+    )

app/routes/comfy_server.py

@@ -0,0 +1,112 @@
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
+from typing import Dict, List, Optional, Any
+import logging
+import aiohttp
+import asyncio
+
+from app.comfy.comfy_server import server_manager
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/comfy", tags=["ComfyUI Server Management"])
+
+
+class ServerRegistrationRequest(BaseModel):
+    """服务器注册请求模型"""
+
+    name: str
+    http_url: str
+    ws_url: str
+
+
+class ServerUnregisterRequest(BaseModel):
+    """服务器注销请求模型"""
+
+    name: str
+
+
+class ServerStatusResponse(BaseModel):
+    """服务器状态响应模型"""
+
+    name: str
+    http_url: str
+    ws_url: str
+    status: str
+    last_health_check: Optional[str] = None
+    current_tasks: int
+    max_concurrent_tasks: int
+    capabilities: Dict[str, Any]
+    metadata: Dict[str, Any]
+
+
+@router.post("/register", response_model=Dict[str, str])
+async def register_server(request: ServerRegistrationRequest):
+    """注册ComfyUI服务器"""
+    try:
+        print(request)
+        success = await server_manager.register_server(
+            name=request.name,
+            http_url=request.http_url,
+            ws_url=request.ws_url,
+            max_concurrent_tasks=1,
+            capabilities={},
+            metadata={},
+        )
+
+        if success:
+            logger.info(f"服务器 {request.name} 注册成功")
+            return {"message": f"服务器 {request.name} 注册成功", "status": "success"}
+        else:
+            raise HTTPException(status_code=400, detail="服务器注册失败")
+
+    except Exception as e:
+        logger.error(f"注册服务器 {request.name} 时发生错误: {e}")
+        raise HTTPException(status_code=500, detail=f"注册失败: {str(e)}")
+
+
+@router.post("/unregister", response_model=Dict[str, str])
+async def unregister_server(request: ServerUnregisterRequest):
+    """注销ComfyUI服务器"""
+    try:
+        success = await server_manager.unregister_server(request.name)
+
+        if success:
+            logger.info(f"服务器 {request.name} 注销成功")
+            return {"message": f"服务器 {request.name} 注销成功", "status": "success"}
+        else:
+            raise HTTPException(status_code=404, detail=f"服务器 {request.name} 不存在")
+
+    except Exception as e:
+        logger.error(f"注销服务器 {request.name} 时发生错误: {e}")
+        raise HTTPException(status_code=500, detail=f"注销失败: {str(e)}")
+
+
+@router.get("/list", response_model=List[ServerStatusResponse])
+async def list_all_servers():
+    """获取所有服务器列表"""
+    try:
+        servers = await server_manager.get_all_servers()
+
+        return [
+            ServerStatusResponse(
+                name=server.name,
+                http_url=server.http_url,
+                ws_url=server.ws_url,
+                status=server.status.value,
+                last_health_check=(
+                    server.last_health_check.isoformat()
+                    if server.last_health_check
+                    else None
+                ),
+                current_tasks=server.current_tasks,
+                max_concurrent_tasks=server.max_concurrent_tasks,
+                capabilities=server.capabilities,
+                metadata=server.metadata,
+            )
+            for server in servers
+        ]
+
+    except Exception as e:
+        logger.error(f"获取服务器列表时发生错误: {e}")
+        raise HTTPException(status_code=500, detail=f"获取列表失败: {str(e)}")

app/routes/run.py

@@ -0,0 +1,167 @@
+import json
+from typing import Dict, Optional, List, Any
+from datetime import datetime, timedelta
+
+from fastapi import APIRouter, Body, HTTPException
+from fastapi.responses import JSONResponse
+
+from app.comfy.comfy_queue import queue_manager
+from app.comfy.comfy_workflow import ComfyWorkflow
+from app.database.api import (
+    get_workflow,
+    get_workflow_run,
+    get_workflow_run_nodes,
+    get_workflow_runs_recent,
+)
+
+run_router = APIRouter(
+    prefix="/api/run",
+    tags=["Run"],
+)
+
+
+@run_router.post("")
+async def run_workflow(
+    workflow_name: str,
+    workflow_version: Optional[str] = None,
+    data: Dict = Body(...),
+):
+    """
+    异步执行工作流。
+    立即返回任务ID，调用者可以通过任务ID查询执行状态。
+    """
+    try:
+        if not workflow_name:
+            raise HTTPException(status_code=400, detail="`workflow_name` 字段是必需的")
+
+        # 获取工作流定义
+        workflow_data = await get_workflow(workflow_name, workflow_version)
+        if not workflow_data:
+            detail = (
+                f"工作流 '{workflow_name}'"
+                + (f" 带版本 '{workflow_version}'" if workflow_version else " (最新版)")
+                + " 未找到。"
+            )
+            raise HTTPException(status_code=404, detail=detail)
+
+        workflow = json.loads(workflow_data["workflow_json"])
+        flow = ComfyWorkflow(workflow_name, workflow)
+
+        # 提交到队列
+        workflow_run_id = await queue_manager.add_task(workflow=flow, request_data=data)
+
+        return JSONResponse(
+            content={
+                "workflow_run_id": workflow_run_id,
+                "status": "queued",
+                "message": "工作流已提交到队列，正在等待执行",
+            },
+            status_code=202,
+        )
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"提交工作流失败: {str(e)}")
+
+
+@run_router.get("")
+async def get_runs(
+    limit: int = 10, status: Optional[str] = None
+) -> List[Dict[str, Any]]:
+    """获取运行列表，支持分页和状态过滤"""
+    try:
+        end_time = datetime.now()
+        start_time = end_time - timedelta(hours=24)
+
+        recent_runs = await get_workflow_runs_recent(start_time, end_time)
+
+        # 如果指定了状态过滤
+        if status:
+            recent_runs = [run for run in recent_runs if run.get("status") == status]
+
+        # 限制返回数量
+        limited_runs = recent_runs[:limit]
+
+        # 格式化返回数据
+        formatted_runs = []
+        for run in limited_runs:
+            formatted_runs.append(
+                {
+                    "id": run.get("id"),  # 使用数据库中的id字段
+                    "workflow_name": run.get("workflow_name"),
+                    "status": run.get("status", "unknown"),
+                    "created_at": run.get("created_at"),
+                    "updated_at": run.get("updated_at"),
+                    "api_spec": run.get("api_spec"),
+                    "result": run.get("result"),  # 添加任务结果
+                    "error_message": run.get("error_message"),  # 添加错误信息
+                    "completed_at": run.get("completed_at"),  # 添加完成时间
+                }
+            )
+
+        return formatted_runs
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"获取运行列表失败: {str(e)}")
+
+
+@run_router.get("/metrics")
+async def get_run_metrics() -> Dict[str, Any]:
+    """获取运行概览统计信息"""
+    try:
+        # 获取最近24小时的任务统计
+        end_time = datetime.now()
+        start_time = end_time - timedelta(hours=24)
+
+        recent_runs = await get_workflow_runs_recent(start_time, end_time)
+
+        # 统计各种状态的任务数量
+        status_counts = {}
+        for run in recent_runs:
+            status = run.get("status", "unknown")
+            status_counts[status] = status_counts.get(status, 0) + 1
+
+        return {
+            "running_tasks": status_counts.get("running", 0),
+            "pending_tasks": status_counts.get("pending", 0),
+            "completed_tasks": status_counts.get("completed", 0),
+            "failed_tasks": status_counts.get("failed", 0),
+            "total_tasks_24h": len(recent_runs),
+            "timestamp": datetime.now().isoformat(),
+        }
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"获取运行概览失败: {str(e)}")
+
+
+@run_router.get("/{workflow_run_id}")
+async def get_run_status(workflow_run_id: str):
+    """
+    获取工作流执行状态。
+    """
+    try:
+        """获取任务状态"""
+        workflow_run = await get_workflow_run(workflow_run_id)
+        if not workflow_run:
+            return {"error": "任务不存在"}
+        nodes = await get_workflow_run_nodes(workflow_run_id)
+
+        result = {
+            "id": workflow_run_id,
+            "status": workflow_run.status,
+            "created_at": workflow_run.created_at,
+            "started_at": workflow_run.started_at,
+            "completed_at": workflow_run.completed_at,
+            "server_url": workflow_run.server_url,
+            "error_message": workflow_run.error_message,
+            "nodes": nodes,
+        }
+
+        # 如果任务完成，从数据库获取结果
+        if workflow_run.status == "completed" and workflow_run.result:
+            try:
+                result["result"] = json.loads(workflow_run.result)
+            except (json.JSONDecodeError, TypeError):
+                result["result"] = None
+
+        return result
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"获取状态失败: {str(e)}")

app/routes/runx/__init__.py

@@ -0,0 +1,4 @@
+from ._base import runx_router
+from .model_with_multi_dress import runx_router as model_with_multi_dress_router
+
+__all__ = ["runx_router", "model_with_multi_dress_router"]

app/routes/runx/_base.py

@@ -0,0 +1,6 @@
+from fastapi import APIRouter
+
+runx_router = APIRouter(
+    prefix="/api/runx",
+    tags=["RunX"],
+)

app/routes/runx/model_with_multi_dress.py

@@ -0,0 +1,127 @@
+import json
+from typing import Optional
+
+from fastapi import Body, HTTPException
+from fastapi.responses import JSONResponse
+from app.comfy.comfy_queue import queue_manager
+from app.comfy.comfy_workflow import ComfyWorkflow
+from app.database.api import get_workflow
+from ._base import runx_router
+
+RUNX_NAME = "model_with_multi_dress"
+RUNX_WORKFLOW_NAME = "测试"
+
+
+@runx_router.post(f"/{RUNX_NAME}")
+async def model_with_multi_dress(
+    data: dict = Body(...),
+    workflow_version: Optional[str] = None,
+):
+    """
+    一个模特，穿多件不同的衣服
+    """
+    workflow_name = RUNX_WORKFLOW_NAME
+
+    # 获取工作流定义
+    workflow_data = await get_workflow(workflow_name, workflow_version)
+    if not workflow_data:
+        detail = (
+            f"工作流 '{workflow_name}'"
+            + (f" 带版本 '{workflow_version}'" if workflow_version else " (最新版)")
+            + " 未找到。"
+        )
+        raise HTTPException(status_code=404, detail=detail)
+
+    workflow = json.loads(workflow_data["workflow_json"])
+    flow = ComfyWorkflow(workflow_name, workflow)
+
+    # 将请求拆分为多个请求
+    batch_data = _convert(data)
+
+    # 提交到队列
+    workflow_run_ids: list[str] = []
+    for item in batch_data:
+        workflow_run_id = await queue_manager.add_task(workflow=flow, request_data=item)
+        workflow_run_ids.append(workflow_run_id)
+
+    return JSONResponse(
+        content={
+            "workflow_run_ids": workflow_run_ids,
+            "status": "queued",
+            "message": "工作流已提交到队列，正在等待执行",
+        },
+        status_code=202,
+    )
+
+
+@runx_router.get(f"/{RUNX_NAME}/json_schema")
+async def model_with_multi_dress_json_schema():
+    """
+    获取工作流定义
+    """
+    return {
+        "type": "object",
+        "properties": {
+            "模特图": {
+                "type": "object",
+                "properties": {
+                    "image": {"type": "string"},
+                },
+            },
+            "模特描述": {
+                "type": "object",
+                "properties": {
+                    "value": {"type": "string"},
+                },
+            },
+            "穿搭图": {
+                "type": "array",
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "image": {"type": "string"},
+                    },
+                },
+            },
+        },
+        "required": ["模特图", "模特描述", "穿搭图"],
+    }
+
+
+def _convert(data: dict) -> list[dict]:
+    """
+    数据格式
+    {
+        "模特图": {"image" : "xxx" },
+        "模特描述" : { "value" : "xxx" },
+        "穿搭图": [
+            {"image" : "xxx" },
+            {"image" : "xxx" }
+        ]
+    }
+    转换为
+    [
+        {
+            "模特图": {"image" : "xxx" },
+            "模特描述" : { "value" : "xxx" },
+            "穿搭图": { "image" : "xxx" }
+        }
+    ]
+    """
+    result: list[dict] = []
+
+    # 获取基础信息
+    model_image = data.get("模特图", {})
+    model_description = data.get("模特描述", {})
+    dress_images = data.get("穿搭图", [])
+
+    # 为每个穿搭图创建一个记录
+    for dress_image in dress_images:
+        record = {
+            "模特图": model_image,
+            "模特描述": model_description,
+            "穿搭图": dress_image,
+        }
+        result.append(record)
+
+    return result

app/routes/service.py

@@ -0,0 +1,96 @@
+import asyncio
+from datetime import datetime
+from typing import List
+
+import aiohttp
+from fastapi import APIRouter, HTTPException, Path
+from pydantic import BaseModel
+
+from app.comfy.comfy_queue import queue_manager
+from app.comfy.comfy_server import server_manager
+from app.config import Settings
+
+settings = Settings()
+
+
+service_router = APIRouter(
+    prefix="/api/service",
+    tags=["Status"],
+)
+
+
+class ServerQueueDetails(BaseModel):
+    running_count: int
+    pending_count: int
+
+
+class ServerStatus(BaseModel):
+    server_index: int
+    http_url: str
+    ws_url: str
+    is_reachable: bool
+    is_free: bool
+    queue_details: ServerQueueDetails
+
+
+class FileDetails(BaseModel):
+    name: str
+    size_kb: float
+    modified_at: datetime
+
+
+class ServerFiles(BaseModel):
+    server_index: int
+    http_url: str
+    input_files: List[FileDetails]
+    output_files: List[FileDetails]
+
+
+@service_router.get("/metrics")
+async def get_metrics():
+    """
+    获取队列状态概览。
+    """
+    try:
+        pending_count = len(queue_manager.pending_tasks)
+        running_count = len(queue_manager.running_tasks)
+
+        return {
+            "pending_tasks": pending_count,
+            "running_tasks": running_count,
+            "total_servers": len(queue_manager.running_tasks),
+            "queue_manager_status": "active",
+        }
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"获取队列状态失败: {str(e)}")
+
+
+@service_router.get("/servers_status", response_model=List[ServerStatus])
+async def get_servers_status():
+    """
+    获取所有已配置的ComfyUI服务器的配置信息和实时状态。
+    """
+    servers = await server_manager.get_all_servers()
+    if not servers:
+        return []
+
+    async with aiohttp.ClientSession() as session:
+        status_tasks = [
+            queue_manager.get_server_status(server, session) for server in servers
+        ]
+        live_statuses = await asyncio.gather(*status_tasks)
+
+    response_list = []
+    for i, server_config in enumerate(servers):
+        status_data = live_statuses[i]
+        response_list.append(
+            ServerStatus(
+                server_index=i,
+                http_url=server_config.http_url,
+                ws_url=server_config.ws_url,
+                is_reachable=status_data["is_reachable"],
+                is_free=status_data["is_free"],
+                queue_details=status_data["queue_details"],
+            )
+        )
+    return response_list

app/routes/workflow.py

@@ -0,0 +1,105 @@
+import json
+from typing import Optional, List
+
+from fastapi import APIRouter, Body, HTTPException, Path
+from fastapi.responses import JSONResponse
+
+from app.comfy.comfy_workflow import ComfyWorkflow
+from app.database.api import (
+    save_workflow,
+    get_all_workflows,
+    delete_workflow,
+    get_workflow,
+)
+
+
+workflow_router = APIRouter(
+    prefix="/api/workflow",
+    tags=["Workflow"],
+)
+
+
+@workflow_router.post("", status_code=201)
+async def publish_workflow_endpoint(data: dict = Body(...)):
+    """
+    发布工作流
+    """
+    try:
+        name, wf_json = data.get("name"), data.get("workflow")
+        if not name or not wf_json:
+            raise HTTPException(
+                status_code=400, detail="`name` and `workflow` fields are required."
+            )
+        await save_workflow(name, json.dumps(wf_json))
+        print(f"Workflow '{name}' published.")
+        return JSONResponse(
+            content={"status": "success", "message": f"Workflow '{name}' published."},
+            status_code=201,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to save workflow: {e}")
+
+
+@workflow_router.get("", response_model=List[dict])
+async def get_all_workflows_endpoint():
+    """
+    获取所有工作流
+    """
+    try:
+        return await get_all_workflows()
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to get workflows: {e}")
+
+
+@workflow_router.delete("/{workflow_name:path}")
+async def delete_workflow_endpoint(
+    workflow_name: str = Path(
+        ...,
+        description="The full, unique name of the workflow to delete, e.g., 'my_workflow [20250101120000]'",
+    )
+):
+    """
+    删除工作流
+    """
+    try:
+        success = await delete_workflow(workflow_name)
+        if success:
+            return {"status": "deleted", "name": workflow_name}
+        else:
+            raise HTTPException(
+                status_code=404, detail=f"Workflow '{workflow_name}' not found"
+            )
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to delete workflow: {e}")
+
+
+@workflow_router.get("/{base_name:path}")
+async def get_one_workflow_endpoint(base_name: str, version: Optional[str] = None):
+    """
+    获取工作流规范
+    """
+    workflow_data = await get_workflow(base_name, version)
+
+    if not workflow_data:
+        detail = (
+            f"Workflow '{base_name}'"
+            + (f" with version '{version}'" if version else " (latest)")
+            + " not found."
+        )
+        raise HTTPException(status_code=404, detail=detail)
+
+    try:
+        workflow = json.loads(workflow_data["workflow_json"])
+        flow = ComfyWorkflow(base_name, workflow)
+        return {
+            "name": flow.workflow_name,
+            "workflow": flow.workflow_data,
+            "api_spec": flow.get_api_spec(),
+            "inputs_json_schema": flow.get_inputs_json_schema(),
+        }
+    except Exception as e:
+        raise HTTPException(
+            status_code=500, detail=f"Failed to parse workflow specification: {e}"
+        )

app/static/css/monitor.css

@@ -0,0 +1,339 @@
+* {
+    margin: 0;
+    padding: 0;
+    box-sizing: border-box;
+}
+
+body {
+    font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, 'Helvetica Neue', Arial, sans-serif;
+    background: linear-gradient(135deg, #f5f7fa 0%, #c3cfe2 100%);
+    min-height: 100vh;
+    padding: 24px;
+    color: #2c3e50;
+    line-height: 1.6;
+}
+
+.container {
+    max-width: 1200px;
+    margin: 0 auto;
+}
+
+.refresh-btn {
+    background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
+    color: white;
+    border: none;
+    padding: 12px 24px;
+    border-radius: 8px;
+    cursor: pointer;
+    font-size: 14px;
+    font-weight: 500;
+    margin-bottom: 24px;
+    box-shadow: 0 4px 15px rgba(102, 126, 234, 0.3);
+    transition: all 0.3s ease;
+}
+
+.refresh-btn:hover {
+    transform: translateY(-2px);
+    box-shadow: 0 6px 20px rgba(102, 126, 234, 0.4);
+}
+
+.stats-grid {
+    display: grid;
+    grid-template-columns: repeat(auto-fit, minmax(200px, 1fr));
+    gap: 20px;
+    margin-bottom: 24px;
+}
+
+.stat-card {
+    background: rgba(255, 255, 255, 0.95);
+    padding: 24px;
+    border-radius: 16px;
+    box-shadow: 0 8px 32px rgba(0, 0, 0, 0.1);
+    text-align: center;
+    backdrop-filter: blur(10px);
+    border: 1px solid rgba(255, 255, 255, 0.2);
+    transition: all 0.3s ease;
+}
+
+.stat-card:hover {
+    transform: translateY(-4px);
+    box-shadow: 0 12px 40px rgba(0, 0, 0, 0.15);
+}
+
+.stat-number {
+    font-size: 2.2em;
+    font-weight: 700;
+    background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
+    -webkit-background-clip: text;
+    -webkit-text-fill-color: transparent;
+    background-clip: text;
+    margin: 8px 0;
+}
+
+.stat-label {
+    color: #7f8c8d;
+    font-size: 0.9em;
+    font-weight: 500;
+    text-transform: uppercase;
+    letter-spacing: 0.5px;
+}
+
+.section {
+    background: rgba(255, 255, 255, 0.95);
+    padding: 24px;
+    border-radius: 16px;
+    box-shadow: 0 8px 32px rgba(0, 0, 0, 0.1);
+    margin-bottom: 24px;
+    backdrop-filter: blur(10px);
+    border: 1px solid rgba(255, 255, 255, 0.2);
+}
+
+.section h3 {
+    margin-bottom: 20px;
+    color: #2c3e50;
+    font-size: 1.3em;
+    font-weight: 600;
+    display: flex;
+    align-items: center;
+    gap: 8px;
+}
+
+.server-status {
+    display: grid;
+    grid-template-columns: repeat(auto-fit, minmax(280px, 1fr));
+    gap: 16px;
+}
+
+.server-card {
+    border: 1px solid #e1e8ed;
+    border-radius: 8px;
+    padding: 12px;
+    background: #f8f9fa;
+    transition: all 0.3s ease;
+    position: relative;
+    overflow: hidden;
+}
+
+.server-card::before {
+    content: '';
+    position: absolute;
+    top: 0;
+    left: 0;
+    right: 0;
+    height: 4px;
+    background: #e1e8ed;
+}
+
+.server-card.online {
+    border-color: #27ae60;
+    background: #f8fff9;
+}
+
+.server-card.online::before {
+    background: #27ae60;
+}
+
+.server-card.offline {
+    border-color: #e74c3c;
+    background: #fff8f8;
+}
+
+.server-card.offline::before {
+    background: #e74c3c;
+}
+
+.server-header {
+    display: flex;
+    align-items: center;
+    gap: 12px;
+    margin-bottom: 8px;
+}
+
+.server-card h4 {
+    color: #2c3e50;
+    margin: 0;
+    font-size: 1em;
+    font-weight: 600;
+    white-space: nowrap;
+}
+
+.server-url {
+    color: #3498db;
+    font-size: 0.8em;
+    font-weight: 500;
+    font-family: monospace;
+    white-space: nowrap;
+    overflow: hidden;
+    text-overflow: ellipsis;
+    max-width: 200px;
+}
+
+.server-info {
+    display: flex;
+    flex-wrap: wrap;
+    gap: 8px;
+    align-items: center;
+    font-size: 0.8em;
+    margin-top: 8px;
+}
+
+.info-item {
+    display: flex;
+    align-items: center;
+    gap: 4px;
+    white-space: nowrap;
+}
+
+.info-separator {
+    color: #e1e8ed;
+    margin: 0 4px;
+}
+
+.status-icon {
+    font-size: 0.9em;
+}
+
+.status-text {
+    font-weight: 600;
+    color: #2c3e50;
+}
+
+.info-label {
+    color: #7f8c8d;
+    font-weight: 500;
+    font-size: 0.75em;
+}
+
+.info-value {
+    color: #2c3e50;
+    font-weight: 600;
+    font-size: 0.75em;
+}
+
+.info-value:has(span[style*="color: #e74c3c"]) {
+    color: #e74c3c;
+}
+
+.info-value:has(span[style*="color: #f39c12"]) {
+    color: #f39c12;
+}
+
+.info-value:has(span[style*="color: #27ae60"]) {
+    color: #27ae60;
+}
+
+.status-indicator {
+    display: inline-block;
+    width: 8px;
+    height: 8px;
+    border-radius: 50%;
+    margin-right: 8px;
+}
+
+.status-online { background-color: #27ae60; }
+.status-offline { background-color: #e74c3c; }
+
+.task-list {
+    max-height: 500px;
+    overflow-y: auto;
+}
+
+.task-item {
+    border: 1px solid #e1e8ed;
+    border-radius: 8px;
+    padding: 16px;
+    margin: 8px 0;
+    background: #f8f9fa;
+    transition: all 0.3s ease;
+}
+
+.task-item:hover {
+    background: #ffffff;
+    box-shadow: 0 4px 12px rgba(0, 0, 0, 0.1);
+}
+
+.task-header {
+    display: flex;
+    justify-content: space-between;
+    align-items: center;
+    margin-bottom: 8px;
+}
+
+.task-name {
+    font-weight: 600;
+    color: #2c3e50;
+}
+
+.task-status {
+    padding: 4px 12px;
+    border-radius: 20px;
+    font-size: 0.8em;
+    font-weight: 600;
+    color: white;
+    text-transform: uppercase;
+    letter-spacing: 0.5px;
+}
+
+.status-pending { background: #f39c12; }
+.status-running { background: #3498db; }
+.status-completed { background: #27ae60; }
+.status-failed { background: #e74c3c; }
+
+.task-meta {
+    color: #7f8c8d;
+    font-size: 0.85em;
+    margin-bottom: 8px;
+}
+
+.task-meta > div {
+    margin-bottom: 4px;
+}
+
+.task-result {
+    color: #27ae60;
+    font-size: 0.85em;
+    margin-top: 8px;
+    padding: 8px;
+    background: #f0fff4;
+    border-radius: 6px;
+    border-left: 3px solid #27ae60;
+    white-space: pre-wrap;
+    word-wrap: break-word;
+    max-height: 300px;
+    overflow-y: auto;
+    font-family: 'Monaco', 'Menlo', 'Ubuntu Mono', monospace;
+    line-height: 1.4;
+}
+
+.task-error {
+    color: #e74c3c;
+    font-size: 0.85em;
+    margin-top: 8px;
+    padding: 8px;
+    background: #fff5f5;
+    border-radius: 6px;
+    border-left: 3px solid #e74c3c;
+    white-space: pre-wrap;
+    word-wrap: break-word;
+    max-height: 200px;
+    overflow-y: auto;
+    font-family: 'Monaco', 'Menlo', 'Ubuntu Mono', monospace;
+    line-height: 1.4;
+}
+
+.error-message {
+    color: #e74c3c;
+    font-size: 0.85em;
+    margin-top: 8px;
+    padding: 8px;
+    background: #fff5f5;
+    border-radius: 6px;
+    border-left: 3px solid #e74c3c;
+}
+
+@media (max-width: 768px) {
+    body { padding: 16px; }
+    .stats-grid { grid-template-columns: repeat(2, 1fr); }
+    .server-status { grid-template-columns: 1fr; }
+    .server-info { grid-template-columns: 1fr; }
+}

app/static/index.html

@@ -0,0 +1,49 @@
+<!DOCTYPE html>
+<html lang="zh-CN">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>ComfyUI 工作流监控</title>
+    <link rel="stylesheet" href="/static/css/monitor.css">
+</head>
+<body>
+    <div class="container">
+        <button class="refresh-btn" onclick="refreshData()">🔄 刷新数据</button>
+        
+        <div class="stats-grid">
+            <div class="stat-card">
+                <div class="stat-number" id="running-tasks">--</div>
+                <div class="stat-label">运行中任务</div>
+            </div>
+            <div class="stat-card">
+                <div class="stat-number" id="pending-tasks">--</div>
+                <div class="stat-label">等待中任务</div>
+            </div>
+            <div class="stat-card">
+                <div class="stat-number" id="completed-tasks">--</div>
+                <div class="stat-label">已完成任务</div>
+            </div>
+            <div class="stat-card">
+                <div class="stat-number" id="total-tasks">--</div>
+                <div class="stat-label">24小时总任务</div>
+            </div>
+        </div>
+        
+        <div class="section">
+            <h3>🖥️ 服务器状态</h3>
+            <div class="server-status" id="server-status">
+                <!-- 服务器状态将在这里动态加载 -->
+            </div>
+        </div>
+        
+        <div class="section">
+            <h3>📋 最近任务</h3>
+            <div class="task-list" id="recent-tasks">
+                <!-- 最近任务将在这里动态加载 -->
+            </div>
+        </div>
+    </div>
+    
+    <script src="/static/js/monitor.js"></script>
+</body>
+</html>

app/static/js/monitor.js

@@ -0,0 +1,209 @@
+async function refreshData() {
+  try {
+    // 获取运行概览 - 更新为新的API路径
+    const runMetrics = await fetch("/api/run/metrics").then((r) => r.json());
+    document.getElementById("running-tasks").textContent =
+      runMetrics.running_tasks;
+    document.getElementById("pending-tasks").textContent =
+      runMetrics.pending_tasks;
+    document.getElementById("completed-tasks").textContent =
+      runMetrics.completed_tasks;
+    document.getElementById("total-tasks").textContent =
+      runMetrics.total_tasks_24h;
+
+    // 获取服务器状态 - 使用现有的list接口
+    const serverStatus = await fetch("/api/comfy/list").then((r) => r.json());
+    updateServerStatus(serverStatus);
+
+    // 获取运行列表 - 更新为新的API路径
+    const runs = await fetch("/api/run").then((r) => r.json());
+    updateRecentTasks(runs);
+  } catch (error) {
+    console.error("刷新数据失败:", error);
+  }
+}
+
+function updateServerStatus(servers) {
+  const container = document.getElementById("server-status");
+  container.innerHTML = "";
+
+  if (servers.length === 0) {
+    container.innerHTML =
+      '<p style="text-align: center; color: #7f8c8d; grid-column: 1/-1;">暂无服务器信息</p>';
+    return;
+  }
+
+  servers.forEach((server) => {
+    const statusClass = server.status === "online" ? "online" : "offline";
+    const statusColor =
+      server.status === "online" ? "status-online" : "status-offline";
+
+    // 格式化时间
+    const formatTime = (timeStr) => {
+      if (!timeStr) return "N/A";
+      try {
+        const date = new Date(timeStr);
+        const now = new Date();
+        const diff = now - date;
+
+        if (diff < 60000) return "刚刚";
+        if (diff < 3600000) return Math.floor(diff / 60000) + "分钟前";
+        if (diff < 86400000) return Math.floor(diff / 3600000) + "小时前";
+        return date.toLocaleDateString();
+      } catch {
+        return timeStr;
+      }
+    };
+
+    // 格式化健康检查时间
+    const formatHealthCheck = (timeStr) => {
+      if (!timeStr) return "N/A";
+      try {
+        const date = new Date(timeStr);
+        const now = new Date();
+        const diff = now - date;
+
+        if (diff > 86400000) return "⚠️ 检查超时"; // 超过1天
+        if (diff > 3600000) return "⚠️ 检查延迟"; // 超过1小时
+        if (diff < 60000) return "✅ 刚刚";
+        if (diff < 3600000) return `✅ ${Math.floor(diff / 60000)}分钟前`;
+        return `✅ ${Math.floor(diff / 3600000)}小时前`;
+      } catch {
+        return timeStr;
+      }
+    };
+
+    container.innerHTML += `
+            <div class="server-card ${statusClass}">
+                <div class="server-header">
+                    <h4>${server.name || "未命名"}</h4>
+                    <span class="server-url">📍 ${server.http_url}</span>
+                </div>
+                <div class="server-info">
+                    <div class="info-item">
+                        <span class="status-icon">${
+                          server.status === "online" ? "🟢" : "🔴"
+                        }</span>
+                        <span class="status-text">${
+                          server.status === "online" ? "在线" : "离线"
+                        }</span>
+                    </div>
+                    <span class="info-separator">|</span>
+                    <div class="info-item">
+                        <span class="status-text">任务: ${
+                          server.current_tasks || 0
+                        }/${server.max_concurrent_tasks || 1}</span>
+                    </div>
+                    <span class="info-separator">|</span>
+                    <div class="info-item">
+                        <span class="status-text">检查: ${formatHealthCheck(
+                          server.last_health_check
+                        )}</span>
+                    </div>
+                </div>
+                ${
+                  server.error
+                    ? `<div class="error-message">${server.error}</div>`
+                    : ""
+                }
+            </div>
+        `;
+  });
+}
+
+function updateRecentTasks(tasks) {
+  const container = document.getElementById("recent-tasks");
+  container.innerHTML = "";
+
+  if (tasks.length === 0) {
+    container.innerHTML =
+      '<p style="text-align: center; color: #7f8c8d;">暂无任务记录</p>';
+    return;
+  }
+
+  tasks.forEach((task) => {
+    const statusClass = "status-" + task.status.toLowerCase();
+    const statusText =
+      {
+        pending: "等待中",
+        running: "运行中",
+        completed: "已完成",
+        failed: "失败",
+      }[task.status] || task.status;
+
+    // 格式化时间
+    const formatTime = (timeStr) => {
+      if (!timeStr) return "N/A";
+      try {
+        return new Date(timeStr).toLocaleString();
+      } catch {
+        return timeStr;
+      }
+    };
+
+    // 格式化结果数据
+    const formatResult = (result) => {
+      if (!result) return "暂无结果";
+      try {
+        const parsed = JSON.parse(result);
+        if (typeof parsed === "object") {
+          // 如果是对象，尝试提取关键信息
+          if (parsed.images && Array.isArray(parsed.images)) {
+            return `生成图片: ${
+              parsed.images.length
+            } 张\n完整结果: ${JSON.stringify(parsed, null, 2)}`;
+          }
+          if (parsed.output && typeof parsed.output === "object") {
+            return `输出数据: ${
+              Object.keys(parsed.output).length
+            } 项\n完整结果: ${JSON.stringify(parsed, null, 2)}`;
+          }
+          return `完整结果: ${JSON.stringify(parsed, null, 2)}`;
+        }
+        return `完整结果: ${String(result)}`;
+      } catch {
+        return `完整结果: ${String(result)}`;
+      }
+    };
+
+    container.innerHTML += `
+            <div class="task-item">
+                <div class="task-header">
+                    <div class="task-name">${
+                      task.workflow_name || "未命名工作流"
+                    }</div>
+                    <span class="task-status ${statusClass}">${statusText}</span>
+                </div>
+                <div class="task-meta">
+                    <div>ID: ${task.id}</div>
+                    <div>创建时间: ${formatTime(task.created_at)}</div>
+                    ${
+                      task.completed_at
+                        ? `<div>完成时间: ${formatTime(
+                            task.completed_at
+                          )}</div>`
+                        : ""
+                    }
+                </div>
+                ${
+                  task.error_message
+                    ? `<div class="task-error">❌ 错误: ${task.error_message}</div>`
+                    : ""
+                }
+                ${
+                  task.result
+                    ? `<div class="task-result">📊 ${formatResult(
+                        task.result
+                      )}</div>`
+                    : ""
+                }
+            </div>
+        `;
+  });
+}
+
+// 页面加载时自动刷新数据
+document.addEventListener("DOMContentLoaded", refreshData);
+
+// 每30秒自动刷新一次
+setInterval(refreshData, 30000);

app/utils/s3_client.py

@@ -0,0 +1,18 @@
+import boto3
+from app.config import Settings
+import asyncio
+settings = Settings()
+
+s3_client = boto3.client('s3', aws_access_key_id=settings.AWS_ACCESS_KEY_ID, aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY, region_name=settings.AWS_REGION_NAME)
+
+async def upload_file_to_s3(file_path: str, bucket: str, object_name: str) -> str:
+    """从本地文件路径上传"""
+    loop = asyncio.get_event_loop()
+    await loop.run_in_executor(None, lambda: s3_client.upload_file(file_path, bucket, object_name))
+    return f"https://cdn.roasmax.cn/{object_name}"
+
+async def upload_bytes_to_s3(file_bytes: bytes, bucket: str, object_name: str) -> str:
+    """直接从内存中的bytes上传 (新函数)"""
+    loop = asyncio.get_event_loop()
+    await loop.run_in_executor(None, lambda: s3_client.put_object(Body=file_bytes, Bucket=bucket, Key=object_name))
+    return f"https://cdn.roasmax.cn/{object_name}"