新增视频 图片生成自定义节点

__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": "获取生成结果 (图片/视频链接)"
 }

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

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": "获取生成结果 (图片/视频链接)"
+}

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": "统一生图任务节点"
+# }

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": "通用文件保存"
+# }

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节点(视频/图像)" }

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": "统一视频生成节点",
+# }

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": "视频下载器 (带预览)"
+# }