comi-logic/ComfyUI-CustomNode
7
0
Fork 0
Code Pull Requests Releases Activity

PERF 重构项目，优化命名和结构

Browse Source 
FIX 补齐requirements
FIX 剔除硬编码
This commit is contained in:
kyj@bowong.ai
2025-07-10 15:32:26 +08:00
parent fe30757baf
commit 1f3c9db743
43 changed files with 1522 additions and 3099 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
__init__.py
View File

@@ -1,35 +1,14 @@
from .nodes.llm_api import LLMChat, LLMChatMultiModalImageUpload, LLMChatMultiModalImageTensor, Jinja2RenderTemplate
from .nodes.compute_video_point import VideoStartPointDurationCompute
from .nodes.cos import COSUpload, COSDownload
from .nodes.face_detect import FaceDetect
from .nodes.face_extract import FaceExtract
from .nodes.heygem import HeyGemF2F, HeyGemF2FFromFile
from .nodes.image import SaveImagePath
from .nodes.load_image_pro import LoadNetImg
from .nodes.log2db import LogToDB
from .nodes.nodes_bfl import (
FluxProUltraImageNode,
FluxKontextProImageNode,
FluxKontextMaxImageNode,
FluxProExpandNode,
FluxProFillNode,
FluxProCannyNode,
FluxProDepthNode
from .nodes.image_face_nodes import FaceDetect, FaceExtract
from .nodes.image_gesture_nodes import JMGestureCorrect
from .nodes.image_nodes import SaveImagePath, LoadNetImg, SaveImageWithOutput
from .nodes.llm_nodes import LLMChat, LLMChatMultiModalImageUpload, LLMChatMultiModalImageTensor, Jinja2RenderTemplate
from .nodes.object_storage_nodes import COSUpload, COSDownload, S3Download, S3Upload, S3UploadURL
from .nodes.text_nodes import StringEmptyJudgement, LoadTextLocal, LoadTextOnline, RandomLineSelector
from .nodes.util_nodes import LogToDB, TaskIdGenerate, TraverseFolder, UnloadAllModels, VodToLocalNode, \
PlugAndPlayWebhook
from .nodes.video_lipsync_nodes import HeyGemF2F, HeyGemF2FFromFile
from .nodes.video_nodes import VideoCut, VideoCutByFramePoint, VideoChangeFPS, VideoStartPointDurationCompute
)
from .nodes.s3 import S3Download, S3Upload, S3UploadURL
from .nodes.string_empty_judgement import StringEmptyJudgement
from .nodes.text import *
from .nodes.traverse_folder import TraverseFolder
from .nodes.unload_all_models import UnloadAllModels
from .nodes.video import VideoCut, VideoCutByFramePoint, VideoChangeFPS
from .nodes.vod2local import VodToLocalNode
from .nodes.prompt_id_generator import TaskIdGenerate
from .nodes.random_line import RandomLineSelector
from .nodes.webhook_forward import PlugAndPlayWebhook, SaveImageWithOutput
# A dictionary that contains all nodes you want to export with their names
# NOTE: names should be globally unique
NODE_CLASS_MAPPINGS = {
"FaceOccDetect": FaceDetect,
"FaceExtract": FaceExtract,
@@ -53,14 +32,6 @@ NODE_CLASS_MAPPINGS = {
"HeyGemF2FFromFile": HeyGemF2FFromFile,
"SaveImagePath": SaveImagePath,
"LoadNetImg": LoadNetImg,
"FluxProUltraImageNode": FluxProUltraImageNode,
# "FluxProImageNode": FluxProImageNode,
"FluxKontextProImageNode": FluxKontextProImageNode,
"FluxKontextMaxImageNode": FluxKontextMaxImageNode,
"FluxProExpandNode": FluxProExpandNode,
"FluxProFillNode": FluxProFillNode,
"FluxProCannyNode": FluxProCannyNode,
"FluxProDepthNode": FluxProDepthNode,
"TaskIdGenerate": TaskIdGenerate,
"RandomLineSelector": RandomLineSelector,
"PlugAndPlayWebhook": PlugAndPlayWebhook,
@@ -68,10 +39,10 @@ NODE_CLASS_MAPPINGS = {
"LLMChat": LLMChat,
"LLMChatMultiModalImageUpload": LLMChatMultiModalImageUpload,
"LLMChatMultiModalImageTensor": LLMChatMultiModalImageTensor,
"Jinja2RenderTemplate": Jinja2RenderTemplate
"Jinja2RenderTemplate": Jinja2RenderTemplate,
"JMGestureCorrect": JMGestureCorrect
}
# A dictionary that contains the friendly/humanly readable titles for the nodes
NODE_DISPLAY_NAME_MAPPINGS = {
"FaceOccDetect": "面部遮挡检测",
"FaceExtract": "面部提取",
@@ -94,15 +65,7 @@ NODE_DISPLAY_NAME_MAPPINGS = {
"HeyGemF2F": "HeyGem口型同步(API, 传入文件Tensor)",
"HeyGemF2FFromFile": "HeyGem口型同步(API, 传入文件路径)",
"SaveImagePath": "保存图片",
"LoadNetImg": "load_net_image",
"FluxProUltraImageNode": "Flux 1.1 [pro] Ultra Image",
# "FluxProImageNode": "Flux 1.1 [pro] Image",
"FluxKontextProImageNode": "Flux.1 Kontext [pro] Image",
"FluxKontextMaxImageNode": "Flux.1 Kontext [max] Image",
"FluxProExpandNode": "Flux.1 Expand Image",
"FluxProFillNode": "Flux.1 Fill Image",
"FluxProCannyNode": "Flux.1 Canny Control Image",
"FluxProDepthNode": "Flux.1 Depth Control Image",
"LoadNetImg": "加载网络图片",
"TaskIdGenerate": "TaskID生成器",
"RandomLineSelector": "随机选择一行内容",
"PlugAndPlayWebhook": "Webhook转发器",
@@ -110,5 +73,6 @@ NODE_DISPLAY_NAME_MAPPINGS = {
"LLMChat": "LLM调用",
"LLMChatMultiModalImageUpload": "多模态LLM调用-图片Path",
"LLMChatMultiModalImageTensor": "多模态LLM调用-图片Tensor",
"Jinja2RenderTemplate": "Jinja2格式Prompt模板渲染"
"Jinja2RenderTemplate": "Jinja2格式Prompt模板渲染",
"JMGestureCorrect": "人物侧身图片转为正面图-即梦"
}

5
comfyui_v2.py → ext/comfyui_modal_deploy.py
View File

@@ -8,14 +8,13 @@ image = (
python_version="3.10"
)
.apt_install("git", "gcc", "libportaudio2", "ffmpeg")
.pip_install("fastapi[standard]==0.115.4") # install web dependencies
.pip_install("comfy_cli==0.0.0",
index_url="https://packages-1747622887395:0ee15474ccd7b27b57ca63a9306327678e6c2631@g-ldyi2063-pypi.pkg.coding.net/dev/packages/simple")
.run_commands(
"comfy --skip-prompt install --fast-deps --nvidia --version 0.3.40"
)
.pip_install_from_pyproject("./pyproject_comfyui.toml")
.run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI-CustomNode.git")
.pip_install_from_pyproject("./pyproject.toml")
.run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI-CustomNode.git", force_build=True)
.run_commands("comfy node install https://github.com/yolain/ComfyUI-Easy-Use.git")
.run_commands("cp -f /root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/ext/nodes_bfl.py /root/comfy/ComfyUI/comfy_api_nodes/nodes_bfl.py")
)

98
ext/cos_utils.py
View File

@@ -1,98 +0,0 @@
# -*- coding:utf-8 -*-
"""
File cos_utils.py
Author silence
Date 2025/6/11 20:37
"""
import mimetypes
import os
from loguru import logger
from qcloud_cos import CosConfig
from qcloud_cos import CosS3Client
# COS配置
COS_BUCKET_NAME = 'sucai-1324682537'
COS_SECRET_ID = 'AKIDsrihIyjZOBsjimt8TsN8yvv1AMh5dB44'
COS_SECRET_KEY = 'CPZcxdk6W39Jd4cGY95wvupoyMd0YFqW'
COS_REGION = 'ap-shanghai'
class CosUploadNode:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"file_path": ("STRING", {"forceInput": True}),
},
"optional": {
"remove_source": ("BOOLEAN", {"default": False}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("upload_url",)
FUNCTION = "upload_to_cos"
CATEGORY = "不忘科技-自定义节点🚩"
def upload_to_cos(self, file_path, remove_source=False):
try:
# 检查文件是否存在
if not os.path.isfile(file_path):
raise Exception(f"文件不存在: {file_path}")
# 获取文件MIME类型和分类
mime_type, _ = mimetypes.guess_type(file_path)
if mime_type:
category = mime_type.split('/')[0]
else:
category = 'unknown'
file_name = os.path.basename(file_path)
object_key = f'tk/{category}/{file_name}'
# 配置COS客户端
config = CosConfig(
Region=COS_REGION,
SecretId=COS_SECRET_ID,
SecretKey=COS_SECRET_KEY
)
client = CosS3Client(config)
# 上传文件
client.upload_file(
Bucket=COS_BUCKET_NAME,
Key=object_key,
LocalFilePath=file_path,
EnableMD5=False
)
# 生成访问URL
upload_url = f'https://{COS_BUCKET_NAME}.cos.{COS_REGION}.myqcloud.com/{object_key}'
logger.info(f'文件上传成功: {upload_url}')
# 如果需要删除源文件
if remove_source and os.path.exists(file_path):
try:
os.remove(file_path)
logger.info(f'源文件已删除: {file_path}')
except Exception as e:
logger.warning(f'删除源文件失败: {e}')
return (upload_url,)
except Exception as e:
error_msg = f'上传失败: {str(e)}'
logger.error(error_msg)
# 返回错误信息而不是抛出异常，这样ComfyUI不会中断流程
return (f"ERROR: {error_msg}",)
# 节点注册
NODE_CLASS_MAPPINGS = {
"CosUploadNode": CosUploadNode
}
NODE_DISPLAY_NAME_MAPPINGS = {
"CosUploadNode": "腾讯COS上传"
}

71
ext/image.py
View File

@@ -1,71 +0,0 @@
import os
import uuid
from PIL import Image
import numpy as np
import torch
class SaveImagePath:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"image_path": ("IMAGE", {"forceInput": True}),
}
}
RETURN_TYPES = ("STRING",)
FUNCTION = "load"
CATEGORY = "不忘科技-自定义节点🚩"
def load(self, image_path):
# 如果是torch.Tensor类型，转换为numpy数组
if isinstance(image_path, torch.Tensor):
image_path = image_path.cpu().numpy()
# 去除多余的维度，如果形状是(1, 1, height, width, channels)或(1, height, width, channels)等情况
while len(image_path.shape) > 3:
image_path = image_path.squeeze(0)
# 如果是通道优先格式 (C, H, W)，转换为通道最后格式 (H, W, C)
if len(image_path.shape) == 3 and image_path.shape[0] <= 4:
image_path = np.transpose(image_path, (1, 2, 0))
# 如果是单通道图像，转换为3通道
if len(image_path.shape) == 2:
image_path = np.stack([image_path] * 3, axis=-1)
# 数据范围和类型转换 - 这是关键修复
if image_path.dtype == np.float32 or image_path.dtype == np.float64:
# ComfyUI图像数据通常是0-1范围的浮点数
if image_path.max() <= 1.0:
# 从0-1范围转换到0-255范围
image_path = (image_path * 255.0).astype(np.uint8)
else:
# 如果已经是0-255范围，直接转换类型
image_path = np.clip(image_path, 0, 255).astype(np.uint8)
elif image_path.dtype != np.uint8:
# 其他数据类型，确保在0-255范围内
image_path = np.clip(image_path, 0, 255).astype(np.uint8)
pil_image = Image.fromarray(image_path)
output_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "output")
# base_dir = os.path.dirname(os.path.abspath(__file__))
# output_dir = '/root/comfy/ComfyUI/output'
if not os.path.exists(output_dir):
os.makedirs(output_dir)
file_name = "%s.png" % str(uuid.uuid4())
p = os.path.join(output_dir, file_name)
pil_image.save(p)
return (p,)
# 节点类定义结束，以下是用于注册节点的字典结构（通常在实际使用中由ComfyUI等框架来解析和注册）
NODE_CLASS_MAPPINGS = {
"SaveImagePath": SaveImagePath
}
NODE_DISPLAY_NAME_MAPPINGS = {
"SaveImagePath": "保存图片路径"
}

62
ext/load_image_pro.py
View File

@@ -1,62 +0,0 @@
from io import BytesIO
import numpy as np
import requests
import torch
from PIL import Image
# 定义节点类
class LoadNetImg:
# 定义节点输入类型
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"image_url": ("STRING", {
"default": "https://example.com/sample.jpg",
"multiline": False
}),
}
}
# 定义节点输出类型
RETURN_TYPES = ("IMAGE",) # 返回图像数据
RETURN_NAMES = ("image",) # 命名返回值
FUNCTION = "load_image_task" # 函数标识符，方便注册多个节点功能
OUTPUT_NODE = False # 不允许该节点直接作为最终输出节点
CATEGORY = "image" # 节点所属类别（在 ComfyUI 界面中分类）
def load_image_task(self, image_url):
try:
if not image_url or not image_url.strip():
raise ValueError("需要提供图片URL")
# 下载网络图片
response = requests.get(image_url)
response.raise_for_status() # 请求失败时抛出异常
image = Image.open(BytesIO(response.content)).convert("RGB")
# 按照官方格式转换图像数据
# Convert to numpy array and normalize to 0-1
image_array = np.array(image).astype(np.float32) / 255.0
# Convert to torch tensor and add batch dimension
image_tensor = torch.from_numpy(image_array)[None,]
return (image_tensor,) # 返回torch张量
except Exception as e:
print(f"Error loading image: {e}")
# 返回一个空的黑色图片作为错误处理
empty_image = torch.zeros((1, 512, 512, 3), dtype=torch.float32)
return (empty_image,)
# 映射节点类和名称
NODE_CLASS_MAPPINGS = {
"LoadNetImg": LoadNetImg, # 将类映射到节点名称
}
NODE_DISPLAY_NAME_MAPPINGS = {
"LoadNetImg": "load_net_image", # 节点显示名称
}

