nodes/synthetai-ComfyUI-JM-KLing-API
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

增加图生视频节点

Browse Source
This commit is contained in:
jueming
2025-04-02 19:45:41 +08:00
parent e57c654914
commit db798c11c7
3 changed files with 481 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
__init__.py
+3
View File
@@ -2,10 +2,12 @@ from .nodes.api_key import KLingAIAPIKey
from .nodes.text2video import KLingAIText2Video
from .nodes.query_status import KLingAIQueryStatus
from .nodes.video_downloader import KLingAIVideoDownloader
from .nodes.image2video import KLingAIImage2Video
NODE_CLASS_MAPPINGS = {
"JM-KLingAI-API/api-key": KLingAIAPIKey,
"JM-KLingAI-API/text2video": KLingAIText2Video,
"JM-KLingAI-API/image2video": KLingAIImage2Video,
"JM-KLingAI-API/query-status": KLingAIQueryStatus,
"JM-KLingAI-API/video-downloader": KLingAIVideoDownloader
}
@@ -13,6 +15,7 @@ NODE_CLASS_MAPPINGS = {
NODE_DISPLAY_NAME_MAPPINGS = {
"JM-KLingAI-API/api-key": "KLingAI API Key",
"JM-KLingAI-API/text2video": "KLingAI Text to Video",
"JM-KLingAI-API/image2video": "KLingAI Image to Video",
"JM-KLingAI-API/query-status": "KLingAI Query Status",
"JM-KLingAI-API/video-downloader": "KLingAI Video Downloader"
}
nodes/image2video.py
+315
View File
@@ -0,0 +1,315 @@
import json
import random
import requests
import base64
import os
from PIL import Image
import io
import numpy as np
import torch
class KLingAIImage2Video:
"""
KLingAI Image to Video Node
创建图生视频任务节点
"""
def __init__(self):
self.api_base = "https://api.klingai.com"
self.endpoint = "/v1/videos/image2video"
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"api_token": ("STRING", {"default": "", "multiline": False}),
"image": ("IMAGE",),
},
"optional": {
"model_name": (["kling-v1", "kling-v1-5", "kling-v1-6"], {"default": "kling-v1"}),
"positive_prompt": ("STRING", {
"default": "",
"multiline": True,
"placeholder": "正向提示词 (最多2500字符)"
}),
"negative_prompt": ("STRING", {
"default": "",
"multiline": True,
"placeholder": "负向提示词 (最多2500字符)"
}),
"cfg_scale": ("FLOAT", {
"default": 0.5,
"min": 0.0,
"max": 1.0,
"step": 0.1
}),
"mode": (["std", "pro"], {"default": "std"}),
"duration": (["5", "10"], {"default": "5"}),
"image_tail": ("IMAGE",),
"use_camera_control": ("BOOLEAN", {"default": False}),
"camera_type": (["simple", "down_back", "forward_up", "right_turn_forward", "left_turn_forward"], {"default": "simple"}),
"camera_horizontal": ("FLOAT", {
"default": 0.0,
"min": -10.0,
"max": 10.0,
"step": 0.5
}),
"camera_vertical": ("FLOAT", {
"default": 0.0,
"min": -10.0,
"max": 10.0,
"step": 0.5
}),
"camera_pan": ("FLOAT", {
"default": 0.0,
"min": -10.0,
"max": 10.0,
"step": 0.5
}),
"camera_tilt": ("FLOAT", {
"default": 0.0,
"min": -10.0,
"max": 10.0,
"step": 0.5
}),
"camera_roll": ("FLOAT", {
"default": 0.0,
"min": -10.0,
"max": 10.0,
"step": 0.5
}),
"camera_zoom": ("FLOAT", {
"default": 0.0,
"min": -10.0,
"max": 10.0,
"step": 0.5
}),
"external_task_id": ("STRING", {
"default": "",
"multiline": False,
"placeholder": "可选：自定义任务ID（需要确保唯一性）"
}),
"callback_url": ("STRING", {
"default": "",
"multiline": False,
"placeholder": "可选：任务结果回调通知地址"
}),
"seed": ("INT", {
"default": -1,
"min": -1,
"max": 0xffffffffffffffff
}),
}
}
RETURN_TYPES = ("STRING", "STRING", "STRING", "STRING", "INT")
RETURN_NAMES = ("task_id", "task_status", "created_at", "updated_at", "seed")
FUNCTION = "create_image2video_task"
CATEGORY = "JM-KLingAI-API/image-2-video"
def tensor_to_pil(self, tensor):
"""将ComfyUI的Tensor图像转换为PIL图像"""
if tensor is None:
return None
# 确保是一个单一的图像，不是批次
if len(tensor.shape) == 4:
tensor = tensor[0]
# 转换为适合PIL的格式
i = 255. * tensor.cpu().numpy()
img = Image.fromarray(np.clip(i, 0, 255).astype(np.uint8))
return img
def image_to_base64(self, image):
"""将ComfyUI图像转换为base64字符串"""
if image is None:
return None
try:
# 处理tensor格式的图像
if isinstance(image, torch.Tensor):
pil_image = self.tensor_to_pil(image)
if pil_image is None:
return None
else:
# 原来的处理方法 (已不再使用，保留以防万一)
img = Image.fromarray((image[0] * 255).astype('uint8'), 'RGB')
pil_image = img
# 转换为base64
buffered = io.BytesIO()
pil_image.save(buffered, format="JPEG")
return base64.b64encode(buffered.getvalue()).decode("utf-8")
except Exception as e:
print(f"图像转换base64错误: {str(e)}")
return None
def get_camera_control(self, camera_type, h, v, pan, tilt, roll, zoom):
"""构建摄像机控制参数"""
camera_control = {"type": camera_type}
# 如果是simple类型，添加config配置
if camera_type == "simple":
# 确保只有一个方向参数不为0
non_zero_params = sum(1 for p in [h, v, pan, tilt, roll, zoom] if abs(p) > 0.001)
if non_zero_params > 1:
print("警告：simple类型摄像机控制只能设置一个方向参数，将使用第一个非零参数")
camera_control["config"] = {
"horizontal": h,
"vertical": v,
"pan": pan,
"tilt": tilt,
"roll": roll,
"zoom": zoom
}
return camera_control
def validate_camera_params(self, camera_type, h, v, pan, tilt, roll, zoom):
"""验证摄像机参数合法性"""
if camera_type == "simple":
non_zero_count = sum(1 for p in [h, v, pan, tilt, roll, zoom] if abs(p) > 0.001)
if non_zero_count > 1:
return False, "simple类型摄像机控制只能设置一个方向参数不为0"
if non_zero_count == 0:
return False, "simple类型摄像机控制至少需要一个方向参数不为0"
return True, ""
def create_image2video_task(self, api_token, image, model_name="kling-v1",
positive_prompt="", negative_prompt="",
cfg_scale=0.5, mode="std", duration="5",
image_tail=None, use_camera_control=False,
camera_type="simple", camera_horizontal=0.0,
camera_vertical=0.0, camera_pan=0.0,
camera_tilt=0.0, camera_roll=0.0,
camera_zoom=0.0, external_task_id="",
callback_url="", seed=-1):
"""
创建图生视频任务
"""
try:
# 验证必要参数
if not api_token:
raise ValueError("API令牌不能为空")
if image is None:
raise ValueError("输入图像不能为空")
# 生成随机种子（本地使用，不发送给API）
if seed == -1:
seed = random.randint(0, 0xffffffffffffffff)
# 验证文本提示词长度
if positive_prompt and len(positive_prompt) > 2500:
raise ValueError("正向提示词不能超过2500字符")
if negative_prompt and len(negative_prompt) > 2500:
raise ValueError("负向提示词不能超过2500字符")
# 验证摄像机参数
if use_camera_control:
valid, msg = self.validate_camera_params(
camera_type, camera_horizontal, camera_vertical,
camera_pan, camera_tilt, camera_roll, camera_zoom
)
if not valid:
raise ValueError(f"摄像机参数错误: {msg}")
# 转换图像为base64
image_base64 = self.image_to_base64(image)
if not image_base64:
raise ValueError("图像转换为base64失败")
image_tail_base64 = self.image_to_base64(image_tail) if image_tail is not None else None
# 准备请求头
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_token.strip()}"
}
# 准备请求体
payload = {
"model_name": model_name,
"image": image_base64,
"cfg_scale": float(cfg_scale),
"mode": mode,
"duration": duration
}
# 添加可选参数
if positive_prompt:
payload["prompt"] = positive_prompt
if negative_prompt:
payload["negative_prompt"] = negative_prompt
if image_tail_base64:
payload["image_tail"] = image_tail_base64
if external_task_id:
payload["external_task_id"] = external_task_id
if callback_url:
payload["callback_url"] = callback_url
# 添加摄像机控制参数
if use_camera_control:
payload["camera_control"] = self.get_camera_control(
camera_type, camera_horizontal, camera_vertical,
camera_pan, camera_tilt, camera_roll, camera_zoom
)
# 发送API请求
url = f"{self.api_base}{self.endpoint}"
print(f"正在发送请求到: {url}")
print(f"使用本地种子: {seed} (仅用于本地，未发送给API)")
response = requests.post(url, headers=headers, json=payload)
response_data = response.json()
print(f"响应状态码: {response.status_code}")
print(f"响应数据: {json.dumps(response_data, indent=2)}")
# 检查错误并提供详细错误信息
if response.status_code != 200:
error_code = response_data.get('code')
error_message = response_data.get('message')
request_id = response_data.get('request_id')
raise Exception(f"API请求失败 (错误码: {error_code}): {error_message} (请求ID: {request_id})")
# 提取响应数据
data = response_data.get("data", {})
task_id = data.get("task_id", "")
task_status = data.get("task_status", "")
created_at = str(data.get("created_at", ""))
updated_at = str(data.get("updated_at", ""))
if not task_id:
raise Exception("API未返回任务ID")
print(f"成功创建图生视频任务，任务ID: {task_id} (本地种子: {seed})")
return (task_id, task_status, created_at, updated_at, seed)
except ValueError as ve:
print(f"参数验证错误: {str(ve)}")
return (None, None, None, None, seed)
except Exception as e:
print(f"创建图生视频任务错误: {str(e)}")
return (None, None, None, None, seed)
def IS_CHANGED(self, api_token, image, model_name="kling-v1",
positive_prompt="", negative_prompt="",
cfg_scale=0.5, mode="std", duration="5",
image_tail=None, use_camera_control=False,
camera_type="simple", camera_horizontal=0.0,
camera_vertical=0.0, camera_pan=0.0,
camera_tilt=0.0, camera_roll=0.0,
camera_zoom=0.0, external_task_id="",
callback_url="", seed=-1):
"""
此方法用于判断节点是否需要重新执行
我们使用种子控制重新执行逻辑
"""
if seed == -1:
return random.randint(0, 0xffffffffffffffff)
return seed
nodes/query_status.py
+163 -52
View File
@@ -1,17 +1,29 @@
import json
import requests
import time
from threading import Thread, Event
# 定义一个带有结果存储的线程类
class TaskStatusThread(Thread):
def __init__(self, target, args):
Thread.__init__(self, target=target, args=args)
self.result = None
def run(self):
self.result = self._target(*self._args)
class KLingAIQueryStatus:
"""
KLingAI Query Task Status Node
Queries the status of a text-to-video generation task
查询文生视频或图生视频任务状态
"""
def __init__(self):
self.api_base = "https://api.klingai.com"
self.endpoint = "/v1/videos/text2video/{}"
self.text2video_endpoint = "/v1/videos/text2video/{}"
self.image2video_endpoint = "/v1/videos/image2video/{}"
self.stop_thread = Event()
self.current_thread = None
@@ -26,7 +38,24 @@ class KLingAIQueryStatus:
"external_task_id": ("STRING", {
"default": "",
"multiline": False,
"placeholder": "Optional: External task ID"
"placeholder": "可选: 自定义任务ID"
}),
"task_type": (["auto", "text2video", "image2video"], {
"default": "auto"
}),
"initial_delay_seconds": ("INT", {
"default": 10,
"min": 0,
"max": 60,
"step": 1,
"display": "slider"
}),
"poll_interval_seconds": ("INT", {
"default": 10,
"min": 5,
"max": 30,
"step": 1,
"display": "slider"
}),
}
}
@@ -36,100 +65,182 @@ class KLingAIQueryStatus:
FUNCTION = "query_task_status"
CATEGORY = "JM-KLingAI-API"
def poll_status(self, api_token, task_id, external_task_id):
def poll_status(self, api_token, task_id, external_task_id, task_type="auto", initial_delay_seconds=10, poll_interval_seconds=10):
"""
Polls the task status every 10 seconds until completion
轮询任务状态直到完成
"""
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_token}"
}
# 初始等待时间，给服务器处理任务的时间
if initial_delay_seconds > 0:
print(f"等待 {initial_delay_seconds} 秒后开始查询任务状态...")
if self.stop_thread.wait(initial_delay_seconds):
return None
print("开始查询任务状态...")
# 确定使用哪个查询ID
query_id = task_id if task_id else external_task_id
# 根据任务类型确定查询端点
endpoints = []
if task_type == "auto":
# 根据任务ID长度初步判断可能是哪种类型
if len(query_id) > 10: # 可灵AI的任务ID通常很长
endpoints = [self.image2video_endpoint, self.text2video_endpoint]
else:
endpoints = [self.text2video_endpoint, self.image2video_endpoint]
elif task_type == "text2video":
endpoints = [self.text2video_endpoint]
elif task_type == "image2video":
endpoints = [self.image2video_endpoint]
# 记录最后找到的有效端点
valid_endpoint = None
while not self.stop_thread.is_set():
try:
# Determine which ID to use
query_id = task_id if task_id else external_task_id
url = f"{self.api_base}{self.endpoint.format(query_id)}"
success = False
# 如果上一次查询找到了有效端点，只使用该端点
if valid_endpoint:
current_endpoints = [valid_endpoint]
else:
current_endpoints = endpoints.copy()
for endpoint in current_endpoints:
try:
url = f"{self.api_base}{endpoint.format(query_id)}"
print(f"查询任务 {query_id} 的状态，使用端点: {endpoint}...")
response = requests.get(url, headers=headers)
# 首先检查响应状态码
if response.status_code == 404:
print(f"在 {endpoint} 未找到任务，尝试其他端点...")
continue
# 解析响应数据
try:
response_data = response.json()
except Exception as e:
print(f"解析响应数据失败: {str(e)}，尝试其他端点...")
continue
# 检查响应中的错误信息
if response.status_code != 200:
error_message = response_data.get('message', '未知错误')
print(f"查询状态错误: {error_message}")
# 任务未找到，这可能是端点错误
if "not found" in error_message.lower():
print(f"在 {endpoint} 未找到任务，尝试其他端点...")
continue
# 其他API错误情况
else:
print(f"API返回错误: {error_message}")
continue
response = requests.get(url, headers=headers)
response_data = response.json()
if response.status_code != 200:
print(f"Error in status query: {response_data.get('message', 'Unknown error')}")
# 获取任务状态
data = response_data.get("data", {})
status = data.get("task_status")
# 如果能获取到状态，说明端点正确
if status:
valid_endpoint = endpoint
success = True
print(f"当前任务状态: {status}")
# 任务成功完成
if status == "succeed":
videos = data.get("task_result", {}).get("videos", [])
if videos and videos[0].get("url"):
return videos[0].get("url")
print("任务成功但未返回视频URL")
return None
# 任务失败
elif status == "failed":
print(f"任务失败: {data.get('task_status_msg', '未知错误')}")
return None
# 任务仍在处理中，中断当前端点循环，等待下次查询
break
else:
print(f"API返回数据中缺少任务状态")
except Exception as e:
print(f"查询端点 {endpoint} 出错: {str(e)}")
# 如果尝试了所有端点但都未成功
if not success and not valid_endpoint:
if not current_endpoints:
print("没有可用的端点进行查询")
return None
data = response_data.get("data", {})
status = data.get("task_status")
print(f"Current task status: {status}")
if status == "succeed":
videos = data.get("task_result", {}).get("videos", [])
if videos:
return videos[0].get("url")
return None
elif status == "failed":
print(f"Task failed: {data.get('task_status_msg', 'Unknown error')}")
return None
# Wait 10 seconds before next poll
self.stop_thread.wait(10)
except Exception as e:
print(f"Error during status polling: {str(e)}")
print("所有端点查询失败，稍后将重试...")
# 按照配置的时间间隔等待再次查询
print(f"等待 {poll_interval_seconds} 秒后再次查询...")
if self.stop_thread.wait(poll_interval_seconds):
return None
return None
def query_task_status(self, api_token, task_id, external_task_id=""):
def query_task_status(self, api_token, task_id, external_task_id="", task_type="auto", initial_delay_seconds=10, poll_interval_seconds=10):
"""
Start polling task status
开始轮询任务状态
"""
try:
# Validate inputs
# 验证输入
if not api_token:
raise ValueError("API token is required")
raise ValueError("API令牌不能为空")
if not task_id and not external_task_id:
raise ValueError("Either task_id or external_task_id must be provided")
raise ValueError("必须提供task_id或external_task_id")
# Stop any existing polling thread
# 显示查询配置信息
print(f"任务ID: {task_id or external_task_id}")
print(f"任务类型: {task_type}")
print(f"初始等待时间: {initial_delay_seconds}秒")
print(f"查询间隔: {poll_interval_seconds}秒")
# 停止任何正在进行的轮询线程
if self.current_thread and self.current_thread.is_alive():
self.stop_thread.set()
self.current_thread.join()
# Reset stop event
# 重置停止事件
self.stop_thread.clear()
# Start new polling thread
self.current_thread = Thread(
# 创建并启动新的轮询线程
self.current_thread = TaskStatusThread(
target=self.poll_status,
args=(api_token, task_id, external_task_id)
args=(api_token, task_id, external_task_id, task_type, initial_delay_seconds, poll_interval_seconds)
)
self.current_thread.start()
# Wait for result
# 等待结果
self.current_thread.join()
# Get result
video_url = self.poll_status(api_token, task_id, external_task_id)
# 获取结果
video_url = self.current_thread.result
if video_url:
print(f"Task completed successfully. Video URL: {video_url}")
print(f"任务成功完成。视频URL: {video_url}")
return (video_url,)
else:
print("Task completed but no video URL available")
print("任务完成但没有可用的视频URL")
return (None,)
except ValueError as ve:
print(f"Validation Error: {str(ve)}")
print(f"参数验证错误: {str(ve)}")
return (None,)
except Exception as e:
print(f"Error querying task status: {str(e)}")
print(f"查询任务状态错误: {str(e)}")
return (None,)
def __del__(self):
"""
Cleanup when node is destroyed
节点销毁时清理资源
"""
if self.current_thread and self.current_thread.is_alive():
self.stop_thread.set()