"""MiniMax-H3 video generation through a New API gateway.""" import json import os import tempfile from typing import Optional import folder_paths from comfy_api.latest import InputImpl, io from ..clients.minimax_h3_client import MiniMaxH3Client from ..utils.config import get_base_url_by_route from ..utils.image_utils import tensor_to_pil from ..utils.minimax_h3_media import ( MAX_REFERENCE_AUDIOS, MAX_REFERENCE_IMAGES, MAX_REFERENCE_VIDEOS, validate_image, validate_reference_audios, validate_reference_videos, ) from ..utils.r2_uploader import upload_audio, upload_image, upload_video MODEL_ID = "MiniMax-H3" MODEL_MAX_ID = "MiniMax-H3-MAX" MODEL_OPTIONS = [MODEL_ID, MODEL_MAX_ID] MODE_TEXT = "文生视频" MODE_FIRST = "首帧图生视频" MODE_LAST = "尾帧图生视频" MODE_FIRST_LAST = "首尾帧生视频" MODE_REFERENCE = "参考素材生视频" RESOLUTION_OPTIONS = ["2K", "768P", "480P"] MODEL_RESOLUTIONS = { MODEL_ID: {"768P", "2K"}, MODEL_MAX_ID: {"480P", "768P"}, } MODEL_DURATION_RANGES = { MODEL_ID: (4, 15), MODEL_MAX_ID: (5, 15), } RATIO_OPTIONS = ["16:9", "21:9", "4:3", "1:1", "3:4", "9:16"] REFERENCE_RATIO_OPTIONS = ["adaptive", *RATIO_OPTIONS] MAX_PROMPT_CHARS = 7000 MAX_REQUEST_BODY_BYTES = 64 * 1024 * 1024 MAX_REFERENCE_MATERIALS = 12 MAX_SEED = 2**31 - 1 def _validate_prompt(prompt: str) -> str: prompt = (prompt or "").strip() if not prompt: raise ValueError("MiniMax H3 提示词不能为空。") if len(prompt) > MAX_PROMPT_CHARS: raise ValueError( f"MiniMax H3 提示词最多 {MAX_PROMPT_CHARS} 字符，当前为 {len(prompt)} 字符。" ) return prompt def _validate_seed(seed: int) -> int: if isinstance(seed, bool) or not isinstance(seed, int) or not 0 <= seed <= MAX_SEED: raise ValueError(f"MiniMax H3 seed 必须是 0～{MAX_SEED} 的整数。") return seed def _validate_generation_options( model: str, resolution: str, duration: int, mode: Optional[str] = None, ) -> None: if model not in MODEL_OPTIONS: raise ValueError(f"不支持的 MiniMax 模型：{model}") allowed_resolutions = MODEL_RESOLUTIONS[model] if resolution not in allowed_resolutions: allowed_text = "、".join( value for value in RESOLUTION_OPTIONS if value in allowed_resolutions ) raise ValueError(f"{model} 分辨率仅支持 {allowed_text}。") minimum_duration, maximum_duration = MODEL_DURATION_RANGES[model] if ( isinstance(duration, bool) or not isinstance(duration, int) or not minimum_duration <= duration <= maximum_duration ): raise ValueError( f"{model} 时长必须是 {minimum_duration}～{maximum_duration} 的整数。" ) if model == MODEL_MAX_ID and mode == MODE_REFERENCE: raise ValueError("MiniMax-H3-MAX 不支持图片、视频或音频参考素材模式。") def _validate_reference_counts( reference_images, reference_videos, reference_audios, ) -> None: if len(reference_images) > MAX_REFERENCE_IMAGES: raise ValueError(f"参考图片最多 {MAX_REFERENCE_IMAGES} 张。") if len(reference_videos) > MAX_REFERENCE_VIDEOS: raise ValueError(f"参考视频最多 {MAX_REFERENCE_VIDEOS} 个。") if len(reference_audios) > MAX_REFERENCE_AUDIOS: raise ValueError(f"参考音频最多 {MAX_REFERENCE_AUDIOS} 个。") total = len(reference_images) + len(reference_videos) + len(reference_audios) if total > MAX_REFERENCE_MATERIALS: raise ValueError( f"参考图片、视频和音频合计最多 {MAX_REFERENCE_MATERIALS} 个，当前为 {total} 个。" ) def _validate_image_tensor(image, label: str): if image is None: raise ValueError(f"{label}不能为空。") ndim = getattr(image, "dim", lambda: None)() if ndim == 4 and int(image.shape[0]) != 1: raise ValueError(f"{label}仅支持单张图片，当前批次包含 {int(image.shape[0])} 张。") def _image_to_pil(image, label: str): _validate_image_tensor(image, label) images = tensor_to_pil(image) if not images: raise ValueError(f"无法读取{label}。") result = images[0] if result.mode not in ("RGB", "RGBA"): result = result.convert("RGB") validate_image(result, label) return result def _validate_body_size(body: dict) -> None: body_size = len( json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode("utf-8") ) if body_size > MAX_REQUEST_BODY_BYTES: raise ValueError( f"MiniMax H3 请求体大小 {body_size / 1024 / 1024:.2f} MB 超过 64 MB 限制。" ) def build_request_body( *, prompt: str, resolution: str, duration: int, mode: str, model: str = MODEL_ID, seed: int = 0, ratio: Optional[str] = None, first_url: Optional[str] = None, last_url: Optional[str] = None, reference_image_urls: Optional[list[str]] = None, reference_video_urls: Optional[list[str]] = None, reference_audio_urls: Optional[list[str]] = None, ) -> dict: """Build and validate the documented MiniMax H3 / H3 Max request body.""" prompt = _validate_prompt(prompt) seed = _validate_seed(seed) _validate_generation_options(model, resolution, duration, mode) reference_image_urls = [url for url in (reference_image_urls or []) if url] reference_video_urls = [url for url in (reference_video_urls or []) if url] reference_audio_urls = [url for url in (reference_audio_urls or []) if url] _validate_reference_counts( reference_image_urls, reference_video_urls, reference_audio_urls, ) has_first_last_material = bool(first_url or last_url) has_reference_material = bool( reference_image_urls or reference_video_urls or reference_audio_urls ) content = [{"type": "text", "text": prompt}] if mode == MODE_TEXT: if has_first_last_material or has_reference_material: raise ValueError("文生视频不能同时使用首尾帧或参考素材。") if ratio not in RATIO_OPTIONS: raise ValueError("MiniMax H3 文生视频必须选择具体宽高比，不能使用 adaptive。") request_ratio = ratio elif mode == MODE_FIRST: if last_url or has_reference_material: raise ValueError("首帧图生视频不能同时使用尾帧或参考素材。") if not first_url: raise ValueError("首帧图生视频必须连接首帧图片。") content.append({ "type": "image_url", "image_url": {"url": first_url}, "role": "first_frame", }) request_ratio = "adaptive" elif mode == MODE_LAST: if first_url or has_reference_material: raise ValueError("尾帧图生视频不能同时使用首帧或参考素材。") if not last_url: raise ValueError("尾帧图生视频必须连接尾帧图片。") content.append({ "type": "image_url", "image_url": {"url": last_url}, "role": "last_frame", }) request_ratio = "adaptive" elif mode == MODE_FIRST_LAST: if has_reference_material: raise ValueError("首尾帧模式和参考素材模式不能混用。") if not first_url or not last_url: raise ValueError("首尾帧生视频必须同时连接首帧和尾帧图片。") content.extend([ { "type": "image_url", "image_url": {"url": first_url}, "role": "first_frame", }, { "type": "image_url", "image_url": {"url": last_url}, "role": "last_frame", }, ]) request_ratio = "adaptive" elif mode == MODE_REFERENCE: if has_first_last_material: raise ValueError("参考素材模式和首尾帧模式不能混用。") if not has_reference_material: raise ValueError("参考素材生视频至少需要一张图片、一个视频或一段音频。") content.extend( { "type": "image_url", "image_url": {"url": url}, "role": "reference_image", } for url in reference_image_urls ) content.extend( { "type": "video_url", "video_url": {"url": url}, "role": "reference_video", } for url in reference_video_urls ) content.extend( { "type": "audio_url", "audio_url": {"url": url}, "role": "reference_audio", } for url in reference_audio_urls ) request_ratio = ratio or "adaptive" if request_ratio not in REFERENCE_RATIO_OPTIONS: raise ValueError(f"MiniMax H3 参考素材模式不支持宽高比：{request_ratio}") else: raise ValueError(f"不支持的 MiniMax H3 生成模式：{mode}") body = { "model": model, "content": content, "resolution": resolution, "duration": duration, "ratio": request_ratio, "seed": seed, } _validate_body_size(body) return body def _build_mode_input(): reference_images = io.Autogrow.Input( "参考图片组", template=io.Autogrow.TemplateNames( input=io.Image.Input("参考图片"), names=[f"参考图片{i}" for i in range(1, MAX_REFERENCE_IMAGES + 1)], min=0, ), tooltip="可动态连接，最多 9 张。", ) reference_videos = io.Autogrow.Input( "参考视频组", template=io.Autogrow.TemplateNames( input=io.Video.Input("参考视频"), names=[f"参考视频{i}" for i in range(1, MAX_REFERENCE_VIDEOS + 1)], min=0, ), tooltip="可动态连接，最多 3 个；每段 2～15 秒，总时长不超过 15 秒。", ) reference_audios = io.Autogrow.Input( "参考音频组", template=io.Autogrow.TemplateNames( input=io.Audio.Input("参考音频"), names=[f"参考音频{i}" for i in range(1, MAX_REFERENCE_AUDIOS + 1)], min=0, ), tooltip="可动态连接，最多 3 段；每段 2～15 秒，总时长不超过 15 秒。", ) return io.DynamicCombo.Input( "生成模式", options=[ io.DynamicCombo.Option( MODE_TEXT, [ io.Combo.Input( "宽高比", options=RATIO_OPTIONS, default="16:9", tooltip="文生视频必须使用具体比例，不能使用 adaptive。", ) ], ), io.DynamicCombo.Option( MODE_FIRST, [io.Image.Input("首帧图片", tooltip="输入图片决定视频比例。")], ), io.DynamicCombo.Option( MODE_LAST, [io.Image.Input("尾帧图片", tooltip="输入图片决定视频比例。")], ), io.DynamicCombo.Option( MODE_FIRST_LAST, [ io.Image.Input("首帧图片"), io.Image.Input("尾帧图片"), ], ), io.DynamicCombo.Option( MODE_REFERENCE, [ reference_images, reference_videos, reference_audios, io.Combo.Input( "宽高比", options=REFERENCE_RATIO_OPTIONS, default="adaptive", tooltip="参考素材模式默认 adaptive，也可指定具体比例。", ), ], ), ], tooltip="首尾帧模式与参考素材模式互斥。", ) def _collect_autogrow(value) -> list: """Collect connected values while tolerating a legacy single input value.""" if value is None: return [] if isinstance(value, dict): return [item for item in value.values() if item is not None] return [value] def _make_progress_callbacks(): try: from comfy.utils import ProgressBar pbar = ProgressBar(100) except Exception: pbar = None last_progress = -1 def set_progress(value: int): nonlocal last_progress value = max(0, min(100, int(value))) if value <= last_progress: return last_progress = value if pbar: pbar.update_absolute(value, 100) def on_stage(stage: str): if stage == "submitting": print("[MiniMax H3] 正在创建任务...") set_progress(0) elif stage.startswith("submitted:"): print(f"[MiniMax H3] 任务已进入队列：{stage.split(':', 1)[1]}") elif stage == "downloading": print("[MiniMax H3] 生成成功，正在立即下载临时 CDN 视频...") elif stage == "done": set_progress(100) def on_progress(progress: int): # New API returns the authoritative task percentage in data.progress. # Mirror it directly in ComfyUI while preventing stale poll responses # from moving the node progress bar backwards. set_progress(progress) return on_stage, on_progress class MiniMaxH3Video(io.ComfyNode): """MiniMax H3 / H3 Max text, frame, and reference video generation.""" @classmethod def define_schema(cls): return io.Schema( node_id="MiniMaxH3Video", display_name="MiniMax H3 / H3 Max 视频生成", category="comfyui_o1key/Video", description=( "通过 New API 网关调用 MiniMax-H3 或 MiniMax-H3-MAX；H3 支持多模态参考，H3 Max 支持文生和首尾帧模式。" ), inputs=[ io.String.Input( "提示词", multiline=True, default="", placeholder="描述画面、动作、镜头与声音...", tooltip="必填，最多 7000 字符。", ), _build_mode_input(), io.Combo.Input( "分辨率", options=RESOLUTION_OPTIONS, default="2K", ), io.Int.Input( "时长", default=5, min=4, max=15, step=1, display_mode=io.NumberDisplay.slider, tooltip="H3 支持 4～15 秒；H3 Max 支持 5～15 秒。", ), io.Combo.Input( "模型", options=MODEL_OPTIONS, default=MODEL_ID, tooltip="H3 支持 768P/2K 和多模态参考；H3 Max 支持 480P/768P，不支持参考素材模式。", ), io.Int.Input( "seed", default=0, min=0, max=MAX_SEED, step=1, display_mode=io.NumberDisplay.number, control_after_generate=io.ControlAfterGenerate.randomize, tooltip="原生随机种子；相同参数与 seed 可用于复现结果。", ), ], outputs=[io.Video.Output(display_name="视频")], not_idempotent=True, search_aliases=["MiniMax H3", "MiniMax H3 Max", "海螺视频", "H3 视频"], accept_all_inputs=True, ) @classmethod async def execute( cls, 提示词, 生成模式, 分辨率, 时长, 模型=MODEL_ID, seed=0, **_kwargs, ) -> io.NodeOutput: prompt = _validate_prompt(提示词) if not isinstance(生成模式, dict): raise ValueError("MiniMax H3 生成模式参数无效。") mode = 生成模式.get("生成模式") _validate_generation_options(模型, 分辨率, int(时长), mode) base_url = get_base_url_by_route() first_url = None last_url = None reference_image_urls = [] reference_video_urls = [] reference_audio_urls = [] if mode == MODE_FIRST: first_image = 生成模式.get("首帧图片") first_url = await upload_image( _image_to_pil(first_image, "首帧图片"), base_url=base_url, ) elif mode == MODE_LAST: last_image = 生成模式.get("尾帧图片") last_url = await upload_image( _image_to_pil(last_image, "尾帧图片"), base_url=base_url, ) elif mode == MODE_FIRST_LAST: first_image = 生成模式.get("首帧图片") last_image = 生成模式.get("尾帧图片") first_url = await upload_image( _image_to_pil(first_image, "首帧图片"), base_url=base_url, ) last_url = await upload_image( _image_to_pil(last_image, "尾帧图片"), base_url=base_url, ) elif mode == MODE_REFERENCE: reference_images = _collect_autogrow( 生成模式.get("参考图片组", 生成模式.get("参考图片")) ) reference_videos = _collect_autogrow( 生成模式.get("参考视频组", 生成模式.get("参考视频")) ) reference_audios = _collect_autogrow( 生成模式.get("参考音频组", 生成模式.get("参考音频")) ) _validate_reference_counts( reference_images, reference_videos, reference_audios, ) if not reference_images and not reference_videos and not reference_audios: raise ValueError("参考素材生视频至少需要连接一种参考素材。") reference_pil_images = [ _image_to_pil(image, f"参考图片{index}") for index, image in enumerate(reference_images, start=1) ] validate_reference_videos(reference_videos) validate_reference_audios(reference_audios) for image in reference_pil_images: reference_image_urls.append(await upload_image(image, base_url=base_url)) for video in reference_videos: reference_video_urls.append(await upload_video(video, base_url=base_url)) for audio in reference_audios: reference_audio_urls.append(await upload_audio(audio, base_url=base_url)) body = build_request_body( prompt=prompt, resolution=分辨率, duration=int(时长), mode=mode, model=模型, seed=int(seed), ratio=生成模式.get("宽高比"), first_url=first_url, last_url=last_url, reference_image_urls=reference_image_urls, reference_video_urls=reference_video_urls, reference_audio_urls=reference_audio_urls, ) temp_dir = folder_paths.get_temp_directory() os.makedirs(temp_dir, exist_ok=True) fd, save_path = tempfile.mkstemp( suffix=".mp4", prefix="minimax_h3_", dir=temp_dir, ) os.close(fd) on_stage, on_progress = _make_progress_callbacks() client = MiniMaxH3Client(base_url=base_url) try: result_path, _task_id = await client.generate_async( body=body, save_path=save_path, on_stage=on_stage, on_progress=on_progress, ) return io.NodeOutput(InputImpl.VideoFromFile(result_path)) except BaseException: try: if os.path.isfile(save_path): os.remove(save_path) except OSError: pass raise NODE_CLASS_MAPPINGS = {"MiniMaxH3Video": MiniMaxH3Video} NODE_DISPLAY_NAME_MAPPINGS = {"MiniMaxH3Video": "MiniMax H3 / H3 Max 视频生成"}