3
ext/nodes_bfl.py
View File

@@ -1,3 +1,4 @@
# 用于替换comfy内置api节点
import io
from inspect import cleandoc
from typing import Union, Optional
@@ -1071,7 +1072,6 @@ class FluxProDepthNode(ComfyNodeABC):
# NOTE: names should be globally unique
NODE_CLASS_MAPPINGS = {
"FluxProUltraImageNode": FluxProUltraImageNode,
# "FluxProImageNode": FluxProImageNode,
"FluxKontextProImageNode": FluxKontextProImageNode,
"FluxKontextMaxImageNode": FluxKontextMaxImageNode,
"FluxProExpandNode": FluxProExpandNode,
@@ -1083,7 +1083,6 @@ NODE_CLASS_MAPPINGS = {
# A dictionary that contains the friendly/humanly readable titles for the nodes
NODE_DISPLAY_NAME_MAPPINGS = {
"FluxProUltraImageNode": "Flux 1.1 [pro] Ultra Image",
# "FluxProImageNode": "Flux 1.1 [pro] Image",
"FluxKontextProImageNode": "Flux.1 Kontext [pro] Image",
"FluxKontextMaxImageNode": "Flux.1 Kontext [max] Image",
"FluxProExpandNode": "Flux.1 Expand Image",

40
ext/prompt_id_generator.py
View File

@@ -1,40 +0,0 @@
import uuid
class TaskIdGenerate:
"""TaskID生成器：用户可传入或自动生成TaskID"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {},
"optional": {
"custom_task_id": ("STRING", {"default": "", "placeholder": "留空则自动生成"}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("task_id",)
FUNCTION = "generate_task_id"
OUTPUT_NODE = False
CATEGORY = "utils"
def generate_task_id(self, custom_task_id=""):
if custom_task_id and custom_task_id.strip():
# 用户输入了自定义ID
task_id = custom_task_id.strip()
print(f"📝 使用自定义TaskID: {task_id}")
else:
# 自动生成UUID
task_id = str(uuid.uuid4())
print(f"🎲 自动生成TaskID: {task_id}")
return (task_id,)
NODE_CLASS_MAPPINGS = {
"TaskIdGenerate": TaskIdGenerate
}
NODE_DISPLAY_NAME_MAPPINGS = {
"TaskIdGenerate": "TaskID生成器"
}

8
pyproject.toml → ext/pyproject.toml
View File

@@ -19,10 +19,12 @@ dependencies = [
"av",
"imageio",
"loguru",
"openai-whisper",
"sentry-sdk",
"pydantic",
"pydantic_settings",
"conformer==0.3.2",
"einops>0.6.1"
"einops>0.6.1",
"tqdm",
"retry",
"pyYAML",
"boto3"
]

77
ext/random_line.py
View File

@@ -1,77 +0,0 @@
import random
import time
class RandomLineSelector:
"""
ComfyUI自定义节点：随机行选择器
从输入的多行文本中随机选择一行输出
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"text": ("STRING", {
"multiline": True,
"default": """
保持人物样貌和衣服不变，改变人物的光影效果。更换背景: 图像呈现出一个温馨的维多利亚复古风格房间，墙面被多样化的艺术画框与照片装饰，展现出浓郁的怀旧氛围。中心的金框全身镜是视觉焦点，上面缠绕着精致花卉装饰，增强了空间的艺术感。左侧靠墙摆放花瓶与烛台，白色地板映衬着跳动的暖黄色光线，营造出柔和而梦幻的氛围。右侧床铺整洁柔软，与一旁的小抽屉柜和台灯形成和谐的组合。整体画面色调以暖黄色为主，搭配田园风格装饰元素，构图均衡，注重细节，体现了舒适浪漫的生活场景。人物重新打光融合到场景中。
保持人物形象和衣服不变，改变人物的光影效果。更换背景: 衣帽间的开放式悬挂区搭配浅色木地板，衣杆上挂满中性色长裙，旁边搁板整齐叠放针织毛衣与围巾。
保持人物形象和衣服不变，改变人物的光影效果。更换背景: 白色衣柜门打开后，分层挂满长裙和连体裤，底部抽屉收纳着色彩鲜艳的运动服与家居裤。
保持人物形象和衣服不变，改变人物的光影效果。更换背景: 玻璃门的衣柜设计体现精致生活，内部悬挂精致衬衫和镂空裙摆，饰品抽屉随时为配件搭配提供轻便选择。
"""
}),
"seed": ("INT", {
"default": 0,
"min": -1,
"max": 0xffffffffffffffff
}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("selected_line",)
FUNCTION = "select_random_line"
CATEGORY = "text"
def select_random_line(self, text, seed):
"""
从输入文本中随机选择一行
Args:
text (str): 输入的多行文本
seed (int): 随机种子
Returns:
tuple: 包含选中行的元组
"""
# 设置随机种子（-1表示使用当前时间戳）
if seed == -1:
random.seed(int(time.time() * 1000000)) # 使用微秒级时间戳
else:
random.seed(seed)
# 过滤掉空行
non_empty_lines = [line.strip() for line in text.split('\n') if line.strip()]
# 如果没有非空行，返回空字符串
if not non_empty_lines:
return ("",)
# 随机选择一行
selected_line = random.choice(non_empty_lines)
return (selected_line,)
# ComfyUI节点映射
NODE_CLASS_MAPPINGS = {
"RandomLineSelector": RandomLineSelector
}
# 节点显示名称映射
NODE_DISPLAY_NAME_MAPPINGS = {
"RandomLineSelector": "随机选择一行内容"
}
# https://bowongai--dev-ui.modal.run/
# https://bowongai--dev-ui.modal.run

187
ext/webhook_forward.py
View File

@@ -1,187 +0,0 @@
# -*- coding:utf-8 -*-
"""
File webhook_forward.py
Author silence
Date 2025/6/13 13:42
"""
import base64
import io
import os
import random
import string
import time
import numpy as np
import requests
from PIL import Image
class PromptIDGenerator:
"""PromptID生成器：用户可传入或自动生成PromptID"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {},
"optional": {
"custom_prompt_id": ("STRING", {"default": "", "placeholder": "留空则自动生成"}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("prompt_id",)
FUNCTION = "generate_prompt_id"
OUTPUT_NODE = False
CATEGORY = "utils"
def generate_prompt_id(self, custom_prompt_id=""):
if custom_prompt_id and custom_prompt_id.strip():
# 用户输入了自定义ID
prompt_id = custom_prompt_id.strip()
print(f"📝 使用自定义PromptID: {prompt_id}")
else:
# 自动生成ID：时间戳 + 随机字符
timestamp = str(int(time.time()))
random_str = ''.join(random.choices(string.ascii_lowercase + string.digits, k=6))
prompt_id = f"prompt_{timestamp}_{random_str}"
print(f"🎲 自动生成PromptID: {prompt_id}")
return (prompt_id,)
class PlugAndPlayWebhook:
"""即插即用Webhook节点：连上线就能转发数据"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"webhook_url": ("STRING", {"default": "http://127.0.0.1:8010/handler/webhook",
"placeholder": "https://your-api.com/webhook"}),
"image_url": ("STRING", {"default": "",
"placeholder": "图片的url"}),
},
"optional": {
"prompt_id": ("STRING", {"default": ""}),
},
"hidden": {"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO", "unique_id": "UNIQUE_ID"},
}
RETURN_TYPES = ()
FUNCTION = "send"
OUTPUT_NODE = True
CATEGORY = "utils"
def send(self, webhook_url, image_url, prompt_id="", prompt=None, extra_pnginfo=None, unique_id=None):
if not webhook_url:
raise ValueError("❌ 请填写Webhook URL！")
# 使用传入的prompt_id，如果没有则用unique_id
final_prompt_id = prompt_id or unique_id or "unknown"
# 准备发送的数据
data = {
"img_base64": image_url,
"format": "png",
"image_url": image_url,
"prompt_id": final_prompt_id,
"timestamp": time.time()
}
# 发送Webhook
try:
response = requests.post(webhook_url, json=data)
response.raise_for_status()
print(f'发送的数据:{data}')
except Exception as e:
print(f"❌ 发送失败: {str(e)}")
# 终端节点，无需返回
return ()
class SaveImageWithOutput:
"""保存图片并输出的节点：既保存图片又能继续传递数据"""
def __init__(self):
self.output_dir = "output"
self.type = "output"
self.prefix_append = ""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"images": ("IMAGE",),
"filename_prefix": ("STRING", {"default": "ComfyUI"}),
},
}
RETURN_TYPES = ("IMAGE", "STRING")
RETURN_NAMES = ("images", "file_path")
FUNCTION = "save_image"
OUTPUT_NODE = False
CATEGORY = "image"
def save_image(self, images, filename_prefix="ComfyUI"):
filename_prefix += self.prefix_append
full_output_folder, filename, counter, subfolder, filename_prefix = self.get_save_image_path(filename_prefix,
self.output_dir)
file_paths = []
for i, image in enumerate(images):
# 转换图片数据
img_array = 255. * image.cpu().numpy()
img = Image.fromarray(np.clip(img_array, 0, 255).astype(np.uint8))
# 生成文件名
file = f"{filename_prefix}_{counter + i:05}_.png"
file_path = os.path.join(full_output_folder, file)
# 保存图片
img.save(file_path, compress_level=4)
file_paths.append(file_path)
print(f"✅ 图片已保存到: {file_paths[0] if file_paths else '未知路径'}")
# 直接返回元组：(原始图片数据, 第一个文件路径)
return (images, file_paths[0] if file_paths else "")
def get_save_image_path(self, filename_prefix, output_dir):
def map_filename(filename):
prefix_len = len(os.path.basename(filename_prefix))
prefix = filename[:prefix_len + 1]
try:
digits = int(filename[prefix_len + 1:].split('_')[0])
except:
digits = 0
return (digits, prefix)
subfolder = ""
full_output_folder = os.path.join(output_dir, subfolder)
if os.path.commonpath((output_dir, os.path.abspath(full_output_folder))) != output_dir:
print("Saving image outside the output folder is not allowed.")
return {}
try:
counter = max(filter(lambda a: a[1][:-1] == filename_prefix and a[1][-1] == "_",
map(map_filename, os.listdir(full_output_folder))))[0] + 1
except ValueError:
counter = 1
except FileNotFoundError:
os.makedirs(full_output_folder, exist_ok=True)
counter = 1
return full_output_folder, filename_prefix, counter, subfolder, filename_prefix
NODE_CLASS_MAPPINGS = {
"PlugAndPlayWebhook": PlugAndPlayWebhook,
"SaveImageWithOutput": SaveImageWithOutput
}
NODE_DISPLAY_NAME_MAPPINGS = {
"PlugAndPlayWebhook": "Webhook转发器",
"SaveImageWithOutput": "保存图片(带输出)"
}

50
nodes/compute_video_point.py
View File

@@ -1,50 +0,0 @@
import re
from datetime import datetime
import loguru
def validate_time_format(time_str):
pattern = r'^([0-1][0-9]|2[0-3]):([0-5][0-9]):([0-5][0-9]|\d{1,2}).(\d{3})$'
return bool(re.match(pattern, time_str))
def get_duration_wave(audio):
waveform, sample_rate = audio["waveform"], audio["sample_rate"]
# 防止话说不完
return waveform.shape[2] / sample_rate
class VideoStartPointDurationCompute:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"start_time": ("STRING", {"forceInput": True}),
"audio": ("AUDIO", {"forceInput": True}),
"end_padding": ("FLOAT", {"forceInput": True, "default": 0.4}),
"fps": ("INT", {"default": 25, "step": 1}),
},
}
RETURN_TYPES = ("FLOAT", "FLOAT",)
RETURN_NAMES = ("起始帧位", "帧数")
FUNCTION = "compute"
CATEGORY = "不忘科技-自定义节点🚩"
def compute(self, start_time, audio, fps, end_padding):
if not validate_time_format(start_time):
raise ValueError("start_time或者end_time时间格式不对（start_time or end_time is not in time format）")
time_format = "%H:%M:%S.%f"
start_dt = datetime.strptime(start_time, time_format)
start_sec = (start_dt - datetime(1900, 1, 1)).total_seconds()
start_point = start_sec * fps
duration = get_duration_wave(audio)
loguru.logger.info("audio duration %.3f s" % duration)
duration = duration + end_padding
loguru.logger.info("audio duration with padding %.3f s" % duration)
return (start_point, duration*fps,)

132
nodes/cos.py
View File

@@ -1,132 +0,0 @@
import json
import os
import urllib
import loguru
import yaml
from qcloud_cos import CosConfig, CosS3Client, CosClientError, CosServiceError
class COSDownload:
"""腾讯云COS下载"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"cos_bucket": ("STRING", {"default": "bwkj-cos-1324682537"}),
"cos_key": ("STRING", {"multiline": True}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("视频存储路径",)
FUNCTION = "download"
CATEGORY = "不忘科技-自定义节点🚩"
def download(self, cos_bucket, cos_key):
cos_key_in = cos_key.replace("/",os.sep)
destination = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"download",
os.path.dirname(cos_key_in),
os.path.basename(cos_key_in)
)
loguru.logger.info(f"COS DOWNLOAD to {destination}")
os.makedirs(
os.path.dirname(destination),
exist_ok=True,
)
for i in range(0, 10):
try:
if "aws_key_id" in list(os.environ.keys()):
yaml_config = os.environ
else:
with open(
os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config.yaml"
),
encoding="utf-8",
mode="r+",
) as f:
yaml_config = yaml.load(f, Loader=yaml.FullLoader)
config = CosConfig(
Region=yaml_config["cos_region"],
SecretId=yaml_config["cos_secret_id"],
SecretKey=yaml_config["cos_secret_key"],
)
client = CosS3Client(config)
response = client.download_file(
Bucket=cos_bucket,
Key=cos_key,
DestFilePath=destination
)
break
except CosClientError or CosServiceError as e:
raise Exception(f"COS下载失败！ bucket {cos_bucket}; key {cos_key}")
return (
destination,
)
class COSUpload:
"""腾讯云COS上传"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"cos_bucket": ("STRING", {"default": "bwkj-cos-1324682537"}),
"path": ("STRING", {"multiline": True}),
"subfolder": ("STRING", {"default": "test"}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("COS文件Key",)
FUNCTION = "upload"
CATEGORY = "不忘科技-自定义节点🚩"
def upload(self, cos_bucket, path, subfolder):
dest_key = "/".join(
[
subfolder,
(
path.split("/")[-1]
if "/" in path
else path.split("\\")[-1]
),
]
)
loguru.logger.info(f"COS UPLOAD {path} to {cos_bucket}/{subfolder}")
for i in range(0, 10):
try:
if "aws_key_id" in list(os.environ.keys()):
yaml_config = os.environ
else:
with open(
os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config.yaml"
),
encoding="utf-8",
mode="r+",
) as f:
yaml_config = yaml.load(f, Loader=yaml.FullLoader)
config = CosConfig(
Region=yaml_config["cos_region"],
SecretId=yaml_config["cos_secret_id"],
SecretKey=yaml_config["cos_secret_key"],
)
client = CosS3Client(config)
response = client.upload_file(
Bucket=cos_bucket,
Key=dest_key,
LocalFilePath=path,
)
break
except CosClientError or CosServiceError as e:
raise Exception(f"COS上传失败 bucket {cos_bucket}; local_path {path}; subfolder {subfolder}")
return (
dest_key,
)

