""" Seedance 2.0 / 2.5 自动过审节点（xinhankr/可美线路） 与 Seedance / SeedanceMultiModal 的差异： - 模型名用 seedance-2.0 / seedance-2.0-fast / seedance-2.0-mini（fast、mini 仅支持 480p/720p） - 界面仅保留多模态、首尾帧两种生成模式 - 多模态无素材时自动作为文生视频；首尾帧根据尾帧是否连接自动选择首帧/首尾帧 - 参考素材可自动创建为素材，也可直接使用手动填写的 asset ID - 首/尾帧和 asset:// 素材使用 content body 端点不变：POST /v1/video/generations、GET /v1/video/generations/{task_id} """ import asyncio import io as py_io import json import os import re import tempfile import aiohttp from PIL import Image from ..clients.seedance_client import SeedanceClient from ..clients.seedance_element_client import SeedanceElementClient from ..utils.config import get_base_url_by_route from ..utils.r2_uploader import upload_image, upload_video, upload_audio from ..utils.image_utils import pil_to_tensor, tensor_to_pil from ..utils.video_task import format_seedance_generation_error from ..utils.o1key_video_catalog import ( SEEDANCE_CAPABILITIES, SEEDANCE_MODEL_MATRIX, SEEDANCE_REFERENCE_AUDIO_MAX_BYTES, SEEDANCE_REFERENCE_IMAGE_MAX_BYTES, SEEDANCE_REFERENCE_VIDEO_MAX_BYTES, normalize_seedance_parameters, resolve_seedance_model, validate_seedance_media_counts, validate_seedance_reference_dimensions, ) from ..utils.o1key_video_jobs import build_seedance_video_body from comfy_api.latest import InputImpl, io _BASE_MODELS = ["seedance 2.0", "seedance 2.0 fast", "seedance 2.0 mini", "seedance 2.5"] _MODEL_ROUTES = ["海外", "国内"] _LEGACY_MODEL_ROUTES = { "海外HC": "海外", "海外破限高并发": "海外", "海外破限": "海外", "海外破限标准": "海外", "海外标准": "海外", } _ASSET_CREATION_MODES = {"国内": "Doubao", "海外": "HC"} _CANONICAL_MODELS = { "seedance 2.0": "seedance-2.0", "seedance 2.0 fast": "seedance-2.0-fast", "seedance 2.0 mini": "seedance-2.0-mini", "seedance 2.5": "seedance-2.5", } _CANONICAL_ROUTES = {"国内": "domestic", "海外": "overseas_hc"} # 主模型 × 模型线路 → 实际模型ID 映射表 _MODEL_MATRIX = { (display_model, display_route): SEEDANCE_MODEL_MATRIX[(model, route)] for display_model, model in _CANONICAL_MODELS.items() for display_route, route in _CANONICAL_ROUTES.items() } _MODEL_CAPABILITIES = { display_model: { **SEEDANCE_CAPABILITIES[model], "routes": set(_MODEL_ROUTES), } for display_model, model in _CANONICAL_MODELS.items() } # 旧模型列表（保留用于旧的 _resolve_model 函数） _MODELS = ["seedance-2.0", "seedance-2.0-fast", "seedance-2.0-mini"] _RESOLUTIONS = ["480p", "720p", "1080p", "4k"] _FAST_RESOLUTIONS = {"480p", "720p"} _LIMITED_RESOLUTION_MODELS = { "seedance-2.0-fast", "seedance-2.0-mini", "dreamina-seedance-2-0-fast-hc", "dreamina-seedance-2-0-mini-hc", "seedance-2-0-fast-260128-d", "seedance-2-0-fast-d-ep", "seedance-2-0-mini-260615-d", "seedance-2-0-mini-260615-d-ep", } def _normalize_model_route(route: str) -> str: """兼容旧工作流保存的线路显示名。""" return _LEGACY_MODEL_ROUTES.get(route, route) def _resolve_model_matrix(base_model: str, route: str) -> str: """矩阵式解析：主模型 + 模型线路 → 实际模型ID""" route = _normalize_model_route(route) canonical_model = _CANONICAL_MODELS.get(base_model) canonical_route = _CANONICAL_ROUTES.get(route) if canonical_model is None or canonical_route is None: raise ValueError(f"{base_model} 不支持模型线路：{route}") return resolve_seedance_model(canonical_model, canonical_route) def _resolve_asset_creation_mode(route: str) -> str: """根据模型线路匹配素材创建方法。""" route = _normalize_model_route(route) mode = _ASSET_CREATION_MODES.get(route) if mode is None: raise ValueError(f"模型线路 {route} 未配置素材创建方法") return mode _RATIOS = ["智能", "16:9", "9:16", "4:3", "3:4", "1:1", "21:9"] _DURATIONS = ["自动"] + [f"{i}秒" for i in range(4, 31)] _MODE_MULTIMODAL = "多模态参考生视频" _MODE_FIRST_FRAME = "图生视频-首帧" _MODE_FIRST_LAST = "图生视频-首尾帧" _MODE_TEXT = "文生视频" _GENERATION_MODES = [ _MODE_MULTIMODAL, _MODE_FIRST_FRAME, _MODE_FIRST_LAST, _MODE_TEXT, ] _UI_MODE_MULTIMODAL = "多模态" _UI_MODE_FIRST_LAST = "首尾帧" _UI_GENERATION_MODES = [_UI_MODE_MULTIMODAL, _UI_MODE_FIRST_LAST] _CANONICAL_GENERATION_MODES = { _MODE_MULTIMODAL: "multimodal", _MODE_FIRST_FRAME: "first_frame", _MODE_FIRST_LAST: "first_last_frame", _MODE_TEXT: "text", _UI_MODE_MULTIMODAL: "multimodal", _UI_MODE_FIRST_LAST: "first_last_frame", } _ASSET_MODE_AUTO = "关闭" _ASSET_MODE_MANUAL = "打开" _ASSET_MODES = [_ASSET_MODE_AUTO, _ASSET_MODE_MANUAL] _SUCCESS_STATUSES = {"succeeded", "success", "completed", "done", "finished"} _FAILURE_STATUSES = {"failed", "fail", "failure", "error", "expired", "cancelled", "canceled"} class SeedanceAutoPass(io.ComfyNode): """Seedance 全能生成视频（根据模型线路自动创建素材）""" @classmethod def define_schema(cls): return io.Schema( node_id="SeedanceAutoPass", display_name="Seedance 全能生成视频", description="支持多模态（含文生视频）和首尾帧（尾帧可选）两种生成模式。", category="comfyui_o1key/Seedance", inputs=[ io.String.Input("提示词", multiline=True, default=""), io.Combo.Input( "生成模式", options=_UI_GENERATION_MODES, default=_UI_MODE_MULTIMODAL, tooltip="多模态无素材时支持文生视频；首尾帧的尾帧图片可以不连接。", ), io.Combo.Input("主模型", options=_BASE_MODELS, default="seedance 2.0"), io.Combo.Input("模型线路", options=_MODEL_ROUTES, default="国内"), io.Combo.Input("分辨率", options=_RESOLUTIONS, default="720p"), io.Combo.Input("宽高比", options=_RATIOS, default="智能"), io.Combo.Input("时长", options=_DURATIONS, default="5秒"), io.Combo.Input("生成音频", options=["关闭", "打开"], default="关闭"), # 当前 ComfyUI 前端会把 DynamicCombo 触发项当作输入插槽处理， # 在创建节点时抛出“Failed to find input socket”。这里使用稳定的 # 普通下拉，并保留全部可选素材插槽；执行时只读取当前模式对应项。 io.Autogrow.Input( "参考图片", optional=True, template=io.Autogrow.TemplateNames( input=io.Image.Input("参考图片"), names=[f"参考图片{i}" for i in range(1, 31)], min=0, ), ), io.Autogrow.Input( "参考视频", optional=True, template=io.Autogrow.TemplateNames( input=io.Video.Input("参考视频"), names=[f"参考视频{i}" for i in range(1, 11)], min=0, ), ), io.Autogrow.Input( "参考音频", optional=True, template=io.Autogrow.TemplateNames( input=io.Audio.Input("参考音频"), names=[f"参考音频{i}" for i in range(1, 11)], min=0, ), ), io.Image.Input("首帧图片", optional=True), io.Image.Input("尾帧图片", optional=True), io.Combo.Input( "素材创建模式", options=_ASSET_MODES, default=_ASSET_MODE_AUTO, tooltip="关闭：隐藏素材 ID 并自动创建连接的素材；打开：显示并使用已有素材 ID。", ), *[ io.String.Input(f"{prefix}{index}", default="", tooltip="手动模式使用；填写一个 Asset ID。") for prefix, maximum in (("图片素材ID", 30), ("视频素材ID", 10), ("音频素材ID", 10)) for index in range(1, maximum + 1) ], # 原高级参数改为普通参数，并统一放在节点最下方。 io.Combo.Input( "联网搜索", options=["关闭", "打开"], default="关闭", tooltip="兼容旧工作流；当前 content 请求不发送联网搜索参数。", ), io.Int.Input( "seed", default=0, min=0, max=0xffffffffffffffff, ), io.Combo.Input( "返回末帧图片", options=["关闭", "打开"], default="关闭", ), ], outputs=[ io.Video.Output(display_name="视频"), io.Image.Output(display_name="末帧图片"), ], ) @staticmethod def _autogrow_values(kwargs, group_name, legacy_prefix, legacy_max): """按定义顺序提取动态输入，并兼容直接调用时传入的旧编号参数。""" group = kwargs.get(group_name) if isinstance(group, dict): return [value for value in group.values() if value is not None] return [ kwargs[f"{legacy_prefix}{index}"] for index in range(1, legacy_max + 1) if kwargs.get(f"{legacy_prefix}{index}") is not None ] @staticmethod def _mode_inputs(kwargs): """提取 DynamicCombo 当前分支；旧工作流/直接调用默认按多模态处理。""" mode_inputs = kwargs.get("生成模式") if isinstance(mode_inputs, dict): mode = mode_inputs.get("生成模式", _UI_MODE_MULTIMODAL) return mode, mode_inputs if isinstance(mode_inputs, str): return mode_inputs, kwargs return _UI_MODE_MULTIMODAL, kwargs @staticmethod def _asset_creation_inputs(kwargs): """读取素材创建分支；旧工作流没有该控件时默认自动创建。""" asset_inputs = kwargs.get("素材创建模式", kwargs.get("素材创建", _ASSET_MODE_AUTO)) if isinstance(asset_inputs, dict): mode = asset_inputs.get("素材创建模式", asset_inputs.get("素材创建", _ASSET_MODE_AUTO)) inputs = asset_inputs else: mode = asset_inputs inputs = kwargs if mode in {"manual", "手动", _ASSET_MODE_MANUAL}: return _ASSET_MODE_MANUAL, inputs return _ASSET_MODE_AUTO, inputs @staticmethod def _parse_asset_ids(value): """接受换行、中英文逗号或分号分隔的 Asset ID。""" if isinstance(value, (list, tuple)): raw_values = value else: raw_values = re.split(r"[\s,，;；]+", str(value or "")) return [str(item).strip() for item in raw_values if str(item).strip()] @classmethod def _manual_asset_ids(cls, inputs, prefix, maximum): """按编号读取单行 ID，并兼容旧版聚合文本框/API 参数。""" values = [ asset_id for index in range(1, maximum + 1) for asset_id in cls._parse_asset_ids(inputs.get(f"{prefix}{index}", "")) ] return values or cls._parse_asset_ids(inputs.get(prefix, "")) @staticmethod def _canonical_generation_mode( generation_mode, image_count, video_count, audio_count, ): """把两种界面模式和旧工作流模式解析为接口的四种语义。""" if generation_mode == _MODE_TEXT: return "text" if generation_mode == _MODE_FIRST_FRAME: return "first_frame" if generation_mode == _MODE_FIRST_LAST: return "first_last_frame" if generation_mode in {_UI_MODE_FIRST_LAST}: return "first_frame" if image_count == 1 else "first_last_frame" if generation_mode in {_UI_MODE_MULTIMODAL, _MODE_MULTIMODAL}: return "multimodal" if image_count or video_count or audio_count else "text" if generation_mode in {"text", "first_frame", "first_last_frame", "multimodal"}: return generation_mode raise ValueError(f"不支持的生成模式：{generation_mode}") @staticmethod def _validate_mode_inputs( generation_mode, base_model, prompt, ref_images, ref_videos, ref_audios, ): """校验两种界面模式，并返回接口使用的实际生成模式。""" canonical_mode = SeedanceAutoPass._canonical_generation_mode( generation_mode, len(ref_images), len(ref_videos), len(ref_audios), ) if canonical_mode == "text": if not prompt: raise ValueError("文生视频模式下提示词不能为空") return canonical_mode if canonical_mode in {"first_frame", "first_last_frame"}: if generation_mode == _MODE_FIRST_FRAME and len(ref_images) != 1: raise ValueError("图生视频-首帧模式必须提供首帧图片") if generation_mode == _MODE_FIRST_LAST and len(ref_images) != 2: raise ValueError("图生视频-首尾帧模式必须同时提供首帧图片和尾帧图片") if len(ref_images) not in {1, 2}: raise ValueError("首尾帧模式必须提供首帧图片，尾帧图片可以不提供") if ref_videos or ref_audios: raise ValueError("首尾帧模式只支持图片素材") return "first_frame" if len(ref_images) == 1 else "first_last_frame" if not (prompt or ref_images or ref_videos or ref_audios): raise ValueError("多模态模式下，提示词和参考素材不能同时为空") if base_model != "seedance 2.5" and ref_audios and not (ref_images or ref_videos): raise ValueError("Seedance 2.0 系列不可单独输入参考音频，须同时提供参考图片或参考视频") return canonical_mode @staticmethod def _validate_dynamic_parameters( base_model, model_route, duration_s, ref_images, ref_videos, ref_audios, ): """按主模型校验线路、时长和动态参考素材数量。""" model_route = _normalize_model_route(model_route) capabilities = _MODEL_CAPABILITIES.get(base_model) if capabilities is None: raise ValueError(f"不支持的主模型：{base_model}") if model_route not in capabilities["routes"]: supported = "、".join(sorted(capabilities["routes"])) raise ValueError(f"{base_model} 仅支持模型线路：{supported}") if duration_s != "自动": try: duration = int(str(duration_s).removesuffix("秒")) except (TypeError, ValueError): raise ValueError(f"无效的时长：{duration_s}") from None minimum = capabilities["duration_min"] maximum = capabilities["duration_max"] if not minimum <= duration <= maximum: raise ValueError(f"{base_model} 的时长仅支持 {minimum}-{maximum} 秒") validate_seedance_media_counts( _CANONICAL_MODELS[base_model], len(ref_images), len(ref_videos), len(ref_audios), model_label=base_model, ) @staticmethod def _normalize_generation_parameters( generation_mode, base_model, model_route, prompt, resolution, ratio, duration_s, gen_audio, return_last, seed, asset_creation_mode="auto", ): """Translate the released Chinese widgets into the shared video catalog.""" route = _normalize_model_route(model_route) try: canonical_model = _CANONICAL_MODELS[base_model] canonical_route = _CANONICAL_ROUTES[route] canonical_mode = generation_mode if canonical_mode not in {"text", "first_frame", "first_last_frame", "multimodal"}: canonical_mode = _CANONICAL_GENERATION_MODES[generation_mode] except KeyError as exc: raise ValueError(f"Seedance 参数无效：{exc.args[0]}") from None return normalize_seedance_parameters({ "provider": "seedance", "model": canonical_model, "route": canonical_route, "generation_mode": canonical_mode, "asset_creation_mode": asset_creation_mode, "prompt": prompt, "resolution": resolution, "aspect_ratio": "auto" if ratio == "智能" else ratio, "duration": "auto" if duration_s == "自动" else str(duration_s).removesuffix("秒"), "generate_audio": gen_audio, "return_last_frame": return_last, "seed": seed, }) @classmethod async def execute(cls, **kwargs): generation_mode, mode_inputs = cls._mode_inputs(kwargs) asset_mode, asset_inputs = cls._asset_creation_inputs(kwargs) manual_assets = asset_mode == _ASSET_MODE_MANUAL prompt = (kwargs.get("提示词", mode_inputs.get("提示词", "")) or "").strip() base_model = kwargs["主模型"] model_route = _normalize_model_route(kwargs["模型线路"]) model = _resolve_model_matrix(base_model, model_route) resolution = kwargs["分辨率"] ratio = kwargs["宽高比"] duration_s = kwargs["时长"] gen_audio = kwargs["生成音频"] == "打开" web_search = kwargs.get("联网搜索", mode_inputs.get("联网搜索", "关闭")) == "打开" return_last = kwargs.get("返回末帧图片", "关闭") == "打开" create_mode = _resolve_asset_creation_mode(model_route) seed = int(kwargs.get("seed", 0)) if generation_mode in {_UI_MODE_MULTIMODAL, _MODE_MULTIMODAL}: ref_images = cls._autogrow_values(mode_inputs, "参考图片", "参考图片", 30) ref_videos = cls._autogrow_values(mode_inputs, "参考视频", "参考视频", 10) ref_audios = cls._autogrow_values(mode_inputs, "参考音频", "参考音频", 10) elif generation_mode in {_UI_MODE_FIRST_LAST, _MODE_FIRST_FRAME, _MODE_FIRST_LAST}: first_frame = mode_inputs.get("首帧图片") last_frame = mode_inputs.get("尾帧图片") if not manual_assets and first_frame is None: raise ValueError("首尾帧模式必须提供首帧图片，尾帧图片可以不提供") ref_images = [first_frame, last_frame] ref_videos = [] ref_audios = [] else: ref_images = [] ref_videos = [] ref_audios = [] ref_images = [value for value in ref_images if value is not None] if manual_assets: image_urls = cls._manual_asset_ids(asset_inputs, "图片素材ID", 30) video_urls = cls._manual_asset_ids(asset_inputs, "视频素材ID", 10) audio_urls = cls._manual_asset_ids(asset_inputs, "音频素材ID", 10) validation_images = image_urls validation_videos = video_urls validation_audios = audio_urls else: image_urls = video_urls = audio_urls = None validation_images = ref_images validation_videos = ref_videos validation_audios = ref_audios canonical_mode = cls._validate_mode_inputs( generation_mode, base_model, prompt, validation_images, validation_videos, validation_audios, ) cls._validate_dynamic_parameters( base_model, model_route, duration_s, validation_images, validation_videos, validation_audios, ) normalized = cls._normalize_generation_parameters( canonical_mode, base_model, model_route, prompt, resolution, ratio, duration_s, gen_audio, return_last, seed, "manual" if manual_assets else "auto", ) if not manual_assets: cls._validate_reference_media(ref_images, ref_videos, ref_audios) base_url = get_base_url_by_route() if not manual_assets: image_urls, video_urls, audio_urls = await cls._create_assets( ref_images, ref_videos, ref_audios, base_url, create_mode ) body = cls._build_body( model, prompt, resolution, ratio, duration_s, gen_audio, web_search, seed, image_urls, video_urls, audio_urls, use_asset_protocol=True, generation_mode=canonical_mode, return_last_frame=normalized["return_last_frame"], asset_creation_mode=normalized["asset_creation_mode"], ) pretty = json.dumps(body, ensure_ascii=False, indent=2) print("[Seedance自动过审] ── 提交请求体 ─────────────────") print(f"[Seedance自动过审] POST {base_url}/v1/video/generations") print(pretty) try: result_path, last_frame_url = await cls._submit_poll_download(body, base_url) except Exception as exc: message = format_seedance_generation_error(exc) if message == str(exc): raise raise RuntimeError(message) from None last_frame = None if return_last and last_frame_url: last_frame = await cls._url_to_tensor(last_frame_url) return io.NodeOutput(InputImpl.VideoFromFile(result_path), last_frame) @staticmethod def _build_body(model, prompt, resolution, ratio, duration_s, gen_audio, web_search, seed, image_urls, video_urls, audio_urls, use_asset_protocol=False, generation_mode=_MODE_MULTIMODAL, return_last_frame=False, asset_creation_mode="auto"): """ 按主站验证过的格式拼装请求体。 - 多模态直传：扁平格式（HTTPS URL） - 自动创建素材：content 格式（支持 asset:// 协议） - 首帧/首尾帧：始终使用 content 格式以携带 frame role """ del web_search, use_asset_protocol canonical_mode = SeedanceAutoPass._canonical_generation_mode( generation_mode, len(image_urls), len(video_urls), len(audio_urls), ) duration = 5 if duration_s == "自动" else int(str(duration_s).removesuffix("秒")) manual_assets = asset_creation_mode == "manual" prepared = { "first_frame": None if manual_assets else (image_urls[0] if image_urls else None), "last_frame": None if manual_assets else (image_urls[1] if len(image_urls) > 1 else None), "reference_images": ( [] if manual_assets or canonical_mode != "multimodal" else list(image_urls) ), "reference_videos": [] if manual_assets else list(video_urls), "reference_audios": [] if manual_assets else list(audio_urls), } return build_seedance_video_body({ "actual_model": model, "prompt": prompt, "generation_mode": canonical_mode, "asset_creation_mode": asset_creation_mode, "assets": { "images": list(image_urls) if manual_assets else [], "videos": list(video_urls) if manual_assets else [], "audios": list(audio_urls) if manual_assets else [], }, "duration": duration, "resolution": resolution, "aspect_ratio": "auto" if ratio == "智能" else ratio, "generate_audio": bool(gen_audio), "return_last_frame": bool(return_last_frame), "seed": int(seed), }, prepared) @classmethod async def _submit_poll_download(cls, body, base_url): """Use the same submit/poll/download client as o1key 视频生成.""" file_handle, save_path = tempfile.mkstemp( suffix=".mp4", prefix="seedance_autopass_", ) os.close(file_handle) client = SeedanceClient() client.base_url = base_url try: return await client.generate_async( body=body, save_path=save_path, use_new_format=True, ) except BaseException: try: if os.path.isfile(save_path): os.remove(save_path) except OSError: pass raise @staticmethod def _extract_video_url(sdata: dict): """上游已调整：成品直链放在 data.result_url（此前是 localhost 占位）。 优先取 result_url；保留递归下钻兜底（沿 data/content/result/videos 键）， 跳过 localhost/127.0.0.1，防上游结构再变。""" def _usable(v): return (isinstance(v, str) and v.startswith(("http://", "https://")) and "localhost" not in v and "127.0.0.1" not in v) # 首选：data.result_url（兼容顶层 result_url） for holder in (sdata.get("data"), sdata): if isinstance(holder, dict) and _usable(holder.get("result_url")): return holder["result_url"] # 兜底：递归下钻找第一个可用直链 def _walk(node): if isinstance(node, dict): for key in ("url", "video_url"): if _usable(node.get(key)): return node.get(key) for key in ("data", "content", "result", "videos"): if key in node: found = _walk(node.get(key)) if found: return found elif isinstance(node, list): for item in node: found = _walk(item) if found: return found return None return _walk(sdata) @staticmethod def _to_first_pil(image): """统一接收 ComfyUI IMAGE tensor 或批量节点加载的 PIL 图片。""" if isinstance(image, Image.Image): return image.convert("RGB") if image.mode != "RGB" else image pil_images = tensor_to_pil(image) if not pil_images: return None pil = pil_images[0] return pil.convert("RGB") if pil.mode != "RGB" else pil @staticmethod def _video_source(video): if hasattr(video, "get_stream_source"): return video.get_stream_source() if isinstance(video, dict): return ( video.get("video") or video.get("path") or video.get("file") or video.get("filename") or video.get("source_path") ) if isinstance(video, (str, os.PathLike, py_io.BytesIO)): return video for attribute in ("source_path", "path", "video", "file", "filename"): if hasattr(video, attribute): return getattr(video, attribute) return None @staticmethod def _stream_size(source, label): if isinstance(source, py_io.BytesIO): return source.getbuffer().nbytes if isinstance(source, (str, os.PathLike)): path = os.fspath(source) if not os.path.isfile(path): raise ValueError(f"{label}文件不存在") return os.path.getsize(path) raise ValueError(f"无法读取{label}文件") @classmethod def _validate_reference_media(cls, ref_images, ref_videos, ref_audios): """Validate every reference before the first upload or asset request.""" for index, image in enumerate(ref_images, start=1): label = f"参考图片{index}" pil = cls._to_first_pil(image) if pil is None: raise ValueError(f"{label}无法读取") validate_seedance_reference_dimensions(pil.width, pil.height, label) buffer = py_io.BytesIO() pil.save(buffer, format="PNG") if not 0 < buffer.getbuffer().nbytes <= SEEDANCE_REFERENCE_IMAGE_MAX_BYTES: raise ValueError(f"{label}文件大小必须在 1 字节到 30MB 之间") for index, video in enumerate(ref_videos, start=1): label = f"参考视频{index}" source = cls._video_source(video) size = cls._stream_size(source, label) if not 0 < size <= SEEDANCE_REFERENCE_VIDEO_MAX_BYTES: raise ValueError(f"{label}文件大小必须在 1 字节到 512MB 之间") if isinstance(source, (str, os.PathLike)): extension = os.path.splitext(os.fspath(source))[1].lower() if extension not in {".mp4", ".mov"}: raise ValueError(f"{label}格式须为 mp4 或 mov") try: if hasattr(video, "get_dimensions"): width, height = video.get_dimensions() else: import av if isinstance(source, py_io.BytesIO): source.seek(0) with av.open(source, mode="r") as container: stream = next( (item for item in container.streams if item.type == "video"), None, ) if stream is None: raise ValueError width, height = stream.width, stream.height except Exception: raise ValueError(f"{label}不是可读取的视频") from None finally: if isinstance(source, py_io.BytesIO): source.seek(0) validate_seedance_reference_dimensions( width, height, label, require_video_pixel_range=True, ) supported_audio = {".wav", ".mp3", ".m4a", ".aac", ".flac", ".ogg"} for index, audio in enumerate(ref_audios, start=1): label = f"参考音频{index}" if isinstance(audio, (str, os.PathLike)): extension = os.path.splitext(os.fspath(audio))[1].lower() if extension not in supported_audio: raise ValueError(f"{label}格式须为 wav/mp3/m4a/aac/flac/ogg") size = cls._stream_size(audio, label) elif isinstance(audio, dict) and audio.get("waveform") is not None: waveform = audio["waveform"] try: sample_count = int(waveform.shape[-1]) except Exception: raise ValueError(f"{label}无法读取") from None size = 44 + sample_count * 2 else: raise ValueError(f"{label}无法读取") if not 0 < size <= SEEDANCE_REFERENCE_AUDIO_MAX_BYTES: raise ValueError(f"{label}文件大小必须在 1 字节到 100MB 之间") @staticmethod async def _url_to_tensor(url): """Download a requested last frame without forwarding API credentials.""" try: async with aiohttp.ClientSession() as session: async with session.get(url) as response: if response.status != 200: return None data = await response.read() if not data or len(data) > 32 * 1024 * 1024: return None with Image.open(py_io.BytesIO(data)) as image: image.load() return pil_to_tensor([image.convert("RGB")]) except Exception as exc: print(f"[Seedance自动过审] 末帧图片下载失败: {exc}") return None @staticmethod async def _upload_assets(ref_images, ref_videos, ref_audios, base_url): """直传模式：上传素材到R2，返回公开URL列表""" image_urls = [] for idx, image in enumerate(ref_images, start=1): pil = SeedanceAutoPass._to_first_pil(image) if pil is None: raise ValueError(f"第 {idx} 张参考图片无法读取") print(f"[Seedance自动过审][直传] 上传参考图片 {idx}/{len(ref_images)}...") image_urls.append(await upload_image(pil, base_url=base_url)) video_urls = [] for idx, v in enumerate(ref_videos, start=1): print(f"[Seedance自动过审][直传] 上传参考视频 {idx}/{len(ref_videos)}...") video_urls.append(await upload_video(v, base_url=base_url)) audio_urls = [] for idx, a in enumerate(ref_audios, start=1): print(f"[Seedance自动过审][直传] 上传参考音频 {idx}/{len(ref_audios)}...") audio_urls.append(await upload_audio(a, base_url=base_url)) return image_urls, video_urls, audio_urls @staticmethod async def _create_assets(ref_images, ref_videos, ref_audios, base_url, create_mode): """先上传到 R2，再调用匹配的素材 API，返回 asset:// 格式的 URL 列表。""" request_types = {"HC": "hc", "Doubao": "doubao"} request_type = request_types.get(create_mode) if request_type is None: raise ValueError(f"不支持的素材创建模式：{create_mode}") element_client = SeedanceElementClient(base_url=base_url) items = [ ("image", index, value, len(ref_images)) for index, value in enumerate(ref_images, start=1) ] + [ ("video", index, value, len(ref_videos)) for index, value in enumerate(ref_videos, start=1) ] + [ ("audio", index, value, len(ref_audios)) for index, value in enumerate(ref_audios, start=1) ] semaphore = asyncio.Semaphore(3) async def prepare(kind, index, value, total): async with semaphore: labels = {"image": "图片", "video": "视频", "audio": "音频"} asset_types = {"image": "Image", "video": "Video", "audio": "Audio"} label = labels[kind] print(f"[Seedance自动过审][自动创建] 上传参考{label} {index}/{total}...") if kind == "image": pil = SeedanceAutoPass._to_first_pil(value) if pil is None: raise ValueError(f"第 {index} 张参考图片无法读取") uploaded_url = await upload_image(pil, base_url=base_url) elif kind == "video": uploaded_url = await upload_video(value, base_url=base_url) else: uploaded_url = await upload_audio(value, base_url=base_url) if not str(uploaded_url).startswith("https://"): raise ValueError(f"参考{label}上传后未获得 HTTPS 公网地址") name = f"参考{label}{index}" result = await element_client.create_hc_asset_and_wait( name=name, asset_url=uploaded_url, asset_type=asset_types[kind], request_type=request_type, ) element_id = str(result.get("Id") or "").strip() if not element_id: raise RuntimeError(f"创建{label}素材失败，未返回 ID") return kind, f"asset://{element_id}" prepared = await asyncio.gather(*(prepare(*item) for item in items)) result = {"image": [], "video": [], "audio": []} for kind, asset_url in prepared: result[kind].append(asset_url) return result["image"], result["video"], result["audio"] NODE_CLASS_MAPPINGS = { "SeedanceAutoPass": SeedanceAutoPass, } NODE_DISPLAY_NAME_MAPPINGS = { "SeedanceAutoPass": "Seedance 全能生成视频", }