diff --git a/__init__.py b/__init__.py index 80573c8..e8bfad6 100644 --- a/__init__.py +++ b/__init__.py @@ -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" } diff --git a/nodes/image2video.py b/nodes/image2video.py new file mode 100644 index 0000000..0d60652 --- /dev/null +++ b/nodes/image2video.py @@ -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 \ No newline at end of file diff --git a/nodes/query_status.py b/nodes/query_status.py index 0fd0272..5e10ccc 100644 --- a/nodes/query_status.py +++ b/nodes/query_status.py @@ -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()