98
nodes/cos_utils.py
View File

@@ -1,98 +0,0 @@
# -*- coding:utf-8 -*-
"""
File cos_utils.py
Author silence
Date 2025/6/11 20:37
"""
import mimetypes
import os
from loguru import logger
from qcloud_cos import CosConfig
from qcloud_cos import CosS3Client
# COS配置
COS_BUCKET_NAME = 'sucai-1324682537'
COS_SECRET_ID = 'AKIDsrihIyjZOBsjimt8TsN8yvv1AMh5dB44'
COS_SECRET_KEY = 'CPZcxdk6W39Jd4cGY95wvupoyMd0YFqW'
COS_REGION = 'ap-shanghai'
class CosUploadNode:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"file_path": ("STRING", {"forceInput": True}),
},
"optional": {
"remove_source": ("BOOLEAN", {"default": False}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("upload_url",)
FUNCTION = "upload_to_cos"
CATEGORY = "不忘科技-自定义节点🚩"
def upload_to_cos(self, file_path, remove_source=False):
try:
# 检查文件是否存在
if not os.path.isfile(file_path):
raise Exception(f"文件不存在: {file_path}")
# 获取文件MIME类型和分类
mime_type, _ = mimetypes.guess_type(file_path)
if mime_type:
category = mime_type.split('/')[0]
else:
category = 'unknown'
file_name = os.path.basename(file_path)
object_key = f'tk/{category}/{file_name}'
# 配置COS客户端
config = CosConfig(
Region=COS_REGION,
SecretId=COS_SECRET_ID,
SecretKey=COS_SECRET_KEY
)
client = CosS3Client(config)
# 上传文件
client.upload_file(
Bucket=COS_BUCKET_NAME,
Key=object_key,
LocalFilePath=file_path,
EnableMD5=False
)
# 生成访问URL
upload_url = f'https://{COS_BUCKET_NAME}.cos.{COS_REGION}.myqcloud.com/{object_key}'
logger.info(f'文件上传成功: {upload_url}')
# 如果需要删除源文件
if remove_source and os.path.exists(file_path):
try:
os.remove(file_path)
logger.info(f'源文件已删除: {file_path}')
except Exception as e:
logger.warning(f'删除源文件失败: {e}')
return (upload_url,)
except Exception as e:
error_msg = f'上传失败: {str(e)}'
logger.error(error_msg)
# 返回错误信息而不是抛出异常，这样ComfyUI不会中断流程
return (f"ERROR: {error_msg}",)
# 节点注册
NODE_CLASS_MAPPINGS = {
"CosUploadNode": CosUploadNode
}
NODE_DISPLAY_NAME_MAPPINGS = {
"CosUploadNode": "腾讯COS上传"
}

76
nodes/face_detect.py
View File

@@ -1,76 +0,0 @@
import json
from .test_single_image import test_node
class FaceDetect:
"""
人脸遮挡检测
"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"image": ("IMAGE",),
"main_seed": (
"INT:seed",
{"default": 0, "min": 0, "max": 0xFFFFFFFFFFFFFFFF},
),
"model": (["convnext_tiny", "convnext_base"],),
"length": ("INT", {"default": 10, "min": 3, "max": 60, "step": 1}),
"threshold": (
"FLOAT",
{"default": 94, "min": 55, "max": 99, "step": 0.1},
),
},
}
RETURN_TYPES = (
"IMAGE",
"IMAGE",
"STRING",
"STRING",
"STRING",
"STRING",
"STRING",
"INT",
"INT",
)
RETURN_NAMES = (
"图像",
"选中人脸",
"分类",
"概率",
"采用帧序号",
"全部帧序列",
"剪辑配置",
"起始帧序号",
"帧数量",
)
FUNCTION = "predict"
CATEGORY = "不忘科技-自定义节点🚩/面部"
def predict(self, image, main_seed, model, length, threshold):
image, image_selected, cls, prob, nums, period = test_node(
image, length=length, thres=threshold, model_name=model
)
print("全部帧序列", period)
if len(period) > 0:
start, end = period[main_seed % len(period)]
config = {"start": start, "end": end}
else:
raise RuntimeError("未找到符合要求的视频片段")
return (
image,
image_selected,
cls,
prob,
nums,
str(period),
json.dumps(config),
start,
end - start + 1,
)

71
nodes/image.py
View File

@@ -1,71 +0,0 @@
import os
import uuid
from PIL import Image
import numpy as np
import torch
class SaveImagePath:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"image_path": ("IMAGE", {"forceInput": True}),
}
}
RETURN_TYPES = ("STRING",)
FUNCTION = "load"
CATEGORY = "不忘科技-自定义节点🚩"
def load(self, image_path):
# 如果是torch.Tensor类型，转换为numpy数组
if isinstance(image_path, torch.Tensor):
image_path = image_path.cpu().numpy()
# 去除多余的维度，如果形状是(1, 1, height, width, channels)或(1, height, width, channels)等情况
while len(image_path.shape) > 3:
image_path = image_path.squeeze(0)
# 如果是通道优先格式 (C, H, W)，转换为通道最后格式 (H, W, C)
if len(image_path.shape) == 3 and image_path.shape[0] <= 4:
image_path = np.transpose(image_path, (1, 2, 0))
# 如果是单通道图像，转换为3通道
if len(image_path.shape) == 2:
image_path = np.stack([image_path] * 3, axis=-1)
# 数据范围和类型转换 - 这是关键修复
if image_path.dtype == np.float32 or image_path.dtype == np.float64:
# ComfyUI图像数据通常是0-1范围的浮点数
if image_path.max() <= 1.0:
# 从0-1范围转换到0-255范围
image_path = (image_path * 255.0).astype(np.uint8)
else:
# 如果已经是0-255范围，直接转换类型
image_path = np.clip(image_path, 0, 255).astype(np.uint8)
elif image_path.dtype != np.uint8:
# 其他数据类型，确保在0-255范围内
image_path = np.clip(image_path, 0, 255).astype(np.uint8)
pil_image = Image.fromarray(image_path)
output_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "output")
# base_dir = os.path.dirname(os.path.abspath(__file__))
# output_dir = '/root/comfy/ComfyUI/output'
if not os.path.exists(output_dir):
os.makedirs(output_dir)
file_name = "%s.png" % str(uuid.uuid4())
p = os.path.join(output_dir, file_name)
pil_image.save(p)
return (p,)
# 节点类定义结束，以下是用于注册节点的字典结构（通常在实际使用中由ComfyUI等框架来解析和注册）
NODE_CLASS_MAPPINGS = {
"SaveImagePath": SaveImagePath
}
NODE_DISPLAY_NAME_MAPPINGS = {
"SaveImagePath": "保存图片路径"
}

79
nodes/face_extract.py → nodes/image_face_nodes.py
View File

@@ -1,3 +1,4 @@
import json
import os
import cv2
@@ -6,6 +7,81 @@ import torch
from comfy import model_management
from ultralytics import YOLO
from ..utils.face_occu_detect import face_occu_detect
class FaceDetect:
"""
人脸遮挡检测
"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"image": ("IMAGE",),
"main_seed": (
"INT:seed",
{"default": 0, "min": 0, "max": 0xFFFFFFFFFFFFFFFF},
),
"model": (["convnext_tiny", "convnext_base"],),
"length": ("INT", {"default": 10, "min": 3, "max": 60, "step": 1}),
"threshold": (
"FLOAT",
{"default": 94, "min": 55, "max": 99, "step": 0.1},
),
},
}
RETURN_TYPES = (
"IMAGE",
"IMAGE",
"STRING",
"STRING",
"STRING",
"STRING",
"STRING",
"INT",
"INT",
)
RETURN_NAMES = (
"图像",
"选中人脸",
"分类",
"概率",
"采用帧序号",
"全部帧序列",
"剪辑配置",
"起始帧序号",
"帧数量",
)
FUNCTION = "predict"
CATEGORY = "不忘科技-自定义节点🚩/图片/人脸"
def predict(self, image, main_seed, model, length, threshold):
image, image_selected, cls, prob, nums, period = face_occu_detect(
image, length=length, thres=threshold, model_name=model
)
print("全部帧序列", period)
if len(period) > 0:
start, end = period[main_seed % len(period)]
config = {"start": start, "end": end}
else:
raise RuntimeError("未找到符合要求的视频片段")
return (
image,
image_selected,
cls,
prob,
nums,
str(period),
json.dumps(config),
start,
end - start + 1,
)
class FaceExtract:
"""人脸提取 By YOLO"""
@@ -23,7 +99,7 @@ class FaceExtract:
FUNCTION = "crop"
CATEGORY = "不忘科技-自定义节点🚩/面部"
CATEGORY = "不忘科技-自定义节点🚩/图片/人脸"
def crop(self, image):
device = model_management.get_torch_device()
@@ -38,7 +114,6 @@ class FaceExtract:
total_images = image_np.shape[0]
out_images = np.ndarray(shape=(total_images, 512, 512, 3))
print("shape", image_np.shape)
print("aaaaa")
idx = 0
for image_item in image_np:
results = model.predict(

317
nodes/image_gesture_nodes.py Normal file
View File

@@ -0,0 +1,317 @@
import io
import json
import os
import subprocess
import tempfile
import time
import uuid
from time import sleep
import numpy as np
import requests
import torch
import yaml
from PIL import Image
from loguru import logger
from qcloud_cos import CosConfig, CosS3Client
from tqdm import tqdm
class JMUtils:
def __init__(self):
if "aws_key_id" in list(os.environ.keys()):
yaml_config = os.environ
else:
with open(
os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config.yaml"
),
encoding="utf-8",
mode="r+",
) as f:
yaml_config = yaml.load(f, Loader=yaml.FullLoader)
self.api_key = yaml_config["jm_api_key"]
self.cos_region = yaml_config["cos_region"]
self.cos_secret_id = yaml_config["cos_secret_id"]
self.cos_secret_key = yaml_config["cos_secret_key"]
self.cos_bucket_name = yaml_config["cos_sucai_bucket_name"]
def submit_task(self, prompt: str, img_url: str, duration: str = "5"):
try:
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}",
}
json_data = {
"model": "doubao-seedance-1-0-lite-i2v-250428",
"content": [
{
"type": "text",
"text": f"{prompt} --resolution 720p --dur {duration} --camerafixed false",
},
{
"type": "image_url",
"image_url": {
"url": img_url,
},
},
],
}
response = requests.post("https://ark.cn-beijing.volces.com/api/v3/contents/generations/tasks", headers=headers, json=json_data)
logger.info(f"submit task: {json.dumps(response.json())}")
resp_json = response.json()
if "id" not in resp_json:
return {"status": False, "data": img_url, "msg": resp_json["error"]["message"]}
else:
job_id = resp_json["id"]
return {"data": job_id, "status": True, "msg": "任务提交成功"}
except Exception as e:
logger.error(e)
return {"data": None, "status": False, "msg": str(e)}
def query_status(self, job_id: str):
resp_dict = {"status": False, "data": None, "msg": ""}
try:
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}",
}
response = requests.get(f"https://ark.cn-beijing.volces.com/api/v3/contents/generations/tasks/{job_id}", headers=headers)
resp_json = response.json()
resp_dict["status"] = resp_json["status"] == "succeeded"
resp_dict["msg"] = resp_json["status"]
resp_dict["data"] = resp_json["content"]["video_url"] if "content" in resp_json else None
except Exception as e:
logger.error(f"error:{str(e)}")
resp_dict["msg"] = str(e)
finally:
return resp_dict
def upload_io_to_cos(self, file: io.IOBase, mime_type: str = "image/png"):
resp_data = {'status': True, 'data': '', 'msg': ''}
category = mime_type.split('/')[0]
suffix = mime_type.split('/')[1]
try:
object_key = f'tk/{category}/{uuid.uuid4()}.{suffix}'
config = CosConfig(Region=self.cos_region, SecretId=self.cos_secret_id, SecretKey=self.cos_secret_key)
client = CosS3Client(config)
_ = client.upload_file_from_buffer(
Bucket=self.cos_bucket_name,
Key=object_key,
Body=file
)
url = f'https://{self.cos_bucket_name}.cos.{self.cos_region}.myqcloud.com/{object_key}'
resp_data['data'] = url
resp_data['msg'] = '上传成功'
except Exception as e:
logger.error(e)
resp_data['status'] = False
resp_data['msg'] = str(e)
return resp_data
def tensor_to_io(srlf, tensor: torch.Tensor):
# 转换为PIL图像
img = Image.fromarray(np.clip(255. * tensor.cpu().numpy().squeeze(), 0, 255).astype(np.uint8))
image_data = io.BytesIO()
img.save(image_data, format='PNG')
image_data.seek(0)
return image_data
def download_video(self, url, timeout=30, retries=3):
"""下载视频到临时文件并返回文件路径"""
for attempt in range(retries):
try:
# 创建临时文件
temp_file = tempfile.NamedTemporaryFile(suffix='.mp4', delete=False)
temp_path = temp_file.name
temp_file.close()
# 下载视频
print(f"开始下载视频 (尝试 {attempt + 1}/{retries})...")
response = requests.get(url, stream=True, timeout=timeout)
response.raise_for_status()
# 获取文件大小
total_size = int(response.headers.get('content-length', 0))
block_size = 1024 # 1 KB
# 使用tqdm显示下载进度
with open(temp_path, 'wb') as f, tqdm(
desc=url.split('/')[-1],
total=total_size,
unit='B',
unit_scale=True,
unit_divisor=1024
) as bar:
for data in response.iter_content(block_size):
size = f.write(data)
bar.update(size)
print(f"视频下载完成: {temp_path}")
return temp_path
except Exception as e:
print(f"下载错误 (尝试 {attempt + 1}/{retries}): {str(e)}")
if attempt < retries - 1:
time.sleep((attempt + 1) * 2)
else:
raise
def jpg_to_tensor(self, image_path, channel_first=False):
"""
将JPG图像转换为PyTorch张量
参数:
- image_path: JPG图像文件路径
- normalize: 是否将像素值归一化到[0.0, 1.0]
- channel_first: 是否将通道维度放在前面 (C, H, W)
返回:
- tensor: PyTorch张量
"""
try:
# 打开图像文件
image = Image.open(image_path).convert('RGB')
# 转换为张量
tensor = torch.from_numpy(np.array(image).astype(np.float32)/255.0)[None,]
return tensor
except Exception as e:
print(f"转换失败: {str(e)}")
raise
def get_last_15th_frame_tensor(self, video_url, cleanup=True):
"""
从视频URL截取倒数第15帧并转换为Tensor
先下载视频到本地临时文件再处理
"""
try:
# 下载视频
video_path = self.download_video(video_url)
# 获取视频总帧数
cmd_frames = [
'ffprobe', '-v', 'error',
'-select_streams', 'v:0',
'-show_entries', 'stream=nb_frames',
'-of', 'default=nokey=1:noprint_wrappers=1',
video_path
]
result = subprocess.run(
cmd_frames,
capture_output=True,
text=True,
check=True
)
# 处理可能的非数字输出
frame_count = result.stdout.strip()
if not frame_count.isdigit():
# 备选方案：通过解码获取帧数
print("无法获取准确帧数，尝试直接解码...")
cmd_decode = [
'ffmpeg', '-i', video_path,
'-f', 'null', '-'
]
decode_result = subprocess.run(
cmd_decode,
capture_output=True,
text=True
)
for line in decode_result.stderr.split('\n'):
if 'frame=' in line:
parts = line.split('frame=')[-1].split()[0]
if parts.isdigit():
frame_count = int(parts)
break
else:
raise ValueError("无法确定视频帧数")
else:
frame_count = int(frame_count)
# 计算目标帧
target_frame = max(0, frame_count - 15)
print(f"视频总帧数: {frame_count}, 目标帧: {target_frame}")
# 截取指定帧
with tempfile.NamedTemporaryFile(suffix='%03d.jpg', delete=True) as frame_file:
frame_path = frame_file.name
cmd_extract = [
'ffmpeg',
'-ss', f'00:00:00',
'-i', video_path,
'-vframes', '1',
'-vf', f'select=eq(n\,{target_frame})',
'-vsync', '0',
'-an', '-y',
frame_path
]
subprocess.run(
cmd_extract,
capture_output=True,
check=True
)
# 转换为Tensor
tensor = self.jpg_to_tensor(frame_path.replace("%03d","001"))
except Exception as e:
raise e
return tensor
class JMGestureCorrect:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"image": ("IMAGE",)
}
}
RETURN_TYPES = ("IMAGE",)
RETURN_NAMES = ("正面图",)
FUNCTION = "gen"
CATEGORY = "不忘科技-自定义节点🚩/图片/姿态"
def gen(self, image:torch.Tensor):
wait_time = 120
interval = 2
client = JMUtils()
image_io = client.tensor_to_io(image)
upload_data = client.upload_io_to_cos(image_io)
if upload_data["status"]:
image_url = upload_data["data"]
else:
raise Exception("上传失败")
prompt = "Stand straight ahead, facing the camera, showing your full body, maintaining a proper posture, keeping the camera still, and ensuring that your head and feet are all within the frame"
submit_data = client.submit_task(prompt, image_url)
if submit_data["status"]:
job_id = submit_data["data"]
else:
raise Exception("即梦任务提交失败")
job_data = None
for idx, _ in enumerate(range(0, wait_time, interval)):
logger.info(f"查询即梦结果 {idx+1}")
query = client.query_status(job_id)
if query["status"]:
job_data = query["data"]
break
else:
if "error" in query["msg"] or "失败" in query["msg"] or "fail" in query["msg"]:
raise Exception("即梦任务失败 {}".format(query["msg"]))
sleep(interval)
if not job_data:
raise Exception("即梦任务等待超时")
return (client.get_last_15th_frame_tensor(job_data),)

