mirror of
https://github.com/comfyanonymous/ComfyUI.git
synced 2026-10-24 02:49:04 +03:00
dc0486a538
* [Partner Nodes] feat(Meshy): add Meshy T2.5 Smart Topology text and image to model nodes Signed-off-by: bigcat88 <bigcat88@icloud.com> * [Partner Nodes] fix(HeyGen): switch HeyGen Voice Turbo price badge to list price on Nov 1 Signed-off-by: bigcat88 <bigcat88@icloud.com> * [Partner Nodes] fix(Meshy): return Smart Topology 3D files without saving them to the output directory Signed-off-by: bigcat88 <bigcat88@icloud.com> --------- Signed-off-by: bigcat88 <bigcat88@icloud.com> Co-authored-by: Purz <97489706+purzbeats@users.noreply.github.com>
1313 lines
56 KiB
Python
1313 lines
56 KiB
Python
import asyncio
|
|
import re
|
|
|
|
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_LANGUAGES,
|
|
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_image_aspect_ratio,
|
|
validate_string,
|
|
)
|
|
from server import PromptServer
|
|
|
|
_VIDEOS_PATH = "/proxy/heygen/v3/videos"
|
|
_MODEL_VIDEOS_PATH = "/proxy/heygen/v3/models/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"
|
|
_MODEL_VOICES_PATH = "/proxy/heygen/v3/models/audio/voices"
|
|
_MODEL_TTS_PATH = "/proxy/heygen/v3/models/audio/tts"
|
|
|
|
_DEFAULT_VOICE_OPTION = "(avatar's default voice)"
|
|
_VOICE_LANGUAGE_OPTIONS = ["auto", *HEYGEN_VOICE_LANGUAGES]
|
|
_NEW_VOICE_RETRIES = 5
|
|
|
|
_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")
|
|
}
|
|
|
|
_REFERENCE_TAG_RE = re.compile(r"(?<!\w)@(image|video|audio)(\d*)(?!\w)", re.IGNORECASE | re.ASCII)
|
|
_REFERENCE_LABELS = {"image": "Picture", "video": "Video", "audio": "Audio"}
|
|
|
|
|
|
def _rewrite_reference_tags(prompt: str, counts: dict[str, int]) -> str:
|
|
def repl(match: re.Match) -> str:
|
|
kind = match.group(1).lower()
|
|
idx = int(match.group(2) or 1)
|
|
if not 1 <= idx <= counts[kind]:
|
|
raise ValueError(
|
|
f"The prompt references @{kind.capitalize()}{idx}, "
|
|
f"but only {counts[kind]} reference {kind} inputs are connected."
|
|
)
|
|
return f"<{_REFERENCE_LABELS[kind]} {idx}>"
|
|
|
|
prev = None
|
|
while prev != prompt:
|
|
prev = prompt
|
|
prompt = _REFERENCE_TAG_RE.sub(repl, prompt)
|
|
return prompt
|
|
|
|
|
|
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], create_path: str, payload: dict) -> dict:
|
|
"""POST a video payload, poll /v3/videos until terminal, and return the final video data."""
|
|
for attempt in range(_NEW_VOICE_RETRIES + 1):
|
|
try:
|
|
created = await sync_op_raw(
|
|
cls,
|
|
ApiEndpoint(path=create_path, method="POST"),
|
|
data=payload,
|
|
)
|
|
break
|
|
except Exception as e:
|
|
if attempt == _NEW_VOICE_RETRIES or "Invalid voice_id" not in str(e):
|
|
raise
|
|
await asyncio.sleep(3)
|
|
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, _VIDEOS_PATH, 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.0143,"max_usd":0.023595,"format":{"suffix":"/second"}}
|
|
: widgets.engine = "avatar_v"
|
|
? {"type":"usd","usd":0.1716,"format":{"suffix":"/second"}}
|
|
: widgets.engine = "avatar_iv"
|
|
? {"type":"range_usd","min_usd":0.055055,"max_usd":0.115115,"format":{"suffix":"/second"}}
|
|
: {"type":"range_usd","min_usd":0.0143,"max_usd":0.1716,"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, _VIDEOS_PATH, 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(
|
|
depends_on=IO.PriceBadgeDepends(widgets=["source"]),
|
|
expr="""{"type":"usd","usd": widgets.source = "photo" ? 1.8876 : 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 a higher 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", "translate_audio_only"]),
|
|
expr="""{"type":"usd","usd": widgets.mode = "precision" ? 0.03575 """
|
|
""": widgets.translate_audio_only = true ? 0.013585 : 0.019305,"""
|
|
""""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 HeyGenReferenceToVideoNode(IO.ComfyNode):
|
|
|
|
@classmethod
|
|
def define_schema(cls) -> IO.Schema:
|
|
return IO.Schema(
|
|
node_id="HeyGenReferenceToVideoNode",
|
|
display_name="HeyGen Video 1.0 Reference to Video",
|
|
category="partner/video/HeyGen",
|
|
description="Generate a video with synchronized dialogue and sound from a text prompt using "
|
|
"HeyGen Video 1.0. Optionally connect up to 12 reference images, videos, and audio clips and "
|
|
"mention them in the prompt as @Image1, @Video1, @Audio1.",
|
|
inputs=[
|
|
IO.DynamicCombo.Input(
|
|
"model",
|
|
options=[
|
|
IO.DynamicCombo.Option(
|
|
"heygen-video-1",
|
|
[
|
|
IO.String.Input(
|
|
"prompt",
|
|
multiline=True,
|
|
default="",
|
|
tooltip="Description of the video, including any dialogue. Refer to "
|
|
"connected references as @Image1, @Video1, @Audio1, numbered per type "
|
|
"in input order.",
|
|
),
|
|
IO.Int.Input(
|
|
"duration",
|
|
default=5,
|
|
min=5,
|
|
max=15,
|
|
step=1,
|
|
display_mode=IO.NumberDisplay.slider,
|
|
tooltip="Duration of the output video in seconds.",
|
|
),
|
|
IO.Combo.Input(
|
|
"resolution",
|
|
options=["768p", "480p", "2k"],
|
|
tooltip="Output resolution. 2k requires the 16:9 or 9:16 aspect ratio; "
|
|
"'auto' works at 2k only without references.",
|
|
),
|
|
IO.Combo.Input(
|
|
"aspect_ratio",
|
|
options=["auto", "16:9", "9:16", "1:1", "4:3", "3:4", "21:9"],
|
|
tooltip="Output aspect ratio. 'auto' is 16:9 without references and "
|
|
"otherwise follows the first reference image (or the first reference "
|
|
"video when no images are connected).",
|
|
),
|
|
IO.Int.Input(
|
|
"seed",
|
|
default=42,
|
|
min=0,
|
|
max=4294967295,
|
|
control_after_generate=True,
|
|
tooltip="Seed for the generation. Results can still vary between runs "
|
|
"with the same seed.",
|
|
),
|
|
IO.Autogrow.Input(
|
|
"reference_images",
|
|
template=IO.Autogrow.TemplateNames(
|
|
IO.Image.Input("reference_image"),
|
|
names=[f"image_{i}" for i in range(1, 10)],
|
|
min=0,
|
|
),
|
|
tooltip="Up to 9 images of people, products, or places to use in the "
|
|
"video; refer to them as @Image1, @Image2, ...",
|
|
),
|
|
IO.Autogrow.Input(
|
|
"reference_videos",
|
|
template=IO.Autogrow.TemplateNames(
|
|
IO.Video.Input("reference_video"),
|
|
names=[f"video_{i}" for i in range(1, 4)],
|
|
min=0,
|
|
),
|
|
tooltip="Up to 3 videos to use as references; refer to them as "
|
|
"@Video1, @Video2, ...",
|
|
),
|
|
IO.Autogrow.Input(
|
|
"reference_audios",
|
|
template=IO.Autogrow.TemplateNames(
|
|
IO.Audio.Input("reference_audio"),
|
|
names=[f"audio_{i}" for i in range(1, 4)],
|
|
min=0,
|
|
),
|
|
tooltip="Up to 3 audio clips, such as a voice for a speaker; refer to "
|
|
"them as @Audio1, @Audio2, ... Requires at least one reference image or "
|
|
"video. A voice reference needs a few seconds of clean speech; clips "
|
|
"shorter than about 2 seconds are usually ignored.",
|
|
),
|
|
],
|
|
),
|
|
],
|
|
),
|
|
],
|
|
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=["model", "model.duration", "model.resolution"],
|
|
input_groups=["model.reference_images", "model.reference_videos"],
|
|
),
|
|
expr="""
|
|
(
|
|
$dur := $lookup(widgets, "model.duration");
|
|
$res := $lookup(widgets, "model.resolution");
|
|
$imgsRaw := $lookup(inputGroups, "model.reference_images");
|
|
$imgs := $imgsRaw ? $imgsRaw : 0;
|
|
$vidsRaw := $lookup(inputGroups, "model.reference_videos");
|
|
$vids := $vidsRaw ? $vidsRaw : 0;
|
|
$rate := ($imgs + $vids) > 0
|
|
? $lookup({"480p": 0.0286, "768p": 0.0429, "2k": 0.1287}, $res)
|
|
: $lookup({"480p": 0.0143, "768p": 0.02145, "2k": 0.06435}, $res);
|
|
$vids > 0
|
|
? {"type":"range_usd","min_usd": $rate * $dur, "max_usd": $rate * ($dur + 5 * $vids)}
|
|
: {"type":"usd","usd": $rate * $dur}
|
|
)
|
|
""",
|
|
),
|
|
)
|
|
|
|
@classmethod
|
|
async def execute(cls, model: dict) -> IO.NodeOutput:
|
|
reference_images = model.get("reference_images", {})
|
|
reference_videos = model.get("reference_videos", {})
|
|
reference_audios = model.get("reference_audios", {})
|
|
if reference_audios and not (reference_images or reference_videos):
|
|
raise ValueError("Reference audio requires at least one reference image or video.")
|
|
total = len(reference_images) + len(reference_videos) + len(reference_audios)
|
|
if total > 12:
|
|
raise ValueError(f"At most 12 references can be connected in total; got {total}.")
|
|
if model["resolution"] == "2k" and model["aspect_ratio"] not in ("16:9", "9:16"):
|
|
if model["aspect_ratio"] != "auto" or reference_images or reference_videos:
|
|
raise ValueError("2k resolution requires the 16:9 or 9:16 aspect ratio.")
|
|
for key, image in reference_images.items():
|
|
if get_number_of_images(image) != 1:
|
|
raise ValueError(f"Reference image input '{key}' must contain exactly one image, not a batch.")
|
|
validate_image_aspect_ratio(image, (1, 4), (4, 1), strict=False)
|
|
prompt = _rewrite_reference_tags(
|
|
model["prompt"],
|
|
{"image": len(reference_images), "video": len(reference_videos), "audio": len(reference_audios)},
|
|
)
|
|
validate_string(prompt, strip_whitespace=True, min_length=1, max_length=32000)
|
|
payload = {
|
|
"model": model["model"],
|
|
"mode": "reference_to_video" if reference_images or reference_videos else "text_to_video",
|
|
"prompt": prompt,
|
|
"duration": model["duration"],
|
|
"resolution": model["resolution"],
|
|
"seed": model["seed"],
|
|
}
|
|
if model["aspect_ratio"] != "auto":
|
|
payload["aspect_ratio"] = model["aspect_ratio"]
|
|
if reference_images:
|
|
urls = await upload_images_to_comfyapi(
|
|
cls, list(reference_images.values()), max_images=9, mime_type="image/png"
|
|
)
|
|
payload["reference_images"] = [{"type": "url", "url": u} for u in urls]
|
|
if reference_videos:
|
|
payload["reference_videos"] = [
|
|
{"type": "url", "url": await upload_video_to_comfyapi(cls, v)} for v in reference_videos.values()
|
|
]
|
|
if reference_audios:
|
|
payload["reference_audio"] = [
|
|
{
|
|
"type": "url",
|
|
"url": await upload_audio_to_comfyapi(
|
|
cls, a, container_format="mp3", codec_name="libmp3lame", mime_type="audio/mpeg"
|
|
),
|
|
}
|
|
for a in reference_audios.values()
|
|
]
|
|
video = await _create_and_poll_video(cls, _MODEL_VIDEOS_PATH, payload)
|
|
return IO.NodeOutput(await download_url_to_video_output(video["video_url"]))
|
|
|
|
|
|
class HeyGenImageToVideoNode(IO.ComfyNode):
|
|
|
|
@classmethod
|
|
def define_schema(cls) -> IO.Schema:
|
|
return IO.Schema(
|
|
node_id="HeyGenImageToVideoNode",
|
|
display_name="HeyGen Video 1.0 Image to Video",
|
|
category="partner/video/HeyGen",
|
|
description="Animate an image into a video with synchronized dialogue and sound using "
|
|
"HeyGen Video 1.0. The image is used as the first frame.",
|
|
inputs=[
|
|
IO.DynamicCombo.Input(
|
|
"model",
|
|
options=[
|
|
IO.DynamicCombo.Option(
|
|
"heygen-video-1",
|
|
[
|
|
IO.Image.Input(
|
|
"image",
|
|
tooltip="First frame of the video. The output keeps the aspect ratio "
|
|
"of this image; crop it to change the shape of the video.",
|
|
),
|
|
IO.String.Input(
|
|
"prompt",
|
|
multiline=True,
|
|
default="",
|
|
tooltip="Description of what happens in the video, including any dialogue.",
|
|
),
|
|
IO.Int.Input(
|
|
"duration",
|
|
default=5,
|
|
min=5,
|
|
max=15,
|
|
step=1,
|
|
display_mode=IO.NumberDisplay.slider,
|
|
tooltip="Duration of the output video in seconds.",
|
|
),
|
|
IO.Combo.Input(
|
|
"resolution",
|
|
options=["768p", "480p", "2k"],
|
|
tooltip="Output resolution.",
|
|
),
|
|
IO.Int.Input(
|
|
"seed",
|
|
default=42,
|
|
min=0,
|
|
max=4294967295,
|
|
control_after_generate=True,
|
|
tooltip="Seed for the generation. Results can still vary between runs "
|
|
"with the same seed.",
|
|
),
|
|
],
|
|
),
|
|
],
|
|
),
|
|
],
|
|
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=["model", "model.duration", "model.resolution"]),
|
|
expr="""
|
|
{"type":"usd","usd": $lookup({"480p": 0.0143, "768p": 0.02145, "2k": 0.06435}, $lookup(widgets, "model.resolution"))
|
|
* $lookup(widgets, "model.duration")}
|
|
""",
|
|
),
|
|
)
|
|
|
|
@classmethod
|
|
async def execute(cls, model: dict) -> IO.NodeOutput:
|
|
validate_string(model["prompt"], strip_whitespace=True, min_length=1, max_length=32000)
|
|
if get_number_of_images(model["image"]) != 1:
|
|
raise ValueError("The image input must contain exactly one image, not a batch.")
|
|
validate_image_aspect_ratio(model["image"], (1, 4), (4, 1), strict=False)
|
|
image_url = await upload_image_to_comfyapi(cls, model["image"], mime_type="image/png")
|
|
payload = {
|
|
"model": model["model"],
|
|
"mode": "image_to_video",
|
|
"prompt": model["prompt"],
|
|
"image": {"type": "url", "url": image_url},
|
|
"duration": model["duration"],
|
|
"resolution": model["resolution"],
|
|
"seed": model["seed"],
|
|
}
|
|
video = await _create_and_poll_video(cls, _MODEL_VIDEOS_PATH, payload)
|
|
return IO.NodeOutput(await download_url_to_video_output(video["video_url"]))
|
|
|
|
|
|
class HeyGenVoiceCloneNode(IO.ComfyNode):
|
|
|
|
@classmethod
|
|
def define_schema(cls) -> IO.Schema:
|
|
return IO.Schema(
|
|
node_id="HeyGenVoiceCloneNode",
|
|
display_name="HeyGen Voice 1.0 Instant Clone",
|
|
category="partner/audio/HeyGen",
|
|
description="Clone a voice from a recording of one speaker with HeyGen Voice 1.0. Connect the voice_id "
|
|
"to HeyGen Voice 1.0 Text to Speech, or to custom_voice_id in HeyGen Avatar Video and HeyGen Talking Photo.",
|
|
inputs=[
|
|
IO.Audio.Input(
|
|
"audio",
|
|
tooltip="Recording of one speaker. Use at least 10 seconds of clear speech; "
|
|
"only the first 3 minutes are used.",
|
|
),
|
|
IO.Combo.Input(
|
|
"language",
|
|
options=_VOICE_LANGUAGE_OPTIONS,
|
|
tooltip="Language spoken in the recording.",
|
|
),
|
|
IO.Combo.Input(
|
|
"similarity",
|
|
options=["medium", "high", "extra_high", "max"],
|
|
default="max",
|
|
tooltip="How closely the voice follows the recording.",
|
|
),
|
|
IO.Boolean.Input("remove_background_noise", default=True),
|
|
IO.Int.Input(
|
|
"seed",
|
|
default=42,
|
|
min=0,
|
|
max=2147483647,
|
|
control_after_generate=IO.ControlAfterGenerate.fixed,
|
|
tooltip="Change the seed to clone the recording again; each clone is a slightly different voice.",
|
|
),
|
|
],
|
|
outputs=[IO.String.Output(display_name="voice_id")],
|
|
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}"""),
|
|
)
|
|
|
|
@classmethod
|
|
async def execute(
|
|
cls,
|
|
audio: Input.Audio,
|
|
language: str,
|
|
similarity: str,
|
|
remove_background_noise: bool,
|
|
seed: int,
|
|
) -> IO.NodeOutput:
|
|
sample_rate = audio["sample_rate"]
|
|
audio_url = await upload_audio_to_comfyapi(
|
|
cls,
|
|
{"waveform": audio["waveform"][..., : 180 * sample_rate], "sample_rate": sample_rate},
|
|
container_format="mp3",
|
|
codec_name="libmp3lame",
|
|
mime_type="audio/mpeg",
|
|
)
|
|
payload = {
|
|
"mode": "instant",
|
|
"name": "ComfyUI Voice",
|
|
"audio": [{"type": "url", "url": audio_url}],
|
|
"similarity": similarity,
|
|
"remove_background_noise": remove_background_noise,
|
|
}
|
|
if language != "auto":
|
|
payload["language"] = HEYGEN_VOICE_LANGUAGES[language]
|
|
created = await sync_op_raw(
|
|
cls,
|
|
ApiEndpoint(path=_MODEL_VOICES_PATH, method="POST"),
|
|
data=payload,
|
|
)
|
|
voice_id = (created.get("data") or {}).get("voice_id")
|
|
if not voice_id:
|
|
raise ValueError(f"HeyGen did not return a voice_id: {created}")
|
|
final = await poll_op_raw(
|
|
cls,
|
|
ApiEndpoint(path=f"{_MODEL_VOICES_PATH}/{voice_id}"),
|
|
status_extractor=lambda r: (r.get("data") or {}).get("status"),
|
|
completed_statuses=["active", "failed"],
|
|
poll_interval=2.0,
|
|
)
|
|
if final["data"]["status"].strip().lower() == "failed":
|
|
raise ValueError(
|
|
f"HeyGen could not create a voice from this recording ({final['data'].get('failure_reason')})."
|
|
)
|
|
PromptServer.instance.send_progress_text(
|
|
f"Please save the voice_id for reuse.\n\nvoice_id: {voice_id}",
|
|
cls.hidden.unique_id,
|
|
)
|
|
return IO.NodeOutput(voice_id)
|
|
|
|
|
|
def _voice_tts_inputs() -> list:
|
|
return [
|
|
IO.String.Input(
|
|
"voice_id",
|
|
default="",
|
|
tooltip="Voice from HeyGen Voice 1.0 Instant Clone.",
|
|
),
|
|
IO.String.Input(
|
|
"text",
|
|
multiline=True,
|
|
default="",
|
|
tooltip="Up to 5000 characters. Speech longer than about 12 minutes fails, "
|
|
"so split very long Chinese or Japanese text.",
|
|
),
|
|
IO.Combo.Input(
|
|
"language",
|
|
options=_VOICE_LANGUAGE_OPTIONS,
|
|
tooltip="Language of the text. Set it for closely related languages such as Serbian or Malay, "
|
|
"which auto-detection can confuse.",
|
|
),
|
|
IO.Float.Input(
|
|
"expressiveness_boost",
|
|
default=0.5,
|
|
min=0.0,
|
|
max=1.0,
|
|
step=0.05,
|
|
tooltip="Higher values are more expressive; lower values stay closer to the original voice.",
|
|
),
|
|
]
|
|
|
|
|
|
class HeyGenVoiceTextToSpeechNode(IO.ComfyNode):
|
|
|
|
@classmethod
|
|
def define_schema(cls) -> IO.Schema:
|
|
return IO.Schema(
|
|
node_id="HeyGenVoiceTextToSpeechNode",
|
|
display_name="HeyGen Voice 1.0 Text to Speech",
|
|
category="partner/audio/HeyGen",
|
|
description="Speak text in a voice cloned with HeyGen Voice 1.0 Instant Clone.",
|
|
inputs=[
|
|
IO.DynamicCombo.Input(
|
|
"model",
|
|
options=[
|
|
IO.DynamicCombo.Option("heygen-voice-1", _voice_tts_inputs()),
|
|
IO.DynamicCombo.Option("heygen-voice-1-turbo", _voice_tts_inputs()),
|
|
],
|
|
),
|
|
],
|
|
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(
|
|
depends_on=IO.PriceBadgeDepends(widgets=["model", "model.text"]),
|
|
expr="""
|
|
(
|
|
$t := $lookup(widgets, "model.text");
|
|
$promo := $now() < "2026-11-01T00:00:00.000Z";
|
|
$rate := widgets.model = "heygen-voice-1-turbo"
|
|
? ($promo ? 0.00572 : 0.010725)
|
|
: ($promo ? 0.02145 : 0.0429);
|
|
$type($t) = "string" and $length($t) > 0
|
|
? {"type":"usd","usd": $length($t) * $rate / 1000}
|
|
: {"type":"usd","usd": $rate, "format":{"approximate":true,"suffix":"/1K chars"}}
|
|
)
|
|
""",
|
|
),
|
|
)
|
|
|
|
@classmethod
|
|
async def execute(cls, model: dict) -> IO.NodeOutput:
|
|
voice_id = model["voice_id"].strip()
|
|
if not voice_id:
|
|
raise ValueError("Connect a voice_id from HeyGen Voice 1.0 Instant Clone.")
|
|
text = model["text"].strip()
|
|
validate_string(text, field_name="text", min_length=1, max_length=5000)
|
|
payload = {
|
|
"model": model["model"],
|
|
"voice_id": voice_id,
|
|
"text": text,
|
|
"expressiveness_boost": model["expressiveness_boost"],
|
|
}
|
|
if model["language"] != "auto":
|
|
payload["language"] = HEYGEN_VOICE_LANGUAGES[model["language"]]
|
|
response = await sync_op_raw(
|
|
cls,
|
|
ApiEndpoint(path=_MODEL_TTS_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,
|
|
HeyGenReferenceToVideoNode,
|
|
HeyGenImageToVideoNode,
|
|
HeyGenVoiceCloneNode,
|
|
HeyGenVoiceTextToSpeechNode,
|
|
]
|
|
|
|
|
|
async def comfy_entrypoint() -> HeyGenExtension:
|
|
return HeyGenExtension()
|