1099 lines
36 KiB
Python
1099 lines
36 KiB
Python
from .api import Client, ImageGenerator, ImageExpander, Image2Video, Video2Audio, Text2Audio, \
|
||
Text2Video, CameraControl, CameraControlConfig, KolorsVurtualTryOn, VideoExtend, LipSync, LipSyncInput, EffectInput, \
|
||
Effects
|
||
import base64
|
||
import io
|
||
import os
|
||
import re
|
||
import numpy
|
||
import PIL
|
||
import requests
|
||
import torch
|
||
from collections.abc import Iterable
|
||
import configparser
|
||
import folder_paths
|
||
from comfy_extras.nodes_audio import LoadAudio
|
||
from folder_paths import get_temp_directory
|
||
import time
|
||
import urllib.parse
|
||
from pathlib import Path
|
||
|
||
current_dir = os.path.dirname(os.path.abspath(__file__))
|
||
parent_dir = os.path.dirname(current_dir)
|
||
config_path = os.path.join(parent_dir, 'config.ini')
|
||
config = configparser.ConfigParser()
|
||
config.read(config_path)
|
||
|
||
|
||
def _fetch_image(url, stream=True):
|
||
return requests.get(url, stream=stream).content
|
||
|
||
|
||
def _tensor2images(tensor):
|
||
np_imgs = numpy.clip(tensor.cpu().numpy() * 255.0, 0.0, 255.0).astype(numpy.uint8)
|
||
return [PIL.Image.fromarray(np_img) for np_img in np_imgs]
|
||
|
||
|
||
def _images2tensor(images):
|
||
if isinstance(images, Iterable):
|
||
return torch.stack([torch.from_numpy(numpy.array(image)).float() / 255.0 for image in images])
|
||
return torch.from_numpy(numpy.array(images)).unsqueeze(0).float() / 255.0
|
||
|
||
|
||
def _decode_image(data_bytes, rtn_mask=False):
|
||
with io.BytesIO(data_bytes) as bytes_io:
|
||
img = PIL.Image.open(bytes_io)
|
||
if not rtn_mask:
|
||
img = img.convert('RGB')
|
||
elif 'A' in img.getbands():
|
||
img = img.getchannel('A')
|
||
else:
|
||
img = None
|
||
return img
|
||
|
||
|
||
def _encode_image(img, mask=None):
|
||
if mask is not None:
|
||
img = img.copy()
|
||
img.putalpha(mask)
|
||
with io.BytesIO() as bytes_io:
|
||
if mask is not None:
|
||
img.save(bytes_io, format='PNG')
|
||
else:
|
||
img.save(bytes_io, format='JPEG')
|
||
data_bytes = bytes_io.getvalue()
|
||
return data_bytes
|
||
|
||
|
||
def _image_to_base64(image):
|
||
if image is None:
|
||
return None
|
||
return base64.b64encode(_encode_image(_tensor2images(image)[0])).decode("utf-8")
|
||
|
||
|
||
def _load_audio_from_url(audio_url, save_directory, filename_prefix="audio"):
|
||
try:
|
||
response = requests.get(
|
||
audio_url,
|
||
timeout=30,
|
||
stream=True,
|
||
headers={'User-Agent': 'Mozilla/5.0'}
|
||
)
|
||
response.raise_for_status()
|
||
|
||
parsed_url = urllib.parse.urlparse(audio_url)
|
||
ext = Path(parsed_url.path).suffix or '.mp3'
|
||
|
||
timestamp = int(time.time() * 1000) # 毫秒级时间戳
|
||
filename = f"{filename_prefix}_{timestamp}{ext}"
|
||
|
||
file_path = os.path.join(save_directory, filename)
|
||
|
||
os.makedirs(save_directory, exist_ok=True)
|
||
|
||
with open(file_path, 'wb') as f:
|
||
for chunk in response.iter_content(chunk_size=8192):
|
||
if chunk:
|
||
f.write(chunk)
|
||
|
||
return file_path # 返回完整路径
|
||
|
||
except requests.exceptions.Timeout:
|
||
raise Exception(f"time out: {audio_url}")
|
||
except requests.exceptions.HTTPError as e:
|
||
raise Exception(f"http error {e.response.status_code}: {audio_url}")
|
||
except requests.exceptions.RequestException as e:
|
||
raise Exception(f"download failed: {e}")
|
||
except IOError as e:
|
||
raise Exception(f"save failed: {e}")
|
||
except Exception as e:
|
||
raise Exception(f"other failed: {e}")
|
||
|
||
|
||
class KLingAIAPIClient:
|
||
|
||
def __init__(self):
|
||
pass
|
||
|
||
@classmethod
|
||
def INPUT_TYPES(s):
|
||
return {
|
||
"required": {
|
||
"access_key": ("STRING", {"multiline": False, "default": ""}),
|
||
"secret_key": ("STRING", {"multiline": False, "default": ""}),
|
||
"poll_interval": ("INT", {"default": "1"}),
|
||
"area": (["global", "china"],),
|
||
},
|
||
}
|
||
|
||
RETURN_TYPES = ("KLING_AI_API_CLIENT",)
|
||
RETURN_NAMES = ("client",)
|
||
|
||
FUNCTION = "create_client"
|
||
|
||
OUTPUT_NODE = True
|
||
|
||
CATEGORY = "KLingAI"
|
||
|
||
def create_client(self, access_key, secret_key, poll_interval, area):
|
||
|
||
in_china = (area == "china")
|
||
|
||
if access_key == "" or secret_key == "":
|
||
try:
|
||
klingai_api_access_key = config['API']['KLINGAI_API_ACCESS_KEY']
|
||
klingai_api_scerct_key = config['API']['KLINGAI_API_SECRET_KEY']
|
||
if klingai_api_access_key == '':
|
||
raise ValueError('ACCESS_KEY is empty')
|
||
if klingai_api_scerct_key == '':
|
||
raise ValueError('SECRET_KEY is empty')
|
||
|
||
except KeyError:
|
||
raise ValueError('unable to find ACCESS_KEY or SECRET_KEY in config.ini')
|
||
|
||
client = Client(klingai_api_access_key, klingai_api_scerct_key, in_china)
|
||
else:
|
||
client = Client(access_key, secret_key, in_china)
|
||
|
||
client.poll_interval = poll_interval
|
||
return (client,)
|
||
|
||
|
||
class ImageGeneratorNode:
|
||
@classmethod
|
||
def INPUT_TYPES(s):
|
||
return {
|
||
"required": {
|
||
"client": ("KLING_AI_API_CLIENT",),
|
||
"model": (["kling-v1", "kling-v1-5", "kling-v2", "kling-v2-new", "kling-v2-1"],),
|
||
"prompt": ("STRING", {"multiline": True, "default": ""}),
|
||
},
|
||
"optional": {
|
||
"negative_prompt": ("STRING", {"multiline": True, "default": ""}),
|
||
"image": ("IMAGE",),
|
||
"image_reference": (["None", "subject", "face"],),
|
||
"image_fidelity": ("FLOAT", {
|
||
"default": 0.5,
|
||
"min": 0.0,
|
||
"max": 1.0,
|
||
"step": 0.1,
|
||
"round": 0.01,
|
||
"display": "number",
|
||
"lazy": True
|
||
}),
|
||
"human_fidelity": ("FLOAT", {
|
||
"default": 0.45,
|
||
"min": 0.0,
|
||
"max": 1.0,
|
||
"step": 0.1,
|
||
"round": 0.01,
|
||
"display": "number",
|
||
"lazy": True
|
||
}),
|
||
"image_num": ("INT", {
|
||
"default": 1,
|
||
"min": 0,
|
||
"max": 9,
|
||
"step": 1,
|
||
"display": "number",
|
||
"lazy": True
|
||
}),
|
||
"aspect_ratio": (["16:9", "9:16", "1:1", "4:3", "3:4", "3:2", "2:3"],),
|
||
"resolution": ("STRING",
|
||
["1k", "2k", ],
|
||
),
|
||
}
|
||
}
|
||
|
||
RETURN_TYPES = ("IMAGE",)
|
||
RETURN_NAMES = ("image",)
|
||
|
||
FUNCTION = "generate"
|
||
|
||
OUTPUT_NODE = False
|
||
|
||
CATEGORY = "KLingAI"
|
||
|
||
def generate(self,
|
||
client,
|
||
model,
|
||
prompt,
|
||
negative_prompt=None,
|
||
image=None,
|
||
image_reference="None",
|
||
image_fidelity=None,
|
||
human_fidelity=None,
|
||
resolution=None,
|
||
image_num=None,
|
||
aspect_ratio=None):
|
||
generator = ImageGenerator()
|
||
generator.model_name = model
|
||
generator.prompt = prompt
|
||
generator.negative_prompt = negative_prompt
|
||
generator.resolution = resolution
|
||
generator.image = _image_to_base64(image)
|
||
generator.image_fidelity = image_fidelity
|
||
generator.aspect_ratio = aspect_ratio
|
||
generator.n = image_num
|
||
generator.human_fidelity = human_fidelity
|
||
|
||
if model == "kling-v2-1":
|
||
generator.human_fidelity = None
|
||
generator.image_fidelity = None
|
||
|
||
if image_reference != 'None':
|
||
generator.image_reference = image_reference
|
||
|
||
response = generator.run(client)
|
||
|
||
imgs = None
|
||
for image_info in response.task_result.images:
|
||
img = _images2tensor(_decode_image(_fetch_image(image_info.url)))
|
||
if imgs is None:
|
||
imgs = img
|
||
else:
|
||
imgs = torch.cat([imgs, img], dim=0)
|
||
print(f'KLing API output: {image_info.url}')
|
||
|
||
return (imgs,)
|
||
|
||
|
||
class ImageExpanderNode:
|
||
|
||
# TODO ?
|
||
@classmethod
|
||
def INPUT_TYPES(s):
|
||
|
||
expansion_ratio_parameter = {
|
||
"default": 0,
|
||
"min": 0,
|
||
"max": 2,
|
||
}
|
||
|
||
return {
|
||
"required": {
|
||
"client": ("KLING_AI_API_CLIENT",),
|
||
"image": ("IMAGE",),
|
||
"up_expansion_ratio": ("FLOAT", expansion_ratio_parameter),
|
||
"down_expansion_ratio": ("FLOAT", expansion_ratio_parameter),
|
||
"left_expansion_ratio": ("FLOAT", expansion_ratio_parameter),
|
||
"right_expansion_ratio": ("FLOAT", expansion_ratio_parameter),
|
||
|
||
},
|
||
"optional": {
|
||
"prompt": ("STRING", {"multiline": True, "default": ""}),
|
||
"image_num": ("INT", {
|
||
"default": 1,
|
||
"min": 0,
|
||
"max": 9,
|
||
"step": 1,
|
||
"display": "number",
|
||
"lazy": True
|
||
}),
|
||
}
|
||
|
||
}
|
||
|
||
RETURN_TYPES = ("IMAGE",)
|
||
RETURN_NAMES = ("image",)
|
||
|
||
FUNCTION = "generate"
|
||
|
||
OUTPUT_NODE = False
|
||
|
||
CATEGORY = "KLingAI"
|
||
|
||
def generate(self,
|
||
client,
|
||
image,
|
||
prompt=None,
|
||
image_num=None,
|
||
up_expansion_ratio=None,
|
||
down_expansion_ratio=None,
|
||
left_expansion_ratio=None,
|
||
right_expansion_ratio=None,
|
||
):
|
||
generator = ImageExpander()
|
||
|
||
generator.image = _image_to_base64(image)
|
||
generator.up_expansion_ratio = up_expansion_ratio
|
||
generator.down_expansion_ratio = down_expansion_ratio
|
||
generator.left_expansion_ratio = left_expansion_ratio
|
||
generator.right_expansion_ratio = right_expansion_ratio
|
||
generator.prompt = prompt
|
||
generator.n = image_num
|
||
response = generator.run(client)
|
||
|
||
imgs = None
|
||
for image_info in response.task_result.images:
|
||
img = _images2tensor(_decode_image(_fetch_image(image_info.url)))
|
||
if imgs is None:
|
||
imgs = img
|
||
else:
|
||
imgs = torch.cat([imgs, img], dim=0)
|
||
|
||
print(f'KLing API output: {image_info.url}')
|
||
|
||
return (imgs,)
|
||
|
||
|
||
class Image2VideoNode:
|
||
@classmethod
|
||
def INPUT_TYPES(s):
|
||
return {
|
||
"required": {
|
||
"client": ("KLING_AI_API_CLIENT",),
|
||
"model": (
|
||
["kling-v1", "kling-v1-5", "kling-v1-6", "kling-v2-master", "kling-v2-1", "kling-v2-1-master"],),
|
||
},
|
||
"optional": {
|
||
"image": ("IMAGE",),
|
||
"image_tail": ("IMAGE",),
|
||
"prompt": ("STRING", {"multiline": True, "default": ""}),
|
||
"negative_prompt": ("STRING", {"multiline": True, "default": ""}),
|
||
"cfg_scale": ("FLOAT", {
|
||
"default": 0.5,
|
||
"min": 0.0,
|
||
"max": 1.0,
|
||
"step": 0.1,
|
||
"round": 0.01,
|
||
"display": "number",
|
||
"lazy": True
|
||
}),
|
||
"mode": (["std", "pro"],),
|
||
"duration": (["5", "10"],),
|
||
"camera_control_type": (
|
||
["None", "simple", "down_back", "forward_up", "right_turn_forward", "left_turn_forward"],),
|
||
"camera_control_config": (["horizontal", "vertical", "pan", "tilt", "roll", "zoom"],),
|
||
"camera_control_value": ("FLOAT", {
|
||
"default": 0.5,
|
||
"min": -10.0,
|
||
"max": 10.0,
|
||
"step": 1.0,
|
||
"round": 1.0,
|
||
"display": "number",
|
||
"lazy": True
|
||
})
|
||
}
|
||
}
|
||
|
||
RETURN_TYPES = ("STRING", "STRING")
|
||
RETURN_NAMES = ("url", "video_id")
|
||
|
||
FUNCTION = "generate"
|
||
|
||
OUTPUT_NODE = False
|
||
|
||
CATEGORY = "KLingAI"
|
||
|
||
def generate(self,
|
||
client,
|
||
model,
|
||
image=None,
|
||
image_tail=None,
|
||
prompt=None,
|
||
negative_prompt=None,
|
||
cfg_scale=None,
|
||
mode=None,
|
||
duration=None,
|
||
camera_control_type=None,
|
||
camera_control_config=None,
|
||
camera_control_value=None):
|
||
|
||
generator = Image2Video()
|
||
generator.model_name = model
|
||
generator.image = _image_to_base64(image)
|
||
generator.image_tail = _image_to_base64(image_tail)
|
||
generator.prompt = prompt
|
||
generator.negative_prompt = negative_prompt
|
||
generator.cfg_scale = cfg_scale
|
||
generator.mode = mode
|
||
generator.duration = duration
|
||
|
||
if camera_control_type != 'None':
|
||
generator.camera_control = CameraControl()
|
||
generator.camera_control.type = camera_control_type
|
||
|
||
if generator.camera_control.type == "simple":
|
||
generator.camera_control.config = CameraControlConfig()
|
||
if camera_control_config == "horizontal":
|
||
generator.camera_control.config.horizontal = camera_control_value
|
||
if camera_control_config == "vertical":
|
||
generator.camera_control.config.vertical = camera_control_value
|
||
if camera_control_config == "pan":
|
||
generator.camera_control.config.pan = camera_control_value
|
||
if camera_control_config == "tilt":
|
||
generator.camera_control.config.tilt = camera_control_value
|
||
if camera_control_config == "roll":
|
||
generator.camera_control.config.roll = camera_control_value
|
||
if camera_control_config == "zoom":
|
||
generator.camera_control.config.zoom = camera_control_value
|
||
|
||
response = generator.run(client)
|
||
|
||
for video_info in response.task_result.videos:
|
||
print(f'KLing API output video id: {video_info.id}, url: {video_info.url}')
|
||
return (video_info.url, video_info.id)
|
||
|
||
return ('', '')
|
||
|
||
|
||
class Text2VideoNode:
|
||
@classmethod
|
||
def INPUT_TYPES(s):
|
||
return {
|
||
"required": {
|
||
"client": ("KLING_AI_API_CLIENT",),
|
||
"model": (["kling-v1", "kling-v1-6", "kling-v2-master", "kling-v2-1", "kling-v2-1-master"],),
|
||
"prompt": ("STRING", {"multiline": True, "default": ""}),
|
||
},
|
||
"optional": {
|
||
"negative_prompt": ("STRING", {"multiline": True, "default": ""}),
|
||
"cfg_scale": ("FLOAT", {
|
||
"default": 0.5,
|
||
"min": 0.0,
|
||
"max": 1.0,
|
||
"step": 0.1,
|
||
"round": 0.01,
|
||
"display": "number",
|
||
"lazy": True
|
||
}),
|
||
"mode": (["std", "pro"],),
|
||
"aspect_ratio": (["16:9", "9:16", "1:1"],),
|
||
"duration": (["5", "10"],),
|
||
"camera_control_type": (
|
||
["None", "simple", "down_back", "forward_up", "right_turn_forward", "left_turn_forward"],),
|
||
"camera_control_config": (["horizontal", "vertical", "pan", "tilt", "roll", "zoom"],),
|
||
"camera_control_value": ("FLOAT", {
|
||
"default": 0.5,
|
||
"min": -10.0,
|
||
"max": 10.0,
|
||
"step": 1.0,
|
||
"round": 1.0,
|
||
"display": "number",
|
||
"lazy": True
|
||
})
|
||
}
|
||
}
|
||
|
||
RETURN_TYPES = ("STRING", "STRING")
|
||
RETURN_NAMES = ("url", "video_id")
|
||
|
||
FUNCTION = "generate"
|
||
|
||
OUTPUT_NODE = False
|
||
|
||
CATEGORY = "KLingAI"
|
||
|
||
def generate(self,
|
||
client,
|
||
model,
|
||
prompt,
|
||
negative_prompt=None,
|
||
cfg_scale=None,
|
||
mode=None,
|
||
aspect_ratio=None,
|
||
duration=None,
|
||
camera_control_type=None,
|
||
camera_control_config=None,
|
||
camera_control_value=None):
|
||
|
||
generator = Text2Video()
|
||
generator.model_name = model
|
||
generator.prompt = prompt
|
||
generator.negative_prompt = negative_prompt
|
||
generator.cfg_scale = cfg_scale
|
||
generator.mode = mode
|
||
generator.aspect_ratio = aspect_ratio
|
||
generator.duration = duration
|
||
|
||
if camera_control_type != 'None':
|
||
generator.camera_control = CameraControl()
|
||
generator.camera_control.type = camera_control_type
|
||
|
||
if generator.camera_control.type == "simple":
|
||
generator.camera_control.config = CameraControlConfig()
|
||
if camera_control_config == "horizontal":
|
||
generator.camera_control.config.horizontal = camera_control_value
|
||
if camera_control_config == "vertical":
|
||
generator.camera_control.config.vertical = camera_control_value
|
||
if camera_control_config == "pan":
|
||
generator.camera_control.config.pan = camera_control_value
|
||
if camera_control_config == "tilt":
|
||
generator.camera_control.config.tilt = camera_control_value
|
||
if camera_control_config == "roll":
|
||
generator.camera_control.config.roll = camera_control_value
|
||
if camera_control_config == "zoom":
|
||
generator.camera_control.config.zoom = camera_control_value
|
||
|
||
response = generator.run(client)
|
||
|
||
for video_info in response.task_result.videos:
|
||
print(f'KLing API output video id: {video_info.id}, url: {video_info.url}')
|
||
return (video_info.url, video_info.id)
|
||
|
||
return ('', '')
|
||
|
||
|
||
class KolorsVirtualTryOnNode:
|
||
@classmethod
|
||
def INPUT_TYPES(s):
|
||
return {
|
||
"required": {
|
||
"client": ("KLING_AI_API_CLIENT",),
|
||
"model_name": (["kolors-virtual-try-on-v1", "kolors-virtual-try-on-v1-5"],),
|
||
"human_image": ("IMAGE",),
|
||
"cloth_image": ("IMAGE",),
|
||
},
|
||
}
|
||
|
||
RETURN_TYPES = ("IMAGE",)
|
||
RETURN_NAMES = ("image",)
|
||
|
||
FUNCTION = "generate"
|
||
|
||
OUTPUT_NODE = False
|
||
|
||
CATEGORY = "KLingAI"
|
||
|
||
def generate(self,
|
||
client,
|
||
model_name,
|
||
human_image,
|
||
cloth_image=None):
|
||
generator = KolorsVurtualTryOn()
|
||
generator.model_name = model_name
|
||
generator.human_image = _image_to_base64(human_image)
|
||
generator.cloth_image = _image_to_base64(cloth_image)
|
||
|
||
response = generator.run(client)
|
||
|
||
for image_info in response.task_result.images:
|
||
img = _images2tensor(_decode_image(_fetch_image(image_info.url)))
|
||
print(f'KLing API output: {image_info.url}')
|
||
return (img,)
|
||
|
||
|
||
class PreviewVideo:
|
||
@classmethod
|
||
def INPUT_TYPES(cls):
|
||
return {
|
||
"required": {
|
||
"video_url": ("STRING", {"forceInput": True}),
|
||
"filename_prefix": ("STRING", {"default": "KLingAI"}),
|
||
"save_output": ("BOOLEAN", {"default": True}),
|
||
}
|
||
}
|
||
|
||
OUTPUT_NODE = True
|
||
FUNCTION = "run"
|
||
CATEGORY = "KLingAI"
|
||
|
||
RETURN_TYPES = ("STRING",)
|
||
RETURN_NAMES = ("file_path",)
|
||
|
||
def run(self, video_url, filename_prefix, save_output):
|
||
if not save_output:
|
||
return {"ui": {"video_url": [video_url]}, "result": ('',)}
|
||
|
||
output_dir = folder_paths.get_output_directory()
|
||
(
|
||
full_output_folder,
|
||
filename,
|
||
_,
|
||
_,
|
||
_,
|
||
) = folder_paths.get_save_image_path(filename_prefix, output_dir)
|
||
|
||
max_counter = 0
|
||
|
||
matcher = re.compile(f"{re.escape(filename)}_(\\d+)\\D*\\..+", re.IGNORECASE)
|
||
for existing_file in os.listdir(full_output_folder):
|
||
match = matcher.fullmatch(existing_file)
|
||
if match:
|
||
file_counter = int(match.group(1))
|
||
if file_counter > max_counter:
|
||
max_counter = file_counter
|
||
|
||
counter = max_counter + 1
|
||
file = f"{filename}_{counter:05}.mp4"
|
||
file_path = os.path.join(full_output_folder, file)
|
||
|
||
if type(video_url) == list:
|
||
video_url = video_url[0]
|
||
open(file_path, "wb").write(_fetch_image(video_url))
|
||
|
||
return {"ui": {"video_url": [video_url]}, "result": (file_path,)}
|
||
|
||
|
||
class PreviewAudio(LoadAudio):
|
||
|
||
@classmethod
|
||
def INPUT_TYPES(cls):
|
||
return {
|
||
"required": {
|
||
"audio_url": ("STRING", {
|
||
"forceInput": True,
|
||
"default": ""
|
||
}),
|
||
"filename_prefix": ("STRING", {
|
||
"default": "KLingAI"
|
||
}),
|
||
"save_output": ("BOOLEAN", {
|
||
"default": True
|
||
}),
|
||
}
|
||
}
|
||
|
||
RETURN_TYPES = ("AUDIO", "STRING")
|
||
RETURN_NAMES = ("audio", "file_path")
|
||
FUNCTION = "run"
|
||
CATEGORY = "KLingAI"
|
||
OUTPUT_NODE = True
|
||
|
||
@classmethod
|
||
def VALIDATE_INPUTS(cls, **kwargs):
|
||
return True
|
||
|
||
def run(self, audio_url, filename_prefix, save_output):
|
||
|
||
try:
|
||
if not save_output:
|
||
return {
|
||
"ui": {"audio_url": [audio_url]},
|
||
"result": (None, '')
|
||
}
|
||
|
||
temp_directory = get_temp_directory()
|
||
|
||
saved_file_path = _load_audio_from_url(
|
||
audio_url=audio_url,
|
||
save_directory=temp_directory,
|
||
filename_prefix=filename_prefix
|
||
)
|
||
|
||
audio_result = super().load(saved_file_path)
|
||
|
||
audio_data = audio_result[0]
|
||
|
||
return {
|
||
"ui": {
|
||
"audio_url": [audio_url],
|
||
"file_path": [saved_file_path]
|
||
},
|
||
"result": (audio_data, saved_file_path)
|
||
}
|
||
|
||
except Exception as e:
|
||
error_msg = f"[PreviewAudio] Error: {str(e)}"
|
||
print(error_msg)
|
||
raise Exception(error_msg)
|
||
|
||
|
||
class VideoExtendNode:
|
||
@classmethod
|
||
def INPUT_TYPES(cls):
|
||
return {
|
||
"required": {
|
||
"client": ("KLING_AI_API_CLIENT",),
|
||
"video_id": ("STRING", {"multiline": False, "default": ""}),
|
||
"prompt": ("STRING", {"multiline": True, "default": ""}),
|
||
}
|
||
}
|
||
|
||
OUTPUT_NODE = True
|
||
FUNCTION = "run"
|
||
CATEGORY = "KLingAI"
|
||
|
||
RETURN_TYPES = ("STRING", "STRING")
|
||
RETURN_NAMES = ("url", "video_id")
|
||
|
||
def run(self, client, video_id, prompt):
|
||
generator = VideoExtend()
|
||
generator.video_id = video_id
|
||
generator.prompt = prompt
|
||
|
||
response = generator.run(client)
|
||
|
||
for video_info in response.task_result.videos:
|
||
print(f'KLing API output video id: {video_info.id}, url: {video_info.url}')
|
||
return (video_info.url, video_info.id)
|
||
|
||
return ('', '')
|
||
|
||
|
||
class LipSyncTextInputNode:
|
||
audio_types = {
|
||
"阳光少年": "genshin_vindi2",
|
||
"懂事小弟": "zhinen_xuesheng",
|
||
"运动少年": "tiyuxi_xuedi",
|
||
"青春少女": "ai_shatang",
|
||
"温柔小妹": "genshin_klee2",
|
||
"元气少女": "genshin_kirara",
|
||
"阳光男生": "ai_kaiya",
|
||
"幽默小哥": "tiexin_nanyou",
|
||
"文艺小哥": "ai_chenjiahao_712",
|
||
"甜美邻家": "girlfriend_1_speech02",
|
||
"温柔姐姐": "chat1_female_new-3",
|
||
"职场女青": "girlfriend_2_speech02",
|
||
"活泼男童": "cartoon-boy-07",
|
||
"俏皮女童": "cartoon-girl-01",
|
||
"稳重老爸": "ai_huangyaoshi_712",
|
||
"温柔妈妈": "you_pingjing",
|
||
"严肃上司": "ai_laoguowang_712",
|
||
"优雅贵妇": "chengshu_jiejie",
|
||
"慈祥爷爷": "zhuxi_speech02",
|
||
"唠叨爷爷": "uk_oldman3",
|
||
"唠叨奶奶": "laopopo_speech02",
|
||
"和蔼奶奶": "heainainai_speech02",
|
||
"东北老铁": "dongbeilaotie_speech02",
|
||
"重庆小伙": "chongqingxiaohuo_speech02",
|
||
"四川妹子": "chuanmeizi_speech02",
|
||
"潮汕大叔": "chaoshandashu_speech02",
|
||
"台湾男生": "ai_taiwan_man2_speech02",
|
||
"西安掌柜": "xianzhanggui_speech02",
|
||
"天津姐姐": "tianjinjiejie_speech02",
|
||
"新闻播报男": "diyinnansang_DB_CN_M_04-v2",
|
||
"译制片男": "yizhipiannan-v1",
|
||
"元气少女": "guanxiaofang-v2",
|
||
"撒娇女友": "tianmeixuemei-v1",
|
||
"刀片烟嗓": "daopianyansang-v1",
|
||
"乖巧正太": "mengwa-v1",
|
||
|
||
"Sunny": "genshin_vindi2",
|
||
"Sage": "zhinen_xuesheng",
|
||
"Ace": "AOT",
|
||
"Blossom": "ai_shatang",
|
||
"Peppy": "genshin_klee2",
|
||
"Dove": "genshin_kirara",
|
||
"Shine": "ai_kaiya",
|
||
"Anchor": "oversea_male1",
|
||
"Lyric": "ai_chenjiahao_712",
|
||
"Melody": "girlfriend_4_speech02",
|
||
"Tender": "chat1_female_new-3",
|
||
"Siren": "chat_0407_5-1",
|
||
"Zippy": "cartoon-boy-07",
|
||
"Bud": "uk_boy1",
|
||
"Sprite": "cartoon-girl-01",
|
||
"Candy": "PeppaPig_platform",
|
||
"Beacon": "ai_huangzhong_712",
|
||
"Rock": "ai_huangyaoshi_712",
|
||
"Titan": "ai_laoguowang_712",
|
||
"Grace": "chengshu_jiejie",
|
||
"Helen": "you_pingjing",
|
||
"Lore": "calm_story1",
|
||
"Crag": "uk_man2",
|
||
"Prattle": "laopopo_speech02",
|
||
"Hearth": "heainainai_speech02",
|
||
"The Reader": "reader_en_m-v1",
|
||
"Commercial Lady": "commercial_lady_en_f-v1"
|
||
}
|
||
|
||
@classmethod
|
||
def INPUT_TYPES(cls):
|
||
return {
|
||
"required": {
|
||
"text": ("STRING", {"multiline": True, "default": ""}),
|
||
"voice_id": (list(LipSyncTextInputNode.audio_types.keys()), {"multiline": False, "default": ""}),
|
||
"voice_language": (["zh", "en"], {"default": "zh"}),
|
||
"voice_speed": ("FLOAT", {
|
||
"default": 1.0,
|
||
"min": 0.8,
|
||
"max": 2.0,
|
||
"step": 0.1,
|
||
"round": 0.01,
|
||
"display": "number",
|
||
"lazy": True
|
||
})
|
||
}
|
||
}
|
||
|
||
OUTPUT_NODE = True
|
||
FUNCTION = "run"
|
||
CATEGORY = "KLingAI"
|
||
|
||
RETURN_TYPES = ("KLING_AI_API_LIPSYNC_INPUT",)
|
||
RETURN_NAMES = ("input",)
|
||
|
||
def run(self, text, voice_id, voice_language, voice_speed):
|
||
input = LipSyncInput()
|
||
input.mode = "text2video"
|
||
input.text = text
|
||
if voice_id in LipSyncTextInputNode.audio_types:
|
||
input.voice_id = LipSyncTextInputNode.audio_types[voice_id]
|
||
else:
|
||
input.voice_id = voice_id
|
||
|
||
input.voice_language = voice_language
|
||
input.voice_speed = voice_speed
|
||
|
||
return (input,)
|
||
|
||
|
||
class LipSyncAudioInputNode:
|
||
@classmethod
|
||
def INPUT_TYPES(cls):
|
||
return {
|
||
"optional": {
|
||
"audio_file": ("STRING", {"multiline": False, "default": ""}),
|
||
"audio_url": ("STRING", {"multiline": False, "default": ""}),
|
||
},
|
||
}
|
||
|
||
OUTPUT_NODE = True
|
||
FUNCTION = "run"
|
||
CATEGORY = "KLingAI"
|
||
|
||
RETURN_TYPES = ("KLING_AI_API_LIPSYNC_INPUT",)
|
||
RETURN_NAMES = ("input",)
|
||
|
||
def run(self, audio_file, audio_url):
|
||
input = LipSyncInput()
|
||
input.mode = "audio2video"
|
||
if audio_file is not None and len(audio_file) > 0:
|
||
input.audio_type = "file"
|
||
if os.path.exists(audio_file):
|
||
with open(audio_file, 'rb') as file:
|
||
file_data = file.read()
|
||
input.audio_file = base64.b64encode(file_data).decode('utf-8')
|
||
else:
|
||
raise Exception(f"Audio file not found: {audio_file}")
|
||
|
||
if audio_url is not None and len(audio_url) > 0:
|
||
input.audio_type = "url"
|
||
input.audio_url = audio_url
|
||
|
||
return (input,)
|
||
|
||
|
||
class LipSyncNode:
|
||
@classmethod
|
||
def INPUT_TYPES(cls):
|
||
return {
|
||
"required": {
|
||
"client": ("KLING_AI_API_CLIENT",),
|
||
"input": ("KLING_AI_API_LIPSYNC_INPUT",),
|
||
"face_id": ("STRING", {"multiline": False, "default": ""})
|
||
},
|
||
"optional": {
|
||
"video_id": ("STRING", {"multiline": False, "default": ""}),
|
||
"video_url": ("STRING", {"multiline": False, "default": ""}),
|
||
}
|
||
}
|
||
|
||
OUTPUT_NODE = True
|
||
FUNCTION = "run"
|
||
CATEGORY = "KLingAI"
|
||
|
||
RETURN_TYPES = ("STRING", "STRING")
|
||
RETURN_NAMES = ("url", "video_id")
|
||
|
||
def run(self, client, input, video_id=None, video_url=None):
|
||
if video_id is None and video_url is None:
|
||
raise Exception(f"Please input video_id or video_url.")
|
||
|
||
generator = LipSync()
|
||
input.video_id = video_id
|
||
input.video_url = video_url
|
||
generator.input = input
|
||
|
||
response = generator.run(client)
|
||
|
||
for video_info in response.task_result.videos:
|
||
print(f'KLing API output video id: {video_info.id}, url: {video_info.url}')
|
||
return (video_info.url, video_info.id)
|
||
|
||
return ('', '')
|
||
|
||
|
||
class EffectNode:
|
||
@classmethod
|
||
def INPUT_TYPES(cls):
|
||
return {
|
||
"required": {
|
||
"client": ("KLING_AI_API_CLIENT",),
|
||
"effect_scene": (
|
||
["baseball", "inner_voice", "a_list_look", "memory_alive", "trampoline", "trampoline_night",
|
||
"pucker_up", "guess_what", "feed_mooncake", "rampage_ape", "flyer", "dishwasher",
|
||
"pet_chinese_opera", "magic_fireball", "gallery_ring", "pet_moto_rider", "muscle_pet",
|
||
"squeeze_scream",
|
||
"pet_delivery", "running_man", "disappear", "mythic_style", "steampunk", "c4d_cartoon",
|
||
"3d_cartoon_1",
|
||
"3d_cartoon_2", "eagle_snatch", "hug_from_past", "firework", 'media_interview', "pet_lion",
|
||
"pet_chef",
|
||
"santa_gifts", "santa_hug", "girlfriend", "boyfriend", "heart_gesture_1", "pet_wizard",
|
||
"smoke_smoke", "thumbs_up",
|
||
"instant_kid", "dollar_rain", "cry_cry", "building_collapse", "gun_shot", "mushroom", "double_gun",
|
||
"pet_warrior",
|
||
"lightning_power", "jesus_hug", "shark_alert", "long_hair", "lie_flat", "polar_bear_hug",
|
||
"brown_bear_hug",
|
||
"jazz_jazz", "office_escape_plow", "fly_fly", "watermelon_bomb", "pet_dance", "boss_coming",
|
||
"wool_curly",
|
||
"pet_bee", "marry_me", "swing_swing", "day_to_night", "piggy_morph", "wig_out", "car_explosion",
|
||
"ski_ski",
|
||
"tiger_hug", "siblings", "construction_worker", "let’s_ride", "snatched", "magic_broom",
|
||
"felt_felt", "jumpdrop", "celebration", "splashsplash", "surfsurf", "fairy_wing", "angel_wing",
|
||
"dark_wing", "skateskate", "plushcut", "jelly_press", "jelly_slice", "jelly_squish",
|
||
"jelly_jiggle",
|
||
"pixelpixel", "yearbook", "instant_film", "anime_figure", "rocketrocket", "bloombloom",
|
||
"dizzydizzy", "fuzzyfuzzy",
|
||
"squish", "expansion", "hug", "kiss", "heart_gesture", "fight"
|
||
],),
|
||
"model_name": (["kling-v1", "kling-v1-5", "kling-v1-6"],),
|
||
"mode": (["std", "pro"],),
|
||
"duration": (["5", "10"],),
|
||
"image0": ("IMAGE",),
|
||
},
|
||
"optional": {
|
||
"image1": ("IMAGE",)
|
||
}
|
||
}
|
||
|
||
OUTPUT_NODE = True
|
||
FUNCTION = "run"
|
||
CATEGORY = "KLingAI"
|
||
|
||
RETURN_TYPES = ("STRING", "STRING")
|
||
RETURN_NAMES = ("url", "video_id")
|
||
|
||
def run(self, client, effect_scene, model_name, mode, duration, image0, image1=None):
|
||
|
||
generator = Effects()
|
||
generator.effect_scene = effect_scene
|
||
generator.input = EffectInput()
|
||
if effect_scene in ["hug", "kiss", "heart_gesture", "fight"]:
|
||
generator.input.model_name = 'kling-v1-6' if effect_scene == 'fight' else model_name
|
||
generator.input.mode = mode
|
||
generator.input.duration = duration
|
||
if image1 == None or image0 == None:
|
||
raise Exception("This effect needs two images.")
|
||
generator.input.images = [_image_to_base64(image0), _image_to_base64(image1)]
|
||
else:
|
||
generator.input.mode = mode
|
||
generator.input.duration = duration
|
||
if image1 != None and image0 != None:
|
||
raise Exception("This effect needs one image.")
|
||
generator.input.image = _image_to_base64(image0) if image1 == None else _image_to_base64(image1)
|
||
|
||
response = generator.run(client)
|
||
|
||
for video_info in response.task_result.videos:
|
||
print(f'KLing API output video id: {video_info.id}, url: {video_info.url}')
|
||
return (video_info.url, video_info.id)
|
||
|
||
return ('', '')
|
||
|
||
|
||
class Video2AudioNode:
|
||
@classmethod
|
||
def INPUT_TYPES(s):
|
||
return {
|
||
"required": {
|
||
"client": ("KLING_AI_API_CLIENT",),
|
||
"video_id": ("STRING", {"multiline": False, "default": ""}),
|
||
"video_url": ("STRING", {"multiline": False, "default": ""}),
|
||
},
|
||
"optional": {
|
||
"sound_effect_prompt": ("STRING", {"multiline": True, "default": ""}),
|
||
"bgm_prompt": ("STRING", {"multiline": True, "default": ""}),
|
||
"asmr_mod": ("BOOLEAN", {"multiline": False, "default": False}),
|
||
|
||
}
|
||
}
|
||
|
||
RETURN_TYPES = ("STRING", "STRING", "STRING", "STRING")
|
||
RETURN_NAMES = ("videos_id", "videos_url", "audio_id", "audio_url_mp3")
|
||
|
||
FUNCTION = "generate"
|
||
|
||
OUTPUT_NODE = False
|
||
|
||
CATEGORY = "KLingAI"
|
||
|
||
def generate(self,
|
||
client,
|
||
video_id,
|
||
video_url,
|
||
sound_effect_prompt=None,
|
||
bgm_prompt=None,
|
||
asmr_mode=False,
|
||
):
|
||
|
||
generator = Video2Audio()
|
||
generator.video_id = video_id
|
||
generator.video_url = video_url
|
||
|
||
if video_id is None and video_url is None:
|
||
raise Exception("Please input video_id or video_url")
|
||
|
||
if video_id is not None and video_url is not None:
|
||
raise Exception("Please input one of video_id or video_url")
|
||
|
||
generator.sound_effect_prompt = sound_effect_prompt
|
||
generator.bgm_prompt = bgm_prompt
|
||
generator.asmr_mode = asmr_mode
|
||
|
||
response = generator.run(client)
|
||
|
||
results = []
|
||
for audio_info in response.task_result.audios:
|
||
results.append({
|
||
"videos_id": audio_info["videos_id"],
|
||
"videos_url": audio_info["videos_url"],
|
||
"audio_id": audio_info["audio_id"],
|
||
"audio_url": audio_info["audio_url"]
|
||
})
|
||
|
||
return (results)
|
||
|
||
|
||
class MultiImagesToVideoNode:
|
||
pass
|
||
|
||
|
||
class TextToAudioNode:
|
||
@classmethod
|
||
def INPUT_TYPES(s):
|
||
return {
|
||
"required": {
|
||
"client": ("KLING_AI_API_CLIENT",),
|
||
"prompt": ("STRING", {"multiline": True, "default": ""}),
|
||
"duration": ("FLOAT", {
|
||
"default": 3.0,
|
||
"min": 3.0,
|
||
"max": 10.0,
|
||
"step": 0.1,
|
||
"display": "number",
|
||
},),
|
||
},
|
||
}
|
||
|
||
RETURN_TYPES = ("STRING", "STRING")
|
||
RETURN_NAMES = ("id", "url")
|
||
|
||
FUNCTION = "generate"
|
||
|
||
OUTPUT_NODE = True
|
||
|
||
CATEGORY = "KLingAI"
|
||
|
||
def generate(self,
|
||
client,
|
||
prompt,
|
||
duration,
|
||
):
|
||
generator = Text2Audio()
|
||
|
||
generator.prompt = prompt
|
||
generator.duration = duration
|
||
response = generator.run(client)
|
||
|
||
url_mp3 = getattr(response.task_result.audios[0], 'url_mp3', None)
|
||
if not isinstance(url_mp3, str) or not url_mp3.strip():
|
||
raise Exception(f"url_mp3 无效，当前值为：{url_mp3}")
|
||
|
||
audio_id = getattr(response.task_result.audios[0], 'audio_id', None)
|
||
print(f"成功提取：audio_id={audio_id}, url_mp3={url_mp3}")
|
||
|
||
return (audio_id, url_mp3)
|