188
nodes/image_nodes.py Normal file
View File

@@ -0,0 +1,188 @@
import os
import uuid
from io import BytesIO
import numpy as np
import requests
import torch
from PIL import Image
# 定义节点类
class LoadNetImg:
# 定义节点输入类型
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"image_url": ("STRING", {
"default": "https://example.com/sample.jpg",
"multiline": False
}),
}
}
# 定义节点输出类型
RETURN_TYPES = ("IMAGE",) # 返回图像数据
RETURN_NAMES = ("image",) # 命名返回值
FUNCTION = "load_image_task" # 函数标识符，方便注册多个节点功能
OUTPUT_NODE = False # 不允许该节点直接作为最终输出节点
CATEGORY = "不忘科技-自定义节点🚩/图片" # 节点所属类别（在 ComfyUI 界面中分类）
def load_image_task(self, image_url):
try:
if not image_url or not image_url.strip():
raise ValueError("需要提供图片URL")
# 下载网络图片
response = requests.get(image_url)
response.raise_for_status() # 请求失败时抛出异常
image = Image.open(BytesIO(response.content)).convert("RGB")
# 按照官方格式转换图像数据
# Convert to numpy array and normalize to 0-1
image_array = np.array(image).astype(np.float32) / 255.0
# Convert to torch tensor and add batch dimension
image_tensor = torch.from_numpy(image_array)[None,]
return (image_tensor,) # 返回torch张量
except Exception as e:
print(f"Error loading image: {e}")
# 返回一个空的黑色图片作为错误处理
empty_image = torch.zeros((1, 512, 512, 3), dtype=torch.float32)
return (empty_image,)
class SaveImagePath:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"image_path": ("IMAGE", {"forceInput": True}),
}
}
RETURN_TYPES = ("STRING",)
FUNCTION = "load"
CATEGORY = "不忘科技-自定义节点🚩/图片"
def load(self, image_path):
# 如果是torch.Tensor类型，转换为numpy数组
if isinstance(image_path, torch.Tensor):
image_path = image_path.cpu().numpy()
# 去除多余的维度，如果形状是(1, 1, height, width, channels)或(1, height, width, channels)等情况
while len(image_path.shape) > 3:
image_path = image_path.squeeze(0)
# 如果是通道优先格式 (C, H, W)，转换为通道最后格式 (H, W, C)
if len(image_path.shape) == 3 and image_path.shape[0] <= 4:
image_path = np.transpose(image_path, (1, 2, 0))
# 如果是单通道图像，转换为3通道
if len(image_path.shape) == 2:
image_path = np.stack([image_path] * 3, axis=-1)
# 数据范围和类型转换 - 这是关键修复
if image_path.dtype == np.float32 or image_path.dtype == np.float64:
# ComfyUI图像数据通常是0-1范围的浮点数
if image_path.max() <= 1.0:
# 从0-1范围转换到0-255范围
image_path = (image_path * 255.0).astype(np.uint8)
else:
# 如果已经是0-255范围，直接转换类型
image_path = np.clip(image_path, 0, 255).astype(np.uint8)
elif image_path.dtype != np.uint8:
# 其他数据类型，确保在0-255范围内
image_path = np.clip(image_path, 0, 255).astype(np.uint8)
pil_image = Image.fromarray(image_path)
output_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "output")
# base_dir = os.path.dirname(os.path.abspath(__file__))
# output_dir = '/root/comfy/ComfyUI/output'
if not os.path.exists(output_dir):
os.makedirs(output_dir)
file_name = "%s.png" % str(uuid.uuid4())
p = os.path.join(output_dir, file_name)
pil_image.save(p)
return (p,)
class SaveImageWithOutput:
"""保存图片并输出的节点：既保存图片又能继续传递数据"""
def __init__(self):
self.output_dir = "output"
self.type = "output"
self.prefix_append = ""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"images": ("IMAGE",),
"filename_prefix": ("STRING", {"default": "ComfyUI"}),
},
}
RETURN_TYPES = ("IMAGE", "STRING")
RETURN_NAMES = ("images", "file_path")
FUNCTION = "save_image"
OUTPUT_NODE = False
CATEGORY = "不忘科技-自定义节点🚩/图片"
def save_image(self, images, filename_prefix="ComfyUI"):
filename_prefix += self.prefix_append
full_output_folder, filename, counter, subfolder, filename_prefix = self.get_save_image_path(filename_prefix,
self.output_dir)
file_paths = []
for i, image in enumerate(images):
# 转换图片数据
img_array = 255. * image.cpu().numpy()
img = Image.fromarray(np.clip(img_array, 0, 255).astype(np.uint8))
# 生成文件名
file = f"{filename_prefix}_{counter + i:05}_.png"
file_path = os.path.join(full_output_folder, file)
# 保存图片
img.save(file_path, compress_level=4)
file_paths.append(file_path)
print(f"✅ 图片已保存到: {file_paths[0] if file_paths else '未知路径'}")
# 直接返回元组：(原始图片数据, 第一个文件路径)
return (images, file_paths[0] if file_paths else "")
def get_save_image_path(self, filename_prefix, output_dir):
def map_filename(filename):
prefix_len = len(os.path.basename(filename_prefix))
prefix = filename[:prefix_len + 1]
try:
digits = int(filename[prefix_len + 1:].split('_')[0])
except:
digits = 0
return (digits, prefix)
subfolder = ""
full_output_folder = os.path.join(output_dir, subfolder)
if os.path.commonpath((output_dir, os.path.abspath(full_output_folder))) != output_dir:
print("Saving image outside the output folder is not allowed.")
return {}
try:
counter = max(filter(lambda a: a[1][:-1] == filename_prefix and a[1][-1] == "_",
map(map_filename, os.listdir(full_output_folder))))[0] + 1
except ValueError:
counter = 1
except FileNotFoundError:
os.makedirs(full_output_folder, exist_ok=True)
counter = 1
return full_output_folder, filename_prefix, counter, subfolder, filename_prefix

11
nodes/llm_api.py → nodes/llm_nodes.py
View File

