935 lines
30 KiB
Python
935 lines
30 KiB
Python
# import time
|
|
# from pathlib import Path
|
|
# from sync import Sync
|
|
# from sync.common import Audio, Video, GenerationOptions
|
|
# from sync.core.api_error import ApiError
|
|
|
|
# class SyncLipsyncNode:
|
|
# @classmethod
|
|
# def INPUT_TYPES(cls):
|
|
# return {
|
|
# "required": {
|
|
# "video_url": ("STRING", {"default": "https://assets.sync.so/docs/example-video.mp4"}),
|
|
# "audio_url": ("STRING", {"default": "https://assets.sync.so/docs/example-audio.wav"}),
|
|
# "api_key": ("STRING", {"default": ""}),
|
|
# "poll_interval": ("FLOAT", {"default": 5.0, "min": 1.0, "max": 60.0}),
|
|
# }
|
|
# }
|
|
|
|
# RETURN_TYPES = ("STRING",)
|
|
# RETURN_NAMES = ("output_url",)
|
|
# FUNCTION = "lipsync_generate"
|
|
# CATEGORY = "Sync.so"
|
|
# OUTPUT_NODE = True # ✅ This makes it an output node
|
|
|
|
# def lipsync_generate(self, video_url, audio_url, api_key, poll_interval):
|
|
# client = Sync(base_url="https://api.sync.so", api_key=api_key).generations
|
|
|
|
# try:
|
|
# response = client.create(
|
|
# input=[Video(url=video_url), Audio(url=audio_url)],
|
|
# model="lipsync-2",
|
|
# options=GenerationOptions(sync_mode="cut_off"),
|
|
# )
|
|
# except ApiError as e:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Error creating generation: {e.status_code} - {e.body}",)
|
|
# }
|
|
|
|
# job_id = response.id
|
|
# status = None
|
|
# while status not in ['COMPLETED', 'FAILED']:
|
|
# time.sleep(poll_interval)
|
|
# generation = client.get(job_id)
|
|
# status = generation.status
|
|
|
|
# if status == 'COMPLETED':
|
|
# output_url = generation.output_url
|
|
# filename = f"sync_lipsync_output_{int(time.time())}.txt"
|
|
# output_dir = Path("output")
|
|
# output_dir.mkdir(exist_ok=True)
|
|
# output_path = output_dir / filename
|
|
|
|
# with open(output_path, "w") as f:
|
|
# f.write(output_url)
|
|
|
|
# return {
|
|
# "ui": {
|
|
# "texts": [{
|
|
# "filename": filename,
|
|
# "subfolder": "",
|
|
# "type": "output"
|
|
# }]
|
|
# },
|
|
# "result": (output_url,)
|
|
# }
|
|
# else:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Generation failed for job {job_id}",)
|
|
# }
|
|
|
|
# # Node registration for ComfyUI
|
|
# NODE_CLASS_MAPPINGS = {
|
|
# "SyncLipsyncNode": SyncLipsyncNode,
|
|
# }
|
|
|
|
# NODE_DISPLAY_NAME_MAPPINGS = {
|
|
# "SyncLipsyncNode": "Sync.so Lipsync Generator"
|
|
# }
|
|
|
|
|
|
|
|
|
|
#-----------------------------------------------------------------------------------------------
|
|
#Save the video
|
|
|
|
# import time
|
|
# import requests
|
|
# from pathlib import Path
|
|
# from sync import Sync
|
|
# from sync.common import Audio, Video, GenerationOptions
|
|
# from sync.core.api_error import ApiError
|
|
|
|
|
|
# class SyncLipsyncNode:
|
|
# @classmethod
|
|
# def INPUT_TYPES(cls):
|
|
# return {
|
|
# "required": {
|
|
# "video_url": ("STRING", {"default": "https://assets.sync.so/docs/example-video.mp4"}),
|
|
# "audio_url": ("STRING", {"default": "https://assets.sync.so/docs/example-audio.wav"}),
|
|
# "api_key": ("STRING", {"default": ""}),
|
|
# "poll_interval": ("FLOAT", {"default": 5.0, "min": 1.0, "max": 60.0}),
|
|
# }
|
|
# }
|
|
|
|
# RETURN_TYPES = ("STRING",)
|
|
# RETURN_NAMES = ("output_path",)
|
|
# FUNCTION = "lipsync_generate"
|
|
# CATEGORY = "Sync.so"
|
|
# OUTPUT_NODE = True
|
|
|
|
# def lipsync_generate(self, video_url, audio_url, api_key, poll_interval):
|
|
# client = Sync(base_url="https://api.sync.so", api_key=api_key).generations
|
|
|
|
# try:
|
|
# response = client.create(
|
|
# input=[Video(url=video_url), Audio(url=audio_url)],
|
|
# model="lipsync-2",
|
|
# options=GenerationOptions(sync_mode="cut_off"),
|
|
# )
|
|
# except ApiError as e:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Error creating generation: {e.status_code} - {e.body}",)
|
|
# }
|
|
|
|
# job_id = response.id
|
|
# status = None
|
|
# while status not in ['COMPLETED', 'FAILED']:
|
|
# time.sleep(poll_interval)
|
|
# generation = client.get(job_id)
|
|
# status = generation.status
|
|
|
|
# if status == 'COMPLETED':
|
|
# output_url = generation.output_url
|
|
# timestamp = int(time.time())
|
|
# output_dir = Path("output")
|
|
# output_dir.mkdir(exist_ok=True)
|
|
# output_filename = f"sync_output_{timestamp}.mp4"
|
|
# output_path = output_dir / output_filename
|
|
|
|
# try:
|
|
# r = requests.get(output_url)
|
|
# r.raise_for_status()
|
|
# with open(output_path, "wb") as f:
|
|
# f.write(r.content)
|
|
# except Exception as e:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Failed to download video: {str(e)}",)
|
|
# }
|
|
|
|
# return {
|
|
# "ui": {
|
|
# "videos": [{
|
|
# "filename": output_filename,
|
|
# "subfolder": "",
|
|
# "type": "output"
|
|
# }]
|
|
# },
|
|
# "result": (str(output_path),)
|
|
# }
|
|
# else:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Generation failed for job {job_id}",)
|
|
# }
|
|
|
|
|
|
# # Node registration for ComfyUI
|
|
# NODE_CLASS_MAPPINGS = {
|
|
# "SyncLipsyncNode": SyncLipsyncNode,
|
|
# }
|
|
|
|
# NODE_DISPLAY_NAME_MAPPINGS = {
|
|
# "SyncLipsyncNode": "Sync.so Lipsync Generator"
|
|
# }
|
|
|
|
# print("✅ Sync.so node with video preview loaded.")
|
|
|
|
# With multiple model support
|
|
|
|
# import time
|
|
# import requests
|
|
# from pathlib import Path
|
|
# from sync import Sync
|
|
# from sync.common import Audio, Video, GenerationOptions
|
|
# from sync.core.api_error import ApiError
|
|
|
|
|
|
# class SyncLipsyncNode:
|
|
# @classmethod
|
|
# def INPUT_TYPES(cls):
|
|
# return {
|
|
# "required": {
|
|
# "video_url": ("STRING", {"default": "https://assets.sync.so/docs/example-video.mp4"}),
|
|
# "audio_url": ("STRING", {"default": "https://assets.sync.so/docs/example-audio.wav"}),
|
|
# "api_key": ("STRING", {"default": ""}),
|
|
# "poll_interval": ("FLOAT", {"default": 5.0, "min": 1.0, "max": 60.0}),
|
|
# "model": (["lipsync-2", "lipsync-1.9.0-beta"],),
|
|
# }
|
|
# }
|
|
|
|
# RETURN_TYPES = ("STRING",)
|
|
# RETURN_NAMES = ("output_path",)
|
|
# FUNCTION = "lipsync_generate"
|
|
# CATEGORY = "Sync.so"
|
|
# OUTPUT_NODE = True
|
|
|
|
# def lipsync_generate(self, video_url, audio_url, api_key, poll_interval, model):
|
|
# client = Sync(base_url="https://api.sync.so", api_key=api_key).generations
|
|
|
|
# try:
|
|
# response = client.create(
|
|
# input=[Video(url=video_url), Audio(url=audio_url)],
|
|
# model=model,
|
|
# options=GenerationOptions(sync_mode="cut_off"),
|
|
# )
|
|
# except ApiError as e:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Error creating generation: {e.status_code} - {e.body}",)
|
|
# }
|
|
|
|
# job_id = response.id
|
|
# status = None
|
|
# while status not in ['COMPLETED', 'FAILED']:
|
|
# time.sleep(poll_interval)
|
|
# generation = client.get(job_id)
|
|
# status = generation.status
|
|
|
|
# if status == 'COMPLETED':
|
|
# output_url = generation.output_url
|
|
# timestamp = int(time.time())
|
|
# output_dir = Path("output")
|
|
# output_dir.mkdir(exist_ok=True)
|
|
# output_filename = f"sync_output_{timestamp}.mp4"
|
|
# output_path = output_dir / output_filename
|
|
|
|
# try:
|
|
# r = requests.get(output_url)
|
|
# r.raise_for_status()
|
|
# with open(output_path, "wb") as f:
|
|
# f.write(r.content)
|
|
# except Exception as e:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Failed to download video: {str(e)}",)
|
|
# }
|
|
|
|
# return {
|
|
# "ui": {
|
|
# "videos": [{
|
|
# "filename": output_filename,
|
|
# "subfolder": "",
|
|
# "type": "output"
|
|
# }]
|
|
# },
|
|
# "result": (str(output_path),)
|
|
# }
|
|
# else:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Generation failed for job {job_id}",)
|
|
# }
|
|
|
|
|
|
# # Node registration for ComfyUI
|
|
# NODE_CLASS_MAPPINGS = {
|
|
# "SyncLipsyncNode": SyncLipsyncNode,
|
|
# }
|
|
|
|
# NODE_DISPLAY_NAME_MAPPINGS = {
|
|
# "SyncLipsyncNode": "Sync.so Lipsync Generator"
|
|
# }
|
|
|
|
# print("✅ Sync.so node with video preview loaded.")
|
|
|
|
|
|
|
|
|
|
# import time
|
|
# import requests
|
|
# from pathlib import Path
|
|
# from sync import Sync
|
|
# from sync.common import Audio, Video, GenerationOptions
|
|
# from sync.core.api_error import ApiError
|
|
|
|
|
|
# class SyncLipsyncNode:
|
|
# @classmethod
|
|
# def INPUT_TYPES(cls):
|
|
# return {
|
|
# "required": {
|
|
# "video_url": ("STRING", {"default": "https://assets.sync.so/docs/example-video.mp4"}),
|
|
# "audio_url": ("STRING", {"default": "https://assets.sync.so/docs/example-audio.wav"}),
|
|
# "api_key": ("STRING", {"default": ""}),
|
|
# "poll_interval": ("FLOAT", {"default": 5.0, "min": 1.0, "max": 60.0}),
|
|
# "model": (["lipsync-2", "lipsync-1.9.0-beta"],),
|
|
# "segment_secs": ("STRING", {"default": ""}),
|
|
# "segment_frames": ("STRING", {"default": ""}),
|
|
# }
|
|
# }
|
|
|
|
# RETURN_TYPES = ("STRING",)
|
|
# RETURN_NAMES = ("output_path",)
|
|
# FUNCTION = "lipsync_generate"
|
|
# CATEGORY = "Sync.so"
|
|
# OUTPUT_NODE = True
|
|
|
|
# def lipsync_generate(self, video_url, audio_url, api_key, poll_interval, model, segment_secs, segment_frames):
|
|
# client = Sync(base_url="https://api.sync.so", api_key=api_key).generations
|
|
|
|
# # Prepare video kwargs
|
|
# video_kwargs = {}
|
|
# if segment_secs:
|
|
# try:
|
|
# video_kwargs["segments_secs"] = eval(segment_secs)
|
|
# except Exception:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": ("Invalid format for segment_secs. Example: [[5.0, 10.0]]",)
|
|
# }
|
|
|
|
# if segment_frames:
|
|
# try:
|
|
# video_kwargs["segments_frames"] = eval(segment_frames)
|
|
# except Exception:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": ("Invalid format for segment_frames. Example: [[100, 200]]",)
|
|
# }
|
|
|
|
# try:
|
|
# response = client.create(
|
|
# input=[Video(url=video_url, **video_kwargs), Audio(url=audio_url)],
|
|
# model=model,
|
|
# options=GenerationOptions(sync_mode="cut_off"),
|
|
# )
|
|
# except ApiError as e:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Error creating generation: {e.status_code} - {e.body}",)
|
|
# }
|
|
|
|
# job_id = response.id
|
|
# status = None
|
|
# while status not in ['COMPLETED', 'FAILED']:
|
|
# time.sleep(poll_interval)
|
|
# generation = client.get(job_id)
|
|
# status = generation.status
|
|
|
|
# if status == 'COMPLETED':
|
|
# output_url = generation.output_url
|
|
# timestamp = int(time.time())
|
|
# output_dir = Path("output")
|
|
# output_dir.mkdir(exist_ok=True)
|
|
# output_filename = f"sync_output_{timestamp}.mp4"
|
|
# output_path = output_dir / output_filename
|
|
|
|
# try:
|
|
# r = requests.get(output_url)
|
|
# r.raise_for_status()
|
|
# with open(output_path, "wb") as f:
|
|
# f.write(r.content)
|
|
# except Exception as e:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Failed to download video: {str(e)}",)
|
|
# }
|
|
|
|
# return {
|
|
# "ui": {
|
|
# "videos": [{
|
|
# "filename": output_filename,
|
|
# "subfolder": "",
|
|
# "type": "output"
|
|
# }]
|
|
# },
|
|
# "result": (str(output_path),)
|
|
# }
|
|
# else:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Generation failed for job {job_id}",)
|
|
# }
|
|
|
|
|
|
# # Node registration for ComfyUI
|
|
# NODE_CLASS_MAPPINGS = {
|
|
# "SyncLipsyncNode": SyncLipsyncNode,
|
|
# }
|
|
|
|
# NODE_DISPLAY_NAME_MAPPINGS = {
|
|
# "SyncLipsyncNode": "Sync.so Lipsync Generator"
|
|
# }
|
|
|
|
# print("✅ Sync.so node with video preview loaded.")
|
|
|
|
|
|
|
|
|
|
# IS this the END!!!! Filhaal ye acha hai
|
|
|
|
|
|
|
|
import time
|
|
import requests
|
|
from pathlib import Path
|
|
from sync import Sync
|
|
from sync.common import Audio, Video, GenerationOptions
|
|
from sync.core.api_error import ApiError
|
|
|
|
|
|
class SyncLipsyncNode:
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"required": {
|
|
"video_url": ("STRING", {"default": "https://assets.sync.so/docs/example-video.mp4"}),
|
|
"audio_url": ("STRING", {"default": "https://assets.sync.so/docs/example-audio.wav"}),
|
|
"api_key": ("STRING", {"default": ""}),
|
|
"poll_interval": ("FLOAT", {"default": 5.0, "min": 1.0, "max": 60.0}),
|
|
"model": (["lipsync-2", "lipsync-1.9.0-beta"],),
|
|
"segment_secs": ("STRING", {"default": ""}),
|
|
"segment_frames": ("STRING", {"default": ""}),
|
|
"sync_mode": (["loop", "bounce", "cut_off", "silence", "remap"], {"default": "cut_off"}),
|
|
"temperature": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 1.0}),
|
|
"active_speaker": ("BOOLEAN", {"default": False}),
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("STRING",)
|
|
RETURN_NAMES = ("output_path",)
|
|
FUNCTION = "lipsync_generate"
|
|
CATEGORY = "Sync.so"
|
|
OUTPUT_NODE = True
|
|
|
|
def lipsync_generate(
|
|
self,
|
|
video_url,
|
|
audio_url,
|
|
api_key,
|
|
poll_interval,
|
|
model,
|
|
segment_secs,
|
|
segment_frames,
|
|
sync_mode,
|
|
temperature,
|
|
active_speaker,
|
|
):
|
|
client = Sync(base_url="https://api.sync.so", api_key=api_key).generations
|
|
|
|
# Prepare video kwargs
|
|
video_kwargs = {}
|
|
if segment_secs:
|
|
try:
|
|
video_kwargs["segments_secs"] = eval(segment_secs)
|
|
except Exception:
|
|
return {
|
|
"ui": {"texts": []},
|
|
"result": ("Invalid format for segment_secs. Example: [[5.0, 10.0]]",)
|
|
}
|
|
|
|
if segment_frames:
|
|
try:
|
|
video_kwargs["segments_frames"] = eval(segment_frames)
|
|
except Exception:
|
|
return {
|
|
"ui": {"texts": []},
|
|
"result": ("Invalid format for segment_frames. Example: [[100, 200]]",)
|
|
}
|
|
|
|
try:
|
|
response = client.create(
|
|
input=[Video(url=video_url, **video_kwargs), Audio(url=audio_url)],
|
|
model=model,
|
|
options=GenerationOptions(
|
|
sync_mode=sync_mode,
|
|
temperature=temperature,
|
|
active_speaker=active_speaker,
|
|
),
|
|
)
|
|
except ApiError as e:
|
|
return {
|
|
"ui": {"texts": []},
|
|
"result": (f"Error creating generation: {e.status_code} - {e.body}",)
|
|
}
|
|
|
|
job_id = response.id
|
|
status = None
|
|
while status not in ['COMPLETED', 'FAILED']:
|
|
time.sleep(poll_interval)
|
|
generation = client.get(job_id)
|
|
status = generation.status
|
|
|
|
if status == 'COMPLETED':
|
|
output_url = generation.output_url
|
|
timestamp = int(time.time())
|
|
output_dir = Path("output")
|
|
output_dir.mkdir(exist_ok=True)
|
|
output_filename = f"sync_output_{timestamp}.mp4"
|
|
output_path = output_dir / output_filename
|
|
|
|
try:
|
|
r = requests.get(output_url)
|
|
r.raise_for_status()
|
|
with open(output_path, "wb") as f:
|
|
f.write(r.content)
|
|
except Exception as e:
|
|
return {
|
|
"ui": {"texts": []},
|
|
"result": (f"Failed to download video: {str(e)}",)
|
|
}
|
|
|
|
return {
|
|
"ui": {
|
|
"videos": [{
|
|
"filename": output_filename,
|
|
"subfolder": "",
|
|
"type": "output"
|
|
}]
|
|
},
|
|
"result": (str(output_path),)
|
|
}
|
|
else:
|
|
return {
|
|
"ui": {"texts": []},
|
|
"result": (f"Generation failed for job {job_id}",)
|
|
}
|
|
|
|
|
|
# Node registration for ComfyUI
|
|
NODE_CLASS_MAPPINGS = {
|
|
"SyncLipsyncNode": SyncLipsyncNode,
|
|
}
|
|
|
|
NODE_DISPLAY_NAME_MAPPINGS = {
|
|
"SyncLipsyncNode": "Sync.so Lipsync Generator"
|
|
}
|
|
|
|
print("✅ Sync.so node with video preview loaded.")
|
|
|
|
###ENDDDDDD
|
|
|
|
# import time
|
|
# import json
|
|
# import requests
|
|
# from pathlib import Path
|
|
# from sync import Sync
|
|
# from sync.common import Audio, Video, GenerationOptions
|
|
# from sync.core.api_error import ApiError
|
|
|
|
|
|
# class SyncLipsyncNode:
|
|
# @classmethod
|
|
# def INPUT_TYPES(cls):
|
|
# return {
|
|
# "required": {
|
|
# "video_url": ("STRING", {"default": "https://assets.sync.so/docs/example-video.mp4"}),
|
|
# "audio_url": ("STRING", {"default": ""}), # optional if TTS used
|
|
# "tts_text": ("STRING", {"default": ""}), # text for TTS
|
|
# "tts_provider": ("STRING", {"default": ""}), # JSON string for TTS provider config
|
|
# "api_key": ("STRING", {"default": ""}),
|
|
# "poll_interval": ("FLOAT", {"default": 5.0, "min": 1.0, "max": 60.0}),
|
|
# "model": (["lipsync-2", "lipsync-1.9.0-beta"],),
|
|
# "segment_secs": ("STRING", {"default": ""}),
|
|
# "segment_frames": ("STRING", {"default": ""}),
|
|
# "sync_mode": (["loop", "bounce", "cut_off", "silence", "remap"], {"default": "cut_off"}),
|
|
# "temperature": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 1.0}),
|
|
# "active_speaker": ("BOOLEAN", {"default": False}),
|
|
# }
|
|
# }
|
|
|
|
# RETURN_TYPES = ("STRING",)
|
|
# RETURN_NAMES = ("output_path",)
|
|
# FUNCTION = "lipsync_generate"
|
|
# CATEGORY = "Sync.so"
|
|
# OUTPUT_NODE = True
|
|
|
|
# def lipsync_generate(
|
|
# self,
|
|
# video_url,
|
|
# audio_url,
|
|
# tts_text,
|
|
# tts_provider,
|
|
# api_key,
|
|
# poll_interval,
|
|
# model,
|
|
# segment_secs,
|
|
# segment_frames,
|
|
# sync_mode,
|
|
# temperature,
|
|
# active_speaker,
|
|
# ):
|
|
# client = Sync(base_url="https://api.sync.so", api_key=api_key).generations
|
|
|
|
# # Prepare video kwargs
|
|
# video_kwargs = {}
|
|
# if segment_secs:
|
|
# try:
|
|
# video_kwargs["segments_secs"] = eval(segment_secs)
|
|
# except Exception:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": ("Invalid format for segment_secs. Example: [[5.0, 10.0]]",)
|
|
# }
|
|
|
|
# if segment_frames:
|
|
# try:
|
|
# video_kwargs["segments_frames"] = eval(segment_frames)
|
|
# except Exception:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": ("Invalid format for segment_frames. Example: [[100, 200]]",)
|
|
# }
|
|
|
|
# # Prepare inputs: video + (audio or TTS)
|
|
# inputs = [Video(url=video_url, **video_kwargs)]
|
|
|
|
# if tts_text.strip():
|
|
# # Use TTS input
|
|
# try:
|
|
# provider_config = json.loads(tts_provider) if tts_provider else {}
|
|
# except Exception as e:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Invalid JSON format for tts_provider: {str(e)}",)
|
|
# }
|
|
# inputs.append(
|
|
# {
|
|
# "type": "text",
|
|
# "provider": provider_config
|
|
# }
|
|
# )
|
|
# # Note: Sync SDK might require a special class or raw dict. Using dict as per spec.
|
|
# elif audio_url.strip():
|
|
# # Use audio URL input
|
|
# inputs.append(Audio(url=audio_url))
|
|
# else:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": ("Either audio_url or tts_text with tts_provider must be provided.",)
|
|
# }
|
|
|
|
# try:
|
|
# response = client.create(
|
|
# input=inputs,
|
|
# model=model,
|
|
# options=GenerationOptions(
|
|
# sync_mode=sync_mode,
|
|
# temperature=temperature,
|
|
# active_speaker=active_speaker,
|
|
# ),
|
|
# )
|
|
# except ApiError as e:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Error creating generation: {e.status_code} - {e.body}",)
|
|
# }
|
|
|
|
# job_id = response.id
|
|
# status = None
|
|
# while status not in ['COMPLETED', 'FAILED']:
|
|
# time.sleep(poll_interval)
|
|
# generation = client.get(job_id)
|
|
# status = generation.status
|
|
|
|
# if status == 'COMPLETED':
|
|
# output_url = generation.output_url
|
|
# timestamp = int(time.time())
|
|
# output_dir = Path("output")
|
|
# output_dir.mkdir(exist_ok=True)
|
|
# output_filename = f"sync_output_{timestamp}.mp4"
|
|
# output_path = output_dir / output_filename
|
|
|
|
# try:
|
|
# r = requests.get(output_url)
|
|
# r.raise_for_status()
|
|
# with open(output_path, "wb") as f:
|
|
# f.write(r.content)
|
|
# except Exception as e:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Failed to download video: {str(e)}",)
|
|
# }
|
|
|
|
# return {
|
|
# "ui": {
|
|
# "videos": [{
|
|
# "filename": output_filename,
|
|
# "subfolder": "",
|
|
# "type": "output"
|
|
# }]
|
|
# },
|
|
# "result": (str(output_path),)
|
|
# }
|
|
# else:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Generation failed for job {job_id}",)
|
|
# }
|
|
|
|
|
|
# # Node registration for ComfyUI
|
|
# NODE_CLASS_MAPPINGS = {
|
|
# "SyncLipsyncNode": SyncLipsyncNode,
|
|
# }
|
|
|
|
# NODE_DISPLAY_NAME_MAPPINGS = {
|
|
# "SyncLipsyncNode": "Sync.so Lipsync Generator"
|
|
# }
|
|
|
|
# print("✅ Sync.so node with video preview loaded.")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
#---------------------------------------------------------------------------------------------------------------------------
|
|
|
|
# import time
|
|
# from pathlib import Path
|
|
# from sync import Sync
|
|
# from sync.common import Audio, Video, GenerationOptions
|
|
# from sync.core.api_error import ApiError
|
|
|
|
# class SyncLipsyncNode:
|
|
# @classmethod
|
|
# def INPUT_TYPES(cls):
|
|
# return {
|
|
# "required": {
|
|
# "video": ("VIDEO",),
|
|
# "audio": ("AUDIO",),
|
|
# "api_key": ("STRING", {"default": ""}),
|
|
# "poll_interval": ("FLOAT", {"default": 5.0, "min": 1.0, "max": 60.0}),
|
|
# }
|
|
# }
|
|
|
|
# RETURN_TYPES = ("STRING",)
|
|
# RETURN_NAMES = ("output_url",)
|
|
# FUNCTION = "lipsync_generate"
|
|
# CATEGORY = "Sync.so"
|
|
# OUTPUT_NODE = True
|
|
|
|
# def lipsync_generate(self, video, audio, api_key, poll_interval):
|
|
# # Extract file paths from VIDEO/AUDIO objects
|
|
# # video_path = Path(video["filepath"]).resolve().as_uri()
|
|
# # audio_path = Path(audio["filepath"]).resolve().as_uri()
|
|
# video_path = Path(video["path"]).resolve().as_uri()
|
|
# audio_path = Path(audio["path"]).resolve().as_uri()
|
|
|
|
|
|
|
|
# client = Sync(base_url="https://api.sync.so", api_key=api_key).generations
|
|
|
|
# try:
|
|
# response = client.create(
|
|
# input=[Video(url=video_path), Audio(url=audio_path)],
|
|
# model="lipsync-2",
|
|
# options=GenerationOptions(sync_mode="cut_off"),
|
|
# )
|
|
# except ApiError as e:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Error creating generation: {e.status_code} - {e.body}",)
|
|
# }
|
|
|
|
# job_id = response.id
|
|
# status = None
|
|
# while status not in ['COMPLETED', 'FAILED']:
|
|
# time.sleep(poll_interval)
|
|
# generation = client.get(job_id)
|
|
# status = generation.status
|
|
|
|
# if status == 'COMPLETED':
|
|
# output_url = generation.output_url
|
|
# filename = f"sync_lipsync_output_{int(time.time())}.txt"
|
|
# output_dir = Path("output")
|
|
# output_dir.mkdir(exist_ok=True)
|
|
# output_path = output_dir / filename
|
|
|
|
# with open(output_path, "w") as f:
|
|
# f.write(output_url)
|
|
|
|
# return {
|
|
# "ui": {
|
|
# "texts": [{
|
|
# "filename": filename,
|
|
# "subfolder": "",
|
|
# "type": "output"
|
|
# }]
|
|
# },
|
|
# "result": (output_url,)
|
|
# }
|
|
# else:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Generation failed for job {job_id}",)
|
|
# }
|
|
|
|
# # Register the node
|
|
# NODE_CLASS_MAPPINGS = {
|
|
# "SyncLipsyncNode": SyncLipsyncNode,
|
|
# }
|
|
|
|
# NODE_DISPLAY_NAME_MAPPINGS = {
|
|
# "SyncLipsyncNode": "Sync.so Lipsync Generator"
|
|
# }
|
|
# import time
|
|
# from pathlib import Path
|
|
# from sync import Sync
|
|
# from sync.common import Audio, Video, GenerationOptions
|
|
# from sync.core.api_error import ApiError
|
|
|
|
# import soundfile as sf
|
|
# import torch
|
|
# import tempfile
|
|
|
|
# class SyncLipsyncNode:
|
|
# @classmethod
|
|
# def INPUT_TYPES(cls):
|
|
# return {
|
|
# "required": {
|
|
# "video": ("VIDEO",),
|
|
# "audio": ("AUDIO",),
|
|
# "api_key": ("STRING", {"default": ""}),
|
|
# "poll_interval": ("FLOAT", {"default": 5.0, "min": 1.0, "max": 60.0}),
|
|
# }
|
|
# }
|
|
|
|
# RETURN_TYPES = ("STRING",)
|
|
# RETURN_NAMES = ("output_url",)
|
|
# FUNCTION = "lipsync_generate"
|
|
# CATEGORY = "Sync.so"
|
|
# OUTPUT_NODE = True
|
|
|
|
# def save_audio_to_file(self, audio_dict):
|
|
# waveform = audio_dict["waveform"]
|
|
# sample_rate = audio_dict["sample_rate"]
|
|
|
|
# # Convert tensor to numpy array with shape (samples, channels)
|
|
# if isinstance(waveform, torch.Tensor):
|
|
# waveform = waveform.squeeze().permute(1, 0).cpu().numpy()
|
|
# else:
|
|
# waveform = waveform.T # Just in case it's numpy already
|
|
|
|
# # Save to temporary WAV file
|
|
# tmp_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
|
|
# sf.write(tmp_file.name, waveform, sample_rate)
|
|
# tmp_file.close()
|
|
# return tmp_file.name
|
|
|
|
# def lipsync_generate(self, video, audio, api_key, poll_interval):
|
|
# # Extract video path from VideoFromFile object (private attribute)
|
|
# video_path_str = getattr(video, "_VideoFromFile__file", None)
|
|
# if not video_path_str:
|
|
# raise RuntimeError("Video path not found")
|
|
|
|
# video_path = Path(video_path_str).resolve().as_uri()
|
|
|
|
# # Save audio waveform dict to WAV file and get path
|
|
# audio_file_path = self.save_audio_to_file(audio)
|
|
# audio_path = Path(audio_file_path).resolve().as_uri()
|
|
|
|
# client = Sync(base_url="https://api.sync.so", api_key=api_key).generations
|
|
|
|
# try:
|
|
# response = client.create(
|
|
# input=[Video(url=video_path), Audio(url=audio_path)],
|
|
# model="lipsync-2",
|
|
# options=GenerationOptions(sync_mode="cut_off"),
|
|
# )
|
|
# except ApiError as e:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Error creating generation: {e.status_code} - {e.body}",)
|
|
# }
|
|
|
|
# job_id = response.id
|
|
# status = None
|
|
# while status not in ['COMPLETED', 'FAILED']:
|
|
# time.sleep(poll_interval)
|
|
# generation = client.get(job_id)
|
|
# status = generation.status
|
|
|
|
# if status == 'COMPLETED':
|
|
# output_url = generation.output_url
|
|
# filename = f"sync_lipsync_output_{int(time.time())}.txt"
|
|
# output_dir = Path("output")
|
|
# output_dir.mkdir(exist_ok=True)
|
|
# output_path = output_dir / filename
|
|
|
|
# with open(output_path, "w") as f:
|
|
# f.write(output_url)
|
|
|
|
# return {
|
|
# "ui": {
|
|
# "texts": [{
|
|
# "filename": filename,
|
|
# "subfolder": "",
|
|
# "type": "output"
|
|
# }]
|
|
# },
|
|
# "result": (output_url,)
|
|
# }
|
|
# else:
|
|
# return {
|
|
# "ui": {"texts": []},
|
|
# "result": (f"Generation failed for job {job_id}",)
|
|
# }
|
|
|
|
# # Register the node
|
|
# NODE_CLASS_MAPPINGS = {
|
|
# "SyncLipsyncNode": SyncLipsyncNode,
|
|
# }
|
|
|
|
# NODE_DISPLAY_NAME_MAPPINGS = {
|
|
# "SyncLipsyncNode": "Sync.so Lipsync Generator"
|
|
# }
|