diff --git a/__init__.py b/__init__.py index 68e2848..2d6f81e 100644 --- a/__init__.py +++ b/__init__.py @@ -12,6 +12,13 @@ from .nodes.video_lipsync_nodes import HeyGemF2F, HeyGemF2FFromFile from .nodes.video_nodes import VideoCut, VideoCutByFramePoint, VideoChangeFPS, VideoStartPointDurationCompute, \ VideoMerge +from .nodes.union_llm_node import LLMUionNode +from .nodes.img_agent import ImgSubmitNode +from .nodes.video_agent import VideoSubmitNode +from .nodes.save_node import ExtSaveNode +from .nodes.video_preview import VideoDownloaderNode +from .nodes.fetch_task_result import FetchTaskResult + NODE_CLASS_MAPPINGS = { "FaceOccDetect": FaceDetect, "FaceExtract": FaceExtract, @@ -50,7 +57,13 @@ NODE_CLASS_MAPPINGS = { "ModalMidJourneyDescribeImage": ModalMidJourneyDescribeImage, "JMCustom": JMCustom, "VideoMerge": VideoMerge, - "SaveImageAnywhere": SaveImageAnywhere + "SaveImageAnywhere": SaveImageAnywhere, + "LLMUionNode": LLMUionNode, + "ImgSubmitNode": ImgSubmitNode, + "VideoSubmitNode": VideoSubmitNode, + "ExtSaveNode": ExtSaveNode, + "VideoDownloaderNode": VideoDownloaderNode, + "FetchTaskResult": FetchTaskResult } NODE_DISPLAY_NAME_MAPPINGS = { @@ -90,6 +103,12 @@ NODE_DISPLAY_NAME_MAPPINGS = { "ModalMidJourneyGenerateImage": "Prompt生/修图-MJ", "ModalMidJourneyDescribeImage": "反推生图提示词-MJ", "JMCustom": "Prompt生视频", - "VideoMerge":"顺序合并视频", - "SaveImageAnywhere": "保存图片-任意路径" + "VideoMerge": "顺序合并视频", + "SaveImageAnywhere": "保存图片-任意路径", + "LLMUionNode": "LLM多模态节点", + "ImgSubmitNode": "提交图片生成", + "VideoSubmitNode": "提交视频生成", + "ExtSaveNode": "通用文件保存", + "VideoDownloaderNode": "视频下载", + "FetchTaskResult": "获取生成结果 (图片/视频链接)" } diff --git a/config.yaml b/config.yaml new file mode 100644 index 0000000..ee62ab0 --- /dev/null +++ b/config.yaml @@ -0,0 +1,7 @@ +aws_access_key: kfAqoOmIiyiywi25xaAkJUQbZ/EKDnzvI6NRCW1l +aws_key_id: AKIAYRH5NGRSWHN2L4M6 +cos_region: ap-shanghai +cos_secret_id: AKIDsrihIyjZOBsjimt8TsN8yvv1AMh5dB44 +cos_secret_key: CPZcxdk6W39Jd4cGY95wvupoyMd0YFqW +jm_api_key: 21575c22-14aa-40ca-8aa8-f00ca27a3a17 +cos_sucai_bucket_name: sucai-1324682537 \ No newline at end of file diff --git a/nodes/fetch_task_result.py b/nodes/fetch_task_result.py new file mode 100644 index 0000000..2e9d298 --- /dev/null +++ b/nodes/fetch_task_result.py @@ -0,0 +1,145 @@ +import comfy.utils +import folder_paths +import time +import requests +import torch +import numpy as np +from PIL import Image +import base64 +from io import BytesIO +import json +from urllib.parse import urlparse +import os + + +class FetchTaskResult: + # 1. 定义环境 URL 映射 + ENV_URLS = { + "prod": "https://bowongai-prod--text-video-agent-fastapi-app.modal.run", + "test": "https://bowongai-test--text-video-agent-fastapi-app.modal.run", + "dev": "https://bowongai-dev--text-video-agent-fastapi-app.modal.run" + } + + @classmethod + def INPUT_TYPES(s): + return { + "required": { + "env": (list(s.ENV_URLS.keys()),), # 创建一个包含 "prod", "test", "dev" 的下拉列表 + "task_id": ("STRING", {"default": ""}), + "interval": ("INT", {"default": 2, "min": 1, "max": 60}), + "timeout": ("INT", {"default": 300, "min": 10, "max": 3600}), + }, + } + + RETURN_TYPES = ("IMAGE", "STRING", "STRING") + RETURN_NAMES = ("images", "video_urls", "raw_response") + FUNCTION = "execute" + + CATEGORY = "不忘科技-自定义节点🚩/utils/获取结果" + + def execute(self, env, task_id, interval, timeout): + # 4. 根据选择的 env 从映射中获取 base_url + base_url = self.ENV_URLS[env] + + if not task_id: + raise ValueError("Task ID 不能为空 (Task ID cannot be empty)") + + headers = {} # 如果需要，可在此处添加 headers + + start_time = time.time() + while time.time() - start_time < timeout: + try: + params = {'task_id': task_id} + print(f"[{env}] 正在轮询: {base_url}/api/custom/task/status?task_id={task_id}") + response = requests.get( + f'{base_url}/api/custom/task/status', params=params, headers=headers) + response.raise_for_status() + + data_ = response.json() + print(f'原始响应结果:{data_}') + api_status = data_.get('status') + data = data_.get('data', []) + + if isinstance(api_status, bool): + if not api_status: + raise ValueError(f'{data_["msg"]}') + print(f"任务 {task_id} 成功完成。正在分流处理媒体...") + + image_tensors, video_urls = self.dispatch_media(data) + + final_images = torch.cat(image_tensors, dim=0) if image_tensors else torch.empty(0, 64, 64, 3, + dtype=torch.float32) + final_urls = "\n".join(video_urls) + raw_response_str = json.dumps(data_, indent=2, ensure_ascii=False) + + print(f"处理完成: {len(image_tensors)} 个图像, {len(video_urls)} 个视频URL。") + return (final_images, final_urls, raw_response_str) + + print(f"任务未完成。API返回状态: {api_status}。将在 {interval} 秒后重试...") + time.sleep(interval) + + except requests.exceptions.RequestException as e: + print(f"请求 API 失败: {e}. {interval} 秒后重试...") + time.sleep(interval) + except Exception as e: + print(f"处理任务时发生未知错误: {e}") + raise e + + raise TimeoutError(f"轮询任务 {task_id} 超时 ({timeout} 秒)。") + + def tensor_from_pil(self, img_pil): + return torch.from_numpy(np.array(img_pil).astype(np.float32) / 255.0)[None,] + + def dispatch_media(self, data): + if not isinstance(data, list): + return [], [] + + image_tensors = [] + video_urls = [] + + IMAGE_EXTS = ['.png', '.jpg', '.jpeg', '.bmp', '.webp'] + VIDEO_EXTS = ['.mp4', '.webm', '.mkv', '.avi', '.mov'] + + for i, item in enumerate(data): + if not isinstance(item, str): continue + + # 方案 A: 检查是否为 URL + if item.startswith(('http://', 'https://')): + try: + url_path = urlparse(item).path + ext = os.path.splitext(url_path)[1].lower() + + if ext in IMAGE_EXTS: + print(f" -> 识别到图片URL，正在下载和处理...") + response = requests.get(item) + response.raise_for_status() + img = Image.open(BytesIO(response.content)).convert("RGB") + image_tensors.append(self.tensor_from_pil(img)) + + elif ext in VIDEO_EXTS: + print(f" -> 识别到视频URL，直接返回链接。") + video_urls.append(item) + + else: + print(f" -> 识别到未知类型URL '{item}'，已跳过。") + + except Exception as e: + print(f" -> 处理URL时出错: {e}") + else: + try: + print(f" -> 尝试作为 Base64 图片解码...") + img_data = base64.b64decode(item) + img = Image.open(BytesIO(img_data)).convert("RGB") + image_tensors.append(self.tensor_from_pil(img)) + except Exception: + print(f" -> 解码失败，该项不是有效的媒体。") + + return image_tensors, video_urls + +NODE_CLASS_MAPPINGS = { + "FetchTaskResult": FetchTaskResult +} + +NODE_DISPLAY_NAME_MAPPINGS = { + "FetchTaskResult": "获取生成结果 (图片/视频链接)" +} diff --git a/nodes/img_agent.py b/nodes/img_agent.py new file mode 100644 index 0000000..dcd5724 --- /dev/null +++ b/nodes/img_agent.py @@ -0,0 +1,219 @@ +# -*- coding:utf-8 -*- +""" +File img_agent.py +Author silence +Date 2025/9/6 +""" + +import json +import requests +import os +import folder_paths +import mimetypes +from PIL import Image +import numpy as np +import torch +import io +import re + +try: + from loguru import logger +except ImportError: + import logging + + logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') + logger = logging.getLogger("ImgSubmitNode_Final") + print("提示: loguru 未安装，使用内置logging。建议安装以获得更好的日志体验: pip install loguru") + + +def fetch_and_process_image_models(): + """ + 在节点加载时从API获取生图模型列表，并存储其配置用于后端校验。 + """ + image_model_urls = { + "prod": "https://bowongai-prod--text-video-agent-fastapi-app.modal.run/api/custom/model/list?category=image", + "dev": "https://bowongai-dev--text-video-agent-fastapi-app.modal.run/api/custom/model/list?category=image", + "test": "https://bowongai-test--text-video-agent-fastapi-app.modal.run/api/custom/model/list?category=image" + } + + model_data = { + "configs": {}, + "full_display_list": [], + "display_to_tech_name": {}, + "temp_list_for_sorting": [] + } + + try: + response = None + for env, url in image_model_urls.items(): + try: + response = requests.get(url, timeout=10) + response.raise_for_status() + logger.info(f"成功从 [{env}] 环境获取生图模型列表。") + break + except requests.exceptions.RequestException: + logger.warning(f"无法从 [{env}] 环境获取模型列表，尝试下一个...") + continue + + if not response: + raise ConnectionError("所有环境的模型列表API都无法访问。") + + data = response.json() + if not data.get("status") or "data" not in data: + raise ValueError(f"API响应格式错误: {data.get('msg', '未知错误')}") + + for model in data["data"]: + tech_name = model.get("model_name") + if not tech_name: continue + + # --- 核心修正：不再手动添加任何前缀 --- + + # 1. 直接从API获取description，并用strip()清理首尾空格 + description_from_api = str(model.get("description", tech_name)).strip() + + # 2. 生成最终的显示名称 (description本身已包含前缀) + display_name = f"{description_from_api} ({tech_name})" + + # 3. 仅根据mode分配排序键，用于内部排序 + mode = model.get("mode") + sort_key = 99 + if mode == "t2i": + sort_key = 0 + elif mode == "i2i": + sort_key = 1 + elif mode == "both": + sort_key = 2 + + # 4. 存储所有信息 + model_data["configs"][tech_name] = model + model_data["display_to_tech_name"][display_name] = tech_name + model_data["temp_list_for_sorting"].append((sort_key, display_name)) + + model_data["temp_list_for_sorting"].sort(key=lambda x: (x[0], x[1])) + model_data["full_display_list"] = [item[1] for item in model_data["temp_list_for_sorting"]] + + except Exception as e: + logger.error(f"加载生图模型数据失败: {e}") + + if not model_data["full_display_list"]: + model_data["full_display_list"] = ["错误：无法加载模型"] + + return model_data + + +IMAGE_MODEL_DATA = fetch_and_process_image_models() + + +class ImgSubmitNode: + MODEL_DATA = IMAGE_MODEL_DATA + + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + "model_name_display": (cls.MODEL_DATA["full_display_list"],), + "prompt": ("STRING", {"multiline": True, "default": ""}), + "aspect_ratio": ("STRING", {"multiline": False, "default": "1:1"}), + "environment": (["prod", "dev", "test"], {"default": "prod"}), + }, + "optional": { + "image": ("IMAGE",), + "image_filename": ("STRING", {"multiline": False, "default": ""}), + } + } + + RETURN_TYPES = ("STRING",) + RETURN_NAMES = ("data",) + FUNCTION = "submit_task" + CATEGORY = "不忘科技-自定义节点🚩/api/图片生成" + + def _get_base_url_and_tech_name(self, environment, model_name_display): + env_map = { + "prod": "https://bowongai-prod--text-video-agent-fastapi-app.modal.run", + "dev": "https://bowongai-dev--text-video-agent-fastapi-app.modal.run", + "test": "https://bowongai-test--text-video-agent-fastapi-app.modal.run" + } + base_url = env_map.get(environment, env_map["prod"]) + tech_name = self.MODEL_DATA["display_to_tech_name"].get(model_name_display) + if not tech_name: + match = re.search(r'\((.*?)\)', model_name_display) + tech_name = match.group(1) if match else model_name_display + logger.info(f"环境: [{environment}], 模型: '{model_name_display}' -> '{tech_name}'") + return base_url, tech_name + + def submit_task(self, model_name_display, prompt, aspect_ratio, environment, image_filename=None, image=None): + try: + base_url, tech_name = self._get_base_url_and_tech_name(environment, model_name_display) + model_config = self.MODEL_DATA["configs"].get(tech_name) + if not model_config: + raise ValueError(f"无法找到模型 '{tech_name}' 的配置。") + + def validate_and_correct_parameter(param_name, user_value, supported_values): + if not supported_values: return user_value + if user_value in supported_values: return user_value + + default_value = supported_values[0] + logger.warning( + f"参数警告！模型 '{tech_name}' 不支持 '{param_name}': '{user_value}'。" + f"已自动替换为支持的默认值: '{default_value}'。支持的选项: {supported_values}" + ) + return default_value + + final_ar = validate_and_correct_parameter("宽高比", aspect_ratio, model_config.get("supported_ar", [])) + + headers = {'accept': 'application/json'} + payload = {'prompt': prompt, 'model_name': tech_name, 'aspect_ratio': final_ar, + 'mode': 'turbo', 'webhook_flag': 'false'} + files_to_send = {} + file_obj = None + + if image is not None: + logger.info(f"检测到 IMAGE (Tensor) 输入，优先处理。") + img_tensor = image[0] + img_np = np.clip(255. * img_tensor.cpu().numpy(), 0, 255).astype(np.uint8) + pil_image = Image.fromarray(img_np) + buffer = io.BytesIO() + pil_image.save(buffer, format="PNG") + buffer.seek(0) + files_to_send['img_file'] = ('image_from_workflow.png', buffer, 'image/png') + elif image_filename and image_filename.strip(): + logger.info(f"处理文件名: {image_filename}") + full_path = folder_paths.get_full_path("input", image_filename.strip()) + if not (full_path and os.path.exists(full_path)): + return (f"错误: 在ComfyUI的input文件夹中未找到文件 '{image_filename}'",) + filename = os.path.basename(full_path) + mime_type, _ = mimetypes.guess_type(full_path) or ('application/octet-stream', None) + file_obj = open(full_path, 'rb') + files_to_send['img_file'] = (filename, file_obj, mime_type) + else: + logger.info("未提供任何图像输入，以纯文本模式运行。") + + api_endpoint = f'{base_url}/api/custom/image/submit/task' + logger.info(f"向端点 {api_endpoint} 发送请求...") + + response = requests.post( + api_endpoint, headers=headers, data=payload, files=files_to_send, timeout=60 + ) + response.raise_for_status() + response_json = response.json() + logger.info(f"任务提交成功，完整响应: {json.dumps(response_json, indent=2, ensure_ascii=False)}") + + if response_json.get('status') is True: + return (str(response_json.get('data', "错误: 状态为true但缺少data字段")),) + else: + return (json.dumps(response_json, indent=4, ensure_ascii=False),) + + except Exception as e: + logger.error(f"任务处理失败: {e}") + return (f"错误: {str(e)}",) + finally: + if file_obj: + file_obj.close() + +# NODE_CLASS_MAPPINGS = { +# "ImgSubmitNode": ImgSubmitNode +# } +# +# NODE_DISPLAY_NAME_MAPPINGS = { +# "ImgSubmitNode": "统一生图任务节点" +# } diff --git a/nodes/save_node.py b/nodes/save_node.py new file mode 100644 index 0000000..cc70fc9 --- /dev/null +++ b/nodes/save_node.py @@ -0,0 +1,169 @@ +import mimetypes +import os +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import datetime +from urllib.parse import urlparse + +# 导入 ComfyUI 的路径管理器 +import folder_paths +import requests +import torch +from PIL import Image + + +class ExtSaveNode: + + def __init__(self): + self.executor = ThreadPoolExecutor(max_workers=10) + self.output_dir = folder_paths.get_output_directory() + + @classmethod + def INPUT_TYPES(s): + return { + "required": {}, + "optional": { + # multiline=True 可以让UI中的输入框更大，但处理逻辑已兼容多行 + "url_input": ("STRING", {"multiline": True, "default": ""}), + "image_tensor_input": ("IMAGE",), + "subdirectory": ("STRING", {"multiline": False, "default": ""}), + "download_file_type": (["auto", "image", "video", "other"],), + "image_file_prefix": ("STRING", {"multiline": False, "default": "ComfyUI_Image_"}), + "image_file_format": (["png", "jpeg"],), + "jpeg_quality": ("INT", {"default": 90, "min": 1, "max": 100}), + } + } + + RETURN_TYPES = ("STRING", "STRING") + RETURN_NAMES = ("downloaded_paths", "saved_image_paths") + FUNCTION = "process_inputs" + CATEGORY = "不忘科技-自定义节点🚩/utils/文件保存" + + def _get_save_path(self, subdirectory: str) -> str: + full_path = os.path.join(self.output_dir, subdirectory) + os.makedirs(full_path, exist_ok=True) + return full_path + + def _download_file_threaded(self, url, save_path, file_type): + try: + parsed_url = urlparse(url) + filename = os.path.basename(parsed_url.path) + + if not filename or "." not in filename: + try: + with requests.head(url, allow_redirects=True, timeout=60) as h: + h.raise_for_status() + content_type = h.headers.get('content-type') + ext = mimetypes.guess_extension(content_type) if content_type else None + final_ext = ext if ext else "" + filename = f"downloaded_file_{os.urandom(4).hex()}{final_ext}" + except requests.exceptions.RequestException as e: + print(f"Could not determine filename from headers for {url}: {e}") + filename = f"downloaded_file_{os.urandom(4).hex()}" + + file_path = os.path.join(save_path, filename) + + if os.path.exists(file_path): + name, ext = os.path.splitext(filename) + timestamp = datetime.now().strftime("_%Y%m%d%H%M%S%f")[:-3] + filename = f"{name}{timestamp}{ext}" + file_path = os.path.join(save_path, filename) + + print(f"Starting download of {url} to {file_path}") + with requests.get(url, stream=True, timeout=300) as r: + r.raise_for_status() + with open(file_path, 'wb') as f: + for chunk in r.iter_content(chunk_size=8192): + f.write(chunk) + print(f"Finished downloading {url} to {file_path}") + return file_path + except Exception as e: + print(f"Error downloading {url}: {e}") + return f"Download Error: {e}" + + def _save_image_tensor(self, images: torch.Tensor, save_path: str, file_prefix: str, file_format: str, + jpeg_quality: int): + """保存图像Tensor的核心逻辑""" + saved_paths = [] + for i, image_tensor in enumerate(images): + try: + img_np = (image_tensor.cpu().numpy() * 255).astype('uint8') + + if img_np.shape[2] == 1: + img_pil = Image.fromarray(img_np.squeeze(axis=2), mode='L') + elif img_np.shape[2] == 4: + img_pil = Image.fromarray(img_np, mode='RGBA') + else: + img_pil = Image.fromarray(img_np, mode='RGB') + + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:-3] + filename = f"{file_prefix}{timestamp}_{i}.{file_format}" + full_path = os.path.join(save_path, filename) + + if file_format == "png": + img_pil.save(full_path, format="PNG") + elif file_format == "jpeg": + img_pil.save(full_path, format="JPEG", quality=jpeg_quality) + + saved_paths.append(full_path) + except Exception as e: + print(f"Error saving image tensor {i}: {e}") + saved_paths.append(f"Save Error: {e}") + + return ", ".join(saved_paths) + + def process_inputs(self, + url_input: str = "", + image_tensor_input: torch.Tensor = None, + subdirectory: str = "", + download_file_type: str = "auto", + image_file_prefix: str = "ComfyUI_Image_", + image_file_format: str = "png", + jpeg_quality: int = 90): + + downloaded_paths_output = "" + saved_image_paths_output = "" + + final_save_path = self._get_save_path(subdirectory) + + if url_input: + url_input = url_input.strip() + if '\n' in url_input: + lines = [line.strip() for line in url_input.strip().split('\n')] + else: + lines = [line.strip() for line in url_input.strip().split()] + urls_to_download = [line for line in lines if line.startswith(('http://', 'https://'))] + + if urls_to_download: + print(f"Found {len(urls_to_download)} URLs to download. Saving to: {final_save_path}") + + futures = { + self.executor.submit(self._download_file_threaded, url, final_save_path, download_file_type): url + for url in urls_to_download} + + downloaded_paths = [] + for future in as_completed(futures): + result_path = future.result() + downloaded_paths.append(result_path) + + downloaded_paths_output = ", ".join(downloaded_paths) + if image_tensor_input is not None and isinstance(image_tensor_input, + torch.Tensor) and image_tensor_input.numel() > 0: + print(f"Detected Image Tensor input, will save to: {final_save_path}") + saved_image_paths_output = self._save_image_tensor( + image_tensor_input, + final_save_path, + image_file_prefix, + image_file_format, + jpeg_quality + ) + + return (downloaded_paths_output, saved_image_paths_output) + + +# NODE_CLASS_MAPPINGS = { +# "UniversalSaver": ExtSaveNode +# } +# +# NODE_DISPLAY_NAME_MAPPINGS = { +# "UniversalSaver": "通用文件保存" +# } \ No newline at end of file diff --git a/nodes/union_llm_node.py b/nodes/union_llm_node.py new file mode 100644 index 0000000..45dbfd4 --- /dev/null +++ b/nodes/union_llm_node.py @@ -0,0 +1,168 @@ +# -*- coding:utf-8 -*- +""" +File union_llm_node.py +Author silence +Date 2025/9/5 +""" +import os +import requests +import base64 +import mimetypes +import torch +import numpy as np +from PIL import Image +import folder_paths + +tensor_to_file_map = {} + + +class LLMUionNode: + """ + 一个聚合LLM节点。最终修复版，根据用户指正，彻底重构了执行逻辑， + 确保代码的清晰、正确和稳定。 + """ + MODELS = ['gemini-2.5-flash', 'gemini-2.5-pro', "gpt-4o-1120", "gpt-4.1"] + + ENVIRONMENTS = ["prod", "dev", "test"] + ENV_URLS = { + "prod": 'https://bowongai-prod--text-video-agent-fastapi-app.modal.run', + "dev": 'https://bowongai-dev--text-video-agent-fastapi-app.modal.run', + "test": 'https://bowongai-test--text-video-agent-fastapi-app.modal.run' + } + + @classmethod + def INPUT_TYPES(s): + return { + "required": { + "model_name": (s.MODELS,), + "prompt": ("STRING", { "multiline": True, "default": "详细描述这个视频" }), + }, + "optional": { + "video_input": ("*",), + "image": ("IMAGE",), + "environment": (s.ENVIRONMENTS,), + "timeout": ("INT", {"default": 300, "min": 10, "max": 1200}), + } + } + + RETURN_TYPES = ("STRING",) + RETURN_NAMES = ("text",) + FUNCTION = "execute" + CATEGORY = "不忘科技-自定义节点🚩/LLM" + + def tensor_to_pil(self, tensor): + if tensor is None: return None + image_np = tensor[0].cpu().numpy() + image_np = (image_np * 255).astype(np.uint8) + return Image.fromarray(image_np) + + def save_pil_to_temp(self, pil_image): + output_dir = folder_paths.get_temp_directory() + (full_output_folder, filename, counter, _, _) = folder_paths.get_save_image_path("llm_temp_image", output_dir) + filepath = os.path.join(full_output_folder, f"{filename}_{counter:05}.png") + pil_image.save(filepath, 'PNG') + return filepath + + # --- API 处理函数无需改变, 它们接收文件路径 --- + def handler_google_analytics(self, prompt: str, model_id: str, media_file_path: str, base_url: str, timeout: int): + headers = {'accept': 'application/json'} + files = {'prompt': (None, prompt), 'model_id': (None, model_id)} + if media_file_path and os.path.exists(media_file_path): + files['img_file'] = (os.path.basename(media_file_path), open(media_file_path, 'rb'), mimetypes.guess_type(media_file_path)[0] or 'application/octet-stream') + try: + response = requests.post(f'{base_url}/api/llm/google/analysis', headers=headers, files=files, timeout=timeout) + response.raise_for_status() + resp_json = response.json() + result = resp_json.get('data') if resp_json else None + return result or f"API返回成功，但没有有效的 'data' 内容。 响应: {response.text}" + except requests.RequestException as e: + return f"Error calling Gemini API: {str(e)}" + + def handler_other_llm(self, model_name: str, prompt: str, media_path: str, timeout: int): + messages_content = [{"type": "text", "text": prompt}] + if media_path and os.path.exists(media_path): + try: + with open(media_path, "rb") as media_file: + base64_media = base64.b64encode(media_file.read()).decode('utf-8') + mime_type = mimetypes.guess_type(media_path)[0] or "application/octet-stream" + data_url = f"data:{mime_type};base64,{base64_media}" + messages_content.append({"type": "image_url", "image_url": {"url": data_url}}) + except Exception as e: + return f"Error encoding media file: {str(e)}" + + json_payload = {"model": model_name, "messages": [{"role": "user", "content": messages_content}], "temperature": 0.7, "max_tokens": 4096} + try: + resp = requests.post("https://gateway.bowong.cc/chat/completions", headers={"Content-Type": "application/json", "Authorization": "Bearer auth-bowong7777"}, json=json_payload, timeout=timeout) + resp.raise_for_status() + resp_json = resp.json() + if 'choices' in resp_json and resp_json['choices']: + return resp_json['choices'][0]['message']['content'] + else: + return f'Call LLM failed: {resp_json.get("error", {}).get("message", "LLM API returned no choices.")}' + except requests.RequestException as e: + return f"Error calling other LLM API: {str(e)}" + + def execute(self, model_name: str, prompt: str, environment: str = "prod", + video_input: object = None, image: torch.Tensor = None, timeout=300): + + base_url = self.ENV_URLS.get(environment, self.ENV_URLS["prod"]) + media_path = None + + # --- **最终的、唯一的、正确的修复逻辑** --- + + # 优先级 1: 处理 video_input + if video_input is not None: + unwrapped_input = video_input[0] if isinstance(video_input, (list, tuple)) and video_input else video_input + + # 检查是否是支持 save_to() 的视频对象 + if hasattr(unwrapped_input, 'save_to'): + try: + output_dir = folder_paths.get_temp_directory() + (full_output_folder, filename, counter, _, _) = folder_paths.get_save_image_path("llm_temp_video", output_dir) + temp_video_path = os.path.join(full_output_folder, f"{filename}_{counter:05}.mp4") + + print(f"检测到视频对象，使用 save_to() 保存到: {temp_video_path}") + unwrapped_input.save_to(temp_video_path) + + if os.path.exists(temp_video_path): + media_path = temp_video_path + else: + return (f"错误: 调用 save_to() 后文件未成功创建。",) + except Exception as e: + return (f"调用 save_to() 时出错: {e}",) + + # 兼容处理字符串输入的情况 + elif isinstance(unwrapped_input, str): + filename = unwrapped_input + print(f"检测到字符串输入，作为文件名处理: '{filename}'") + full_path = folder_paths.get_full_path("input", filename) + if full_path and os.path.exists(full_path): + media_path = full_path + else: + return (f"错误: 无法在 'input' 文件夹中找到文件 '{filename}'。",) + + # 优先级 2: 如果没有处理 video_input，再处理 image + elif image is not None: + print("检测到图像输入, 正在保存为临时文件...") + pil_image = self.tensor_to_pil(image) + media_path = self.save_pil_to_temp(pil_image) + + # 优先级 3: 纯文本模式 + else: + print("未提供媒体文件, 以纯文本模式运行。") + + if media_path: + print(f"成功解析媒体文件路径: {media_path}") + + # 分发到 API handlers + model_name = model_name.strip() + if model_name.startswith('gemini'): + result = self.handler_google_analytics(prompt, model_name, media_path, base_url=base_url, timeout=timeout) + else: + result = self.handler_other_llm(model_name, prompt, media_path, timeout=timeout) + + return (result,) + + +# NODE_CLASS_MAPPINGS = { "LLMUionNode": LLMUionNode } +# NODE_DISPLAY_NAME_MAPPINGS = { "LLMUionNode": "聚合LLM节点(视频/图像)" } \ No newline at end of file diff --git a/nodes/video_agent.py b/nodes/video_agent.py new file mode 100644 index 0000000..f168571 --- /dev/null +++ b/nodes/video_agent.py @@ -0,0 +1,261 @@ +# -*- coding: utf-8 -*- +""" + File video_agent.py + Author charon + Date 2025/9/4 23:01 + """ +import io +import re +import time + +import numpy as np +import requests +from PIL import Image + +try: + from loguru import logger +except ImportError: + import logging + + logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') + logger = logging.getLogger("VideoAPINode_Final") + print("提示: loguru 未安装，使用内置logging。建议安装以获得更好的日志体验: pip install loguru") + + +def fetch_and_process_models(): + video_urls = { + "prod": "https://bowongai-prod--text-video-agent-fastapi-app.modal.run/api/custom/model/list?category=video", + "dev": "https://bowongai-dev--text-video-agent-fastapi-app.modal.run/api/custom/model/list?category=video", + "test": "https://bowongai-test--text-video-agent-fastapi-app.modal.run/api/custom/model/list?category=video" + } + frame_urls = { + "prod": "https://bowongai-prod--text-video-agent-fastapi-app.modal.run/api/custom/extend/model/list", + "dev": "https://bowongai-dev--text-video-agent-fastapi-app.modal.run/api/custom/extend/model/list", + "test": "https://bowongai-test--text-video-agent-fastapi-app.modal.run/api/custom/extend/model/list" + } + + model_data = { + "configs": {}, + "full_display_list": [], + "display_to_tech_name": {}, + "temp_list_for_sorting": [] + } + + def process_response(response, is_frame_api_source=False): + data = response.json() + if not data.get("status") or "data" not in data: + raise ValueError(f"API响应格式错误: {data.get('msg', '未知错误')}") + + for model in data["data"]: + original_tech_name = model.get("model_name") + mode = model.get("mode") + if not original_tech_name: continue + + tech_name = f"frame/{original_tech_name}" if is_frame_api_source else original_tech_name + description = model.get("description", tech_name) + display_name = f"{description} ({tech_name})" + + model_data["configs"][tech_name] = model + model_data["display_to_tech_name"][display_name] = tech_name + + sort_key = 99 + if is_frame_api_source: + sort_key = 3 + elif mode == "i2v": + sort_key = 2 + elif mode == "both": + sort_key = 1 + elif mode == "t2v": + sort_key = 0 + model_data["temp_list_for_sorting"].append((sort_key, display_name)) + + try: + video_response = None + for u in video_urls.values(): + try: + video_response = requests.get(u, timeout=10, headers={ + 'accept': 'application/json'}) + video_response.raise_for_status() + break + except: + continue + if video_response: process_response(video_response, is_frame_api_source=False) + except Exception as e: + logger.error(f"常规模型加载失败: {e}") + try: + frame_response = None + for u in frame_urls.values(): + try: + frame_response = requests.get(u, timeout=10, headers={ + 'accept': 'application/json'}) + frame_response.raise_for_status() + break + except: + continue + if frame_response: process_response(frame_response, is_frame_api_source=True) + except Exception as e: + logger.error(f"首尾帧模型加载失败: {e}") + + model_data["temp_list_for_sorting"].sort(key=lambda x: x[0]) + model_data["full_display_list"] = [item[1] for item in model_data["temp_list_for_sorting"]] + + if not model_data["full_display_list"]: model_data["full_display_list"] = ["错误：无法加载模型"] + + return model_data + + +MODEL_DATA = fetch_and_process_models() + + +class VideoSubmitNode: + MODEL_DATA = MODEL_DATA + RETURN_TYPES = ("STRING",) + RETURN_NAMES = ("data",) + CATEGORY = "不忘科技-自定义节点🚩/api/视频生成" + + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + "model_name_display": (cls.MODEL_DATA["full_display_list"],), + "prompt": ("STRING", {"multiline": True, "default": ""}), + "aspect_ratio": ("STRING", {"multiline": False, "default": "9:16"}), + "duration": ("STRING", {"multiline": False, "default": "5"}), + "resolution": ("STRING", {"multiline": False, "default": "720p"}), + "environment": (["prod", "dev", "test"], {"default": "prod"}), + }, + "optional": { + "head_image": ("IMAGE", {"description": "首帧图片"}), + "tail_image": ("IMAGE", {"description": "尾帧图片"}), + } + } + + FUNCTION = "submit_task" + + def _get_base_url_and_tech_name(self, environment, model_name_display): + env_map = {"prod": "https://bowongai-prod--text-video-agent-fastapi-app.modal.run", + "dev": "https://bowongai-dev--text-video-agent-fastapi-app.modal.run", + "test": "https://bowongai-test--text-video-agent-fastapi-app.modal.run"} + base_url = env_map.get(environment, env_map["prod"]) + tech_name = self.MODEL_DATA["display_to_tech_name"].get(model_name_display) or ( + re.search(r'\((.*?)\)', model_name_display).group(1) if re.search(r'\((.*?)\)', + model_name_display) else model_name_display) + logger.info(f"模型: '{model_name_display}' -> '{tech_name}'") + return base_url, tech_name + + def _upload_file_2cdn(self, tensor_img, base_url: str): + img_tensor = tensor_img[0] + img_np = np.clip(255. * img_tensor.cpu().numpy(), 0, 255).astype(np.uint8) + pil_image = Image.fromarray(img_np) + buffer = io.BytesIO() + pil_image.save(buffer, format="PNG") + buffer.seek(0) + file_name = f'{time.time_ns()}.png' + mime_type = 'image/png' + files = {'file': (file_name, buffer, mime_type)} + response = requests.post(f'{base_url}/api/file/upload/s3', headers={'accept': 'application/json'}, files=files); + response.raise_for_status() + resp_json = response.json() + if resp_json.get('status'): + return resp_json.get('data') + else: + raise ValueError(resp_json.get('msg', '上传文件失败')) + + def _handler_base_video_task(self, prompt, model_name, aspect_ratio, duration, resolution, base_url, + head_image=None): + headers = {'accept': 'application/json'} + payload = {'prompt': (None, prompt), 'model_name': (None, model_name), 'duration': (None, duration), + 'resolution': (None, resolution), 'aspect_ratio': (None, aspect_ratio), + 'webhook_flag': (None, 'false')} + files = {} + if head_image is not None: + img_tensor = head_image[0] + img_np = np.clip(255. * img_tensor.cpu().numpy(), 0, 255).astype(np.uint8) + pil_image = Image.fromarray(img_np) + buffer = io.BytesIO() + pil_image.save(buffer, format="PNG") + buffer.seek(0) + files['img_file'] = (f'{time.time_ns()}.png', buffer, 'image/png') + files.update(payload) + api_endpoint = f'{base_url}/api/custom/video/submit/task' + response = requests.post(api_endpoint, headers=headers, files=files, timeout=90) + response.raise_for_status() + resp_json = response.json() + if resp_json.get('status'): + return resp_json.get('data') + else: + error_msg = resp_json.get('msg', '未知API错误') + raise ValueError(f"API返回失败: {error_msg}") + + def _handler_frame_video_task(self, prompt, model_name, aspect_ratio, duration, resolution, base_url, head_image, + tail_image): + model_name_for_api = model_name.replace('frame/', '') + head_img_url = self._upload_file_2cdn(head_image, base_url) + tail_img_url = self._upload_file_2cdn(tail_image, base_url) + data = {'prompt': prompt, 'head_img_url': head_img_url, 'tail_img_url': tail_img_url, + 'model_name': model_name_for_api, 'duration': duration, 'aspect_ratio': aspect_ratio, + 'resolution': resolution, 'webhook_flag': 'false'} + response = requests.post(f'{base_url}/api/custom/extend/frame/submit/task', + headers={'accept': 'application/json'}, data=data) + response.raise_for_status() + resp_json = response.json() + if resp_json.get('status'): + return resp_json.get('data') + else: + raise RuntimeError(resp_json.get('msg', '任务失败')) + + def submit_task(self, model_name_display, prompt, aspect_ratio, duration, resolution, environment, head_image=None, + tail_image=None): + try: + base_url, tech_name = self._get_base_url_and_tech_name(environment, model_name_display) + model_config = self.MODEL_DATA["configs"].get(tech_name) + if not model_config: raise ValueError(f"无法找到模型 '{tech_name}' 的配置。") + + is_frame_model = tech_name.startswith('frame/') + + def validate_and_correct_parameter(param_name, user_value, supported_values): + if not supported_values: + return user_value + if user_value in supported_values: + return user_value + default_value = supported_values[0] + logger.warning( + f"参数警告！模型 '{tech_name}' 不支持 '{param_name}': '{user_value}'。" + f"已自动替换为支持的默认值: '{default_value}'。支持的选项: {supported_values}" + ) + return default_value + + final_ar = aspect_ratio + final_res = resolution + final_dur = validate_and_correct_parameter("时长", duration, model_config.get("supported_duration", [])) + + if is_frame_model: + if head_image is None or tail_image is None: raise ValueError( + "您选择了[首尾帧]模型，必须同时提供 'head_image' 和 'tail_image' 输入。") + result = self._handler_frame_video_task(prompt, tech_name, final_ar, final_dur, final_res, base_url, + head_image, tail_image) + else: + image_to_pass = None + true_model_mode = model_config.get('mode') + if true_model_mode == 'i2v': + if head_image is None: raise ValueError("您选择了[图]模型，必须提供 'head_image' 输入。") + image_to_pass = head_image + elif true_model_mode == 't2v': + if head_image is not None: logger.warning("您选择了[文]模型，连接的'head_image'将被忽略。") + elif true_model_mode == 'both': + image_to_pass = head_image + result = self._handler_base_video_task(prompt, tech_name, final_ar, final_dur, final_res, base_url, + image_to_pass) + + return (result,) + except Exception as e: + logger.error(f"任务处理失败: {e}") + return (f"错误: {str(e)}",) + + +# NODE_CLASS_MAPPINGS = { +# "VideoSubmitNode": VideoSubmitNode, +# } +# NODE_DISPLAY_NAME_MAPPINGS = { +# "VideoSubmitNode": "统一视频生成节点", +# } diff --git a/nodes/video_preview.py b/nodes/video_preview.py new file mode 100644 index 0000000..57e68a3 --- /dev/null +++ b/nodes/video_preview.py @@ -0,0 +1,98 @@ +# -*- coding: utf-8 -*- +""" + File video_preview.py + Author charon + Date 2025/9/6 07:01 + """ +import os +import requests +import urllib.parse +from uuid import uuid4 +import folder_paths + + +class VideoDownloaderNode: + OUTPUT_DIR = folder_paths.get_input_directory() + + def __init__(self): + pass + + @classmethod + def INPUT_TYPES(s): + """ + 定义节点的输入参数。 + """ + return { + "required": { + "url": ("STRING", { + "multiline": False, + "default": "视频链接" + }), + "filename": ("STRING", { + "multiline": False, + "default": "" + }), + } + } + + RETURN_TYPES = ("STRING",) + RETURN_NAMES = ("file_path",) + FUNCTION = "download_video" + CATEGORY = "不忘科技-自定义节点🚩/utils/下载视频" + + def download_video(self, url, filename=""): + if not url or not url.strip().startswith('http'): + print("[VideoDownloader] 无效的URL，跳过下载。") + return ("",) + + try: + print(f"[VideoDownloader] 开始从 {url} 下载...") + response = requests.get(url, stream=True, timeout=10) + response.raise_for_status() + + if not filename.strip(): + try: + parsed_url = urllib.parse.urlparse(url) + filename = os.path.basename(parsed_url.path) + if not filename: raise ValueError + except (ValueError, AttributeError): + content_type = response.headers.get('content-type') + ext = '.mp4' + if content_type and '/' in content_type: + mime_type = content_type.split('/')[1] + if len(mime_type) < 5: # 简单的扩展名检查 + ext = '.' + mime_type + filename = f"downloaded_video_{uuid4().hex[:8]}{ext}" + + # 清理文件名，防止路径问题 + safe_filename = "".join(c for c in filename if c.isalnum() or c in ('.', '_', '-')).strip() + if not safe_filename: safe_filename = f"safe_video_{uuid4().hex[:8]}.mp4" + + file_path = os.path.join(self.OUTPUT_DIR, safe_filename) + + with open(file_path, 'wb') as f: + for chunk in response.iter_content(chunk_size=8192): + f.write(chunk) + + print(f"[VideoDownloader] 视频已成功下载到: {file_path}") + + ui_preview = { + "videos": [{ + "filename": safe_filename, + "subfolder": "", + "type": "input" + }] + } + return {"ui": ui_preview, "result": (file_path,)} + + except requests.exceptions.RequestException as e: + print(f"[VideoDownloader] 下载视频时出错: {e}") + return ("",) + +# NODE_CLASS_MAPPINGS = { +# "VideoDownloaderNode": VideoDownloaderNode +# } +# +# NODE_DISPLAY_NAME_MAPPINGS = { +# "VideoDownloaderNode": "视频下载器 (带预览)" +# } \ No newline at end of file