@@ -7,6 +7,7 @@ import re
from mimetypes import guess_type
from typing import Any, Union
import folder_paths
import httpx
import numpy as np
import torch
@@ -14,8 +15,6 @@ from PIL import Image
from jinja2 import Template, StrictUndefined
from retry import retry
import folder_paths
def find_value_recursive(key:str, data:Union[dict, list]) -> str | None | Any:
if isinstance(data, dict):
@@ -73,7 +72,7 @@ class LLMChat:
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("llm输出",)
FUNCTION = "chat"
CATEGORY = "不忘科技-自定义节点🚩/llm"
CATEGORY = "不忘科技-自定义节点🚩/LLM"
def chat(self, llm_provider:str, prompt:str, temperature:float, max_tokens:int, timeout:int):
@retry(Exception, tries=3, delay=1)
@@ -130,7 +129,7 @@ class LLMChatMultiModalImageUpload:
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("llm输出",)
FUNCTION = "chat"
CATEGORY = "不忘科技-自定义节点🚩/llm"
CATEGORY = "不忘科技-自定义节点🚩/LLM"
def chat(self, llm_provider:str, prompt:str, image, temperature:float, max_tokens:int, timeout:int):
@retry(Exception, tries=3, delay=1)
@@ -194,7 +193,7 @@ class LLMChatMultiModalImageTensor:
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("llm输出",)
FUNCTION = "chat"
CATEGORY = "不忘科技-自定义节点🚩/llm"
CATEGORY = "不忘科技-自定义节点🚩/LLM"
def chat(self, llm_provider:str, prompt:str, image:torch.Tensor, temperature:float, max_tokens:int, timeout:int):
@retry(Exception, tries=3, delay=1)
@@ -247,7 +246,7 @@ class Jinja2RenderTemplate:
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("prompt",)
FUNCTION = "render_prompt"
CATEGORY = "不忘科技-自定义节点🚩/llm"
CATEGORY = "不忘科技-自定义节点🚩/LLM"
def render_prompt(self, template: str, kv_map: str) -> tuple:
"""

62
nodes/load_image_pro.py
View File

@@ -1,62 +0,0 @@
from io import BytesIO
import numpy as np
import requests
import torch
from PIL import Image
# 定义节点类
class LoadNetImg:
# 定义节点输入类型
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"image_url": ("STRING", {
"default": "https://example.com/sample.jpg",
"multiline": False
}),
}
}
# 定义节点输出类型
RETURN_TYPES = ("IMAGE",) # 返回图像数据
RETURN_NAMES = ("image",) # 命名返回值
FUNCTION = "load_image_task" # 函数标识符，方便注册多个节点功能
OUTPUT_NODE = False # 不允许该节点直接作为最终输出节点
CATEGORY = "image" # 节点所属类别（在 ComfyUI 界面中分类）
def load_image_task(self, image_url):
try:
if not image_url or not image_url.strip():
raise ValueError("需要提供图片URL")
# 下载网络图片
response = requests.get(image_url)
response.raise_for_status() # 请求失败时抛出异常
image = Image.open(BytesIO(response.content)).convert("RGB")
# 按照官方格式转换图像数据
# Convert to numpy array and normalize to 0-1
image_array = np.array(image).astype(np.float32) / 255.0
# Convert to torch tensor and add batch dimension
image_tensor = torch.from_numpy(image_array)[None,]
return (image_tensor,) # 返回torch张量
except Exception as e:
print(f"Error loading image: {e}")
# 返回一个空的黑色图片作为错误处理
empty_image = torch.zeros((1, 512, 512, 3), dtype=torch.float32)
return (empty_image,)
# 映射节点类和名称
NODE_CLASS_MAPPINGS = {
"LoadNetImg": LoadNetImg, # 将类映射到节点名称
}
NODE_DISPLAY_NAME_MAPPINGS = {
"LoadNetImg": "load_net_image", # 节点显示名称
}

66
nodes/log2db.py
View File

@@ -1,66 +0,0 @@
import json
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import server
from .table import Task
class LogToDB:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"job_id": ("STRING", {"forceInput": True}),
"log": ("STRING", {"forceInput": True}),
"status": ("INT", {"default": 1, "max": 1}),
},
"hidden": {
"unique_id": "UNIQUE_ID",
}
}
RETURN_TYPES = ("STRING",)
FUNCTION = "log2db"
OUTPUT_NODE = True
OUTPUT_IS_LIST = (True,)
# OUTPUT_NODE = False
CATEGORY = "不忘科技-自定义节点🚩"
def log2db(self, log, status, unique_id):
# 获取comfy服务器队列信息
(_, prompt_id, prompt, extra_data, outputs_to_execute) = next(
iter(server.PromptServer.instance.prompt_queue.currently_running.values()))
job_id = extra_data["client_id"]
engine = create_engine(
"mysql+pymysql://root:*k3&5xxG6oqHJM@sh-cdb-1xspb808.sql.tencentcdb.com:28795/comfy",
echo=True
)
# Base.metadata.create_all(engine)
session = sessionmaker(bind=engine)()
# 查询
tasks = session.query(Task).filter(Task.prompt_id == prompt_id).all()
print(prompt)
result = {
"curr_node_id": str(unique_id),
"last_node_id": list(prompt.keys())[-1],
"node_output": str(log)
}
if len(tasks) == 0:
# 不存在插入
task = Task(prompt_id=prompt_id, job_id=job_id, result=json.dumps(result), status=status)
session.add(task)
elif len(tasks) == 1:
# 存在更新
session.query(Task).filter(Task.prompt_id == prompt_id).update({"result": json.dumps(result),
"status": status})
else:
# 异常报错
raise RuntimeError("状态数据库prompt_id不唯一, 无法记录状态！")
session.commit()
return {"ui": {"text": json.dumps(result)}, "result": (json.dumps(result),)}

1093
nodes/nodes_bfl.py
View File

File diff suppressed because it is too large Load Diff

269
nodes/object_storage_nodes.py Normal file
View File

@@ -0,0 +1,269 @@
import os
import boto3
import loguru
import yaml
from qcloud_cos import CosConfig, CosS3Client, CosClientError, CosServiceError
class COSDownload:
"""腾讯云COS下载"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"cos_bucket": ("STRING", {"default": "bwkj-cos-1324682537"}),
"cos_key": ("STRING", {"multiline": True}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("视频存储路径",)
FUNCTION = "download"
CATEGORY = "不忘科技-自定义节点🚩/对象存储/COS"
def download(self, cos_bucket, cos_key):
cos_key_in = cos_key.replace("/", os.sep)
destination = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"download",
os.path.dirname(cos_key_in),
os.path.basename(cos_key_in)
)
loguru.logger.info(f"COS DOWNLOAD to {destination}")
os.makedirs(
os.path.dirname(destination),
exist_ok=True,
)
for i in range(0, 10):
try:
if "aws_key_id" in list(os.environ.keys()):
yaml_config = os.environ
else:
with open(
os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config.yaml"
),
encoding="utf-8",
mode="r+",
) as f:
yaml_config = yaml.load(f, Loader=yaml.FullLoader)
config = CosConfig(
Region=yaml_config["cos_region"],
SecretId=yaml_config["cos_secret_id"],
SecretKey=yaml_config["cos_secret_key"],
)
client = CosS3Client(config)
response = client.download_file(
Bucket=cos_bucket,
Key=cos_key,
DestFilePath=destination
)
break
except CosClientError or CosServiceError as e:
raise Exception(f"COS下载失败！ bucket {cos_bucket}; key {cos_key}")
return (
destination,
)
class COSUpload:
"""腾讯云COS上传"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"cos_bucket": ("STRING", {"default": "bwkj-cos-1324682537"}),
"path": ("STRING", {"multiline": True}),
"subfolder": ("STRING", {"default": "test"}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("COS文件Key",)
FUNCTION = "upload"
CATEGORY = "不忘科技-自定义节点🚩/对象存储/COS"
def upload(self, cos_bucket, path, subfolder):
dest_key = "/".join(
[
subfolder,
(
path.split("/")[-1]
if "/" in path
else path.split("\\")[-1]
),
]
)
loguru.logger.info(f"COS UPLOAD {path} to {cos_bucket}/{subfolder}")
for i in range(0, 10):
try:
if "aws_key_id" in list(os.environ.keys()):
yaml_config = os.environ
else:
with open(
os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config.yaml"
),
encoding="utf-8",
mode="r+",
) as f:
yaml_config = yaml.load(f, Loader=yaml.FullLoader)
config = CosConfig(
Region=yaml_config["cos_region"],
SecretId=yaml_config["cos_secret_id"],
SecretKey=yaml_config["cos_secret_key"],
)
client = CosS3Client(config)
response = client.upload_file(
Bucket=cos_bucket,
Key=dest_key,
LocalFilePath=path,
)
break
except CosClientError or CosServiceError as e:
raise Exception(f"COS上传失败 bucket {cos_bucket}; local_path {path}; subfolder {subfolder}")
return (
dest_key,
)
class S3Download:
"""AWS S3下载"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"s3_bucket": ("STRING", {"default": "bw-comfyui-input"}),
"s3_key": ("STRING", {"multiline": True}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("视频存储路径",)
FUNCTION = "download"
CATEGORY = "不忘科技-自定义节点🚩/对象存储/S3"
def download(self, s3_bucket, s3_key):
s3_key_in = s3_key.replace("/", os.sep)
destination = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"download",
os.path.dirname(s3_key_in),
os.path.basename(s3_key_in),
)
loguru.logger.info(f"S3 DOWNLOAD to {destination}")
os.makedirs(
os.path.dirname(destination),
exist_ok=True,
)
try:
if "aws_key_id" in list(os.environ.keys()):
yaml_config = os.environ
else:
with open(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config.yaml"),
encoding="utf-8", mode="r+") as f:
yaml_config = yaml.load(f, Loader=yaml.FullLoader)
client = boto3.client("s3", aws_access_key_id=yaml_config["aws_key_id"],
aws_secret_access_key=yaml_config["aws_access_key"])
client.download_file(s3_bucket, s3_key, destination)
except Exception as e:
raise Exception(f"S3下载失败！ bucket {s3_bucket}; key {s3_key}")
return (destination,)
class S3Upload:
"""AWS S3上传"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"s3_bucket": ("STRING", {"default": "bw-comfyui-output"}),
"path": ("STRING", {"multiline": True}),
"subfolder": ("STRING", {"default": "test"}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("S3文件Key",)
FUNCTION = "upload"
CATEGORY = "不忘科技-自定义节点🚩/对象存储/S3"
def upload(self, s3_bucket, path, subfolder):
loguru.logger.info(f"S3 UPLOAD {path} to {s3_bucket}/{subfolder}")
try:
if "aws_key_id" in list(os.environ.keys()):
yaml_config = os.environ
else:
with open(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config.yaml"),
encoding="utf-8", mode="r+") as f:
yaml_config = yaml.load(f, Loader=yaml.FullLoader)
client = boto3.client("s3", aws_access_key_id=yaml_config["aws_key_id"],
aws_secret_access_key=yaml_config["aws_access_key"])
dest_key = "/".join(
[
subfolder,
(
path.split("/")[-1]
if "/" in path
else path.split("\\")[-1]
),
]
)
client.upload_file(path, s3_bucket, dest_key)
except Exception as e:
raise Exception(f"S3上传失败！ bucket {s3_bucket}; local_path {path}; subfolder {subfolder}")
return (dest_key,)
class S3UploadURL:
"""AWS S3上传 返回URL"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"path": ("STRING", {"multiline": True}),
"subfolder": ("STRING", {"default": "test"}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("S3文件Key",)
FUNCTION = "upload"
CATEGORY = "不忘科技-自定义节点🚩/对象存储/S3"
def upload(self, path, subfolder):
s3_bucket = "modal-media-cache"
loguru.logger.info(f"S3 UPLOAD {path} to {s3_bucket}/{subfolder}")
try:
if "aws_key_id" in list(os.environ.keys()):
yaml_config = os.environ
else:
with open(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config.yaml"),
encoding="utf-8", mode="r+") as f:
yaml_config = yaml.load(f, Loader=yaml.FullLoader)
client = boto3.client("s3", aws_access_key_id=yaml_config["aws_key_id"],
aws_secret_access_key=yaml_config["aws_access_key"])
dest_key = "/".join(
[
subfolder,
(
path.split("/")[-1]
if "/" in path
else path.split("\\")[-1]
),
]
)
client.upload_file(path, s3_bucket, dest_key)
except Exception as e:
raise Exception(f"S3上传失败！ bucket {s3_bucket}; local_path {path}; subfolder {subfolder}")
url = f"https://cdn.roasmax.cn/{dest_key}"
return (url,)

40
nodes/prompt_id_generator.py
View File

@@ -1,40 +0,0 @@
import uuid
class TaskIdGenerate:
"""TaskID生成器：用户可传入或自动生成TaskID"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {},
"optional": {
"custom_task_id": ("STRING", {"default": "", "placeholder": "留空则自动生成"}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("task_id",)
FUNCTION = "generate_task_id"
OUTPUT_NODE = False
CATEGORY = "utils"
def generate_task_id(self, custom_task_id=""):
if custom_task_id and custom_task_id.strip():
# 用户输入了自定义ID
task_id = custom_task_id.strip()
print(f"📝 使用自定义TaskID: {task_id}")
else:
# 自动生成UUID
task_id = str(uuid.uuid4())
print(f"🎲 自动生成TaskID: {task_id}")
return (task_id,)
NODE_CLASS_MAPPINGS = {
"TaskIdGenerate": TaskIdGenerate
}
NODE_DISPLAY_NAME_MAPPINGS = {
"TaskIdGenerate": "TaskID生成器"
}

77
nodes/random_line.py
View File

@@ -1,77 +0,0 @@
import random
import time
class RandomLineSelector:
"""
ComfyUI自定义节点：随机行选择器
从输入的多行文本中随机选择一行输出
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"text": ("STRING", {
"multiline": True,
"default": """
保持人物样貌和衣服不变，改变人物的光影效果。更换背景: 图像呈现出一个温馨的维多利亚复古风格房间，墙面被多样化的艺术画框与照片装饰，展现出浓郁的怀旧氛围。中心的金框全身镜是视觉焦点，上面缠绕着精致花卉装饰，增强了空间的艺术感。左侧靠墙摆放花瓶与烛台，白色地板映衬着跳动的暖黄色光线，营造出柔和而梦幻的氛围。右侧床铺整洁柔软，与一旁的小抽屉柜和台灯形成和谐的组合。整体画面色调以暖黄色为主，搭配田园风格装饰元素，构图均衡，注重细节，体现了舒适浪漫的生活场景。人物重新打光融合到场景中。
保持人物形象和衣服不变，改变人物的光影效果。更换背景: 衣帽间的开放式悬挂区搭配浅色木地板，衣杆上挂满中性色长裙，旁边搁板整齐叠放针织毛衣与围巾。
保持人物形象和衣服不变，改变人物的光影效果。更换背景: 白色衣柜门打开后，分层挂满长裙和连体裤，底部抽屉收纳着色彩鲜艳的运动服与家居裤。
保持人物形象和衣服不变，改变人物的光影效果。更换背景: 玻璃门的衣柜设计体现精致生活，内部悬挂精致衬衫和镂空裙摆，饰品抽屉随时为配件搭配提供轻便选择。
"""
}),
"seed": ("INT", {
"default": 0,
"min": -1,
"max": 0xffffffffffffffff
}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("selected_line",)
FUNCTION = "select_random_line"
CATEGORY = "text"
def select_random_line(self, text, seed):
"""
从输入文本中随机选择一行
Args:
text (str): 输入的多行文本
seed (int): 随机种子
Returns:
tuple: 包含选中行的元组
"""
# 设置随机种子（-1表示使用当前时间戳）
if seed == -1:
random.seed(int(time.time() * 1000000)) # 使用微秒级时间戳
else:
random.seed(seed)
# 过滤掉空行
non_empty_lines = [line.strip() for line in text.split('\n') if line.strip()]
# 如果没有非空行，返回空字符串
if not non_empty_lines:
return ("",)
# 随机选择一行
selected_line = random.choice(non_empty_lines)
return (selected_line,)
# ComfyUI节点映射
NODE_CLASS_MAPPINGS = {
"RandomLineSelector": RandomLineSelector
}
# 节点显示名称映射
NODE_DISPLAY_NAME_MAPPINGS = {
"RandomLineSelector": "随机选择一行内容"
}
# https://bowongai--dev-ui.modal.run/
# https://bowongai--dev-ui.modal.run

146
nodes/s3.py
View File

@@ -1,146 +0,0 @@
import os
import boto3
import loguru
import yaml
class S3Download:
"""AWS S3下载"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"s3_bucket": ("STRING", {"default": "bw-comfyui-input"}),
"s3_key": ("STRING", {"multiline": True}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("视频存储路径",)
FUNCTION = "download"
CATEGORY = "不忘科技-自定义节点🚩/S3"
def download(self, s3_bucket, s3_key):
s3_key_in = s3_key.replace("/",os.sep)
destination = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"download",
os.path.dirname(s3_key_in),
os.path.basename(s3_key_in),
)
loguru.logger.info(f"S3 DOWNLOAD to {destination}")
os.makedirs(
os.path.dirname(destination),
exist_ok=True,
)
try:
if "aws_key_id" in list(os.environ.keys()):
yaml_config = os.environ
else:
with open(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config.yaml"),encoding="utf-8",mode="r+") as f:
yaml_config = yaml.load(f, Loader=yaml.FullLoader)
client = boto3.client("s3",aws_access_key_id=yaml_config["aws_key_id"],aws_secret_access_key=yaml_config["aws_access_key"])
client.download_file(s3_bucket, s3_key, destination)
except Exception as e:
raise Exception(f"S3下载失败！ bucket {s3_bucket}; key {s3_key}")
return (destination,)
class S3Upload:
"""AWS S3上传"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"s3_bucket": ("STRING", {"default": "bw-comfyui-output"}),
"path": ("STRING", {"multiline": True}),
"subfolder": ("STRING", {"default": "test"}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("S3文件Key",)
FUNCTION = "upload"
CATEGORY = "不忘科技-自定义节点🚩/S3"
def upload(self, s3_bucket, path, subfolder):
loguru.logger.info(f"S3 UPLOAD {path} to {s3_bucket}/{subfolder}")
try:
if "aws_key_id" in list(os.environ.keys()):
yaml_config = os.environ
else:
with open(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config.yaml"),
encoding="utf-8", mode="r+") as f:
yaml_config = yaml.load(f, Loader=yaml.FullLoader)
client = boto3.client("s3", aws_access_key_id=yaml_config["aws_key_id"],
aws_secret_access_key=yaml_config["aws_access_key"])
dest_key = "/".join(
[
subfolder,
(
path.split("/")[-1]
if "/" in path
else path.split("\\")[-1]
),
]
)
client.upload_file(path, s3_bucket, dest_key)
except Exception as e:
raise Exception(f"S3上传失败！ bucket {s3_bucket}; local_path {path}; subfolder {subfolder}")
return (dest_key,)
class S3UploadURL:
"""AWS S3上传 返回URL"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"path": ("STRING", {"multiline": True}),
"subfolder": ("STRING", {"default": "test"}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("S3文件Key",)
FUNCTION = "upload"
CATEGORY = "不忘科技-自定义节点🚩/S3"
def upload(self, path, subfolder):
s3_bucket = "modal-media-cache"
loguru.logger.info(f"S3 UPLOAD {path} to {s3_bucket}/{subfolder}")
try:
if "aws_key_id" in list(os.environ.keys()):
yaml_config = os.environ
else:
with open(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config.yaml"),
encoding="utf-8", mode="r+") as f:
yaml_config = yaml.load(f, Loader=yaml.FullLoader)
client = boto3.client("s3", aws_access_key_id=yaml_config["aws_key_id"],
aws_secret_access_key=yaml_config["aws_access_key"])
dest_key = "/".join(
[
subfolder,
(
path.split("/")[-1]
if "/" in path
else path.split("\\")[-1]
),
]
)
client.upload_file(path, s3_bucket, dest_key)
except Exception as e:
raise Exception(f"S3上传失败！ bucket {s3_bucket}; local_path {path}; subfolder {subfolder}")
url = f"https://cdn.roasmax.cn/{dest_key}"
return (url,)

21
nodes/string_empty_judgement.py
View File

@@ -1,21 +0,0 @@
class StringEmptyJudgement:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"input": ("STRING", {"forceInput": True}),
},
}
RETURN_TYPES = ("BOOLEAN", )
RETURN_NAMES = ("是否为空", )
FUNCTION = "compute"
CATEGORY = "不忘科技-自定义节点🚩"
def compute(self, input):
if len(input) == 0:
return (True, )
else:
return (False, )

127
nodes/text.py
View File

@@ -1,127 +0,0 @@
import json
import os
import folder_paths
def get_allowed_dirs():
dir = os.path.abspath(os.path.join(__file__, "../../user"))
file = os.path.join(dir, "text_file_dirs.json")
with open(file, "r") as f:
return json.loads(f.read())
def get_valid_dirs():
return get_allowed_dirs().keys()
def get_dir_from_name(name):
dirs = get_allowed_dirs()
if name not in dirs:
raise KeyError(name + " dir not found")
path = dirs[name]
path = path.replace("$input", folder_paths.get_input_directory())
path = path.replace("$output", folder_paths.get_output_directory())
path = path.replace("$temp", folder_paths.get_temp_directory())
return path
def is_child_dir(parent_path, child_path):
parent_path = os.path.abspath(parent_path)
child_path = os.path.abspath(child_path)
return os.path.commonpath([parent_path]) == os.path.commonpath([parent_path, child_path])
def get_real_path(dir):
dir = dir.replace("/**/", "/")
dir = os.path.abspath(dir)
dir = os.path.split(dir)[0]
return dir
def get_file(root_dir, file):
if file == "[none]" or not file or not file.strip():
raise ValueError("No file")
root_dir = get_dir_from_name(root_dir)
root_dir = get_real_path(root_dir)
if not os.path.exists(root_dir):
os.mkdir(root_dir)
full_path = os.path.join(root_dir, file)
if not is_child_dir(root_dir, full_path):
raise ReferenceError()
return full_path
class LoadTextLocal:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"root_dir": (list(get_valid_dirs()), {}),
"file": (["[none]"], {
"pysssss.binding": [{
"source": "root_dir",
"callback": [{
"type": "set",
"target": "$this.disabled",
"value": True
}, {
"type": "fetch",
"url": "/pysssss/text-file/{$source.value}",
"then": [{
"type": "set",
"target": "$this.options.values",
"value": "$result"
}, {
"type": "validate-combo"
}, {
"type": "set",
"target": "$this.disabled",
"value": False
}]
}],
}]
}),
"encoding": ("STRING", {"default": "utf-8"}),
},
}
RETURN_TYPES = ("STRING",)
FUNCTION = "load"
CATEGORY = "不忘科技-自定义节点🚩"
@classmethod
def VALIDATE_INPUTS(self, root_dir, file, **kwargs):
if file == "[none]" or not file or not file.strip():
return True
get_file(root_dir, file)
return True
def load(self, root_dir, file, encoding):
with open(get_file(root_dir,file), "r", encoding=encoding) as f:
return (f.read(),)
class LoadTextOnline:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"file_path":("STRING", {"default": "input/"}),
"encoding": ("STRING", {"default": "utf-8"}),
}
}
RETURN_TYPES = ("STRING",)
FUNCTION = "load"
CATEGORY = "不忘科技-自定义节点🚩"
def load(self, file_path, encoding):
with open(file_path, "r", encoding=encoding) as f:
return (f.read(),)

216
nodes/text_nodes.py Normal file
View File

@@ -0,0 +1,216 @@
import os
import random
import time
import folder_paths
def get_allowed_dirs():
return {
"input": "$input/**/*.txt",
"output": "$output/**/*.txt",
"temp": "$temp/**/*.txt"
}
def get_valid_dirs():
return get_allowed_dirs().keys()
def get_dir_from_name(name):
dirs = get_allowed_dirs()
if name not in dirs:
raise KeyError(name + " dir not found")
path = dirs[name]
path = path.replace("$input", folder_paths.get_input_directory())
path = path.replace("$output", folder_paths.get_output_directory())
path = path.replace("$temp", folder_paths.get_temp_directory())
return path
def is_child_dir(parent_path, child_path):
parent_path = os.path.abspath(parent_path)
child_path = os.path.abspath(child_path)
return os.path.commonpath([parent_path]) == os.path.commonpath([parent_path, child_path])
def get_real_path(dir):
dir = dir.replace("/**/", "/")
dir = os.path.abspath(dir)
dir = os.path.split(dir)[0]
return dir
def get_file(root_dir, file):
if file == "[none]" or not file or not file.strip():
raise ValueError("No file")
root_dir = get_dir_from_name(root_dir)
root_dir = get_real_path(root_dir)
if not os.path.exists(root_dir):
os.mkdir(root_dir)
full_path = os.path.join(root_dir, file)
if not is_child_dir(root_dir, full_path):
raise ReferenceError()
return full_path
class LoadTextLocal:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"root_dir": (list(get_valid_dirs()), {}),
"file": (["[none]"], {
"pysssss.binding": [{
"source": "root_dir",
"callback": [{
"type": "set",
"target": "$this.disabled",
"value": True
}, {
"type": "fetch",
"url": "/pysssss/text-file/{$source.value}",
"then": [{
"type": "set",
"target": "$this.options.values",
"value": "$result"
}, {
"type": "validate-combo"
}, {
"type": "set",
"target": "$this.disabled",
"value": False
}]
}],
}]
}),
"encoding": ("STRING", {"default": "utf-8"}),
},
}
RETURN_TYPES = ("STRING",)
FUNCTION = "load"
CATEGORY = "不忘科技-自定义节点🚩/文本"
@classmethod
def VALIDATE_INPUTS(self, root_dir, file, **kwargs):
if file == "[none]" or not file or not file.strip():
return True
get_file(root_dir, file)
return True
def load(self, root_dir, file, encoding):
with open(get_file(root_dir, file), "r", encoding=encoding) as f:
return (f.read(),)
class LoadTextOnline:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"file_path": ("STRING", {"default": "input/"}),
"encoding": ("STRING", {"default": "utf-8"}),
}
}
RETURN_TYPES = ("STRING",)
FUNCTION = "load"
CATEGORY = "不忘科技-自定义节点🚩/文本"
def load(self, file_path, encoding):
with open(file_path, "r", encoding=encoding) as f:
return (f.read(),)
class StringEmptyJudgement:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"input": ("STRING", {"forceInput": True}),
},
}
RETURN_TYPES = ("BOOLEAN",)
RETURN_NAMES = ("是否为空",)
FUNCTION = "compute"
CATEGORY = "不忘科技-自定义节点🚩/文本"
def compute(self, input):
if len(input) == 0:
return (True,)
else:
return (False,)
class RandomLineSelector:
"""
ComfyUI自定义节点：随机行选择器
从输入的多行文本中随机选择一行输出
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"text": ("STRING", {
"multiline": True,
"default": """
保持人物样貌和衣服不变，改变人物的光影效果。更换背景: 图像呈现出一个温馨的维多利亚复古风格房间，墙面被多样化的艺术画框与照片装饰，展现出浓郁的怀旧氛围。中心的金框全身镜是视觉焦点，上面缠绕着精致花卉装饰，增强了空间的艺术感。左侧靠墙摆放花瓶与烛台，白色地板映衬着跳动的暖黄色光线，营造出柔和而梦幻的氛围。右侧床铺整洁柔软，与一旁的小抽屉柜和台灯形成和谐的组合。整体画面色调以暖黄色为主，搭配田园风格装饰元素，构图均衡，注重细节，体现了舒适浪漫的生活场景。人物重新打光融合到场景中。
保持人物形象和衣服不变，改变人物的光影效果。更换背景: 衣帽间的开放式悬挂区搭配浅色木地板，衣杆上挂满中性色长裙，旁边搁板整齐叠放针织毛衣与围巾。
保持人物形象和衣服不变，改变人物的光影效果。更换背景: 白色衣柜门打开后，分层挂满长裙和连体裤，底部抽屉收纳着色彩鲜艳的运动服与家居裤。
保持人物形象和衣服不变，改变人物的光影效果。更换背景: 玻璃门的衣柜设计体现精致生活，内部悬挂精致衬衫和镂空裙摆，饰品抽屉随时为配件搭配提供轻便选择。
"""
}),
"seed": ("INT", {
"default": 0,
"min": -1,
"max": 0xffffffffffffffff
}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("selected_line",)
FUNCTION = "select_random_line"
CATEGORY = "不忘科技-自定义节点🚩/文本"
def select_random_line(self, text, seed):
"""
从输入文本中随机选择一行
Args:
text (str): 输入的多行文本
seed (int): 随机种子
Returns:
tuple: 包含选中行的元组
"""
# 设置随机种子（-1表示使用当前时间戳）
if seed == -1:
random.seed(int(time.time() * 1000000)) # 使用微秒级时间戳
else:
random.seed(seed)
# 过滤掉空行
non_empty_lines = [line.strip() for line in text.split('\n') if line.strip()]
# 如果没有非空行，返回空字符串
if not non_empty_lines:
return ("",)
# 随机选择一行
selected_line = random.choice(non_empty_lines)
return (selected_line,)

39
nodes/traverse_folder.py
View File

@@ -1,39 +0,0 @@
import glob
import os
class AnyType(str):
"""A special class that is always equal in not equal comparisons. Credit to pythongosssss"""
def __ne__(self, __value: object) -> bool:
return False
any = AnyType("*")
class TraverseFolder:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"folder": ("STRING", {"default": r"E:\comfy\ComfyUI\input\s3", "required": True}),
"subfix": ("STRING", {"default":".mp4", "required": True}),
"recursive": ("BOOLEAN", {"default": True, "required": True}),
"idx": (
"INT",
{"default": 0, "min": 0, "max": 0xFFFFFF},
),
},
}
RETURN_TYPES = ("STRING", )
RETURN_NAMES = ("文件路径",)
FUNCTION = "compute"
CATEGORY = "不忘科技-自定义节点🚩"
def compute(self, folder, subfix, recursive, idx):
files = glob.glob(os.path.join(folder, r"**\*%s" % subfix), recursive=recursive)
if len(files) == 0:
raise RuntimeError("No Files Found")
return (str(files[idx % len(files)]),)

32
nodes/unload_all_models.py
View File

@@ -1,32 +0,0 @@
import comfy.model_management
class AnyType(str):
"""A special class that is always equal in not equal comparisons. Credit to pythongosssss"""
def __ne__(self, __value: object) -> bool:
return False
any = AnyType("*")
class UnloadAllModels:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"any":(any,{"forceInput": True})
},
"optional": {},
}
RETURN_TYPES = ()
FUNCTION = "unload_models"
CATEGORY = "不忘科技-自定义节点🚩"
OUTPUT_NODE = True
def unload_models(self,any=None):
# 卸载所有已加载的模型
comfy.model_management.soft_empty_cache()
comfy.model_management.unload_all_models()
comfy.model_management.soft_empty_cache()
return ()

