publisher/comfyui_o1key
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 新增启动欢迎通知、流式预览节点及多项功能更新

Browse Source 
- 新增启动弹窗通知（绿色主题，支持关闭）
- 新增 StreamPreview 流式文本预览节点
- 新增 fileUpload、updateNotifier 前端 JS 模块
- 重构多个 client，统一错误处理
- 删除废弃节点 batch_nano_banana_v2、quan_neng_sheng_tu 等
- 将 .config 纳入版本控制（已清空密钥）
This commit is contained in:
Jony
2026-04-13 00:49:51 +08:00
parent bbc5f4a2c4
commit 92bcf65d14
28 changed files with 1381 additions and 3518 deletions
Download Patch File Download Diff File Expand all files Collapse all files
clients/base_client.py
+100 -28
View File
@@ -79,6 +79,15 @@ class BaseAPIClient(ABC):
"""
pass
def _make_session(self) -> aiohttp.ClientSession:
"""
创建统一的 aiohttp ClientSession，全局禁用 SSL 验证。
所有需要独立创建 session 的地方都应调用此方法，
避免因客户端系统缺少根证书导致 SSLCertVerificationError。
"""
connector = aiohttp.TCPConnector(ssl=False, limit=0, limit_per_host=0)
return aiohttp.ClientSession(connector=connector)
def get_headers(self, use_bearer_token: bool = False) -> Dict[str, str]:
"""
获取请求头
@@ -142,67 +151,132 @@ class BaseAPIClient(ABC):
) -> Dict[str, Any]:
"""
发送异步 HTTP 请求（带详细计时）
Args:
endpoint: API 端点
request_body: 请求体
session: aiohttp 会话（可选）
use_bearer_token: 是否使用 Bearer Token 认证
timeout: 超时时间（秒）- 已废弃，由服务器端控制
timeout: 超时时间（秒），默认 900 秒
Returns:
响应 JSON
Raises:
RuntimeError: 请求失败时
InterruptProcessingException: 用户点击终止按钮时
"""
import time
# 尝试导入 ComfyUI 中断机制
try:
from comfy.model_management import processing_interrupted, InterruptProcessingException
_interrupt_available = True
except ImportError:
_interrupt_available = False
url = f"{self.base_url}{endpoint}"
headers = self.get_headers(use_bearer_token)
# 检查请求大小
self.check_request_size(request_body)
close_session = False
if session is None:
session = aiohttp.ClientSession()
session = self._make_session()
close_session = True
try:
# 连接计时
# 设置请求超时：连接超时 30s，读取超时 900s（防止服务器出图后卡住）
_timeout_seconds = timeout if timeout is not None else 900
_aiohttp_timeout = aiohttp.ClientTimeout(
total=_timeout_seconds,
connect=30,
sock_read=_timeout_seconds
)
async def _do_request():
connect_start = time.time()
async with session.post(url, json=request_body, headers=headers) as response:
async with session.post(url, json=request_body, headers=headers, timeout=_aiohttp_timeout) as response:
connect_time = time.time() - connect_start
if response.status != 200:
error_text = await response.text()
raise RuntimeError(error_text)
# 接收响应体
wait_start = time.time()
response_data = await response.json()
download_time = time.time() - wait_start
# 附加计时信息到响应数据（供上层使用）
response_size = len(str(response_data))
if not isinstance(response_data, dict):
response_data = {"data": response_data}
# 将计时信息存储在响应的元数据中
response_data["_timing"] = {
"connect_time": connect_time,
"download_time": download_time,
"response_size": response_size
}
return response_data
async def _poll_interrupt():
"""每 0.5s 轮询一次中断标志"""
while True:
await asyncio.sleep(0.5)
if processing_interrupted():
return
try:
if _interrupt_available:
request_task = asyncio.ensure_future(_do_request())
interrupt_task = asyncio.ensure_future(_poll_interrupt())
done, pending = await asyncio.wait(
[request_task, interrupt_task],
return_when=asyncio.FIRST_COMPLETED
)
# 取消未完成的任务
for t in pending:
t.cancel()
try:
await t
except (asyncio.CancelledError, Exception):
pass
# 判断是哪个先完成
if interrupt_task in done and request_task not in done:
raise InterruptProcessingException()
# 请求完成，取出结果（可能含异常）
return request_task.result()
else:
return await _do_request()
except InterruptProcessingException:
raise
except aiohttp.ServerTimeoutError as e:
raise RuntimeError(
f"请求超时！等待服务器响应超过 {_timeout_seconds} 秒。\n"
f"服务器可能仍在生成图片，请稍后重试，或检查网络连接。"
) from e
except aiohttp.ClientConnectorError as e:
raise RuntimeError(
f"无法连接到服务器：{str(e)}\n"
f"请检查网络连接是否正常。"
) from e
except asyncio.TimeoutError as e:
raise RuntimeError(
f"请求超时！等待服务器响应超过 {_timeout_seconds} 秒。\n"
f"服务器可能仍在生成图片，请稍后重试，或检查网络连接。"
) from e
finally:
if close_session:
await session.close()
async def request_get_async(
self,
endpoint: str,
@@ -230,9 +304,9 @@ class BaseAPIClient(ABC):
close_session = False
if session is None:
session = aiohttp.ClientSession()
session = self._make_session()
close_session = True
try:
async with session.get(url, headers=headers) as response:
if response.status != 200:
@@ -327,9 +401,7 @@ class BaseAPIClient(ABC):
total = len(requests)
# 创建无限制的连接器
connector = aiohttp.TCPConnector(limit=0, limit_per_host=0)
async with aiohttp.ClientSession(connector=connector) as session:
async with self._make_session() as session:
tasks = []
for req in requests:
clients/gemini_client.py
+27 -67
View File
@@ -141,7 +141,6 @@ class GeminiAPIClient(BaseAPIClient):
resolution: str = "2K",
enable_grounding: bool = False,
enable_image_search: bool = False,
candidate_count: int = 1,
**kwargs
) -> Dict[str, Any]:
"""
@@ -154,7 +153,6 @@ class GeminiAPIClient(BaseAPIClient):
resolution: 分辨率
enable_grounding: 是否启用 Google Search Grounding
enable_image_search: 是否同时启用 Google Image Search（仅 Gemini 3.1 Flash 支持）
candidate_count: 单次请求返回的候选图数量，默认 1
Returns:
请求体字典
@@ -184,7 +182,6 @@ class GeminiAPIClient(BaseAPIClient):
}
],
"generationConfig": {
"candidateCount": candidate_count,
"responseModalities": ["TEXT", "IMAGE"],
"imageConfig": {
"aspectRatio": aspect_ratio,
@@ -303,7 +300,7 @@ class GeminiAPIClient(BaseAPIClient):
# 需要关闭 session 的标记
close_session = False
if session is None:
session = aiohttp.ClientSession()
session = self._make_session()
close_session = True
try:
@@ -438,7 +435,6 @@ class GeminiAPIClient(BaseAPIClient):
debug_request: bool = False,
enable_grounding: bool = False,
enable_image_search: bool = False,
candidate_count: int = 1
) -> tuple[List[Image.Image], Dict[str, Any]]:
"""
单次异步生成请求（极简单行日志）
@@ -465,7 +461,7 @@ class GeminiAPIClient(BaseAPIClient):
total_start = time.time()
# 任务前缀
task_prefix = f"[{task_index}/{total_tasks}]" if task_index is not None and total_tasks else ""
task_prefix = f"[{task_index}/{total_tasks}] " if task_index is not None and total_tasks else ""
# ========== 1. 构建请求 ==========
build_start = time.time()
@@ -477,7 +473,6 @@ class GeminiAPIClient(BaseAPIClient):
resolution=resolution,
enable_grounding=enable_grounding,
enable_image_search=enable_image_search,
candidate_count=candidate_count
)
build_time = time.time() - build_start
@@ -499,7 +494,7 @@ class GeminiAPIClient(BaseAPIClient):
safe_request = _truncate_base64_req(request_body)
print(
f"\n{'='*60}\n"
f"[请求体日志] 任务 {task_prefix or '?'} 发送请求体：\n"
f"[请求体日志] {task_prefix}发送请求体：\n"
f"端点: {endpoint}\n"
f"{_json.dumps(safe_request, ensure_ascii=False, indent=2)}\n"
f"{'='*60}\n"
@@ -541,7 +536,7 @@ class GeminiAPIClient(BaseAPIClient):
safe_response = _truncate_base64(response)
print(
f"\n{'='*60}\n"
f"[调试日志] 任务 {task_prefix or '?'} 完整 API 响应：\n"
f"[调试日志] {task_prefix}完整 API 响应：\n"
f"{_json.dumps(safe_response, ensure_ascii=False, indent=2)}\n"
f"{'='*60}\n"
)
@@ -554,7 +549,7 @@ class GeminiAPIClient(BaseAPIClient):
except Exception as e:
parse_time = time.time() - parse_start
error_first_line = str(e).split('\n')[0]
print(f"{task_prefix} 请求 {size_str} → API {request_time:.1f}s → 解析失败: {error_first_line} ✗")
print(f"{task_prefix}请求 {size_str} → API {request_time:.1f}s → 解析失败: {error_first_line} ✗")
raise
parse_time = time.time() - parse_start
@@ -578,7 +573,7 @@ class GeminiAPIClient(BaseAPIClient):
download_info = f"{img_size_str}"
# 单行输出
print(f"{task_prefix} 请求 {size_str} → API {request_time:.1f}s → {download_info} ✓")
print(f"{task_prefix}请求 {size_str} → API {request_time:.1f}s → {download_info} ✓")
# 返回结果和计时信息
total_time = time.time() - total_start
@@ -605,7 +600,6 @@ class GeminiAPIClient(BaseAPIClient):
debug_request: bool = False,
enable_grounding: bool = False,
enable_image_search: bool = False,
candidate_count: int = 1
) -> List[Image.Image]:
"""
批量全并发生成 - 改进版：支持分批处理和内存管理
@@ -622,7 +616,6 @@ class GeminiAPIClient(BaseAPIClient):
debug_request: 是否打印发送的请求体
enable_grounding: 是否启用 Google Search Grounding
enable_image_search: 是否同时启用 Google Image Search
candidate_count: 单次请求返回的候选图数量
Returns:
生成的图像列表
@@ -635,28 +628,23 @@ class GeminiAPIClient(BaseAPIClient):
success_count = 0
fail_count = 0
first_error = None # 保存第一个错误
# 分批处理配置
max_concurrent = 10 # 最大并发数
save_batch_size = 10 # 分批保存大小
# 计算需要多少批次
num_batches = (batch_size + max_concurrent - 1) // max_concurrent
print(f"GeminiClient: 批量生成 {batch_size} 张图片，并发数: {max_concurrent}，分 {num_batches} 批执行")
connector = aiohttp.TCPConnector(limit=0, limit_per_host=0)
connector = aiohttp.TCPConnector(ssl=False, limit=0, limit_per_host=0)
async with aiohttp.ClientSession(connector=connector) as session:
# 分批执行
for batch_idx in range(num_batches):
batch_start = batch_idx * max_concurrent
batch_end = min(batch_start + max_concurrent, batch_size)
batch_size_current = batch_end - batch_start
if num_batches > 1:
print(f"GeminiClient: 执行第 {batch_idx + 1}/{num_batches} 批 ({batch_start + 1}-{batch_end})...")
# 创建当前批次的任务
tasks = []
for i in range(batch_size_current):
@@ -675,7 +663,6 @@ class GeminiAPIClient(BaseAPIClient):
debug_request=debug_request,
enable_grounding=enable_grounding,
enable_image_search=enable_image_search,
candidate_count=candidate_count
),
name=f"task_{task_index}"
)
@@ -698,14 +685,11 @@ class GeminiAPIClient(BaseAPIClient):
all_images.append(img)
success_count += 1
# 通知进度
if progress_callback:
progress_callback(completed, batch_size, True, None)
# 每成功生成一张图片就打印日志
print(f"GeminiClient: 任务 {completed}/{batch_size} 成功生成图片 ✓")
except Exception as e:
fail_count += 1
# 保存第一个错误（用于后续抛出）
@@ -719,25 +703,20 @@ class GeminiAPIClient(BaseAPIClient):
# 当前批次完成后，立即清理内存
if batch_images:
print(f"GeminiClient: 第 {batch_idx + 1} 批完成，生成 {len(batch_images)} 张图片")
# 强制垃圾回收，释放内存
import gc
gc.collect()
# 短暂暂停，让系统处理内存
await asyncio.sleep(0.1)
# 清空当前批次图片引用，帮助垃圾回收
batch_images = []
# 最终结果检查
if not all_images:
# 如果有保存的原始错误，直接抛出原始错误
if first_error:
raise first_error
raise RuntimeError(f"批量生成失败，{fail_count} 个请求全部失败")
print(f"GeminiClient: 批量生成完成，成功 {success_count}/{batch_size}，失败 {fail_count}")
return all_images
@@ -754,7 +733,6 @@ class GeminiAPIClient(BaseAPIClient):
debug_request: bool = False,
enable_grounding: bool = False,
enable_image_search: bool = False,
candidate_count: int = 1
) -> List[Image.Image]:
"""
同步生成接口（用于 ComfyUI）
@@ -771,7 +749,6 @@ class GeminiAPIClient(BaseAPIClient):
debug_request: 是否打印发送的请求体
enable_grounding: 是否启用 Google Search Grounding
enable_image_search: 是否同时启用 Google Image Search
candidate_count: 单次请求返回的候选图数量
Returns:
生成的图像列表
@@ -788,7 +765,6 @@ class GeminiAPIClient(BaseAPIClient):
debug_request=debug_request,
enable_grounding=enable_grounding,
enable_image_search=enable_image_search,
candidate_count=candidate_count
)
return self.run_async_in_thread(coro)
@@ -837,13 +813,11 @@ class GeminiAPIClient(BaseAPIClient):
fail_count = 0
first_error = None # 保存第一个错误
total_tasks = len(prompts) * images_per_prompt
# 分批处理配置
max_concurrent = 10 # 最大并发数
print(f"GeminiClient: 多提示词批量生成，共 {total_tasks} 个任务，{len(prompts)} 个提示词，每个 {images_per_prompt} 张")
connector = aiohttp.TCPConnector(limit=0, limit_per_host=0)
connector = aiohttp.TCPConnector(ssl=False, limit=0, limit_per_host=0)
async with aiohttp.ClientSession(connector=connector) as session:
# 创建所有任务
@@ -874,15 +848,12 @@ class GeminiAPIClient(BaseAPIClient):
# 分批处理：每10个任务为一组
batch_size = max_concurrent
num_batches = (total_tasks + batch_size - 1) // batch_size
for batch_idx in range(num_batches):
batch_start = batch_idx * batch_size
batch_end = min(batch_start + batch_size, total_tasks)
batch_tasks = tasks[batch_start:batch_end]
if num_batches > 1:
print(f"GeminiClient: 执行第 {batch_idx + 1}/{num_batches} 批 ({batch_start + 1}-{batch_end})...")
# 收集当前批次的结果
batch_images = []
@@ -898,36 +869,26 @@ class GeminiAPIClient(BaseAPIClient):
all_images.append(img)
success_count += 1
# 通知进度
if progress_callback:
progress_callback(completed, total_tasks, True, None)
# 每成功生成一张图片就打印日志
print(f"GeminiClient: 任务 {completed}/{total_tasks} 成功生成图片 ✓")
except Exception as e:
fail_count += 1
# 保存第一个错误（用于后续抛出）
if first_error is None:
first_error = e
error_msg = str(e)
# 传递完整的错误信息（用于排查问题）
if progress_callback:
progress_callback(completed, total_tasks, False, error_msg)
print(f"GeminiClient: 任务 {completed}/{total_tasks} 失败 ✗")
# 当前批次完成后，立即清理内存
if batch_images:
print(f"GeminiClient: 第 {batch_idx + 1} 批完成，生成 {len(batch_images)} 张图片")
# 强制垃圾回收，释放内存
import gc
gc.collect()
# 短暂暂停，让系统处理内存
await asyncio.sleep(0.1)
# 清空当前批次图片引用，帮助垃圾回收
@@ -938,8 +899,7 @@ class GeminiAPIClient(BaseAPIClient):
if first_error:
raise first_error
raise RuntimeError(f"批量生成失败，{fail_count} 个请求全部失败")
print(f"GeminiClient: 多提示词批量生成完成，成功 {success_count}/{total_tasks}，失败 {fail_count}")
return all_images
def generate_multi_prompts_sync(
clients/kling_client.py
+2 -2
View File
@@ -137,7 +137,7 @@ class KlingClient:
on_progress: Optional[Callable[[int], None]] = None,
) -> str:
"""提交 → 轮询 → 下载，返回本地文件路径"""
connector = aiohttp.TCPConnector(force_close=True)
connector = aiohttp.TCPConnector(ssl=False, force_close=True)
async with aiohttp.ClientSession(connector=connector) as session:
if on_stage:
on_stage("submitting")
@@ -196,7 +196,7 @@ class KlingClient:
"Content-Type": "application/json"}
interval = self.POLL_INITIAL_INTERVAL
connector = aiohttp.TCPConnector(force_close=True)
connector = aiohttp.TCPConnector(ssl=False, force_close=True)
async with aiohttp.ClientSession(connector=connector) as session:
# 1. 提交
clients/openai_client.py
+2 -2
View File
@@ -309,7 +309,7 @@ class OpenAIAPIClient(BaseAPIClient):
close_session = False
if session is None:
session = aiohttp.ClientSession()
session = self._make_session()
close_session = True
try:
@@ -677,7 +677,7 @@ class OpenAIAPIClient(BaseAPIClient):
print(f"OpenAIClient: 批量生成 {batch_size} 张，并发数: {max_concurrent}，分 {num_batches} 批")
connector = aiohttp.TCPConnector(limit=0, limit_per_host=0)
connector = aiohttp.TCPConnector(ssl=False, limit=0, limit_per_host=0)
async with aiohttp.ClientSession(connector=connector) as session:
for batch_idx in range(num_batches):
clients/seedance_client.py
+1 -1
View File
@@ -169,7 +169,7 @@ class SeedanceClient:
on_progress: Optional[Callable[[int], None]] = None,
) -> tuple:
"""提交 → 轮询 → 下载，返回 (本地视频路径, 末帧图片URL或None)"""
connector = aiohttp.TCPConnector(force_close=True)
connector = aiohttp.TCPConnector(ssl=False, force_close=True)
async with aiohttp.ClientSession(connector=connector) as session:
# 提交
clients/sora_client.py
+5 -5
View File
@@ -132,7 +132,7 @@ class SoraClient(BaseAPIClient):
close_session = False
if session is None:
session = aiohttp.ClientSession()
session = self._make_session()
close_session = True
try:
@@ -174,7 +174,7 @@ class SoraClient(BaseAPIClient):
close_session = False
if session is None:
session = aiohttp.ClientSession()
session = self._make_session()
close_session = True
interval = self.POLL_INITIAL_INTERVAL
@@ -242,7 +242,7 @@ class SoraClient(BaseAPIClient):
close_session = False
if session is None:
session = aiohttp.ClientSession()
session = self._make_session()
close_session = True
try:
@@ -299,7 +299,7 @@ class SoraClient(BaseAPIClient):
"""
async def _run():
connector = aiohttp.TCPConnector(limit=0)
connector = aiohttp.TCPConnector(ssl=False, limit=0)
async with aiohttp.ClientSession(connector=connector) as session:
# 1. 提交任务
if on_stage:
@@ -408,7 +408,7 @@ class SoraClient(BaseAPIClient):
成功生成的视频路径列表
"""
batch_size = len(save_paths)
connector = aiohttp.TCPConnector(limit=0)
connector = aiohttp.TCPConnector(ssl=False, limit=0)
async with aiohttp.ClientSession(connector=connector) as session:
tasks = [
clients/veo_client.py
+5 -5
View File
@@ -160,7 +160,7 @@ class VeoClient(BaseAPIClient):
close_session = False
if session is None:
session = aiohttp.ClientSession()
session = self._make_session()
close_session = True
try:
@@ -202,7 +202,7 @@ class VeoClient(BaseAPIClient):
close_session = False
if session is None:
session = aiohttp.ClientSession()
session = self._make_session()
close_session = True
interval = self.POLL_INITIAL_INTERVAL
@@ -265,7 +265,7 @@ class VeoClient(BaseAPIClient):
close_session = False
if session is None:
session = aiohttp.ClientSession()
session = self._make_session()
close_session = True
try:
@@ -318,7 +318,7 @@ class VeoClient(BaseAPIClient):
"""
async def _run():
connector = aiohttp.TCPConnector(limit=0)
connector = aiohttp.TCPConnector(ssl=False, limit=0)
async with aiohttp.ClientSession(connector=connector) as session:
# 1. 提交任务
if on_stage:
@@ -383,7 +383,7 @@ class VeoClient(BaseAPIClient):
"""
async def _run():
batch_size = len(save_paths)
connector = aiohttp.TCPConnector(limit=0)
connector = aiohttp.TCPConnector(ssl=False, limit=0)
async def generate_one(save_path: str):
return await self._generate_one_video_async(