nodes/ucloud-comfyui-plugin
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #6 from mfzzf/main

Browse Source 
feat(modelverse): add Gemini Pro Image and Vidu & Sora video generation support
This commit is contained in:
jason
2026-01-04 13:59:18 +08:00
committed by GitHub
parent 2be7d1e881 89f0ef3fe6
commit af42216bd0
15 changed files with 1200 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3 -1
View File
@@ -4,4 +4,6 @@ py/__pycache__
py/modelverse_api/__pycache__
py/modelverse_api/requests/__pycache__
/venv
.idea/
.idea/
.ace-tool/
docs
py/gemini_pro_image.py
+149
View File
@@ -0,0 +1,149 @@
"""
Gemini 3 Pro Image Preview (Nano Banana Pro) ComfyUI Node.
Professional asset production with advanced features:
- High-resolution output (1K, 2K, 4K)
- Aspect ratio control
- Google Search grounding
- Up to 14 reference images
"""
import base64
import torch
from typing import Optional, List, Dict, Any
from comfy.comfy_types.node_typing import IO
from .modelverse_api.client import ModelverseClient
from .modelverse_api.requests.gemini_pro_image import GeminiProImageRequest
from .modelverse_api.utils import decode_image, images2tensor
def _extract_images_from_gemini_response(resp: Dict[str, Any]) -> List[torch.Tensor]:
"""Extract image tensors from Gemini API response."""
tensors: List[torch.Tensor] = []
candidates = resp.get("candidates", [])
for cand in candidates:
content = cand.get("content", {})
parts = content.get("parts", [])
for part in parts:
inline = part.get("inlineData")
if inline:
data_b64 = inline.get("data")
if data_b64:
raw = base64.b64decode(data_b64)
pil_img = decode_image(raw)
tensors.append(images2tensor(pil_img))
# Also print any text response
if part.get("text"):
print("INFO: Gemini Pro Image text response:", part.get("text"))
return tensors
# Aspect ratio options
ASPECT_RATIOS = ["auto", "1:1", "2:3", "3:2", "3:4", "4:3", "4:5", "5:4", "9:16", "16:9", "21:9"]
# Resolution options
IMAGE_SIZES = ["1K", "2K", "4K"]
class GeminiProImageNode:
"""
Gemini 3 Pro Image Preview (Nano Banana Pro) via Modelverse API.
Professional asset production with:
- High-resolution output (1K, 2K, 4K)
- Aspect ratio control
- Google Search grounding for real-time info
- Up to 14 reference images
Endpoint: /v1beta/models/gemini-3-pro-image-preview:generateContent
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"client": ("MODELVERSE_API_CLIENT",),
"prompt": (IO.STRING, {"multiline": True, "default": "Create a professional product photo"}),
"mime_type": (["image/png", "image/jpeg"], {"default": "image/png"}),
"aspect_ratio": (ASPECT_RATIOS, {"default": "auto"}),
"image_size": (IMAGE_SIZES, {"default": "1K"}),
"use_google_search": (IO.BOOLEAN, {"default": False, "tooltip": "Enable Google Search grounding for real-time info"}),
"num_requests": (IO.INT, {"default": 1, "min": 1, "max": 10, "step": 1, "display": "number"}),
},
"optional": {
"image1": (IO.IMAGE, {"default": None, "tooltip": "Reference image 1"}),
"image2": (IO.IMAGE, {"default": None, "tooltip": "Reference image 2"}),
"image3": (IO.IMAGE, {"default": None, "tooltip": "Reference image 3"}),
"image4": (IO.IMAGE, {"default": None, "tooltip": "Reference image 4"}),
"image5": (IO.IMAGE, {"default": None, "tooltip": "Reference image 5"}),
},
}
RETURN_TYPES = (IO.IMAGE,)
RETURN_NAMES = ("image",)
CATEGORY = "UCLOUD_MODELVERSE"
FUNCTION = "execute"
async def execute(
self,
client,
prompt: str,
mime_type: str = "image/png",
aspect_ratio: str = "auto",
image_size: str = "1K",
use_google_search: bool = False,
num_requests: int = 1,
image1=None,
image2=None,
image3=None,
image4=None,
image5=None,
):
if not prompt:
raise ValueError("Prompt is required")
mv_client = ModelverseClient(client["api_key"])
# Collect input images
images = []
for img in [image1, image2, image3, image4, image5]:
if img is not None:
images.append(img)
# Process aspect_ratio
ar = aspect_ratio if aspect_ratio != "auto" else None
outputs: List[torch.Tensor] = []
for i in range(num_requests):
req = GeminiProImageRequest(
prompt=prompt,
images=images if images else None,
mime_type=mime_type,
aspect_ratio=ar,
image_size=image_size,
use_google_search=use_google_search,
)
payload = req.build_payload()
resp = mv_client.post(req.API_PATH, payload)
if isinstance(resp, dict) and resp.get("error"):
err = resp.get("error")
raise Exception(f"GeminiProImage error: {err.get('message', 'Unknown error')}")
tensors = _extract_images_from_gemini_response(resp)
if not tensors:
print("WARN:", "No image data found in Gemini Pro response; check console for details.")
continue
outputs.append(tensors[0])
if not outputs:
raise Exception("No images generated from Gemini 3 Pro Image Preview")
return (torch.cat(outputs, dim=0),)
NODE_CLASS_MAPPINGS = {
'Gemini Pro Image (Nano Banana Pro)': GeminiProImageNode,
}
NODE_DISPLAY_NAME_MAPPINGS = {
'Gemini Pro Image (Nano Banana Pro)': 'Gemini 3 Pro Image Preview',
}
py/modelverse_api/requests/gemini_pro_image.py
+121
View File
@@ -0,0 +1,121 @@
"""
Gemini 3 Pro Image Preview (Nano Banana Pro) request builder.
Supports text-to-image and image editing with advanced features like
aspect ratio, resolution control, and Google Search grounding.
"""
import io
import base64
import numpy as np
from PIL import Image
from typing import Optional, List, Dict, Any
from pydantic import Field
from torch import Tensor
from ..utils import BaseRequest
def _tensor_to_base64(image: Tensor, mime_type: str = "image/png") -> Dict[str, str]:
"""Convert a ComfyUI image tensor (HWC or 4D with batch) to base64 without data URI."""
if image is None:
return None
# If batch, take first frame
if hasattr(image, 'shape') and len(image.shape) == 4:
image = image[0]
# Tensor (H, W, C) in 0..1 -> uint8
np_img = np.clip(255.0 * image.cpu().numpy(), 0, 255).astype(np.uint8)
pil_img = Image.fromarray(np_img)
fmt = 'PNG' if mime_type.lower().endswith('png') else 'JPEG'
with io.BytesIO() as bio:
pil_img.save(bio, format=fmt)
data = bio.getvalue()
return {
"mimeType": mime_type,
"data": base64.b64encode(data).decode("utf-8"),
}
class GeminiProImageRequest(BaseRequest):
"""
Request builder for Gemini 3 Pro Image Preview (Nano Banana Pro) generateContent endpoint.
Supports text-to-image, image editing, and advanced features like aspect ratio,
resolution control (1K/2K/4K), and Google Search grounding.
"""
API_PATH = "/v1beta/models/gemini-3-pro-image-preview:generateContent"
prompt: str = Field(..., description="Text prompt")
images: Optional[List[Tensor]] = Field(default=None, description="Optional input images (up to 14)")
mime_type: str = Field(default="image/png", description="MIME type for inline image data")
aspect_ratio: Optional[str] = Field(default=None, description="Aspect ratio: 1:1, 2:3, 3:2, 3:4, 4:3, 4:5, 5:4, 9:16, 16:9, 21:9")
image_size: Optional[str] = Field(default=None, description="Image resolution: 1K, 2K, 4K")
use_google_search: bool = Field(default=False, description="Enable Google Search grounding")
def __init__(
self,
prompt: str,
images: Optional[List[Tensor]] = None,
mime_type: str = "image/png",
aspect_ratio: Optional[str] = None,
image_size: Optional[str] = None,
use_google_search: bool = False,
**kwargs
):
super().__init__(**kwargs)
self.prompt = prompt
self.images = images
self.mime_type = mime_type
self.aspect_ratio = aspect_ratio
self.image_size = image_size
self.use_google_search = use_google_search
def build_payload(self) -> Dict[str, Any]:
parts: List[Dict[str, Any]] = []
# Add text prompt
if self.prompt:
parts.append({"text": self.prompt})
# Add images (up to 14)
if self.images:
for img in self.images[:14]: # Limit to 14 images
if isinstance(img, Tensor):
parts.append({
"inlineData": _tensor_to_base64(img, self.mime_type)
})
payload = {
"contents": [
{
"role": "user",
"parts": parts,
}
],
"generationConfig": {
"responseModalities": ["TEXT", "IMAGE"],
},
}
# Add imageConfig if aspect_ratio or image_size specified
if self.aspect_ratio or self.image_size:
image_config = {}
if self.aspect_ratio:
image_config["aspectRatio"] = self.aspect_ratio
if self.image_size:
image_config["imageSize"] = self.image_size
payload["generationConfig"]["imageConfig"] = image_config
# Add Google Search tool if enabled
if self.use_google_search:
payload["tools"] = [{"google_search": {}}]
return self._remove_empty_fields(payload)
def field_required(self):
return ["prompt"]
def field_order(self):
return ["prompt", "images", "mime_type", "aspect_ratio", "image_size", "use_google_search"]
py/preview_video.py
+72
View File
@@ -0,0 +1,72 @@
import os
import re
import requests
import folder_paths
from comfy.comfy_types.node_typing import IO
from comfy_api_nodes.util import download_url_to_video_output
class ModelversePreviewVideo:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"video_url": (IO.STRING, {"forceInput": True}),
"filename_prefix": (IO.STRING, {"default": "Modelverse"}),
"save_output": (IO.BOOLEAN, {"default": True}),
}
}
OUTPUT_NODE = True
FUNCTION = "run"
CATEGORY = "UCLOUD_MODELVERSE"
RETURN_TYPES = (IO.VIDEO,)
RETURN_NAMES = ("video",)
async def run(self, video_url, filename_prefix, save_output):
if type(video_url) == list:
video_url = video_url[0]
# Save to file if requested (do this first before download_url_to_video_output consumes the data)
if save_output:
output_dir = folder_paths.get_output_directory()
(
full_output_folder,
filename,
_,
_,
_,
) = folder_paths.get_save_image_path(filename_prefix, output_dir)
max_counter = 0
matcher = re.compile(f"{re.escape(filename)}_(\\d+)\\D*\\..+", re.IGNORECASE)
for existing_file in os.listdir(full_output_folder):
match = matcher.fullmatch(existing_file)
if match:
file_counter = int(match.group(1))
if file_counter > max_counter:
max_counter = file_counter
counter = max_counter + 1
file = f"{filename}_{counter:05}.mp4"
file_path = os.path.join(full_output_folder, file)
response = requests.get(video_url, timeout=120)
response.raise_for_status()
with open(file_path, "wb") as f:
f.write(response.content)
# Download video and get VIDEO output
video = await download_url_to_video_output(video_url)
return {"ui": {"video_url": [video_url]}, "result": (video,)}
NODE_CLASS_MAPPINGS = {
"Modelverse_PreviewVideo": ModelversePreviewVideo,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"Modelverse_PreviewVideo": "Modelverse Preview Video",
}
py/sora_i2v.py
+122
View File
@@ -0,0 +1,122 @@
"""
OpenAI Sora2 Img2Video - 图生视频模型
Models: openai/sora-2/image-to-video, openai/sora-2/image-to-video-pro
"""
import time
from .modelverse_api.client import ModelverseClient
from .modelverse_api.utils import image_to_base64
from comfy.comfy_types.node_typing import IO
MODELS = ["openai/sora-2/image-to-video", "openai/sora-2/image-to-video-pro"]
RESOLUTIONS_PRO = ["720p", "1080p"]
DURATIONS = [4, 8, 12]
class SoraImg2VideoNode:
"""
OpenAI Sora2 Img2Video - 图生视频
Models: image-to-video (普通版), image-to-video-pro (Pro版，支持1080p)
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"client": ("MODELVERSE_API_CLIENT",),
"model": (MODELS, {"default": "openai/sora-2/image-to-video", "tooltip": "image-to-video: 普通版, image-to-video-pro: Pro版(支持1080p)"}),
},
"optional": {
"first_frame_image": (IO.IMAGE, {"tooltip": "首帧图片"}),
"first_frame_url": (IO.STRING, {"default": "", "tooltip": "首帧图片URL (与image二选一)"}),
"prompt": (IO.STRING, {"multiline": True, "default": "", "tooltip": "提示词，用于指导视频生成"}),
"resolution": (RESOLUTIONS_PRO, {"default": "720p", "tooltip": "分辨率 (Pro版支持1080p)"}),
"duration": (DURATIONS, {"default": 4, "tooltip": "视频时长(秒): 4, 8, 12"}),
}
}
RETURN_TYPES = (IO.STRING, IO.STRING)
RETURN_NAMES = ("url", "task_id")
FUNCTION = "generate"
CATEGORY = "UCLOUD_MODELVERSE/Sora"
def generate(self, client, model, first_frame_image=None, first_frame_url="",
prompt="", resolution="720p", duration=4):
api_key = client.get("api_key")
if not api_key:
raise ValueError("API key is not set")
# Validate resolution for non-pro model
if model == "openai/sora-2/image-to-video" and resolution == "1080p":
raise ValueError("普通版不支持 1080p，请使用 Pro 版")
mv_client = ModelverseClient(api_key)
# Validate first frame input
has_url = first_frame_url and first_frame_url.strip()
has_image = first_frame_image is not None
if has_url and has_image:
raise ValueError("请提供 first_frame_image 或 first_frame_url，不能同时提供")
if not has_url and not has_image:
raise ValueError("必须提供 first_frame_image 或 first_frame_url")
task_input = {}
if has_url:
task_input["first_frame_url"] = first_frame_url.strip()
else:
task_input["first_frame_url"] = image_to_base64(first_frame_image)
if prompt and prompt.strip():
task_input["prompt"] = prompt.strip()
parameters = {
"duration": duration,
}
# Pro版支持resolution参数
if model == "openai/sora-2/image-to-video-pro":
parameters["resolution"] = resolution
# Submit task
submit_res = mv_client.submit_task(model, task_input, parameters)
task_id = submit_res.get("output", {}).get("task_id")
if not task_id:
raise Exception(f"Failed to submit task: {submit_res}")
print(f"Sora I2V task submitted: {task_id}")
# Poll for result
video_url = self._poll_task(mv_client, task_id)
return (video_url, task_id)
def _poll_task(self, mv_client, task_id, max_retries=180):
for i in range(max_retries):
status_res = mv_client.get_task_status(task_id)
task_status = status_res.get("output", {}).get("task_status")
if task_status == "Success":
urls = status_res.get("output", {}).get("urls", [])
if urls:
return urls[0]
raise Exception("Task succeeded but no video URL returned")
elif task_status == "Failure":
error = status_res.get("output", {}).get("error_message", "Unknown error")
raise Exception(f"Task failed: {error}")
elif task_status in ["Pending", "Running"]:
print(f"Task {task_id}: {task_status} ({i+1}/{max_retries})")
time.sleep(5)
else:
raise Exception(f"Unknown status: {task_status}")
raise Exception("Task timed out")
NODE_CLASS_MAPPINGS = {
"Sora_Img2Video": SoraImg2VideoNode,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"Sora_Img2Video": "Sora Img2Video",
}
py/sora_t2v.py
+99
View File
@@ -0,0 +1,99 @@
"""
OpenAI Sora2 Text2Video - 文生视频模型
Models: openai/sora-2/text-to-video, openai/sora-2/text-to-video-pro
"""
import time
from .modelverse_api.client import ModelverseClient
from comfy.comfy_types.node_typing import IO
MODELS = ["openai/sora-2/text-to-video", "openai/sora-2/text-to-video-pro"]
SIZES = ["720x1280", "1280x720"]
SIZES_PRO = ["720x1280", "1280x720", "1024x1792", "1792x1024"]
DURATIONS = [4, 8, 12]
class SoraText2VideoNode:
"""
OpenAI Sora2 Text2Video - 文生视频
Models: text-to-video (普通版), text-to-video-pro (Pro版，支持更多分辨率)
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"client": ("MODELVERSE_API_CLIENT",),
"model": (MODELS, {"default": "openai/sora-2/text-to-video", "tooltip": "text-to-video: 普通版, text-to-video-pro: Pro版"}),
"prompt": (IO.STRING, {"multiline": True, "default": "A beautiful girl is dancing", "tooltip": "提示词，用于指导视频生成"}),
},
"optional": {
"size": (SIZES_PRO, {"default": "720x1280", "tooltip": "视频尺寸 (Pro版支持更多选项)"}),
"duration": (DURATIONS, {"default": 4, "tooltip": "视频时长(秒): 4, 8, 12"}),
}
}
RETURN_TYPES = (IO.STRING, IO.STRING)
RETURN_NAMES = ("url", "task_id")
FUNCTION = "generate"
CATEGORY = "UCLOUD_MODELVERSE/Sora"
def generate(self, client, model, prompt, size="720x1280", duration=4):
api_key = client.get("api_key")
if not api_key:
raise ValueError("API key is not set")
# Validate size for non-pro model
if model == "openai/sora-2/text-to-video" and size not in SIZES:
raise ValueError(f"普通版仅支持 {SIZES}，请使用 Pro 版获取更多分辨率选项")
mv_client = ModelverseClient(api_key)
task_input = {"prompt": prompt}
parameters = {
"size": size,
"duration": duration,
}
# Submit task
submit_res = mv_client.submit_task(model, task_input, parameters)
task_id = submit_res.get("output", {}).get("task_id")
if not task_id:
raise Exception(f"Failed to submit task: {submit_res}")
print(f"Sora T2V task submitted: {task_id}")
# Poll for result
video_url = self._poll_task(mv_client, task_id)
return (video_url, task_id)
def _poll_task(self, mv_client, task_id, max_retries=180):
for i in range(max_retries):
status_res = mv_client.get_task_status(task_id)
task_status = status_res.get("output", {}).get("task_status")
if task_status == "Success":
urls = status_res.get("output", {}).get("urls", [])
if urls:
return urls[0]
raise Exception("Task succeeded but no video URL returned")
elif task_status == "Failure":
error = status_res.get("output", {}).get("error_message", "Unknown error")
raise Exception(f"Task failed: {error}")
elif task_status in ["Pending", "Running"]:
print(f"Task {task_id}: {task_status} ({i+1}/{max_retries})")
time.sleep(5)
else:
raise Exception(f"Unknown status: {task_status}")
raise Exception("Task timed out")
NODE_CLASS_MAPPINGS = {
"Sora_Text2Video": SoraText2VideoNode,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"Sora_Text2Video": "Sora Text2Video",
}
py/vidu_extend.py
+120
View File
@@ -0,0 +1,120 @@
"""
Vidu Extend - 视频延长模型
Models: viduq2-pro, viduq2-turbo
"""
import time
from .modelverse_api.client import ModelverseClient
from .modelverse_api.utils import image_to_base64
from comfy.comfy_types.node_typing import IO
MODELS = ["viduq2-pro", "viduq2-turbo"]
RESOLUTIONS = ["540p", "720p", "1080p"]
class ViduExtendNode:
"""
Vidu Extend - 视频延长
Models: viduq2-pro, viduq2-turbo
输入视频时长需4秒-1分钟，延长1-7秒
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"client": ("MODELVERSE_API_CLIENT",),
"model": (MODELS, {"default": "viduq2-turbo", "tooltip": "viduq2-pro: 效果好, viduq2-turbo: 生成快"}),
"video_url": (IO.STRING, {"default": "", "tooltip": "需要延长的视频URL (4秒-1分钟)"}),
"duration": (IO.INT, {"default": 5, "min": 1, "max": 7, "step": 1, "tooltip": "延长时长(秒), 1-7"}),
"resolution": (RESOLUTIONS, {"default": "720p", "tooltip": "分辨率"}),
},
"optional": {
"last_frame_image": (IO.IMAGE, {"tooltip": "延长到尾帧的参考图像"}),
"last_frame_url": (IO.STRING, {"default": "", "tooltip": "延长到尾帧的参考图像URL"}),
"prompt": (IO.STRING, {"multiline": True, "default": "", "tooltip": "延长提示词，控制延长内容"}),
}
}
RETURN_TYPES = (IO.STRING, IO.STRING)
RETURN_NAMES = ("url", "task_id")
FUNCTION = "generate"
CATEGORY = "UCLOUD_MODELVERSE/Vidu"
def generate(self, client, model, video_url, duration, resolution,
last_frame_image=None, last_frame_url="", prompt=""):
api_key = client.get("api_key")
if not api_key:
raise ValueError("API key is not set")
if not video_url or not video_url.strip():
raise ValueError("必须提供 video_url")
mv_client = ModelverseClient(api_key)
task_input = {
"video_url": video_url.strip(),
}
# Optional last frame
has_last_url = last_frame_url and last_frame_url.strip()
has_last_image = last_frame_image is not None
if has_last_url and has_last_image:
raise ValueError("尾帧：请提供 image 或 url，不能同时提供")
if has_last_url:
task_input["last_frame_url"] = last_frame_url.strip()
elif has_last_image:
task_input["last_frame_url"] = image_to_base64(last_frame_image)
if prompt and prompt.strip():
task_input["prompt"] = prompt.strip()
parameters = {
"vidu_type": "extend",
"duration": duration,
"resolution": resolution,
}
# Submit task
submit_res = mv_client.submit_task(model, task_input, parameters)
task_id = submit_res.get("output", {}).get("task_id")
if not task_id:
raise Exception(f"Failed to submit task: {submit_res}")
print(f"Vidu Extend task submitted: {task_id}")
# Poll for result
video_url_result = self._poll_task(mv_client, task_id)
return (video_url_result, task_id)
def _poll_task(self, mv_client, task_id, max_retries=180):
for i in range(max_retries):
status_res = mv_client.get_task_status(task_id)
task_status = status_res.get("output", {}).get("task_status")
if task_status == "Success":
urls = status_res.get("output", {}).get("urls", [])
if urls:
return urls[0]
raise Exception("Task succeeded but no video URL returned")
elif task_status == "Failure":
error = status_res.get("output", {}).get("error_message", "Unknown error")
raise Exception(f"Task failed: {error}")
elif task_status in ["Pending", "Running"]:
print(f"Task {task_id}: {task_status} ({i+1}/{max_retries})")
time.sleep(5)
else:
raise Exception(f"Unknown status: {task_status}")
raise Exception("Task timed out")
NODE_CLASS_MAPPINGS = {
"Vidu_Extend": ViduExtendNode,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"Vidu_Extend": "Vidu Extend",
}
py/vidu_i2v.py
+122
View File
@@ -0,0 +1,122 @@
"""
Vidu Img2Video - 图生视频模型
Models: viduq2-pro, viduq2-turbo
"""
import time
from .modelverse_api.client import ModelverseClient
from .modelverse_api.utils import image_to_base64
from comfy.comfy_types.node_typing import IO
MODELS = ["viduq2-pro", "viduq2-turbo"]
RESOLUTIONS = ["540p", "720p", "1080p"]
MOVEMENT_AMPLITUDES = ["auto", "small", "medium", "large"]
class ViduImg2VideoNode:
"""
Vidu Img2Video - 图生视频
Models: viduq2-pro (效果好，细节丰富), viduq2-turbo (效果好，生成快)
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"client": ("MODELVERSE_API_CLIENT",),
"model": (MODELS, {"default": "viduq2-pro", "tooltip": "viduq2-pro: 效果好细节丰富, viduq2-turbo: 生成快"}),
"duration": (IO.INT, {"default": 5, "min": 1, "max": 10, "step": 1, "tooltip": "视频时长(秒)"}),
"resolution": (RESOLUTIONS, {"default": "720p", "tooltip": "分辨率"}),
"movement_amplitude": (MOVEMENT_AMPLITUDES, {"default": "auto", "tooltip": "运动幅度"}),
},
"optional": {
"first_frame_image": (IO.IMAGE, {"tooltip": "首帧图片"}),
"first_frame_url": (IO.STRING, {"default": "", "tooltip": "首帧图片URL (与image二选一)"}),
"prompt": (IO.STRING, {"multiline": True, "default": "", "tooltip": "文本提示词，最长2000字符"}),
"seed": (IO.INT, {"default": 0, "min": 0, "max": 2147483647, "tooltip": "随机种子"}),
"bgm": (IO.BOOLEAN, {"default": False, "tooltip": "是否添加背景音乐"}),
}
}
RETURN_TYPES = (IO.STRING, IO.STRING)
RETURN_NAMES = ("url", "task_id")
FUNCTION = "generate"
CATEGORY = "UCLOUD_MODELVERSE/Vidu"
def generate(self, client, model, duration, resolution, movement_amplitude,
first_frame_image=None, first_frame_url="", prompt="", seed=0, bgm=False):
api_key = client.get("api_key")
if not api_key:
raise ValueError("API key is not set")
mv_client = ModelverseClient(api_key)
# Validate first frame input
has_url = first_frame_url and first_frame_url.strip()
has_image = first_frame_image is not None
if has_url and has_image:
raise ValueError("请提供 first_frame_image 或 first_frame_url，不能同时提供")
if not has_url and not has_image:
raise ValueError("必须提供 first_frame_image 或 first_frame_url")
task_input = {}
if has_url:
task_input["first_frame_url"] = first_frame_url.strip()
else:
task_input["first_frame_url"] = image_to_base64(first_frame_image)
if prompt and prompt.strip():
task_input["prompt"] = prompt.strip()
parameters = {
"vidu_type": "img2video",
"duration": duration,
"resolution": resolution,
"movement_amplitude": movement_amplitude,
"seed": seed,
"bgm": bgm,
}
# Submit task
submit_res = mv_client.submit_task(model, task_input, parameters)
task_id = submit_res.get("output", {}).get("task_id")
if not task_id:
raise Exception(f"Failed to submit task: {submit_res}")
print(f"Vidu I2V task submitted: {task_id}")
# Poll for result
video_url = self._poll_task(mv_client, task_id)
return (video_url, task_id)
def _poll_task(self, mv_client, task_id, max_retries=180):
for i in range(max_retries):
status_res = mv_client.get_task_status(task_id)
task_status = status_res.get("output", {}).get("task_status")
if task_status == "Success":
urls = status_res.get("output", {}).get("urls", [])
if urls:
return urls[0]
raise Exception("Task succeeded but no video URL returned")
elif task_status == "Failure":
error = status_res.get("output", {}).get("error_message", "Unknown error")
raise Exception(f"Task failed: {error}")
elif task_status in ["Pending", "Running"]:
print(f"Task {task_id}: {task_status} ({i+1}/{max_retries})")
time.sleep(5)
else:
raise Exception(f"Unknown status: {task_status}")
raise Exception("Task timed out")
NODE_CLASS_MAPPINGS = {
"Vidu_Img2Video": ViduImg2VideoNode,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"Vidu_Img2Video": "Vidu Img2Video",
}
py/vidu_ref2v.py
+136
View File
@@ -0,0 +1,136 @@
"""
Vidu Reference2Video - 参考生视频模型
Model: viduq2
支持1-7张参考图片，生成具备主体一致的视频
"""
import time
from .modelverse_api.client import ModelverseClient
from .modelverse_api.utils import image_to_base64
from comfy.comfy_types.node_typing import IO
ASPECT_RATIOS = ["16:9", "9:16", "3:4", "4:3", "1:1"]
RESOLUTIONS = ["540p", "720p", "1080p"]
class ViduReference2VideoNode:
"""
Vidu Reference2Video - 参考生视频
Model: viduq2
支持1-7张参考图片，生成具备主体一致的视频
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"client": ("MODELVERSE_API_CLIENT",),
"prompt": (IO.STRING, {"multiline": True, "default": "make it dance", "tooltip": "文本提示词，最长2000字符"}),
"duration": (IO.INT, {"default": 5, "min": 1, "max": 10, "step": 1, "tooltip": "视频时长(秒)"}),
"aspect_ratio": (ASPECT_RATIOS, {"default": "16:9", "tooltip": "长宽比"}),
"resolution": (RESOLUTIONS, {"default": "720p", "tooltip": "分辨率"}),
},
"optional": {
"image1": (IO.IMAGE, {"tooltip": "参考图片1"}),
"image2": (IO.IMAGE, {"tooltip": "参考图片2"}),
"image3": (IO.IMAGE, {"tooltip": "参考图片3"}),
"image4": (IO.IMAGE, {"tooltip": "参考图片4"}),
"image5": (IO.IMAGE, {"tooltip": "参考图片5"}),
"image6": (IO.IMAGE, {"tooltip": "参考图片6"}),
"image7": (IO.IMAGE, {"tooltip": "参考图片7"}),
"image_urls": (IO.STRING, {"default": "", "multiline": True, "tooltip": "参考图片URL列表，每行一个"}),
"seed": (IO.INT, {"default": 0, "min": 0, "max": 2147483647, "tooltip": "随机种子"}),
"bgm": (IO.BOOLEAN, {"default": False, "tooltip": "是否添加背景音乐"}),
}
}
RETURN_TYPES = (IO.STRING, IO.STRING)
RETURN_NAMES = ("url", "task_id")
FUNCTION = "generate"
CATEGORY = "UCLOUD_MODELVERSE/Vidu"
def generate(self, client, prompt, duration, aspect_ratio, resolution,
image1=None, image2=None, image3=None, image4=None,
image5=None, image6=None, image7=None,
image_urls="", seed=0, bgm=False):
api_key = client.get("api_key")
if not api_key:
raise ValueError("API key is not set")
mv_client = ModelverseClient(api_key)
# Collect images
images = []
for img in [image1, image2, image3, image4, image5, image6, image7]:
if img is not None:
images.append(image_to_base64(img))
# Add URL images
if image_urls and image_urls.strip():
for url in image_urls.strip().split('\n'):
url = url.strip()
if url:
images.append(url)
if not images:
raise ValueError("至少需要提供1张参考图片")
if len(images) > 7:
print(f"Warning: 最多支持7张参考图片，当前{len(images)}张，将只使用前7张")
images = images[:7]
task_input = {
"images": images,
"prompt": prompt,
}
parameters = {
"vidu_type": "reference2video",
"duration": duration,
"aspect_ratio": aspect_ratio,
"resolution": resolution,
"seed": seed,
"bgm": bgm,
}
# Submit task
submit_res = mv_client.submit_task("viduq2", task_input, parameters)
task_id = submit_res.get("output", {}).get("task_id")
if not task_id:
raise Exception(f"Failed to submit task: {submit_res}")
print(f"Vidu Ref2V task submitted: {task_id}")
# Poll for result
video_url = self._poll_task(mv_client, task_id)
return (video_url, task_id)
def _poll_task(self, mv_client, task_id, max_retries=180):
for i in range(max_retries):
status_res = mv_client.get_task_status(task_id)
task_status = status_res.get("output", {}).get("task_status")
if task_status == "Success":
urls = status_res.get("output", {}).get("urls", [])
if urls:
return urls[0]
raise Exception("Task succeeded but no video URL returned")
elif task_status == "Failure":
error = status_res.get("output", {}).get("error_message", "Unknown error")
raise Exception(f"Task failed: {error}")
elif task_status in ["Pending", "Running"]:
print(f"Task {task_id}: {task_status} ({i+1}/{max_retries})")
time.sleep(5)
else:
raise Exception(f"Unknown status: {task_status}")
raise Exception("Task timed out")
NODE_CLASS_MAPPINGS = {
"Vidu_Reference2Video": ViduReference2VideoNode,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"Vidu_Reference2Video": "Vidu Reference2Video",
}
py/vidu_startend2v.py
+142
View File
@@ -0,0 +1,142 @@
"""
Vidu StartEnd2Video - 首尾帧生视频模型
Models: viduq2-pro, viduq2-turbo
"""
import time
from .modelverse_api.client import ModelverseClient
from .modelverse_api.utils import image_to_base64
from comfy.comfy_types.node_typing import IO
MODELS = ["viduq2-pro", "viduq2-turbo"]
RESOLUTIONS = ["540p", "720p", "1080p"]
MOVEMENT_AMPLITUDES = ["auto", "small", "medium", "large"]
class ViduStartEnd2VideoNode:
"""
Vidu StartEnd2Video - 首尾帧生视频
Models: viduq2-pro (效果好，细节丰富), viduq2-turbo (效果好，生成快)
注意：首尾帧分辨率需相近 (0.8~1.25)
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"client": ("MODELVERSE_API_CLIENT",),
"model": (MODELS, {"default": "viduq2-pro", "tooltip": "viduq2-pro: 效果好细节丰富, viduq2-turbo: 生成快"}),
"duration": (IO.INT, {"default": 5, "min": 1, "max": 8, "step": 1, "tooltip": "视频时长(秒), 1-8"}),
"resolution": (RESOLUTIONS, {"default": "720p", "tooltip": "分辨率"}),
"movement_amplitude": (MOVEMENT_AMPLITUDES, {"default": "auto", "tooltip": "运动幅度"}),
},
"optional": {
"first_frame_image": (IO.IMAGE, {"tooltip": "首帧图片"}),
"first_frame_url": (IO.STRING, {"default": "", "tooltip": "首帧图片URL"}),
"last_frame_image": (IO.IMAGE, {"tooltip": "尾帧图片"}),
"last_frame_url": (IO.STRING, {"default": "", "tooltip": "尾帧图片URL"}),
"prompt": (IO.STRING, {"multiline": True, "default": "", "tooltip": "文本提示词，最长2000字符"}),
"seed": (IO.INT, {"default": 0, "min": 0, "max": 2147483647, "tooltip": "随机种子"}),
"bgm": (IO.BOOLEAN, {"default": False, "tooltip": "是否添加背景音乐"}),
}
}
RETURN_TYPES = (IO.STRING, IO.STRING)
RETURN_NAMES = ("url", "task_id")
FUNCTION = "generate"
CATEGORY = "UCLOUD_MODELVERSE/Vidu"
def generate(self, client, model, duration, resolution, movement_amplitude,
first_frame_image=None, first_frame_url="",
last_frame_image=None, last_frame_url="",
prompt="", seed=0, bgm=False):
api_key = client.get("api_key")
if not api_key:
raise ValueError("API key is not set")
mv_client = ModelverseClient(api_key)
# Validate first frame
has_first_url = first_frame_url and first_frame_url.strip()
has_first_image = first_frame_image is not None
if has_first_url and has_first_image:
raise ValueError("首帧：请提供 image 或 url，不能同时提供")
if not has_first_url and not has_first_image:
raise ValueError("必须提供首帧图片")
# Validate last frame
has_last_url = last_frame_url and last_frame_url.strip()
has_last_image = last_frame_image is not None
if has_last_url and has_last_image:
raise ValueError("尾帧：请提供 image 或 url，不能同时提供")
if not has_last_url and not has_last_image:
raise ValueError("必须提供尾帧图片")
task_input = {}
# First frame
if has_first_url:
task_input["first_frame_url"] = first_frame_url.strip()
else:
task_input["first_frame_url"] = image_to_base64(first_frame_image)
# Last frame
if has_last_url:
task_input["last_frame_url"] = last_frame_url.strip()
else:
task_input["last_frame_url"] = image_to_base64(last_frame_image)
if prompt and prompt.strip():
task_input["prompt"] = prompt.strip()
parameters = {
"vidu_type": "start-end2video",
"duration": duration,
"resolution": resolution,
"movement_amplitude": movement_amplitude,
"seed": seed,
"bgm": bgm,
}
# Submit task
submit_res = mv_client.submit_task(model, task_input, parameters)
task_id = submit_res.get("output", {}).get("task_id")
if not task_id:
raise Exception(f"Failed to submit task: {submit_res}")
print(f"Vidu StartEnd2V task submitted: {task_id}")
# Poll for result
video_url = self._poll_task(mv_client, task_id)
return (video_url, task_id)
def _poll_task(self, mv_client, task_id, max_retries=180):
for i in range(max_retries):
status_res = mv_client.get_task_status(task_id)
task_status = status_res.get("output", {}).get("task_status")
if task_status == "Success":
urls = status_res.get("output", {}).get("urls", [])
if urls:
return urls[0]
raise Exception("Task succeeded but no video URL returned")
elif task_status == "Failure":
error = status_res.get("output", {}).get("error_message", "Unknown error")
raise Exception(f"Task failed: {error}")
elif task_status in ["Pending", "Running"]:
print(f"Task {task_id}: {task_status} ({i+1}/{max_retries})")
time.sleep(5)
else:
raise Exception(f"Unknown status: {task_status}")
raise Exception("Task timed out")
NODE_CLASS_MAPPINGS = {
"Vidu_StartEnd2Video": ViduStartEnd2VideoNode,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"Vidu_StartEnd2Video": "Vidu StartEnd2Video",
}
py/vidu_t2v.py
+101
View File
@@ -0,0 +1,101 @@
"""
Vidu Text2Video - 文生视频模型
Model: viduq2
"""
import time
from .modelverse_api.client import ModelverseClient
from comfy.comfy_types.node_typing import IO
ASPECT_RATIOS = ["16:9", "9:16", "3:4", "4:3", "1:1"]
RESOLUTIONS = ["540p", "720p", "1080p"]
class ViduText2VideoNode:
"""
Vidu Text2Video - 文生视频
Model: viduq2
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"client": ("MODELVERSE_API_CLIENT",),
"prompt": (IO.STRING, {"multiline": True, "default": "A beautiful sunset over the ocean", "tooltip": "文本提示词，最长2000字符"}),
"duration": (IO.INT, {"default": 5, "min": 1, "max": 10, "step": 1, "tooltip": "视频时长(秒)"}),
"aspect_ratio": (ASPECT_RATIOS, {"default": "16:9", "tooltip": "长宽比"}),
"resolution": (RESOLUTIONS, {"default": "720p", "tooltip": "分辨率"}),
},
"optional": {
"seed": (IO.INT, {"default": 0, "min": 0, "max": 2147483647, "tooltip": "随机种子，0表示随机"}),
"guidance_scale": (IO.FLOAT, {"default": 7.5, "min": 1.0, "max": 20.0, "step": 0.5, "tooltip": "引导系数"}),
"bgm": (IO.BOOLEAN, {"default": False, "tooltip": "是否添加背景音乐"}),
}
}
RETURN_TYPES = (IO.STRING, IO.STRING)
RETURN_NAMES = ("url", "task_id")
FUNCTION = "generate"
CATEGORY = "UCLOUD_MODELVERSE/Vidu"
def generate(self, client, prompt, duration, aspect_ratio, resolution, seed=0, guidance_scale=7.5, bgm=False):
api_key = client.get("api_key")
if not api_key:
raise ValueError("API key is not set")
mv_client = ModelverseClient(api_key)
task_input = {"prompt": prompt}
parameters = {
"vidu_type": "text2video",
"duration": duration,
"aspect_ratio": aspect_ratio,
"resolution": resolution,
"seed": seed,
"guidance_scale": guidance_scale,
"bgm": bgm,
}
# Submit task
submit_res = mv_client.submit_task("viduq2", task_input, parameters)
task_id = submit_res.get("output", {}).get("task_id")
if not task_id:
raise Exception(f"Failed to submit task: {submit_res}")
print(f"Vidu T2V task submitted: {task_id}")
# Poll for result
video_url = self._poll_task(mv_client, task_id)
return (video_url, task_id)
def _poll_task(self, mv_client, task_id, max_retries=180):
for i in range(max_retries):
status_res = mv_client.get_task_status(task_id)
task_status = status_res.get("output", {}).get("task_status")
if task_status == "Success":
urls = status_res.get("output", {}).get("urls", [])
if urls:
return urls[0]
raise Exception("Task succeeded but no video URL returned")
elif task_status == "Failure":
error = status_res.get("output", {}).get("error_message", "Unknown error")
raise Exception(f"Task failed: {error}")
elif task_status in ["Pending", "Running"]:
print(f"Task {task_id}: {task_status} ({i+1}/{max_retries})")
time.sleep(5)
else:
raise Exception(f"Unknown status: {task_status}")
raise Exception("Task timed out")
NODE_CLASS_MAPPINGS = {
"Vidu_Text2Video": ViduText2VideoNode,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"Vidu_Text2Video": "Vidu Text2Video",
}
py/wan_ai_i2v.py
+5 -58
View File
@@ -1,29 +1,8 @@
import time
import requests
import os
import tempfile
import io
from .modelverse_api.client import ModelverseClient
from .modelverse_api.utils import image_to_base64
from comfy.comfy_types.node_typing import IO
# Import VideoFromFile class for ComfyUI video handling
try:
# Try the newer import path first
from comfy_extras.nodes_video import VideoFromFile
except ImportError:
try:
# Fallback to older import path
from comfy.model_management import VideoFromFile
except ImportError:
# Final fallback - create a simple wrapper
class VideoFromFile:
def __init__(self, video_io):
self.video_io = video_io
self.video_data = video_io.getvalue() if hasattr(video_io, 'getvalue') else video_io
def get_dimensions(self):
# Return default dimensions if we can't determine them
return (1280, 720) # width, height
class Modelverse_WanAII2V:
def __init__(self):
@@ -47,8 +26,8 @@ class Modelverse_WanAII2V:
}
}
RETURN_TYPES = (IO.VIDEO,)
RETURN_NAMES = ("video",)
RETURN_TYPES = (IO.STRING, IO.STRING)
RETURN_NAMES = ("url", "task_id")
FUNCTION = "generate_video"
CATEGORY = "UCLOUD_MODELVERSE"
@@ -155,40 +134,8 @@ class Modelverse_WanAII2V:
if not video_url:
raise Exception(f"Task timed out after {max_retries * 5} seconds")
# 3. Download video
print(f"Downloading video from: {video_url}")
try:
response = requests.get(video_url, timeout=60)
response.raise_for_status()
except requests.exceptions.RequestException as e:
raise Exception(f"Failed to download video: {e}")
# Extract video data
video_data = response.content
if not video_data:
raise Exception("No video data was returned")
print("Video generation completed successfully")
print(f"Downloaded video data size: {len(video_data)} bytes")
# Convert video data to BytesIO object
video_io = io.BytesIO(video_data)
# Ensure the BytesIO object is at the beginning
video_io.seek(0)
# Return VideoFromFile object
try:
video_obj = VideoFromFile(video_io)
print("VideoFromFile object created successfully")
return (video_obj,)
except Exception as e:
print(f"Error creating VideoFromFile object: {e}")
# If VideoFromFile fails, try returning the raw video data
# This is a fallback that might work with some ComfyUI versions
return (video_data,)
return (video_url, task_id)
NODE_CLASS_MAPPINGS = {
py/wan_ai_t2v.py
+5 -46
View File
@@ -1,20 +1,7 @@
import time
import requests
import os
import tempfile
import io
from .modelverse_api.client import ModelverseClient
from comfy.comfy_types.node_typing import IO
# Import VideoFromFile class for ComfyUI video handling
try:
# Try the newer import path first
from comfy_extras.nodes_video import VideoFromFile
except ImportError:
try:
# Fallback to older import path
from comfy.model_management import VideoFromFile
except:
raise ImportError("VideoFromFile not found")
class Modelverse_WanAIT2V:
def __init__(self):
@@ -33,8 +20,8 @@ class Modelverse_WanAIT2V:
}
}
RETURN_TYPES = (IO.VIDEO,)
RETURN_NAMES = ("video",)
RETURN_TYPES = (IO.STRING, IO.STRING)
RETURN_NAMES = ("url", "task_id")
FUNCTION = "generate_video"
CATEGORY = "UCLOUD_MODELVERSE"
@@ -81,36 +68,8 @@ class Modelverse_WanAIT2V:
time.sleep(5) # Wait for 5 seconds before polling again
else:
raise Exception(f"Unknown task status: {task_status}")
# 3. Download video
response = requests.get(video_url)
response.raise_for_status()
# Extract video data
video_data = response.content
if not video_data:
raise Exception("No video data was returned")
print("Video generation completed successfully")
print(f"Downloaded video data size: {len(video_data)} bytes")
# Convert video data to BytesIO object
video_io = io.BytesIO(video_data)
# Ensure the BytesIO object is at the beginning
video_io.seek(0)
# Return VideoFromFile object
try:
video_obj = VideoFromFile(video_io)
print("VideoFromFile object created successfully")
return (video_obj,)
except Exception as e:
print(f"Error creating VideoFromFile object: {e}")
# If VideoFromFile fails, try returning the raw video data
# This is a fallback that might work with some ComfyUI versions
return (video_data,)
return (video_url, task_id)
NODE_CLASS_MAPPINGS = {
pyproject.toml
+1 -1
View File
@@ -1,7 +1,7 @@
[project]
name = "ucloud-comfyui"
description = "This is a custom node for ComfyUI that allows you to use the UCloud API directly in ComfyUI. UCloud is a high-performance AI computing platform providing powerful image generation models and industry-leading inference speeds. It offers a wide range of AI model options, including popular text-to-image and image editing models, ready to use without local GPU setup."
version = "1.0.7"
version = "1.0.8"
license = {file = "LICENSE"}
dependencies = ["requests", "pillow"]
requirements.txt
+2 -1
View File
@@ -1,3 +1,4 @@
requests >= 2.32.4
pillow >= 11
openai >= 1.107.0
openai >= 1.107.0
GitPython==3.1.45