339
nodes/util_nodes.py Normal file
View File

@@ -0,0 +1,339 @@
import glob
import json
import os
import time
from pathlib import Path
import comfy.model_management
import requests
import server
import yaml
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from tencentcloud.common import credential
from tencentcloud.common.profile.client_profile import ClientProfile
from tencentcloud.common.profile.http_profile import HttpProfile
from tencentcloud.vod.v20180717 import vod_client, models
from ..utils.task_table import Task
class LogToDB:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"job_id": ("STRING", {"forceInput": True}),
"log": ("STRING", {"forceInput": True}),
"status": ("INT", {"default": 1, "max": 1}),
"sql_url": ("STRING", {
"default": "mysql+pymysql://root:root@example.com:3306/test"}),
},
"hidden": {
"unique_id": "UNIQUE_ID",
}
}
RETURN_TYPES = ("STRING",)
FUNCTION = "log2db"
OUTPUT_NODE = True
OUTPUT_IS_LIST = (True,)
# OUTPUT_NODE = False
CATEGORY = "不忘科技-自定义节点🚩/工具"
def log2db(self, log, status, sql_url, unique_id):
# 获取comfy服务器队列信息
(_, prompt_id, prompt, extra_data, outputs_to_execute) = next(
iter(server.PromptServer.instance.prompt_queue.currently_running.values()))
job_id = extra_data["client_id"]
engine = create_engine(
sql_url,
echo=True
)
# Base.metadata.create_all(engine)
session = sessionmaker(bind=engine)()
# 查询
tasks = session.query(Task).filter(Task.prompt_id == prompt_id).all()
print(prompt)
result = {
"curr_node_id": str(unique_id),
"last_node_id": list(prompt.keys())[-1],
"node_output": str(log)
}
if len(tasks) == 0:
# 不存在插入
task = Task(prompt_id=prompt_id, job_id=job_id, result=json.dumps(result), status=status)
session.add(task)
elif len(tasks) == 1:
# 存在更新
session.query(Task).filter(Task.prompt_id == prompt_id).update({"result": json.dumps(result),
"status": status})
else:
# 异常报错
raise RuntimeError("状态数据库prompt_id不唯一, 无法记录状态！")
session.commit()
return {"ui": {"text": json.dumps(result)}, "result": (json.dumps(result),)}
class VodToLocalNode:
def __init__(self):
if "aws_key_id" in list(os.environ.keys()):
yaml_config = os.environ
else:
with open(
os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config.yaml"
),
encoding="utf-8",
mode="r+",
) as f:
yaml_config = yaml.load(f, Loader=yaml.FullLoader)
self.secret_id = yaml_config["cos_secret_id"]
self.secret_key = yaml_config["cos_secret_key"]
self.region = yaml_config["cos_region"]
self.vod_client = self.init_vod_client()
def init_vod_client(self):
"""初始化VOD客户端"""
try:
http_profile = HttpProfile(endpoint="vod.tencentcloudapi.com")
client_profile = ClientProfile(httpProfile=http_profile)
cred = credential.Credential(self.secret_id, self.secret_key)
return vod_client.VodClient(cred, self.region, client_profile)
except Exception as e:
raise RuntimeError(f"VOD client initialization failed: {e}")
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"file_id": ("STRING", {"default": ""}),
"sub_app_id": ("STRING", {"default": ""}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("local_path",)
FUNCTION = "execute"
CATEGORY = "不忘科技-自定义节点🚩/工具"
def execute(self, file_id, sub_app_id):
# 调用下载逻辑
local_path = self.download_vod(file_id, sub_app_id)
print(f"下载成功: {local_path}")
return (local_path,)
def _get_download_url(self, file_id, sub_app_id):
"""获取媒体文件下载地址"""
try:
req = models.DescribeMediaInfosRequest()
req.FileIds = [file_id]
req.SubAppId = int(sub_app_id)
resp = self.vod_client.DescribeMediaInfos(req)
if not resp.MediaInfoSet:
raise ValueError("File not found")
media_info = resp.MediaInfoSet[0]
if not media_info.BasicInfo.MediaUrl:
raise ValueError("No download URL available")
return media_info.BasicInfo.MediaUrl
except Exception as e:
raise RuntimeError(f"Tencent API error: {e}")
def create_directory(self, path):
p = Path(path)
if not p.exists():
p.mkdir(
parents=True, exist_ok=True
) # parents=True会自动创建所有必需的父目录，exist_ok=True表示如果目录已存在则不会引发异常
print(f"目录已创建: {path}")
else:
print(f"目录已存在: {path}")
def download_vod(self, file_id, sub_app_id):
"""
需要补充腾讯云VOD SDK调用逻辑
返回本地文件路径
"""
media_url = self._get_download_url(file_id=file_id, sub_app_id=sub_app_id)
print(f"download from url: {media_url}")
# 生成一个临时目录路径名并创建该目录
self.create_directory(
os.path.join(
os.path.dirname(os.path.abspath(__file__)), "download", f"{sub_app_id}"
)
)
output_dir = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
"download",
f"{sub_app_id}",
f"{file_id}.mp4",
)
# 判断文件是否存在
if os.path.exists(output_dir):
return output_dir
return self._download_file(url=media_url, save_path=output_dir, timeout=60 * 10)
def _download_file(self, url: str, save_path: str, timeout: int = 30):
"""下载文件到本地"""
try:
with requests.get(url, stream=True, timeout=timeout) as response:
response.raise_for_status()
with open(save_path, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
return save_path
except Exception as e:
raise RuntimeError(f"Download error: {e}")
class AnyType(str):
"""A special class that is always equal in not equal comparisons. Credit to pythongosssss"""
def __ne__(self, __value: object) -> bool:
return False
any = AnyType("*")
class UnloadAllModels:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"any": (any, {"forceInput": True})
},
"optional": {},
}
RETURN_TYPES = ()
FUNCTION = "unload_models"
CATEGORY = "不忘科技-自定义节点🚩/工具"
OUTPUT_NODE = True
def unload_models(self, any=None):
# 卸载所有已加载的模型
comfy.model_management.soft_empty_cache()
comfy.model_management.unload_all_models()
comfy.model_management.soft_empty_cache()
return ()
class TraverseFolder:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"folder": ("STRING", {"default": r"E:\comfy\ComfyUI\input\s3", "required": True}),
"subfix": ("STRING", {"default": ".mp4", "required": True}),
"recursive": ("BOOLEAN", {"default": True, "required": True}),
"idx": (
"INT",
{"default": 0, "min": 0, "max": 0xFFFFFF},
),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("文件路径",)
FUNCTION = "compute"
CATEGORY = "不忘科技-自定义节点🚩/工具"
def compute(self, folder, subfix, recursive, idx):
files = glob.glob(os.path.join(folder, r"**\*%s" % subfix), recursive=recursive)
if len(files) == 0:
raise RuntimeError("No Files Found")
return (str(files[idx % len(files)]),)
class PlugAndPlayWebhook:
"""即插即用Webhook节点：连上线就能转发数据"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"webhook_url": ("STRING", {"default": "http://127.0.0.1:8010/handler/webhook",
"placeholder": "https://your-api.com/webhook"}),
"image_url": ("STRING", {"default": "",
"placeholder": "图片的url"}),
},
"optional": {
"prompt_id": ("STRING", {"default": ""}),
},
"hidden": {"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO", "unique_id": "UNIQUE_ID"},
}
RETURN_TYPES = ()
FUNCTION = "send"
OUTPUT_NODE = True
CATEGORY = "不忘科技-自定义节点🚩/工具"
def send(self, webhook_url, image_url, prompt_id="", prompt=None, extra_pnginfo=None, unique_id=None):
if not webhook_url:
raise ValueError("❌ 请填写Webhook URL！")
# 使用传入的prompt_id，如果没有则用unique_id
final_prompt_id = prompt_id or unique_id or "unknown"
# 准备发送的数据
data = {
"img_base64": image_url,
"format": "png",
"image_url": image_url,
"prompt_id": final_prompt_id,
"timestamp": time.time()
}
# 发送Webhook
try:
response = requests.post(webhook_url, json=data)
response.raise_for_status()
print(f'发送的数据:{data}')
except Exception as e:
print(f"❌ 发送失败: {str(e)}")
# 终端节点，无需返回
return ()
class TaskIdGenerate:
"""TaskID生成器：用户可传入或自动生成TaskID"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {},
"optional": {
"custom_task_id": ("STRING", {"default": "", "placeholder": "留空则自动生成"}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("task_id",)
FUNCTION = "generate_task_id"
OUTPUT_NODE = False
CATEGORY = "不忘科技-自定义节点🚩/工具"
def generate_task_id(self, custom_task_id=""):
if custom_task_id and custom_task_id.strip():
# 用户输入了自定义ID
task_id = custom_task_id.strip()
print(f"📝 使用自定义TaskID: {task_id}")
else:
# 自动生成UUID
task_id = str(uuid.uuid4())
print(f"🎲 自动生成TaskID: {task_id}")
return (task_id,)

6
nodes/heygem.py → nodes/video_lipsync_nodes.py
View File

@@ -1,6 +1,7 @@
import json
import os
import time
import traceback
import uuid
import httpx
@@ -8,7 +9,6 @@ import loguru
import torchaudio
import torchvision
from torch import Tensor
import traceback
def task_submit(uid, video, audio, heygem_url):
@@ -140,7 +140,7 @@ class HeyGemF2F:
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("视频存储路径",)
FUNCTION = "f2f"
CATEGORY = "不忘科技-自定义节点🚩/口型同步"
CATEGORY = "不忘科技-自定义节点🚩/视频/口型"
def f2f(self, video:Tensor, audio:dict, heygem_url:str, heygem_temp_path:str, is_Windows:bool):
uid = str(uuid.uuid4())
@@ -206,7 +206,7 @@ class HeyGemF2FFromFile:
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("视频存储路径",)
FUNCTION = "f2f"
CATEGORY = "不忘科技-自定义节点🚩/口型同步"
CATEGORY = "不忘科技-自定义节点🚩/视频/口型"
def f2f(self, video:str, audio:str, heygem_url:str, heygem_temp_path:str, is_Windows:bool):
uid = str(uuid.uuid4())

84
nodes/video.py → nodes/video_nodes.py
View File

@@ -1,9 +1,8 @@
import errno
import glob
import os
import re
import shutil
import subprocess
import sys
import traceback
import uuid
from datetime import datetime
@@ -11,12 +10,10 @@ from datetime import datetime
import ffmpy
import loguru
import torchvision.io
from ffmpy import FFExecutableNotFoundError, FFRuntimeError
video_extensions = ['webm', 'mp4', 'mkv', 'gif', 'mov']
class VideoCut:
"""FFMPEG视频剪辑"""
@@ -24,14 +21,15 @@ class VideoCut:
def INPUT_TYPES(s):
return {
"required": {
"video_path": ("STRING",{"placeholder": "X://insert/path/here.mp4", "vhs_path_extensions": video_extensions}),
"video_path": (
"STRING", {"placeholder": "X://insert/path/here.mp4", "vhs_path_extensions": video_extensions}),
"start": ("STRING", {"default": "00:00:00.000"}),
"end": ("STRING", {"default": "00:00:10.000"}),
},
}
RETURN_TYPES = ("IMAGE","AUDIO")
RETURN_NAMES = ("视频帧","音频")
RETURN_TYPES = ("IMAGE", "AUDIO")
RETURN_NAMES = ("视频帧", "音频")
FUNCTION = "cut"
@@ -121,11 +119,12 @@ class VideoCut:
os.remove(files[0])
except:
pass
return (video, {"waveform":audio,"sample_rate":info["audio_fps"]},)
return (video, {"waveform": audio, "sample_rate": info["audio_fps"]},)
except:
traceback.print_exc()
raise Exception("Cut Failed")
class VideoCutByFramePoint:
"""FFMPEG视频剪辑-帧位"""
@@ -133,7 +132,8 @@ class VideoCutByFramePoint:
def INPUT_TYPES(s):
return {
"required": {
"video_path": ("STRING",{"placeholder": "X://insert/path/here.mp4", "vhs_path_extensions": video_extensions}),
"video_path": (
"STRING", {"placeholder": "X://insert/path/here.mp4", "vhs_path_extensions": video_extensions}),
"start_point": ("FLOAT", {"default": "0.0"}),
"duration": ("FLOAT", {"default": "10.0"}),
"fps": ("INT", {"default": "25"}),
@@ -141,8 +141,8 @@ class VideoCutByFramePoint:
},
}
RETURN_TYPES = ("IMAGE","AUDIO")
RETURN_NAMES = ("视频帧","音频")
RETURN_TYPES = ("IMAGE", "AUDIO")
RETURN_NAMES = ("视频帧", "音频")
FUNCTION = "cut"
@@ -189,9 +189,9 @@ class VideoCutByFramePoint:
outputs={
output: [
"-ss",
"%.3f" % (start_point/fps),
"%.3f" % (start_point / fps),
"-t",
"%.3f" % (duration/fps),
"%.3f" % (duration / fps),
"-c:v",
"libx264",
"-c:a",
@@ -234,11 +234,12 @@ class VideoCutByFramePoint:
os.remove(output)
except:
pass
return (video, {"waveform":audio,"sample_rate":info["audio_fps"]},)
return (video, {"waveform": audio, "sample_rate": info["audio_fps"]},)
except:
traceback.print_exc()
raise Exception("Cut Failed")
class VideoChangeFPS:
"""FFMPEG视频FPS转换"""
@@ -246,7 +247,8 @@ class VideoChangeFPS:
def INPUT_TYPES(s):
return {
"required": {
"video_path": ("STRING",{"placeholder": "X://insert/path/here.mp4", "vhs_path_extensions": video_extensions}),
"video_path": (
"STRING", {"placeholder": "X://insert/path/here.mp4", "vhs_path_extensions": video_extensions}),
"fps": ("INT", {"default": 30}),
},
}
@@ -323,7 +325,7 @@ class VideoChangeFPS:
video_path = "output/" + video_path
loguru.logger.info("Processing video: %s" % video_path)
output_temp = ".".join([video_path.split(".")[-2] + "-%dfps-temp" % fps, video_path.split(".")[-1]])
output = ".".join([video_path.split(".")[-2]+"-%dfps" % fps,video_path.split(".")[-1]])
output = ".".join([video_path.split(".")[-2] + "-%dfps" % fps, video_path.split(".")[-1]])
# 分步执行
self.adjust_video_fps(video_path, output_temp, fps) # 第一步：调整FPS
self.align_audio_to_video(output_temp, output) # 第二步：对齐音频
@@ -333,7 +335,8 @@ class VideoChangeFPS:
final_audio_dur = self.get_media_duration(output, "a")
print("video_duration:", final_video_dur, "audio_duration:", final_audio_dur)
if abs(final_video_dur - final_audio_dur) > 0.01:
loguru.logger.warning(f"音视频长度未对齐！视频长度: {final_video_dur:.3f}s, 音频长度: {final_audio_dur:.3f}s")
loguru.logger.warning(
f"音视频长度未对齐！视频长度: {final_video_dur:.3f}s, 音频长度: {final_audio_dur:.3f}s")
else:
loguru.logger.success(f"处理成功！视频长度: {final_video_dur:.3f}s, 音频长度: {final_audio_dur:.3f}s")
try:
@@ -343,4 +346,49 @@ class VideoChangeFPS:
return (output,)
except:
traceback.print_exc()
raise Exception("ChangeFPS Failed")
raise Exception("ChangeFPS Failed")
def validate_time_format(time_str):
pattern = r'^([0-1][0-9]|2[0-3]):([0-5][0-9]):([0-5][0-9]|\d{1,2}).(\d{3})$'
return bool(re.match(pattern, time_str))
def get_duration_wave(audio):
waveform, sample_rate = audio["waveform"], audio["sample_rate"]
# 防止话说不完
return waveform.shape[2] / sample_rate
class VideoStartPointDurationCompute:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"start_time": ("STRING", {"forceInput": True}),
"audio": ("AUDIO", {"forceInput": True}),
"end_padding": ("FLOAT", {"forceInput": True, "default": 0.4}),
"fps": ("INT", {"default": 25, "step": 1}),
},
}
RETURN_TYPES = ("FLOAT", "FLOAT",)
RETURN_NAMES = ("起始帧位", "帧数")
FUNCTION = "compute"
CATEGORY = "不忘科技-自定义节点🚩/视频"
def compute(self, start_time, audio, fps, end_padding):
if not validate_time_format(start_time):
raise ValueError("start_time或者end_time时间格式不对（start_time or end_time is not in time format）")
time_format = "%H:%M:%S.%f"
start_dt = datetime.strptime(start_time, time_format)
start_sec = (start_dt - datetime(1900, 1, 1)).total_seconds()
start_point = start_sec * fps
duration = get_duration_wave(audio)
loguru.logger.info("audio duration %.3f s" % duration)
duration = duration + end_padding
loguru.logger.info("audio duration with padding %.3f s" % duration)
return (start_point, duration * fps,)

112
nodes/vod2local.py
View File

@@ -1,112 +0,0 @@
import os
from pathlib import Path
import requests
from tencentcloud.common import credential
from tencentcloud.common.profile.client_profile import ClientProfile
from tencentcloud.common.profile.http_profile import HttpProfile
from tencentcloud.vod.v20180717 import vod_client, models
class VodToLocalNode:
def __init__(self):
self.secret_id = "AKIDsrihIyjZOBsjimt8TsN8yvv1AMh5dB44"
self.secret_key = "CPZcxdk6W39Jd4cGY95wvupoyMd0YFqW"
self.vod_client = self.init_vod_client()
def init_vod_client(self):
"""初始化VOD客户端"""
try:
http_profile = HttpProfile(endpoint="vod.tencentcloudapi.com")
client_profile = ClientProfile(httpProfile=http_profile)
cred = credential.Credential(self.secret_id, self.secret_key)
return vod_client.VodClient(cred, "ap-shanghai", client_profile)
except Exception as e:
raise RuntimeError(f"VOD client initialization failed: {e}")
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"file_id": ("STRING", {"default": ""}),
"sub_app_id": ("STRING", {"default": ""}),
}
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("local_path",)
FUNCTION = "execute"
CATEGORY = "video"
def execute(self, file_id, sub_app_id):
# 调用下载逻辑
local_path = self.download_vod(file_id, sub_app_id)
print(f"下载成功: {local_path}")
return (local_path,)
def _get_download_url(self, file_id, sub_app_id):
"""获取媒体文件下载地址"""
try:
req = models.DescribeMediaInfosRequest()
req.FileIds = [file_id]
req.SubAppId = int(sub_app_id)
resp = self.vod_client.DescribeMediaInfos(req)
if not resp.MediaInfoSet:
raise ValueError("File not found")
media_info = resp.MediaInfoSet[0]
if not media_info.BasicInfo.MediaUrl:
raise ValueError("No download URL available")
return media_info.BasicInfo.MediaUrl
except Exception as e:
raise RuntimeError(f"Tencent API error: {e}")
def create_directory(self, path):
p = Path(path)
if not p.exists():
p.mkdir(
parents=True, exist_ok=True
) # parents=True会自动创建所有必需的父目录，exist_ok=True表示如果目录已存在则不会引发异常
print(f"目录已创建: {path}")
else:
print(f"目录已存在: {path}")
def download_vod(self, file_id, sub_app_id):
"""
需要补充腾讯云VOD SDK调用逻辑
返回本地文件路径
"""
media_url = self._get_download_url(file_id=file_id, sub_app_id=sub_app_id)
print(f"download from url: {media_url}")
# 生成一个临时目录路径名并创建该目录
self.create_directory(
os.path.join(
os.path.dirname(os.path.abspath(__file__)), "download", f"{sub_app_id}"
)
)
output_dir = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
"download",
f"{sub_app_id}",
f"{file_id}.mp4",
)
# 判断文件是否存在
if os.path.exists(output_dir):
return output_dir
return self._download_file(url=media_url, save_path=output_dir, timeout=60 * 10)
def _download_file(self, url: str, save_path: str, timeout: int = 30):
"""下载文件到本地"""
try:
with requests.get(url, stream=True, timeout=timeout) as response:
response.raise_for_status()
with open(save_path, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
return save_path
except Exception as e:
raise RuntimeError(f"Download error: {e}")

187
nodes/webhook_forward.py
View File

@@ -1,187 +0,0 @@
# -*- coding:utf-8 -*-
"""
File webhook_forward.py
Author silence
Date 2025/6/13 13:42
"""
import base64
import io
import os
import random
import string
import time
import numpy as np
import requests
from PIL import Image
class PromptIDGenerator:
"""PromptID生成器：用户可传入或自动生成PromptID"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {},
"optional": {
"custom_prompt_id": ("STRING", {"default": "", "placeholder": "留空则自动生成"}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("prompt_id",)
FUNCTION = "generate_prompt_id"
OUTPUT_NODE = False
CATEGORY = "utils"
def generate_prompt_id(self, custom_prompt_id=""):
if custom_prompt_id and custom_prompt_id.strip():
# 用户输入了自定义ID
prompt_id = custom_prompt_id.strip()
print(f"📝 使用自定义PromptID: {prompt_id}")
else:
# 自动生成ID：时间戳 + 随机字符
timestamp = str(int(time.time()))
random_str = ''.join(random.choices(string.ascii_lowercase + string.digits, k=6))
prompt_id = f"prompt_{timestamp}_{random_str}"
print(f"🎲 自动生成PromptID: {prompt_id}")
return (prompt_id,)
class PlugAndPlayWebhook:
"""即插即用Webhook节点：连上线就能转发数据"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"webhook_url": ("STRING", {"default": "http://127.0.0.1:8010/handler/webhook",
"placeholder": "https://your-api.com/webhook"}),
"image_url": ("STRING", {"default": "",
"placeholder": "图片的url"}),
},
"optional": {
"prompt_id": ("STRING", {"default": ""}),
},
"hidden": {"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO", "unique_id": "UNIQUE_ID"},
}
RETURN_TYPES = ()
FUNCTION = "send"
OUTPUT_NODE = True
CATEGORY = "utils"
def send(self, webhook_url, image_url, prompt_id="", prompt=None, extra_pnginfo=None, unique_id=None):
if not webhook_url:
raise ValueError("❌ 请填写Webhook URL！")
# 使用传入的prompt_id，如果没有则用unique_id
final_prompt_id = prompt_id or unique_id or "unknown"
# 准备发送的数据
data = {
"img_base64": image_url,
"format": "png",
"image_url": image_url,
"prompt_id": final_prompt_id,
"timestamp": time.time()
}
# 发送Webhook
try:
response = requests.post(webhook_url, json=data)
response.raise_for_status()
print(f'发送的数据:{data}')
except Exception as e:
print(f"❌ 发送失败: {str(e)}")
# 终端节点，无需返回
return ()
class SaveImageWithOutput:
"""保存图片并输出的节点：既保存图片又能继续传递数据"""
def __init__(self):
self.output_dir = "output"
self.type = "output"
self.prefix_append = ""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"images": ("IMAGE",),
"filename_prefix": ("STRING", {"default": "ComfyUI"}),
},
}
RETURN_TYPES = ("IMAGE", "STRING")
RETURN_NAMES = ("images", "file_path")
FUNCTION = "save_image"
OUTPUT_NODE = False
CATEGORY = "image"
def save_image(self, images, filename_prefix="ComfyUI"):
filename_prefix += self.prefix_append
full_output_folder, filename, counter, subfolder, filename_prefix = self.get_save_image_path(filename_prefix,
self.output_dir)
file_paths = []
for i, image in enumerate(images):
# 转换图片数据
img_array = 255. * image.cpu().numpy()
img = Image.fromarray(np.clip(img_array, 0, 255).astype(np.uint8))
# 生成文件名
file = f"{filename_prefix}_{counter + i:05}_.png"
file_path = os.path.join(full_output_folder, file)
# 保存图片
img.save(file_path, compress_level=4)
file_paths.append(file_path)
print(f"✅ 图片已保存到: {file_paths[0] if file_paths else '未知路径'}")
# 直接返回元组：(原始图片数据, 第一个文件路径)
return (images, file_paths[0] if file_paths else "")
def get_save_image_path(self, filename_prefix, output_dir):
def map_filename(filename):
prefix_len = len(os.path.basename(filename_prefix))
prefix = filename[:prefix_len + 1]
try:
digits = int(filename[prefix_len + 1:].split('_')[0])
except:
digits = 0
return (digits, prefix)
subfolder = ""
full_output_folder = os.path.join(output_dir, subfolder)
if os.path.commonpath((output_dir, os.path.abspath(full_output_folder))) != output_dir:
print("Saving image outside the output folder is not allowed.")
return {}
try:
counter = max(filter(lambda a: a[1][:-1] == filename_prefix and a[1][-1] == "_",
map(map_filename, os.listdir(full_output_folder))))[0] + 1
except ValueError:
counter = 1
except FileNotFoundError:
os.makedirs(full_output_folder, exist_ok=True)
counter = 1
return full_output_folder, filename_prefix, counter, subfolder, filename_prefix
NODE_CLASS_MAPPINGS = {
"PlugAndPlayWebhook": PlugAndPlayWebhook,
"SaveImageWithOutput": SaveImageWithOutput
}
NODE_DISPLAY_NAME_MAPPINGS = {
"PlugAndPlayWebhook": "Webhook转发器",
"SaveImageWithOutput": "保存图片(带输出)"
}

