import uuid import torch from typing_extensions import override from comfy_api.latest import IO, ComfyExtension, Input from comfy_api_nodes.apis.heygen import ( HEYGEN_AVATAR_MAP, HEYGEN_AVATAR_OPTIONS, HEYGEN_TRANSLATE_LANGUAGES, HEYGEN_VOICE_GENERAL_MAP, HEYGEN_VOICE_GENERAL_OPTIONS, HEYGEN_VOICE_TTS_MAP, HEYGEN_VOICE_TTS_OPTIONS, ) from comfy_api_nodes.util import ( ApiEndpoint, audio_bytes_to_audio_input, download_url_as_bytesio, download_url_to_image_tensor, download_url_to_video_output, downscale_image_tensor_by_max_side, get_number_of_images, poll_op_raw, sync_op_raw, upload_audio_to_comfyapi, upload_image_to_comfyapi, upload_images_to_comfyapi, upload_video_to_comfyapi, validate_string, ) from server import PromptServer _VIDEOS_PATH = "/proxy/heygen/v3/videos" _TRANSLATIONS_PATH = "/proxy/heygen/v3/video-translations" _SPEECH_PATH = "/proxy/heygen/v3/voices/speech" _AVATARS_PATH = "/proxy/heygen/v3/avatars" _LOOKS_PATH = "/proxy/heygen/v3/avatars/looks" _DEFAULT_VOICE_OPTION = "(avatar's default voice)" _AVATARS_BY_ENGINE = { e: [label for label, (_aid, _atype, engines) in HEYGEN_AVATAR_MAP.items() if e in engines] for e in ("avatar_iv", "avatar_iii", "avatar_v") } async def _apply_speech_source(cls: type[IO.ComfyNode], payload: dict, speech: dict, require_voice: bool) -> None: """Fill script/audio speech fields of a /v3/videos payload from the DynamicCombo dict.""" if speech["speech"] == "audio": payload["audio_url"] = await upload_audio_to_comfyapi( cls, speech["audio"], container_format="mp3", codec_name="libmp3lame", mime_type="audio/mpeg" ) elif speech["speech"] == "script": validate_string(speech["text"], strip_whitespace=True, min_length=1, max_length=5000) payload["script"] = speech["text"] voice_id = speech.get("custom_voice_id", "").strip() if not voice_id and speech["voice"] != _DEFAULT_VOICE_OPTION: voice_id = HEYGEN_VOICE_GENERAL_MAP[speech["voice"]] if voice_id: payload["voice_id"] = voice_id elif require_voice: raise ValueError("A voice is required when driving the video with a text script.") speed = speech.get("voice_speed", 1.0) if speed != 1.0: payload["voice_settings"] = {"speed": round(speed, 2)} async def _create_and_poll_video(cls: type[IO.ComfyNode], payload: dict) -> dict: """POST a /v3/videos payload, poll until terminal, and return the final video data.""" created = await sync_op_raw( cls, ApiEndpoint(path=_VIDEOS_PATH, method="POST", headers={"Idempotency-Key": uuid.uuid4().hex}), data=payload, ) video_id = (created.get("data") or {}).get("video_id") if not video_id: raise ValueError(f"HeyGen did not return a video_id: {created}") final = await poll_op_raw( cls, ApiEndpoint(path=f"{_VIDEOS_PATH}/{video_id}"), status_extractor=lambda r: (r.get("data") or {}).get("status"), queued_statuses=["pending", "waiting"], poll_interval=5.0, ) data = final["data"] if not data.get("video_url"): raise ValueError(f"HeyGen returned no video_url for video {video_id}.") return data async def _resolve_avatar( cls: type[IO.ComfyNode], avatar_label: str, custom_avatar_id: str, engine_choice: str ) -> tuple[str, str | None]: """Resolve (avatar_id, engine_type) from the combo/override + engine widgets.""" custom_avatar_id = custom_avatar_id.strip() if custom_avatar_id: look = ( await sync_op_raw( cls, ApiEndpoint(path=f"{_LOOKS_PATH}/{custom_avatar_id}"), final_label_on_success=None, ) ).get("data") or {} avatar_id = custom_avatar_id avatar_label = look.get("name") or custom_avatar_id supported = look.get("supported_api_engines") or [] else: avatar_id, avatar_type, supported = HEYGEN_AVATAR_MAP[avatar_label] if engine_choice == "auto": engine = next((e for e in ("avatar_iv", "avatar_iii", "avatar_v") if e in supported), None) else: engine = engine_choice if supported and engine not in supported: raise ValueError( f"Avatar '{avatar_label}' does not support the {engine} engine " f"(supported: {', '.join(supported)}). Set engine to 'auto' to pick " "a compatible engine automatically." ) return avatar_id, engine class HeyGenTalkingPhotoNode(IO.ComfyNode): """Animate a still image of a person into a lip-synced talking video.""" @classmethod def define_schema(cls) -> IO.Schema: return IO.Schema( node_id="HeyGenTalkingPhotoNode", display_name="HeyGen Talking Photo", category="partner/video/HeyGen", description="Animate any image of a person into a lip-synced talking video " "(HeyGen Avatar IV). Drive it with a text script or your own audio.", inputs=[ IO.Image.Input( "image", tooltip="Image of a person to animate. Downscaled automatically if larger than 2K.", ), IO.DynamicCombo.Input( "speech", display_name="speech source", options=[ IO.DynamicCombo.Option( "script", [ IO.String.Input( "text", multiline=True, default="", tooltip="Text for the avatar to speak (up to 5000 characters). " "The generated speech must be at least 1 second long.", ), IO.Combo.Input( "voice", options=HEYGEN_VOICE_GENERAL_OPTIONS, tooltip="Voice for the script (HeyGen's most popular voices).", ), IO.String.Input( "custom_voice_id", default="", optional=True, tooltip="Optional HeyGen voice ID. When set, overrides the voice selected above. " "Any voice from HeyGen's library (2000+) can be used.", ), IO.Float.Input( "voice_speed", default=1.0, min=0.5, max=1.5, step=0.05, optional=True, tooltip="Speech speed multiplier.", ), ], ), IO.DynamicCombo.Option( "audio", [ IO.Audio.Input( "audio", tooltip="Audio for the avatar to lip-sync, up to 10 minutes.", ), ], ), ], tooltip="Drive the avatar with a text script (HeyGen text-to-speech) or your own audio.", ), IO.Combo.Input( "resolution", options=["720p", "1080p"], default="1080p", optional=True, tooltip="Output video resolution.", ), IO.Combo.Input( "aspect_ratio", options=["auto", "16:9", "9:16", "1:1", "4:5", "5:4"], default="auto", optional=True, tooltip="Output aspect ratio. 'auto' follows the input image.", ), IO.Combo.Input( "expressiveness", options=["low", "medium", "high"], default="low", optional=True, tooltip="How expressive the animated face and gestures are.", ), IO.Int.Input( "seed", default=42, min=0, max=2147483647, control_after_generate=True, optional=True, tooltip="Not sent to HeyGen; change it to force a re-run.", ), ], outputs=[IO.Video.Output()], hidden=[ IO.Hidden.auth_token_comfy_org, IO.Hidden.api_key_comfy_org, IO.Hidden.unique_id, ], is_api_node=True, price_badge=IO.PriceBadge( expr="""{"type":"usd","usd":0.0715,"format":{"suffix":"/second"}}""", ), ) @classmethod async def execute( cls, image: Input.Image, speech: dict, resolution: str = "1080p", aspect_ratio: str = "auto", expressiveness: str = "low", seed: int = 0, ) -> IO.NodeOutput: image = downscale_image_tensor_by_max_side(image, max_side=2000) image_url = await upload_image_to_comfyapi(cls, image, mime_type="image/png", total_pixels=None) payload = { "type": "image", "image": {"type": "url", "url": image_url}, "resolution": resolution, "aspect_ratio": aspect_ratio, "expressiveness": expressiveness, "title": "ComfyUI Talking Photo", } await _apply_speech_source(cls, payload, speech, require_voice=True) video = await _create_and_poll_video(cls, payload) return IO.NodeOutput(await download_url_to_video_output(video["video_url"])) class HeyGenAvatarVideoNode(IO.ComfyNode): """Generate a presenter video from a HeyGen avatar look.""" @classmethod def define_schema(cls) -> IO.Schema: return IO.Schema( node_id="HeyGenAvatarVideoNode", display_name="HeyGen Avatar Video", category="partner/video/HeyGen", description="Generate a talking-presenter video from a HeyGen avatar. " "Includes HeyGen's most popular public avatars; any look ID can be supplied as an override.", inputs=[ IO.DynamicCombo.Input( "engine", options=[ IO.DynamicCombo.Option( "auto", [ IO.Combo.Input( "avatar", options=HEYGEN_AVATAR_OPTIONS, tooltip="Avatar look to present the video (curated from HeyGen's " "public library). The best engine the look supports is chosen " "automatically.", ), ], ), IO.DynamicCombo.Option( "avatar_iv", [ IO.Combo.Input( "avatar", options=_AVATARS_BY_ENGINE["avatar_iv"], tooltip="Avatar looks that support the Avatar IV engine.", ), ], ), IO.DynamicCombo.Option( "avatar_iii", [ IO.Combo.Input( "avatar", options=_AVATARS_BY_ENGINE["avatar_iii"], tooltip="Avatar looks that support the Avatar III engine.", ), ], ), IO.DynamicCombo.Option( "avatar_v", [ IO.Combo.Input( "avatar", options=_AVATARS_BY_ENGINE["avatar_v"], tooltip="Avatar looks that support the Avatar V engine.", ), ], ), ], tooltip="Rendering engine; each choice lists only the avatars that support it. " "'auto' offers every avatar and picks its best engine (Avatar IV preferred). " "Avatar V is highest fidelity, Avatar III is the most affordable.", ), IO.String.Input( "custom_avatar_id", default="", optional=True, tooltip="Optional HeyGen avatar look ID. When set, overrides the avatar selected above. " "Any of HeyGen's 3000+ public looks (or your private avatars) can be used.", ), IO.DynamicCombo.Input( "speech", display_name="speech source", options=[ IO.DynamicCombo.Option( "script", [ IO.String.Input( "text", multiline=True, default="", tooltip="Text for the avatar to speak (up to 5000 characters). " "The generated speech must be at least 1 second long.", ), IO.Combo.Input( "voice", options=[_DEFAULT_VOICE_OPTION] + HEYGEN_VOICE_GENERAL_OPTIONS, tooltip="Voice for the script. The default option uses the voice HeyGen assigned to the avatar.", ), IO.String.Input( "custom_voice_id", default="", optional=True, tooltip="Optional HeyGen voice ID. When set, overrides the voice selected above. " "Any voice from HeyGen's library (2000+) can be used.", ), IO.Float.Input( "voice_speed", default=1.0, min=0.5, max=1.5, step=0.05, optional=True, tooltip="Speech speed multiplier.", ), ], ), IO.DynamicCombo.Option( "audio", [ IO.Audio.Input( "audio", tooltip="Audio for the avatar to lip-sync, up to 10 minutes.", ), ], ), ], tooltip="Drive the avatar with a text script (HeyGen text-to-speech) or your own audio.", ), IO.Combo.Input( "resolution", options=["720p", "1080p"], default="1080p", optional=True, tooltip="Output video resolution.", ), IO.Combo.Input( "aspect_ratio", options=["auto", "16:9", "9:16", "1:1", "4:5", "5:4"], default="auto", optional=True, tooltip="Output aspect ratio. 'auto' follows the avatar's source footage.", ), IO.String.Input( "background_color", default="", optional=True, tooltip="Optional solid background color as a hex code (e.g. '#00ff00'). " "Leave empty for the avatar's own background.", ), IO.Int.Input( "seed", default=42, min=0, max=2147483647, control_after_generate=True, optional=True, tooltip="Not sent to HeyGen; change it to force a re-run.", ), ], outputs=[IO.Video.Output()], hidden=[ IO.Hidden.auth_token_comfy_org, IO.Hidden.api_key_comfy_org, IO.Hidden.unique_id, ], is_api_node=True, price_badge=IO.PriceBadge( depends_on=IO.PriceBadgeDepends(widgets=["engine"]), expr=""" widgets.engine = "avatar_iii" ? {"type":"range_usd","min_usd":0.023881,"max_usd":0.061919,"format":{"suffix":"/second"}} : widgets.engine = "avatar_v" ? {"type":"usd","usd":0.095381,"format":{"suffix":"/second"}} : widgets.engine = "avatar_iv" ? {"type":"range_usd","min_usd":0.0715,"max_usd":0.095381,"format":{"suffix":"/second"}} : {"type":"range_usd","min_usd":0.023881,"max_usd":0.095381,"format":{"suffix":"/second"}} """, ), ) @classmethod async def execute( cls, engine: dict, speech: dict, custom_avatar_id: str = "", resolution: str = "1080p", aspect_ratio: str = "auto", background_color: str = "", seed: int = 0, ) -> IO.NodeOutput: avatar_id, engine_type = await _resolve_avatar(cls, engine["avatar"], custom_avatar_id, engine["engine"]) payload = { "type": "avatar", "avatar_id": avatar_id, "resolution": resolution, "aspect_ratio": aspect_ratio, "title": "ComfyUI Avatar Video", } if engine_type: payload["engine"] = {"type": engine_type} background_color = background_color.strip() if background_color: if not background_color.startswith("#"): raise ValueError("background_color must be a hex color code like '#00ff00'.") payload["background"] = {"type": "color", "value": background_color} await _apply_speech_source(cls, payload, speech, require_voice=False) video = await _create_and_poll_video(cls, payload) return IO.NodeOutput(await download_url_to_video_output(video["video_url"])) class HeyGenCreateAvatarNode(IO.ComfyNode): """Create a reusable HeyGen avatar from a photo or a text prompt.""" @classmethod def define_schema(cls) -> IO.Schema: return IO.Schema( node_id="HeyGenCreateAvatarNode", display_name="HeyGen Create Avatar", category="partner/video/HeyGen", description="Create your own reusable HeyGen avatar from a photo of a person or " "from a text prompt (a generated character). Feed the resulting avatar_id into " "HeyGen Avatar Video's custom_avatar_id — and save the ID somewhere to reuse the " "avatar in future workflows.", inputs=[ IO.DynamicCombo.Input( "source", options=[ IO.DynamicCombo.Option( "prompt", [ IO.String.Input( "prompt", multiline=True, default="", tooltip="Description of the avatar to generate (up to 1000 characters).", ), IO.Autogrow.Input( "reference_images", template=IO.Autogrow.TemplateNames( IO.Image.Input("ref_image"), names=[f"ref_image_{i}" for i in range(1, 4)], min=0, ), tooltip="Up to 3 reference images guiding the generated look.", ), ], ), IO.DynamicCombo.Option( "photo", [ IO.Image.Input( "identity_photo", tooltip="Photo of the person to turn into an avatar. " "Downscaled automatically if larger than 2K.", ), ], ), ], tooltip="Generate a new character from a text prompt, or create the avatar " "from a connected photo of a person.", ), ], outputs=[ IO.String.Output( display_name="avatar_id", tooltip="Avatar look ID. Pass it to HeyGen Avatar Video's custom_avatar_id; " "save it to reuse the avatar later.", ), IO.Image.Output(display_name="preview"), ], hidden=[ IO.Hidden.auth_token_comfy_org, IO.Hidden.api_key_comfy_org, IO.Hidden.unique_id, ], is_api_node=True, price_badge=IO.PriceBadge( expr="""{"type":"usd","usd":1.43}""", ), ) @classmethod async def execute( cls, source: dict, ) -> IO.NodeOutput: payload: dict = {"name": "ComfyUI Avatar"} if source["source"] == "photo": image = downscale_image_tensor_by_max_side(source["identity_photo"], max_side=2000) image_url = await upload_image_to_comfyapi(cls, image, mime_type="image/png", total_pixels=None) payload["type"] = "photo" payload["file"] = {"type": "url", "url": image_url} else: validate_string(source["prompt"], strip_whitespace=True, min_length=1, max_length=1000) payload["type"] = "prompt" payload["prompt"] = source["prompt"] ref_tensors = [t for t in (source.get("reference_images") or {}).values() if t is not None] if ref_tensors: n_images = sum(get_number_of_images(t) for t in ref_tensors) if n_images > 3: raise ValueError(f"HeyGen accepts at most 3 reference images; got {n_images}.") scaled = [downscale_image_tensor_by_max_side(t, max_side=2000) for t in ref_tensors] ref_urls = await upload_images_to_comfyapi( cls, scaled, max_images=3, mime_type="image/png", total_pixels=None ) payload["reference_images"] = [{"type": "url", "url": u} for u in ref_urls] created = await sync_op_raw( cls, ApiEndpoint(path=_AVATARS_PATH, method="POST"), data=payload, ) look_id = ((created.get("data") or {}).get("avatar_item") or {}).get("id") if not look_id: raise ValueError(f"HeyGen did not return an avatar: {created}") final = await poll_op_raw( cls, ApiEndpoint(path=f"{_LOOKS_PATH}/{look_id}"), # A missing status means the look needed no training and is ready. status_extractor=lambda r: (r.get("data") or {}).get("status") or "completed", failed_statuses=["failed", "pending_consent"], poll_interval=5.0, ) data = final["data"] if data.get("preview_image_url"): preview = await download_url_to_image_tensor(data["preview_image_url"]) else: preview = torch.zeros(1, 64, 64, 3) PromptServer.instance.send_progress_text( f"Please save the avatar_id for reuse.\n\navatar_id: {look_id}", cls.hidden.unique_id, ) return IO.NodeOutput(look_id, preview) class HeyGenVideoTranslateNode(IO.ComfyNode): """Translate a spoken video into another language with voice cloning and lip sync.""" @classmethod def define_schema(cls) -> IO.Schema: return IO.Schema( node_id="HeyGenVideoTranslateNode", display_name="HeyGen Video Translate", category="partner/video/HeyGen", description="Translate a spoken video into another language. Clones the original " "speaker's voice and re-animates the mouth to match the translated speech.", inputs=[ IO.Video.Input( "video", tooltip="Video with speech to translate.", ), IO.Combo.Input( "output_language", options=HEYGEN_TRANSLATE_LANGUAGES, tooltip="Target language for the translated video.", ), IO.Combo.Input( "mode", options=["speed", "precision"], default="speed", tooltip="'speed' is faster; 'precision' produces higher-quality lip sync at twice the price.", ), IO.Boolean.Input( "translate_audio_only", default=False, optional=True, tooltip="Only swap the audio track, keeping the original mouth movements (no lip sync).", ), IO.Int.Input( "speaker_count", default=0, min=0, max=10, optional=True, tooltip="Number of speakers in the video. 0 = detect automatically.", ), IO.Int.Input( "seed", default=42, min=0, max=2147483647, control_after_generate=True, optional=True, tooltip="Not sent to HeyGen; change it to force a re-run.", ), ], outputs=[IO.Video.Output()], hidden=[ IO.Hidden.auth_token_comfy_org, IO.Hidden.api_key_comfy_org, IO.Hidden.unique_id, ], is_api_node=True, price_badge=IO.PriceBadge( depends_on=IO.PriceBadgeDepends(widgets=["mode"]), expr="""{"type":"usd","usd": widgets.mode = "precision" ? 0.095381 : 0.047619,""" """"format":{"suffix":"/second"}}""", ), ) @classmethod async def execute( cls, video: Input.Video, output_language: str, mode: str, translate_audio_only: bool = False, speaker_count: int = 0, seed: int = 0, ) -> IO.NodeOutput: video_url = await upload_video_to_comfyapi(cls, video) payload = { "video": {"type": "url", "url": video_url}, "output_languages": [output_language], "mode": mode, "translate_audio_only": translate_audio_only, "title": "ComfyUI Video Translate", } if speaker_count > 0: payload["speaker_num"] = speaker_count created = await sync_op_raw( cls, ApiEndpoint(path=_TRANSLATIONS_PATH, method="POST"), data=payload, ) translation_ids = (created.get("data") or {}).get("video_translation_ids") or [] if not translation_ids: raise ValueError(f"HeyGen did not return a translation ID: {created}") final = await poll_op_raw( cls, ApiEndpoint(path=f"{_TRANSLATIONS_PATH}/{translation_ids[0]}"), status_extractor=lambda r: (r.get("data") or {}).get("status"), queued_statuses=["pending"], poll_interval=5.0, ) data = final["data"] if not data.get("video_url"): raise ValueError(f"HeyGen returned no video_url for translation {translation_ids[0]}.") return IO.NodeOutput(await download_url_to_video_output(data["video_url"])) class HeyGenTextToSpeechNode(IO.ComfyNode): """Synthesize speech audio from text with HeyGen's Starfish TTS engine.""" @classmethod def define_schema(cls) -> IO.Schema: return IO.Schema( node_id="HeyGenTextToSpeechNode", display_name="HeyGen Text to Speech", category="partner/audio/HeyGen", description="Generate speech audio from text using HeyGen's Starfish TTS engine. " "Includes HeyGen's most popular voices across 17 languages.", inputs=[ IO.String.Input( "text", multiline=True, default="", tooltip="Text to synthesize (up to 5000 characters). The generated speech " "must be at least 1 second long.", ), IO.Combo.Input( "voice", options=HEYGEN_VOICE_TTS_OPTIONS, tooltip="Voice to use (curated from HeyGen's most popular Starfish-compatible voices).", ), IO.String.Input( "custom_voice_id", default="", optional=True, tooltip="Optional HeyGen voice ID. When set, overrides the voice selected above. " "The voice must support the Starfish engine.", ), IO.Float.Input( "speed", default=1.0, min=0.5, max=2.0, step=0.05, optional=True, tooltip="Speech speed multiplier.", ), IO.Boolean.Input( "ssml", default=False, optional=True, tooltip="Treat the text as SSML markup (for pauses, emphasis, and pronunciation control).", ), IO.Int.Input( "seed", default=42, min=0, max=2147483647, control_after_generate=True, optional=True, tooltip="Not sent to HeyGen; change it to force a re-run.", ), ], outputs=[IO.Audio.Output()], hidden=[ IO.Hidden.auth_token_comfy_org, IO.Hidden.api_key_comfy_org, IO.Hidden.unique_id, ], is_api_node=True, price_badge=IO.PriceBadge( expr="""{"type":"usd","usd":0.00095381,"format":{"approximate":true,"suffix":"/second"}}""", ), ) @classmethod async def execute( cls, text: str, voice: str, custom_voice_id: str = "", speed: float = 1.0, ssml: bool = False, seed: int = 0, ) -> IO.NodeOutput: validate_string(text, strip_whitespace=True, min_length=1, max_length=5000) payload = { "text": text, "voice_id": custom_voice_id.strip() or HEYGEN_VOICE_TTS_MAP[voice], "speed": round(speed, 2), } if ssml: payload["input_type"] = "ssml" response = await sync_op_raw( cls, ApiEndpoint(path=_SPEECH_PATH, method="POST"), data=payload, ) audio_url = (response.get("data") or {}).get("audio_url") if not audio_url: raise ValueError(f"HeyGen did not return an audio_url: {response}") audio_bytes = await download_url_as_bytesio(audio_url) return IO.NodeOutput(audio_bytes_to_audio_input(audio_bytes.getvalue())) class HeyGenExtension(ComfyExtension): @override async def get_node_list(self) -> list[type[IO.ComfyNode]]: return [ HeyGenTalkingPhotoNode, HeyGenAvatarVideoNode, HeyGenCreateAvatarNode, HeyGenVideoTranslateNode, HeyGenTextToSpeechNode, ] async def comfy_entrypoint() -> HeyGenExtension: return HeyGenExtension()