comi-logic/ComfyUI-WorkflowPublisher
7
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: 添加ComfyUI服务器动态IP和端口检测功能，优化服务器注册和状态同步逻辑，增强系统的灵活性和可靠性

Browse Source
This commit is contained in:
iHeyTang
2025-08-18 13:36:34 +08:00
parent 8e818887d5
commit 7484e300ec
3 changed files with 506 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

323
js/publisher.js
View File

@@ -1,16 +1,91 @@
import { app } from "../../scripts/app.js";
import { api } from "../../scripts/api.js";
// 检测ComfyUI服务端口的函数
async function detectComfyUIPort(host) {
try {
const currentPort = window.location.port;
if (currentPort) {
console.log(`从当前页面URL获取端口: ${currentPort}`);
return parseInt(currentPort);
}
} catch (error) {
console.warn("无法从当前页面获取端口:", error);
}
// 默认返回8188
console.warn("无法检测到ComfyUI端口，使用默认端口8188");
return 8188;
}
// 获取本地IP地址的函数
async function getLocalIPAddress() {
try {
// 尝试通过WebRTC获取本地IP地址
const pc = new RTCPeerConnection({
iceServers: [],
});
pc.createDataChannel("");
const offer = await pc.createOffer();
await pc.setLocalDescription(offer);
return new Promise((resolve) => {
pc.onicecandidate = (event) => {
if (event.candidate) {
const ipMatch = event.candidate.candidate.match(
/([0-9]{1,3}(\.[0-9]{1,3}){3})/
);
if (ipMatch) {
const localIP = ipMatch[1];
// 保留有效的内网IP地址
if (
localIP.startsWith("192.168.") ||
localIP.startsWith("10.") ||
localIP.startsWith("172.") ||
localIP === "127.0.0.1"
) {
pc.close();
resolve(localIP);
}
}
}
};
// 超时处理
setTimeout(() => {
pc.close();
resolve("127.0.0.1"); // 如果获取失败，使用默认地址
}, 3000);
});
} catch (error) {
console.warn("无法获取本地IP地址，使用默认地址:", error);
return "127.0.0.1";
}
}
// 本地存储的 key
const COMFY_SERVER_REGISTRATION_KEY = "comfyui_server_registration";
// ComfyUI服务器注册配置
const DEFAULT_SERVER_CONFIG = {
// ComfyUI服务器注册配置 - 使用动态IP地址
let DEFAULT_SERVER_CONFIG = {
name: "comfyui_node_1",
http_url: "http://127.0.0.1:8188",
http_url: "http://127.0.0.1:8188", // 将在初始化时更新为实际IP
workflow_service_host: "http://localhost:8000",
};
// 初始化默认配置
async function initializeDefaultConfig() {
try {
const localIP = await getLocalIPAddress();
const port = await detectComfyUIPort(localIP);
DEFAULT_SERVER_CONFIG.http_url = `http://${localIP}:${port}`;
console.log("自动获取本地IP地址和端口:", localIP, port);
} catch (error) {
console.warn("初始化IP地址和端口失败，使用默认地址:", error);
}
}
// 获取ComfyUI服务器注册配置
function getComfyServerConfig() {
try {
@@ -66,8 +141,7 @@ class ComfyServerManager {
// 注册服务器到工作流服务
async registerServer() {
if (!this.serverConfig.workflow_service_host) {
console.warn("未设置工作流服务地址，无法注册");
return false;
return { success: false, error: "未设置工作流服务地址" };
}
try {
@@ -91,22 +165,26 @@ class ComfyServerManager {
console.log("ComfyUI服务器注册成功:", result);
this.isRegistered = true;
this.startHeartbeat();
return true;
return { success: true };
} else {
const errorText = await response.text();
console.error("ComfyUI服务器注册失败:", response.status, errorText);
return false;
const errorMessage = `HTTP ${response.status}: ${
errorText || "服务器响应错误"
}`;
console.error("ComfyUI服务器注册失败:", errorMessage);
return { success: false, error: errorMessage };
}
} catch (error) {
console.error("注册ComfyUI服务器时发生错误:", error);
return false;
const errorMessage = `网络错误: ${error.message}`;
console.error("注册ComfyUI服务器时发生错误:", errorMessage);
return { success: false, error: errorMessage };
}
}
// 注销服务器
async unregisterServer() {
if (!this.serverConfig.workflow_service_host || !this.isRegistered) {
return false;
return { success: false, error: "未设置工作流服务地址或服务器未注册" };
}
try {
@@ -125,14 +203,19 @@ class ComfyServerManager {
console.log("ComfyUI服务器注销成功");
this.isRegistered = false;
this.stopHeartbeat();
return true;
return { success: true };
} else {
console.error("ComfyUI服务器注销失败:", response.status);
return false;
const errorText = await response.text();
const errorMessage = `HTTP ${response.status}: ${
errorText || "服务器响应错误"
}`;
console.error("ComfyUI服务器注销失败:", errorMessage);
return { success: false, error: errorMessage };
}
} catch (error) {
console.error("注销ComfyUI服务器时发生错误:", error);
return false;
const errorMessage = `网络错误: ${error.message}`;
console.error("注销ComfyUI服务器时发生错误:", errorMessage);
return { success: false, error: errorMessage };
}
}
@@ -206,8 +289,12 @@ class ComfyServerManager {
// 如果已注册，需要重新注册以更新配置
if (this.isRegistered) {
this.unregisterServer().then(() => {
setTimeout(() => this.registerServer(), 1000);
this.unregisterServer().then((result) => {
if (result.success) {
setTimeout(() => this.registerServer(), 1000);
} else {
console.warn("更新配置时注销失败:", result.error);
}
});
}
}
@@ -264,6 +351,46 @@ class ComfyServerManager {
}
return null;
}
// 动态更新IP地址和端口
async updateLocalIPAddress() {
try {
const localIP = await getLocalIPAddress();
const port = await detectComfyUIPort(localIP);
const newHttpUrl = `http://${localIP}:${port}`;
if (this.serverConfig.http_url !== newHttpUrl) {
this.serverConfig.http_url = newHttpUrl;
saveComfyServerConfig(this.serverConfig);
console.log("已更新本地IP地址和端口:", localIP, port);
return { ip: localIP, port: port };
}
return null;
} catch (error) {
console.warn("更新IP地址和端口失败:", error);
return null;
}
}
// 同步注册状态
syncRegistrationStatus() {
if (this.serverConfig.workflow_service_host) {
this.getServerStatus()
.then((status) => {
if (status) {
this.isRegistered = true;
console.log("同步注册状态: 已注册");
} else {
this.isRegistered = false;
console.log("同步注册状态: 未注册");
}
})
.catch((error) => {
console.warn("同步注册状态失败:", error);
this.isRegistered = false;
});
}
}
}
// 自定义 fetchApi 函数，使用ComfyUI服务器配置中的工作流服务地址
@@ -375,6 +502,9 @@ const createButton = (innerHTML, title, { onClick }) => {
app.registerExtension({
name: "Comfy.WorkflowPublisher",
async setup() {
// 初始化默认配置，获取本地IP地址
await initializeDefaultConfig();
const addCustomStyles = () => {
const styleId = "jags-publisher-styles";
if (document.getElementById(styleId)) return;
@@ -589,20 +719,31 @@ app.registerExtension({
<div style="display: flex; flex-direction: column; gap: 15px;">
<div>
<label for="server-name" style="display: block; margin-bottom: 5px;">当前服务节点名称:</label>
<input id="server-name" type="text" value="${serverConfig.name}" placeholder="例如: comfyui_node_1" style="width: 100%; box-sizing: border-box; background: var(--comfy-input-bg); color: var(--fg-color); border: 1px solid var(--border-color); padding: 8px; border-radius: 4px;">
<input id="server-name" type="text" value="${
serverConfig.name
}" placeholder="例如: comfyui_node_1" style="width: 100%; box-sizing: border-box; background: var(--comfy-input-bg); color: var(--fg-color); border: 1px solid var(--border-color); padding: 8px; border-radius: 4px;">
</div>
<div>
<label for="server-http-url" style="display: block; margin-bottom: 5px;">当前服务节点地址:</label>
<input id="server-http-url" type="text" value="${serverConfig.http_url}" placeholder="例如: http://127.0.0.1:8188" style="width: 100%; box-sizing: border-box; background: var(--comfy-input-bg); color: var(--fg-color); border: 1px solid var(--border-color); padding: 8px; border-radius: 4px;">
<small style="color: var(--border-color); font-size: 0.9em;">WebSocket地址将自动生成为: <span id="ws-url-preview">${serverConfig.http_url ? serverConfig.http_url.replace(/^https?:\/\//, 'ws://') + '/ws' : ''}</span></small>
<input id="server-http-url" type="text" value="${
serverConfig.http_url
}" placeholder="例如: http://127.0.0.1:8188" style="width: 100%; box-sizing: border-box; background: var(--comfy-input-bg); color: var(--fg-color); border: 1px solid var(--border-color); padding: 8px; border-radius: 4px;">
<small style="color: var(--border-color); font-size: 0.9em;">WebSocket地址将自动生成为: <span id="ws-url-preview">${
serverConfig.http_url
? serverConfig.http_url.replace(/^https?:\/\//, "ws://") +
"/ws"
: ""
}</span></small>
</div>
<div>
<label for="workflow-service-host" style="display: block; margin-bottom: 5px;">工作流服务地址 (统一管理地址):</label>
<input id="workflow-service-host" type="text" value="${settings.host}" placeholder="例如: http://localhost:8000" style="width: 100%; box-sizing: border-box; background: var(--comfy-input-bg); color: var(--fg-color); border: 1px solid var(--border-color); padding: 8px; border-radius: 4px;">
<input id="workflow-service-host" type="text" value="${
settings.host
}" placeholder="例如: http://localhost:8000" style="width: 100%; box-sizing: border-box; background: var(--comfy-input-bg); color: var(--fg-color); border: 1px solid var(--border-color); padding: 8px; border-radius: 4px;">
</div>
<div id="server-status" style="padding: 10px; border-radius: 4px; background: var(--comfy-input-bg); border: 1px solid var(--border-color);">
@@ -613,6 +754,7 @@ app.registerExtension({
<button id="btn-register" class="jags-button" style="flex: 1; min-width: 120px;">注册服务器</button>
<button id="btn-unregister" class="jags-button" style="flex: 1; min-width: 120px;">注销服务器</button>
<button id="btn-refresh-status" class="jags-button" style="flex: 1; min-width: 120px;">刷新状态</button>
<button id="btn-refresh-ip" class="jags-button" style="flex: 1; min-width: 120px;">刷新IP和端口</button>
</div>
</div>
`;
@@ -627,19 +769,22 @@ app.registerExtension({
const serverManager = new ComfyServerManager();
if (settings.host) {
serverManager.setWorkflowServiceHost(settings.host);
// 同步注册状态
serverManager.syncRegistrationStatus();
}
// 绑定事件
const btnRegister = document.getElementById("btn-register");
const btnUnregister = document.getElementById("btn-unregister");
const btnRefreshStatus = document.getElementById("btn-refresh-status");
const btnRefreshIP = document.getElementById("btn-refresh-ip");
const statusText = document.getElementById("status-text");
// 添加HTTP地址输入框的实时预览功能
const httpUrlInput = document.getElementById("server-http-url");
const wsUrlPreview = document.getElementById("ws-url-preview");
httpUrlInput.addEventListener("input", function() {
httpUrlInput.addEventListener("input", function () {
const httpUrl = this.value;
if (httpUrl) {
try {
@@ -677,15 +822,16 @@ app.registerExtension({
btnRegister.disabled = true;
btnRegister.textContent = "注册中...";
const success = await serverManager.registerServer();
if (success) {
const result = await serverManager.registerServer();
if (result.success) {
statusText.textContent = "已注册";
statusText.style.color = "var(--success-color)";
btnRegister.textContent = "已注册";
btnRegister.disabled = true;
btnUnregister.disabled = false;
} else {
statusText.textContent = "注册失败";
const errorMessage = result.error || "未知错误";
statusText.textContent = `注册失败: ${errorMessage}`;
statusText.style.color = "var(--error-color)";
btnRegister.textContent = "注册服务器";
btnRegister.disabled = false;
@@ -697,8 +843,8 @@ app.registerExtension({
btnUnregister.disabled = true;
btnUnregister.textContent = "注销中...";
const success = await serverManager.unregisterServer();
if (success) {
const result = await serverManager.unregisterServer();
if (result.success) {
statusText.textContent = "已注销";
statusText.style.color = "orange";
btnRegister.disabled = false;
@@ -706,7 +852,8 @@ app.registerExtension({
btnUnregister.disabled = true;
btnUnregister.textContent = "注销服务器";
} else {
statusText.textContent = "注销失败";
const errorMessage = result.error || "未知错误";
statusText.textContent = `注销失败: ${errorMessage}`;
statusText.style.color = "var(--error-color)";
btnUnregister.disabled = false;
btnUnregister.textContent = "注销服务器";
@@ -718,17 +865,56 @@ app.registerExtension({
if (settings.host) {
serverManager.setWorkflowServiceHost(settings.host);
const status = await serverManager.getServerStatus();
// 调试信息
console.log("服务器状态:", status);
if (status && status.last_heartbeat) {
console.log("最后心跳时间原始值:", status.last_heartbeat);
console.log("时间类型:", typeof status.last_heartbeat);
}
if (status) {
statusText.textContent = `在线 - 最后心跳: ${new Date(
status.last_heartbeat
).toLocaleString()}`;
// 安全地处理时间显示
let heartbeatTime = "未知";
if (status.last_heartbeat) {
try {
// 尝试解析ISO格式的时间字符串
const date = new Date(status.last_heartbeat);
if (!isNaN(date.getTime())) {
heartbeatTime = date.toLocaleString("zh-CN", {
year: "numeric",
month: "2-digit",
day: "2-digit",
hour: "2-digit",
minute: "2-digit",
second: "2-digit",
});
} else {
heartbeatTime = "时间格式错误";
}
} catch (error) {
console.warn(
"时间解析失败:",
error,
"原始时间:",
status.last_heartbeat
);
heartbeatTime = "时间解析失败";
}
}
statusText.textContent = `在线 - 最后心跳: ${heartbeatTime}`;
statusText.style.color = "var(--success-color)";
// 同步更新本地注册状态
serverManager.isRegistered = true;
btnRegister.disabled = true;
btnRegister.textContent = "已注册";
btnUnregister.disabled = false;
} else {
statusText.textContent = "未注册";
statusText.style.color = "orange";
// 同步更新本地注册状态
serverManager.isRegistered = false;
btnRegister.disabled = false;
btnRegister.textContent = "注册服务器";
btnUnregister.disabled = true;
@@ -736,10 +922,81 @@ app.registerExtension({
}
};
// 刷新IP地址和端口
btnRefreshIP.onclick = async () => {
btnRefreshIP.disabled = true;
btnRefreshIP.textContent = "刷新中...";
try {
const result = await serverManager.updateLocalIPAddress();
if (result) {
// 更新输入框显示
document.getElementById(
"server-http-url"
).value = `http://${result.ip}:${result.port}`;
// 更新WebSocket预览
const wsUrlPreview = document.getElementById("ws-url-preview");
wsUrlPreview.textContent = `ws://${result.ip}:${result.port}/ws`;
statusText.textContent = `IP地址和端口已更新为: ${result.ip}:${result.port}`;
statusText.style.color = "var(--success-color)";
} else {
statusText.textContent = "IP地址和端口无需更新";
statusText.style.color = "orange";
}
} catch (error) {
statusText.textContent = `IP地址和端口更新失败: ${error.message}`;
statusText.style.color = "var(--error-color)";
} finally {
btnRefreshIP.disabled = false;
btnRefreshIP.textContent = "刷新IP和端口";
}
};
// 初始状态检查
if (settings.host) {
btnRefreshStatus.click();
} else {
// 如果没有设置工作流服务地址，确保按钮状态正确
btnRegister.disabled = false;
btnUnregister.disabled = true;
}
// 自动刷新IP地址和端口
setTimeout(async () => {
try {
const result = await serverManager.updateLocalIPAddress();
if (result) {
// 更新输入框显示
document.getElementById(
"server-http-url"
).value = `http://${result.ip}:${result.port}`;
// 更新WebSocket预览
const wsUrlPreview = document.getElementById("ws-url-preview");
wsUrlPreview.textContent = `ws://${result.ip}:${result.port}/ws`;
console.log(
"对话框显示时自动更新IP地址和端口:",
result.ip,
result.port
);
}
} catch (error) {
console.warn("自动更新IP地址和端口失败:", error);
}
}, 500);
// 延迟同步按钮状态，确保与内部状态一致
setTimeout(() => {
if (serverManager.isRegistered) {
btnRegister.disabled = true;
btnRegister.textContent = "已注册";
btnUnregister.disabled = false;
} else {
btnRegister.disabled = false;
btnRegister.textContent = "注册服务器";
btnUnregister.disabled = true;
}
}, 1000);
} catch (error) {
console.error("显示ComfyUI服务器管理对话框失败:", error);
alert(`显示对话框失败: ${error.message}`);

36
workflow_service/comfy/comfy_server.py
View File

@@ -194,6 +194,33 @@ class ComfyUIServerManager:
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} 已存在，将更新配置")
@@ -217,8 +244,12 @@ class ComfyUIServerManager:
# 保存到数据库
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}")
logger.info(f"服务器 {name} 注册成功: {http_url} (IP: {host}:{port})")
return True
async def unregister_server(self, name: str) -> bool:
@@ -393,7 +424,8 @@ class ComfyUIServerManager:
if is_healthy and server.status != ServerStatus.ONLINE:
server.status = ServerStatus.ONLINE
logger.info(f"服务器 {server.name} 恢复在线")
server.last_heartbeat = current_time # 同步更新心跳时间
logger.info(f"服务器 {server.name} 恢复在线，心跳时间已同步")
elif not is_healthy and server.status == ServerStatus.ONLINE:
server.status = ServerStatus.ERROR
logger.warning(f"服务器 {server.name} 健康检查失败")

260
workflow_service/routes/monitor.py
View File

@@ -11,7 +11,10 @@ import json
logger = logging.getLogger(__name__)
from workflow_service.config import settings
from workflow_service.database.api import get_workflow_runs_recent, get_workflow_run_nodes
from workflow_service.database.api import (
get_workflow_runs_recent,
get_workflow_run_nodes,
)
from workflow_service.comfy.comfy_queue import WorkflowQueueManager
from workflow_service.comfy.comfy_server import server_manager
@@ -20,6 +23,7 @@ monitor_router = APIRouter(prefix="/monitor", tags=["监控"])
# 全局队列管理器实例
queue_manager = WorkflowQueueManager()
@monitor_router.get("/", response_class=HTMLResponse)
async def monitor_dashboard():
"""监控仪表板页面"""
@@ -139,8 +143,8 @@ async def monitor_dashboard():
.server-card {
border: 1px solid #e1e8ed;
border-radius: 12px;
padding: 20px;
border-radius: 8px;
padding: 12px;
background: #f8f9fa;
transition: all 0.3s ease;
position: relative;
@@ -175,34 +179,84 @@ async def monitor_dashboard():
background: #e74c3c;
}
.server-header {
display: flex;
align-items: center;
gap: 12px;
margin-bottom: 8px;
}
.server-card h4 {
color: #2c3e50;
margin-bottom: 12px;
font-size: 1.1em;
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: grid;
grid-template-columns: 1fr 1fr;
gap: 12px;
font-size: 0.9em;
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: 6px;
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 {
@@ -376,26 +430,46 @@ async def monitor_dashboard():
}
};
// 格式化心跳时间，突出显示状态
const formatHeartbeat = (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}">
<h4>${server.http_url}</h4>
<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-indicator ${statusColor}"></span>
<span class="info-label">状态:</span>
<span class="info-value">${server.status === 'online' ? '在线' : '离线'}</span>
<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="info-label">任务:</span>
<span class="info-value">${server.current_tasks || 0}/${server.max_concurrent_tasks || 1}</span>
<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="info-label">心跳:</span>
<span class="info-value">${formatTime(server.last_heartbeat)}</span>
<span class="status-text">心跳: ${formatHeartbeat(server.last_heartbeat)}</span>
</div>
<span class="info-separator">|</span>
<div class="info-item">
<span class="info-label">检查:</span>
<span class="info-value">${formatTime(server.last_health_check)}</span>
<span class="status-text">检查: ${formatTime(server.last_health_check)}</span>
</div>
</div>
${server.error ? `<div class="error-message">${server.error}</div>` : ''}
@@ -448,14 +522,15 @@ async def monitor_dashboard():
"""
return HTMLResponse(content=html_content)
@monitor_router.get("/system-stats")
async def get_system_stats() -> Dict[str, Any]:
"""获取系统统计信息"""
try:
cpu_percent = psutil.cpu_percent(interval=1)
memory = psutil.virtual_memory()
disk = psutil.disk_usage('/')
disk = psutil.disk_usage("/")
return {
"cpu_percent": round(cpu_percent, 1),
"memory_percent": round(memory.percent, 1),
@@ -464,11 +539,12 @@ async def get_system_stats() -> Dict[str, Any]:
"disk_percent": round(disk.percent, 1),
"disk_used_gb": round(disk.used / (1024**3), 2),
"disk_total_gb": round(disk.total / (1024**3), 2),
"timestamp": datetime.now().isoformat()
"timestamp": datetime.now().isoformat(),
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"获取系统统计信息失败: {str(e)}")
@monitor_router.get("/task-stats")
async def get_task_stats() -> Dict[str, Any]:
"""获取任务统计信息"""
@@ -476,139 +552,164 @@ async def get_task_stats() -> Dict[str, Any]:
# 获取最近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 = 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),
"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()
"timestamp": datetime.now().isoformat(),
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"获取任务统计信息失败: {str(e)}")
@monitor_router.get("/server-status")
async def get_server_status() -> List[Dict[str, Any]]:
"""获取服务器状态信息"""
try:
import aiohttp
import asyncio
# 从 ComfyUIServerManager 获取所有注册的服务器
all_servers = await server_manager.get_all_servers()
if not all_servers:
logger.info("当前没有动态注册的服务器")
return []
async def check_server_status(server):
try:
timeout = aiohttp.ClientTimeout(total=5)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(f"{server.http_url}/system_stats") as response:
async with session.get(
f"{server.http_url}/system_stats"
) as response:
if response.status == 200:
return {
"name": server.name,
"http_url": server.http_url,
"ws_url": server.ws_url,
"status": "online",
"response_time": response.headers.get('X-Response-Time', 'N/A'),
"current_tasks": getattr(server, 'current_tasks', 0),
"max_concurrent_tasks": getattr(server, 'max_concurrent_tasks', 1),
"last_heartbeat": getattr(server, 'last_heartbeat', None),
"last_health_check": getattr(server, 'last_health_check', None)
"response_time": response.headers.get(
"X-Response-Time", "N/A"
),
"current_tasks": getattr(server, "current_tasks", 0),
"max_concurrent_tasks": getattr(
server, "max_concurrent_tasks", 1
),
"last_heartbeat": getattr(
server, "last_heartbeat", None
),
"last_health_check": getattr(
server, "last_health_check", None
),
}
else:
return {
"name": server.name,
"http_url": server.http_url,
"ws_url": server.ws_url,
"status": "offline",
"error": f"HTTP {response.status}",
"current_tasks": getattr(server, 'current_tasks', 0),
"max_concurrent_tasks": getattr(server, 'max_concurrent_tasks', 1),
"last_heartbeat": getattr(server, 'last_heartbeat', None),
"last_health_check": getattr(server, 'last_health_check', None)
"current_tasks": getattr(server, "current_tasks", 0),
"max_concurrent_tasks": getattr(
server, "max_concurrent_tasks", 1
),
"last_heartbeat": getattr(
server, "last_heartbeat", None
),
"last_health_check": getattr(
server, "last_health_check", None
),
}
except Exception as e:
return {
"name": server.name,
"http_url": server.http_url,
"ws_url": server.ws_url,
"status": "offline",
"error": str(e),
"current_tasks": getattr(server, 'current_tasks', 0),
"max_concurrent_tasks": getattr(server, 'max_concurrent_tasks', 1),
"last_heartbeat": getattr(server, 'last_heartbeat', None),
"last_health_check": getattr(server, 'last_health_check', None)
"current_tasks": getattr(server, "current_tasks", 0),
"max_concurrent_tasks": getattr(server, "max_concurrent_tasks", 1),
"last_heartbeat": getattr(server, "last_heartbeat", None),
"last_health_check": getattr(server, "last_health_check", None),
}
# 并发检查所有服务器状态
tasks = [check_server_status(server) for server in all_servers]
results = await asyncio.gather(*tasks, return_exceptions=True)
# 过滤掉异常结果
valid_results = []
for result in results:
if isinstance(result, dict):
valid_results.append(result)
return valid_results
except Exception as e:
raise HTTPException(status_code=500, detail=f"获取服务器状态失败: {str(e)}")
@monitor_router.get("/recent-tasks")
async def get_recent_tasks(limit: int = 10) -> 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)
# 限制返回数量
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')
})
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"),
}
)
return formatted_runs
except Exception as e:
raise HTTPException(status_code=500, detail=f"获取最近任务失败: {str(e)}")
@monitor_router.get("/health")
async def health_check() -> Dict[str, Any]:
"""健康检查端点"""
try:
# 检查数据库连接
from workflow_service.database.connection import get_db
db = get_db()
# 从 ComfyUIServerManager 获取服务器信息
all_servers = await server_manager.get_all_servers()
dynamic_servers_count = len(all_servers)
# 检查服务器健康状态
online_servers = [s for s in all_servers if s.status == 'online']
offline_servers = [s for s in all_servers if s.status != 'online']
online_servers = [s for s in all_servers if s.status == "online"]
offline_servers = [s for s in all_servers if s.status != "online"]
return {
"status": "healthy",
"timestamp": datetime.now().isoformat(),
@@ -617,17 +718,18 @@ async def health_check() -> Dict[str, Any]:
"servers": {
"total_dynamic": dynamic_servers_count,
"online": len(online_servers),
"offline": len(offline_servers)
"offline": len(offline_servers),
},
"uptime": "N/A" # 可以添加启动时间跟踪
"uptime": "N/A", # 可以添加启动时间跟踪
}
except Exception as e:
return {
"status": "unhealthy",
"timestamp": datetime.now().isoformat(),
"error": str(e)
"error": str(e),
}
@monitor_router.get("/queue/status")
async def get_queue_status():
"""获取队列状态"""
@@ -636,12 +738,13 @@ async def get_queue_status():
return {
"success": True,
"data": status,
"timestamp": datetime.now().isoformat()
"timestamp": datetime.now().isoformat(),
}
except Exception as e:
logger.error(f"获取队列状态失败: {e}")
raise HTTPException(status_code=500, detail=f"获取队列状态失败: {str(e)}")
@monitor_router.post("/queue/trigger")
async def trigger_queue_processing():
"""手动触发队列处理"""
@@ -650,24 +753,25 @@ async def trigger_queue_processing():
return {
"success": True,
"message": "队列处理已手动触发",
"timestamp": datetime.now().isoformat()
"timestamp": datetime.now().isoformat(),
}
except Exception as e:
logger.error(f"手动触发队列处理失败: {e}")
raise HTTPException(status_code=500, detail=f"手动触发队列处理失败: {str(e)}")
@monitor_router.post("/queue/interval")
async def set_monitor_interval(interval: int):
"""设置队列监控间隔"""
try:
if interval < 1:
raise HTTPException(status_code=400, detail="监控间隔不能小于1秒")
await queue_manager.set_monitor_interval(interval)
return {
"success": True,
"message": f"队列监控间隔已设置为 {interval} 秒",
"timestamp": datetime.now().isoformat()
"timestamp": datetime.now().isoformat(),
}
except Exception as e:
logger.error(f"设置监控间隔失败: {e}")