28
pyproject_comfyui.toml
View File

@@ -1,28 +0,0 @@
[build-system]
requires = ["setuptools"]
build-backend = "setuptools.build_meta"
[project]
name = "modal_comfyui"
version = "1.0.0"
requires-python = ">=3.10"
dependencies = [
"fastapi[standard]==0.115.4",
"cos-python-sdk-v5",
"sqlalchemy",
"ultralytics",
"tencentcloud-sdk-python",
"pymysql",
"Pillow",
"ffmpy",
"opencv-python",
"av",
"imageio",
"loguru",
"openai-whisper",
"sentry-sdk",
"pydantic",
"pydantic_settings",
"conformer==0.3.2",
"einops>0.6.1"
]

21
requirements.txt
View File

@@ -1,13 +1,18 @@
ffmpy
torch
torchvision
Pillow
opencv-python
ultralytics
cos-python-sdk-v5
sqlalchemy
ultralytics
tencentcloud-sdk-python
boto3
pymysql
Pillow
ffmpy
opencv-python
av
imageio
loguru
conformer==0.3.2
einops>0.6.1
tqdm
retry
pyYAML
retry
boto3
Jinja2

5
user/text_file_dirs.json
View File

@@ -1,5 +0,0 @@
{
"input": "$input/**/*.txt",
"output": "$output/**/*.txt",
"temp": "$temp/**/*.txt"
}

10
nodes/test_single_image.py → utils/face_occu_detect.py
View File

@@ -7,8 +7,8 @@ import torch
from torchvision import transforms
from torchvision.transforms import Resize
from .utils import load_weight
from .model import Model
from ..utils.modal_utils import load_weight
from ..utils.model_module import Model
# CONSTANT
MEAN = [0.485, 0.456, 0.406]
@@ -18,7 +18,7 @@ CLASSES = {0: "non-occluded",
1: "occluded"}
def test_image(opt):
def face_occu_detect_single(opt):
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = Model(opt.model, 2, False).to(device)
model = load_weight(model, opt.weight)
@@ -56,7 +56,7 @@ def test_image(opt):
))
def test_node(image: torch.Tensor, length=10, thres=95, model_name="convnext_tiny"):
def face_occu_detect(image: torch.Tensor, length=10, thres=95, model_name="convnext_tiny"):
weight_dic = {
"convnext_tiny": "best_convnext_tiny.pth",
"convnext_base": "best_convnext_base.pth"
@@ -115,4 +115,4 @@ if __name__ == "__main__":
parser.add_argument("--image", type=str, help="Image path")
args = parser.parse_args()
test_image(args)
face_occu_detect_single(args)

0
nodes/utils.py → utils/modal_utils.py
View File

2
nodes/model.py → utils/model_module.py
View File

@@ -1,7 +1,7 @@
from torch import nn
from PIL import ImageFile
from .utils import get_model
from ..utils.modal_utils import get_model
ImageFile.LOAD_TRUNCATED_IMAGES = True

0
nodes/table.py → utils/task_table.py
View File