nodes/7BEII-comfyui-PD_comfy-api-node
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

GPT2

Browse Source 
213
This commit is contained in:
pandy
2026-04-22 17:19:41 +08:00
parent 6b938aaef6
commit a25fbf29c1
6 changed files with 759 additions and 414 deletions
Download Patch File Download Diff File Expand all files Collapse all files
__init__.py
+30 -43
View File
@@ -1,74 +1,61 @@
"""
Comfyui_PDuse - PD ComfyUI 自定义节点套件
包含多种实用的图像处理和工具节点
"""
import importlib
import os
import sys
# 初始化节点映射
NODE_CLASS_MAPPINGS = {}
NODE_DISPLAY_NAME_MAPPINGS = {}
def get_ext_dir(subpath=None, mkdir=False):
"""获取扩展目录路径"""
dir = os.path.dirname(__file__)
def get_ext_dir(subpath=None):
base_dir = os.path.dirname(__file__)
if subpath is not None:
dir = os.path.join(dir, subpath)
return os.path.abspath(dir)
base_dir = os.path.join(base_dir, subpath)
return os.path.abspath(base_dir)
def safe_import_module(module_path, file_name):
"""安全导入模块，避免导入错误导致崩溃"""
try:
imported_module = importlib.import_module(module_path)
# 检查模块是否包含所需的映射
if hasattr(imported_module, 'NODE_CLASS_MAPPINGS'):
if hasattr(imported_module, "NODE_CLASS_MAPPINGS"):
NODE_CLASS_MAPPINGS.update(imported_module.NODE_CLASS_MAPPINGS)
print(f"✅ 成功加载模块: {file_name}")
if hasattr(imported_module, 'NODE_DISPLAY_NAME_MAPPINGS'):
print(f"[PD_NODE] Loaded module: {file_name}")
if hasattr(imported_module, "NODE_DISPLAY_NAME_MAPPINGS"):
NODE_DISPLAY_NAME_MAPPINGS.update(imported_module.NODE_DISPLAY_NAME_MAPPINGS)
return True
except Exception as e:
print(f"❌ 加载模块 {file_name} 失败: {e}")
print(f"[PD_NODE] Failed to load module {file_name}: {e}")
return False
# 动态扫描并加载 py/ 目录下的所有模块
py_dir = get_ext_dir("py")
if os.path.exists(py_dir):
files = os.listdir(py_dir)
for file in files:
if not file.endswith(".py") or file.startswith("_"):
for file_name in os.listdir(py_dir):
if not file_name.endswith(".py") or file_name.startswith("_"):
continue
module_name = os.path.splitext(file)[0]
module_path = f"{__name__}.py.{module_name}"
safe_import_module(module_path, file)
else:
print(f"警告: py目录不存在: {py_dir}")
# Web界面目录
module_name = os.path.splitext(file_name)[0]
module_path = f"{__name__}.py.{module_name}"
safe_import_module(module_path, file_name)
else:
print(f"[PD_NODE] Missing py directory: {py_dir}")
WEB_DIRECTORY = "./js"
# 显示加载信息
if NODE_CLASS_MAPPINGS:
print("=" * 50)
print("🎨 Comfyui_PDuse 节点套件加载完成")
print("[PD_NODE] Comfyui-PD_comfy-api-node loaded")
print("=" * 50)
print(f"📦 总计加载节点数量: {len(NODE_CLASS_MAPPINGS)}")
print("📋 已加载的节点:")
for name, display_name in NODE_DISPLAY_NAME_MAPPINGS.items():
print(f" • {display_name}")
print(f"[PD_NODE] Total nodes: {len(NODE_CLASS_MAPPINGS)}")
for node_name, display_name in NODE_DISPLAY_NAME_MAPPINGS.items():
print(f"[PD_NODE] {node_name} -> {display_name}")
print("=" * 50)
else:
print("⚠️ 警告: 没有找到任何可用的节点")
print("[PD_NODE] Warning: no nodes were loaded")
# 导出变量
__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", "WEB_DIRECTORY"]
py/IMAGE_GPT2_apikey.py
+211
View File
@@ -0,0 +1,211 @@
"""
PD: OpenAI GPT Image 2 Node (ComfyUI API Key)
"""
import time
import requests
import torch
from ._image_gpt_common import (
GPT_IMAGE_2_PRESET_SIZES,
calculate_token_price_2,
decode_response_images,
empty_image,
estimate_price_range_2,
format_rmb_range,
format_rmb_value,
mask_to_png_bytes,
resolve_gpt_image_2_size,
tensor_to_bytes,
validate_gpt_image_2_size,
)
class PDOpenAIGPTImage2APIKey:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"api_key": ("STRING", {
"default": "",
"multiline": False,
"placeholder": "Your ComfyUI API Key"
}),
"prompt": ("STRING", {
"multiline": True,
"default": "",
"tooltip": "Text prompt for GPT Image 2"
}),
"model": (["gpt-image-2"], {
"default": "gpt-image-2"
}),
"seed": ("INT", {
"default": 0,
"min": 0,
"max": 2**31 - 1,
"step": 1,
"tooltip": "not implemented yet in backend"
}),
"quality": (["auto", "low", "medium", "high"], {
"default": "auto",
"tooltip": "GPT Image 2 quality"
}),
"background": (["auto", "opaque"], {
"default": "auto",
"tooltip": "GPT Image 2 does not support transparent background"
}),
"size": (GPT_IMAGE_2_PRESET_SIZES, {
"default": "1024x1024",
"tooltip": "Preset output size"
}),
"custom_width": ("INT", {
"default": 0,
"min": 0,
"max": 3840,
"step": 16,
"tooltip": "Override width when both width and height are non-zero"
}),
"custom_height": ("INT", {
"default": 0,
"min": 0,
"max": 3840,
"step": 16,
"tooltip": "Override height when both width and height are non-zero"
}),
"num_images": ("INT", {
"default": 1,
"min": 1,
"max": 8,
"step": 1,
"tooltip": "How many images to generate"
}),
},
"optional": {
"image": ("IMAGE", {
"tooltip": "Optional reference image for image editing."
}),
"mask": ("MASK", {
"tooltip": "Optional mask for inpainting (white areas will be replaced)"
}),
},
"hidden": {
"unique_id": "UNIQUE_ID",
}
}
RETURN_TYPES = ("IMAGE", "STRING")
RETURN_NAMES = ("image", "info")
FUNCTION = "generate_image"
CATEGORY = "PD_Tools/Image_Generation"
def generate_image(self, api_key, prompt, model, seed, quality, background, size, custom_width, custom_height, num_images, unique_id, image=None, mask=None):
if not api_key or not api_key.strip():
error_msg = "Error: ComfyUI API Key is required"
print(f"[IMAGE_GPT2_APIKEY] {error_msg}")
return (empty_image(), error_msg)
try:
resolved_size = validate_gpt_image_2_size(size, custom_width, custom_height)
except Exception as e:
return (empty_image(), f"Error: {str(e)}")
api_key = api_key.strip()
is_edit_mode = image is not None
base_url = "https://api.comfy.org/proxy/openai/images/edits" if is_edit_mode else "https://api.comfy.org/proxy/openai/images/generations"
start_time = time.time()
try:
if is_edit_mode:
files = {
"image": ("image.png", tensor_to_bytes(image), "image/png")
}
if mask is not None:
files["mask"] = ("mask.png", mask_to_png_bytes(mask), "image/png")
data = {
"model": model,
"prompt": prompt,
"n": str(num_images),
"size": resolved_size,
"quality": quality,
"background": background,
"moderation": "low"
}
headers = {
"X-API-KEY": api_key,
"Accept": "application/json",
"User-Agent": "ComfyUI-PD-Node/2.1"
}
response = requests.post(base_url, headers=headers, files=files, data=data, timeout=300)
else:
payload = {
"model": model,
"prompt": prompt,
"n": num_images,
"size": resolved_size,
"quality": quality,
"background": background,
"moderation": "low"
}
headers = {
"X-API-KEY": api_key,
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "ComfyUI-PD-Node/2.1"
}
response = requests.post(base_url, headers=headers, json=payload, timeout=300)
if response.status_code != 200:
error_msg = f"API Error {response.status_code}: {response.text[:200]}"
print(f"[IMAGE_GPT2_APIKEY] {error_msg}")
return (empty_image(), error_msg)
result = response.json()
image_tensors = decode_response_images(result)
if not image_tensors:
error_msg = "Error: Failed to process output images"
print(f"[IMAGE_GPT2_APIKEY] {error_msg}")
return (empty_image(), error_msg)
final_image = torch.cat(image_tensors, dim=0)
duration = time.time() - start_time
min_cost, max_cost = estimate_price_range_2(quality, num_images)
token_cost = calculate_token_price_2(num_images)
size_note = resolve_gpt_image_2_size(size, custom_width, custom_height)
info_str = (
f"模型: {model}\n"
f"显示名称: OpenAI GPT Image 2\n"
f"模式: {'编辑' if is_edit_mode else '生成'}\n"
f"状态: 成功\n"
f"耗时: {duration:.2f} 秒\n"
f"图片数量: {len(image_tensors)}\n"
f"尺寸预设: {size}\n"
f"实际尺寸: {size_note}\n"
f"质量: {quality}\n"
f"背景: {background}\n"
f"预估价格区间(USD): ~${min_cost:.3f}-${max_cost:.3f}\n"
f"预估价格区间(RMB): {format_rmb_range(min_cost, max_cost)}\n"
f"Token 公式估算(USD): ${token_cost:.4f}\n"
f"Token 公式估算(RMB): {format_rmb_value(token_cost)}"
)
return (final_image, info_str)
except requests.exceptions.Timeout:
return (empty_image(), "Error: Request timeout (300s)")
except requests.exceptions.ConnectionError as e:
return (empty_image(), f"Error: Connection failed - {str(e)}")
except Exception as e:
import traceback
traceback.print_exc()
return (empty_image(), f"Error: {str(e)}")
NODE_CLASS_MAPPINGS = {
"PDOpenAIGPTImage2APIKey": PDOpenAIGPTImage2APIKey
}
NODE_DISPLAY_NAME_MAPPINGS = {
"PDOpenAIGPTImage2APIKey": "PD: GPT Image 2 (apikey)"
}
py/IMAGE_GPT2_authtoken.py
+209
View File
@@ -0,0 +1,209 @@
"""
PD: OpenAI GPT Image 2 Node (ComfyUI Auth Token)
"""
import time
import requests
import torch
from ._image_gpt_common import (
GPT_IMAGE_2_PRESET_SIZES,
calculate_token_price_2,
decode_response_images,
empty_image,
estimate_price_range_2,
format_rmb_range,
format_rmb_value,
mask_to_png_bytes,
resolve_gpt_image_2_size,
tensor_to_bytes,
validate_gpt_image_2_size,
)
class PDOpenAIGPTImage2AuthToken:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"auth_token": ("STRING", {
"default": "",
"multiline": False,
"placeholder": "Paste your Auth Token here"
}),
"prompt": ("STRING", {
"multiline": True,
"default": "",
"tooltip": "Describe what you want to generate or edit"
}),
"model": (["gpt-image-2"], {
"default": "gpt-image-2",
"tooltip": "Locked GPT Image 2 model"
}),
"quality": (["auto", "low", "medium", "high"], {
"default": "auto",
"tooltip": "GPT Image 2 quality"
}),
"background": (["auto", "opaque"], {
"default": "auto",
"tooltip": "GPT Image 2 does not support transparent background"
}),
"size": (GPT_IMAGE_2_PRESET_SIZES, {
"default": "1024x1024",
"tooltip": "Preset output size"
}),
"custom_width": ("INT", {
"default": 0,
"min": 0,
"max": 3840,
"step": 16,
"tooltip": "Override width when both width and height are non-zero"
}),
"custom_height": ("INT", {
"default": 0,
"min": 0,
"max": 3840,
"step": 16,
"tooltip": "Override height when both width and height are non-zero"
}),
"num_images": ("INT", {
"default": 1,
"min": 1,
"max": 8,
"step": 1,
"tooltip": "How many images to generate"
}),
"seed": ("INT", {
"default": 0,
"min": 0,
"max": 0xFFFFFFFFFFFFFFFF
}),
},
"optional": {
"image": ("IMAGE", {"default": None, "tooltip": "Input image for editing"}),
"mask": ("MASK", {"default": None, "tooltip": "Mask for inpainting"}),
},
"hidden": {
"unique_id": "UNIQUE_ID",
}
}
RETURN_TYPES = ("IMAGE", "STRING")
RETURN_NAMES = ("image", "info")
FUNCTION = "generate_image"
CATEGORY = "PD_Tools/Image_Generation"
def generate_image(self, auth_token, prompt, model, quality, background, size, custom_width, custom_height, num_images, seed, unique_id, image=None, mask=None):
if not auth_token or not auth_token.strip():
error_msg = "Error: Auth Token is required"
print(f"[IMAGE_GPT2_AUTHTOKEN] {error_msg}")
return (empty_image(), error_msg)
try:
resolved_size = validate_gpt_image_2_size(size, custom_width, custom_height)
except Exception as e:
return (empty_image(), f"Error: {str(e)}")
auth_token = auth_token.strip()
if auth_token.lower().startswith("bearer "):
auth_token = auth_token[7:].strip()
base_url = "https://api.comfy.org"
is_edit_mode = image is not None
start_time = time.time()
try:
if is_edit_mode:
url = f"{base_url}/proxy/openai/images/edits"
files = {
"image": ("image.png", tensor_to_bytes(image), "image/png")
}
if mask is not None:
files["mask"] = ("mask.png", mask_to_png_bytes(mask), "image/png")
data = {
"model": model,
"prompt": prompt,
"n": str(num_images),
"size": resolved_size,
"quality": quality,
"background": background,
"moderation": "low"
}
headers = {
"Authorization": f"Bearer {auth_token}",
"User-Agent": "ComfyUI-PD-Node/2.1"
}
response = requests.post(url, headers=headers, files=files, data=data, timeout=300)
else:
url = f"{base_url}/proxy/openai/images/generations"
payload = {
"model": model,
"prompt": prompt,
"n": num_images,
"size": resolved_size,
"quality": quality,
"background": background,
"moderation": "low"
}
headers = {
"Authorization": f"Bearer {auth_token}",
"Content-Type": "application/json",
"User-Agent": "ComfyUI-PD-Node/2.1"
}
response = requests.post(url, headers=headers, json=payload, timeout=300)
if response.status_code != 200:
error_msg = f"API Error {response.status_code}: {response.text[:200]}"
print(f"[IMAGE_GPT2_AUTHTOKEN] {error_msg}")
return (empty_image(), error_msg)
result = response.json()
image_tensors = decode_response_images(result)
if not image_tensors:
error_msg = "Error: Failed to process output images"
print(f"[IMAGE_GPT2_AUTHTOKEN] {error_msg}")
return (empty_image(), error_msg)
final_image = torch.cat(image_tensors, dim=0)
duration = time.time() - start_time
min_cost, max_cost = estimate_price_range_2(quality, num_images)
token_cost = calculate_token_price_2(num_images)
size_note = resolve_gpt_image_2_size(size, custom_width, custom_height)
info_str = (
f"模型: {model}\n"
f"显示名称: OpenAI GPT Image 2\n"
f"模式: {'编辑' if is_edit_mode else '生成'}\n"
f"状态: 成功\n"
f"耗时: {duration:.2f} 秒\n"
f"图片数量: {len(image_tensors)}\n"
f"尺寸预设: {size}\n"
f"实际尺寸: {size_note}\n"
f"质量: {quality}\n"
f"背景: {background}\n"
f"预估价格区间(USD): ~${min_cost:.3f}-${max_cost:.3f}\n"
f"预估价格区间(RMB): {format_rmb_range(min_cost, max_cost)}\n"
f"Token 公式估算(USD): ${token_cost:.4f}\n"
f"Token 公式估算(RMB): {format_rmb_value(token_cost)}"
)
return (final_image, info_str)
except requests.exceptions.Timeout:
return (empty_image(), "Error: Request timeout (300s)")
except requests.exceptions.ConnectionError as e:
return (empty_image(), f"Error: Connection failed - {str(e)}")
except Exception as e:
import traceback
traceback.print_exc()
return (empty_image(), f"Error: {str(e)}")
NODE_CLASS_MAPPINGS = {
"PDOpenAIGPTImage2AuthToken": PDOpenAIGPTImage2AuthToken
}
NODE_DISPLAY_NAME_MAPPINGS = {
"PDOpenAIGPTImage2AuthToken": "PD: GPT Image 2 (comfyui_AuthToken)"
}
py/IMAGE_GPT_apikey.py
+63 -186
View File
@@ -1,86 +1,47 @@
"""
PD: OpenAI GPT Image Node (ComfyUI API Key)
使用 ComfyUI API Key 调用 ComfyUI 代理 API
参数和官方节点完全一样
PD: OpenAI GPT Image 1 & 1.5 Node (ComfyUI API Key)
"""
import time
import torch
import requests
import time
import io
import base64
import numpy as np
from PIL import Image
def tensor_to_bytes(tensor, max_pixels=2048*2048):
"""将 Tensor 转换为 PNG BytesIO，并进行下采样"""
if tensor.ndim == 4:
tensor = tensor[0]
# 下采样大图片
height, width = tensor.shape[:2]
current_pixels = height * width
if current_pixels > max_pixels:
scale = (max_pixels / current_pixels) ** 0.5
new_height = int(height * scale)
new_width = int(width * scale)
# 使用 PIL 进行高质量缩放
array = np.clip(255. * tensor.cpu().numpy(), 0, 255).astype(np.uint8)
image = Image.fromarray(array)
image = image.resize((new_width, new_height), Image.Resampling.LANCZOS)
else:
array = np.clip(255. * tensor.cpu().numpy(), 0, 255).astype(np.uint8)
image = Image.fromarray(array)
buffered = io.BytesIO()
image.save(buffered, format="PNG")
buffered.seek(0)
return buffered
def load_image_to_tensor(image_data):
"""将图片数据（URL 或 bytes）转换为 Tensor"""
try:
if isinstance(image_data, str) and image_data.startswith("http"):
response = requests.get(image_data, timeout=30)
response.raise_for_status()
img = Image.open(io.BytesIO(response.content))
else:
img = Image.open(io.BytesIO(image_data))
img = img.convert("RGB")
image_np = np.array(img).astype(np.float32) / 255.0
return torch.from_numpy(image_np).unsqueeze(0)
except Exception as e:
print(f"[IMAGE_GPT_APIKEY] Error loading image: {e}")
return torch.zeros((1, 512, 512, 3))
from ._image_gpt_common import (
calculate_token_price_1_or_1_5,
decode_response_images,
empty_image,
estimate_price_range_1_or_1_5,
format_rmb_range,
format_rmb_value,
load_image_to_tensor,
mask_to_png_bytes,
tensor_to_bytes,
)
class PDOpenAIGPTImageAPIKey:
"""
PD: OpenAI GPT Image (API Key)
使用 ComfyUI API Key 调用 ComfyUI 代理 API
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"api_key": ("STRING", {
"default": "",
"multiline": False,
"default": "",
"multiline": False,
"placeholder": "Your ComfyUI API Key"
}),
"prompt": ("STRING", {
"multiline": True,
"default": "",
"tooltip": "Text prompt for GPT Image"
"multiline": True,
"default": "",
"tooltip": "Text prompt for GPT Image 1 or 1.5"
}),
"model": (["gpt-image-1", "gpt-image-1.5"], {
"default": "gpt-image-1.5",
"tooltip": "GPT Image model version"
}),
"seed": ("INT", {
"default": 0,
"min": 0,
"default": 0,
"min": 0,
"max": 2**31 - 1,
"step": 1,
"tooltip": "not implemented yet in backend"
@@ -98,9 +59,9 @@ class PDOpenAIGPTImageAPIKey:
"tooltip": "Image size"
}),
"n": ("INT", {
"default": 1,
"min": 1,
"max": 8,
"default": 1,
"min": 1,
"max": 8,
"step": 1,
"tooltip": "How many images to generate"
}),
@@ -112,9 +73,6 @@ class PDOpenAIGPTImageAPIKey:
"mask": ("MASK", {
"tooltip": "Optional mask for inpainting (white areas will be replaced)"
}),
"model": (["gpt-image-1", "gpt-image-1.5"], {
"default": "gpt-image-1"
}),
},
"hidden": {
"unique_id": "UNIQUE_ID",
@@ -126,57 +84,25 @@ class PDOpenAIGPTImageAPIKey:
FUNCTION = "generate_image"
CATEGORY = "PD_Tools/Image_Generation"
def generate_image(self, api_key, prompt, seed, quality, background, size, n, unique_id, image=None, mask=None, model="gpt-image-1"):
"""生成或编辑图片"""
# 验证 API Key
def generate_image(self, api_key, prompt, model, seed, quality, background, size, n, unique_id, image=None, mask=None):
if not api_key or not api_key.strip():
error_msg = "Error: ComfyUI API Key is required"
print(f"[IMAGE_GPT_APIKEY] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
# 清理 API Key
api_key = api_key.strip()
return (empty_image(), error_msg)
# 判断模式：编辑 vs 生成
api_key = api_key.strip()
is_edit_mode = image is not None
# ComfyUI 代理 API URL
if is_edit_mode:
base_url = "https://api.comfy.org/proxy/openai/images/edits"
else:
base_url = "https://api.comfy.org/proxy/openai/images/generations"
base_url = "https://api.comfy.org/proxy/openai/images/edits" if is_edit_mode else "https://api.comfy.org/proxy/openai/images/generations"
start_time = time.time()
try:
if is_edit_mode:
# 图片编辑模式 - 使用 multipart/form-data
# 准备文件
files = {}
img_bytes = tensor_to_bytes(image)
files['image'] = ('image.png', img_bytes, 'image/png')
# 处理 mask
files = {
"image": ("image.png", tensor_to_bytes(image), "image/png")
}
if mask is not None:
m_tensor = mask
if m_tensor.ndim == 3:
m_tensor = m_tensor[0]
mask_h, mask_w = m_tensor.shape
rgba = np.zeros((mask_h, mask_w, 4), dtype=np.uint8)
mask_np = m_tensor.cpu().numpy()
alpha = ((1.0 - mask_np) * 255).astype(np.uint8)
rgba[..., 3] = alpha
mask_pil = Image.fromarray(rgba)
mask_buffer = io.BytesIO()
mask_pil.save(mask_buffer, format="PNG")
mask_buffer.seek(0)
files['mask'] = ('mask.png', mask_buffer, 'image/png')
files["mask"] = ("mask.png", mask_to_png_bytes(mask), "image/png")
# 表单数据 - multipart/form-data 模式下所有值都应该是字符串
data = {
"model": model,
"prompt": prompt,
@@ -186,22 +112,13 @@ class PDOpenAIGPTImageAPIKey:
"background": background,
"moderation": "low"
}
# 请求头
headers = {
"X-API-KEY": api_key,
"Accept": "application/json",
"User-Agent": "ComfyUI-PD-Node/2.0"
"User-Agent": "ComfyUI-PD-Node/2.1"
}
print(f"[IMAGE_GPT_APIKEY] Sending edit request")
print(f"[IMAGE_GPT_APIKEY] Model: {model}, Size: {size}, Quality: {quality}")
print(f"[IMAGE_GPT_APIKEY] Image size: {image.shape}, Mask: {'Yes' if mask is not None else 'No'}")
response = requests.post(base_url, headers=headers, files=files, data=data, timeout=300)
else:
# 文生图模式 - 使用 JSON
payload = {
"model": model,
"prompt": prompt,
@@ -209,104 +126,64 @@ class PDOpenAIGPTImageAPIKey:
"size": size,
"quality": quality,
"background": background,
"response_format": "url"
"moderation": "low"
}
headers = {
"X-API-KEY": api_key,
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "ComfyUI-PD-Node/2.0"
"User-Agent": "ComfyUI-PD-Node/2.1"
}
print(f"[IMAGE_GPT_APIKEY] Sending generation request")
print(f"[IMAGE_GPT_APIKEY] Model: {model}, Size: {size}, Quality: {quality}")
print(f"[IMAGE_GPT_APIKEY] Prompt length: {len(prompt)} chars")
response = requests.post(base_url, headers=headers, json=payload, timeout=300)
# 检查响应
if response.status_code != 200:
error_msg = f"API Error {response.status_code}: {response.text[:200]}"
print(f"[IMAGE_GPT_APIKEY] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
return (empty_image(), error_msg)
result = response.json()
data_list = result.get("data", [])
if not data_list:
error_msg = "Error: No image data in response"
print(f"[IMAGE_GPT_APIKEY] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
# 提取图片
image_tensors = []
for item in data_list:
if "url" in item:
img_tensor = load_image_to_tensor(item["url"])
image_tensors.append(img_tensor)
elif "b64_json" in item:
img_data = base64.b64decode(item["b64_json"])
img_tensor = load_image_to_tensor(img_data)
image_tensors.append(img_tensor)
image_tensors = decode_response_images(result)
if not image_tensors:
error_msg = "Error: Failed to process output images"
print(f"[IMAGE_GPT_APIKEY] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
return (empty_image(), error_msg)
final_image = torch.cat(image_tensors, dim=0)
# 计算耗时和成本
end_time = time.time()
duration = end_time - start_time
# 估算成本
cost_map = {
"gpt-image-1": {"low": 0.02, "medium": 0.04, "high": 0.08},
"gpt-image-1.5": {"low": 0.03, "medium": 0.06, "high": 0.12}
}
cost = cost_map.get(model, {}).get(quality, 0.04)
total_cost = cost * n
duration = time.time() - start_time
min_cost, max_cost = estimate_price_range_1_or_1_5(model, quality, n)
token_cost = calculate_token_price_1_or_1_5(model, n)
info_str = (
f"Model: {model}\n"
f"Mode: {'Edit' if is_edit_mode else 'Generate'}\n"
f"Status: Success\n"
f"Time: {duration:.2f}s\n"
f"Images: {len(image_tensors)}\n"
f"Size: {size}\n"
f"Quality: {quality}\n"
f"Background: {background}\n"
f"Est. Cost: ${total_cost:.3f}"
f"模型: {model}\n"
f"显示名称: OpenAI GPT Image 1 & 1.5\n"
f"模式: {'编辑' if is_edit_mode else '生成'}\n"
f"状态: 成功\n"
f"耗时: {duration:.2f} 秒\n"
f"图片数量: {len(image_tensors)}\n"
f"尺寸: {size}\n"
f"质量: {quality}\n"
f"背景: {background}\n"
f"预估价格区间(USD): ${min_cost:.3f}-${max_cost:.3f}\n"
f"预估价格区间(RMB): {format_rmb_range(min_cost, max_cost)}\n"
f"Token 公式估算(USD): ${token_cost:.4f}\n"
f"Token 公式估算(RMB): {format_rmb_value(token_cost)}"
)
print(f"[IMAGE_GPT_APIKEY] Success! Generated {len(image_tensors)} images in {duration:.2f}s")
return (final_image, info_str)
except requests.exceptions.Timeout:
error_msg = "Error: Request timeout (300s)"
print(f"[IMAGE_GPT_APIKEY] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
return (empty_image(), "Error: Request timeout (300s)")
except requests.exceptions.ConnectionError as e:
error_msg = f"Error: Connection failed - {str(e)}"
print(f"[IMAGE_GPT_APIKEY] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
return (empty_image(), f"Error: Connection failed - {str(e)}")
except Exception as e:
error_msg = f"Error: {str(e)}"
print(f"[IMAGE_GPT_APIKEY] {error_msg}")
import traceback
traceback.print_exc()
return (torch.zeros((1, 512, 512, 3)), error_msg)
return (empty_image(), f"Error: {str(e)}")
# 节点注册
NODE_CLASS_MAPPINGS = {
"PDOpenAIGPTImageAPIKey": PDOpenAIGPTImageAPIKey
}
NODE_DISPLAY_NAME_MAPPINGS = {
"PDOpenAIGPTImageAPIKey": "PD: GPT Image (apikey)"
"PDOpenAIGPTImageAPIKey": "PD: GPT Image 1 & 1.5 (apikey)"
}
py/IMAGE_GPT_authtoken.py
+57 -185
View File
@@ -1,85 +1,41 @@
"""
PD: OpenAI GPT Image Node (ComfyUI Auth Token)
基于 gemini_image_api 的 Auth Token 认证方式
支持文生图和图片编辑
PD: OpenAI GPT Image 1 & 1.5 Node (ComfyUI Auth Token)
"""
import torch
import requests
import time
import io
import base64
import numpy as np
from PIL import Image
import requests
import torch
def tensor_to_bytes(tensor, max_pixels=2048*2048):
"""将 Tensor 转换为 PNG BytesIO，并进行下采样"""
if tensor.ndim == 4:
tensor = tensor[0]
# 下采样大图片
height, width = tensor.shape[:2]
current_pixels = height * width
if current_pixels > max_pixels:
scale = (max_pixels / current_pixels) ** 0.5
new_height = int(height * scale)
new_width = int(width * scale)
# 使用 PIL 进行高质量缩放
array = np.clip(255. * tensor.cpu().numpy(), 0, 255).astype(np.uint8)
image = Image.fromarray(array)
image = image.resize((new_width, new_height), Image.Resampling.LANCZOS)
else:
array = np.clip(255. * tensor.cpu().numpy(), 0, 255).astype(np.uint8)
image = Image.fromarray(array)
buffered = io.BytesIO()
image.save(buffered, format="PNG")
buffered.seek(0)
return buffered
def load_image_to_tensor(image_data):
"""将图片数据（URL 或 bytes）转换为 Tensor"""
try:
if isinstance(image_data, str) and image_data.startswith("http"):
response = requests.get(image_data, timeout=30)
response.raise_for_status()
img = Image.open(io.BytesIO(response.content))
else:
img = Image.open(io.BytesIO(image_data))
img = img.convert("RGB")
image_np = np.array(img).astype(np.float32) / 255.0
return torch.from_numpy(image_np).unsqueeze(0)
except Exception as e:
print(f"[IMAGE_GPT_AUTHTOKEN] Error loading image: {e}")
return torch.zeros((1, 512, 512, 3))
from ._image_gpt_common import (
calculate_token_price_1_or_1_5,
decode_response_images,
empty_image,
estimate_price_range_1_or_1_5,
format_rmb_range,
format_rmb_value,
mask_to_png_bytes,
tensor_to_bytes,
)
class PDOpenAIGPTImageAuthToken:
"""
PD: OpenAI GPT Image (Auth Token)
使用 ComfyUI Auth Token 调用 GPT Image API
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"auth_token": ("STRING", {
"default": "",
"multiline": False,
"default": "",
"multiline": False,
"placeholder": "Paste your Auth Token here"
}),
"prompt": ("STRING", {
"multiline": True,
"default": "",
"multiline": True,
"default": "",
"tooltip": "Describe what you want to generate or edit"
}),
"model": (["gpt-image-1", "gpt-image-1.5"], {
"default": "gpt-image-1",
"default": "gpt-image-1.5",
"tooltip": "GPT Image model version"
}),
"quality": (["low", "medium", "high"], {
@@ -88,22 +44,22 @@ class PDOpenAIGPTImageAuthToken:
}),
"background": (["auto", "opaque", "transparent"], {
"default": "auto",
"tooltip": "Background transparency (GPT Image only)"
"tooltip": "Background transparency"
}),
"size": (["auto", "1024x1024", "1024x1536", "1536x1024"], {
"default": "auto",
"tooltip": "Image size"
}),
"n": ("INT", {
"default": 1,
"min": 1,
"max": 8,
"default": 1,
"min": 1,
"max": 8,
"step": 1,
"tooltip": "Number of images to generate"
}),
"seed": ("INT", {
"default": 0,
"min": 0,
"default": 0,
"min": 0,
"max": 0xFFFFFFFFFFFFFFFF
}),
},
@@ -122,57 +78,28 @@ class PDOpenAIGPTImageAuthToken:
CATEGORY = "PD_Tools/Image_Generation"
def generate_image(self, auth_token, prompt, model, quality, background, size, n, seed, unique_id, image=None, mask=None):
"""生成或编辑图片"""
# 验证 Token
if not auth_token or not auth_token.strip():
error_msg = "Error: Auth Token is required"
print(f"[IMAGE_GPT_AUTHTOKEN] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
# 清理 Token
return (empty_image(), error_msg)
auth_token = auth_token.strip()
if auth_token.lower().startswith("bearer "):
auth_token = auth_token[7:].strip()
# API 基础地址
base_url = "https://api.comfy.org"
# 判断模式：编辑 vs 生成
is_edit_mode = image is not None
start_time = time.time()
try:
if is_edit_mode:
# 图片编辑模式 - 使用 multipart/form-data
url = f"{base_url}/proxy/openai/images/edits"
# 准备文件
files = {}
img_bytes = tensor_to_bytes(image)
files['image'] = ('image.png', img_bytes, 'image/png')
# 处理 mask
files = {
"image": ("image.png", tensor_to_bytes(image), "image/png")
}
if mask is not None:
m_tensor = mask
if m_tensor.ndim == 3:
m_tensor = m_tensor[0]
mask_h, mask_w = m_tensor.shape
rgba = np.zeros((mask_h, mask_w, 4), dtype=np.uint8)
mask_np = m_tensor.cpu().numpy()
alpha = ((1.0 - mask_np) * 255).astype(np.uint8)
rgba[..., 3] = alpha
mask_pil = Image.fromarray(rgba)
mask_buffer = io.BytesIO()
mask_pil.save(mask_buffer, format="PNG")
mask_buffer.seek(0)
files['mask'] = ('mask.png', mask_buffer, 'image/png')
files["mask"] = ("mask.png", mask_to_png_bytes(mask), "image/png")
# 表单数据 - multipart/form-data 模式下所有值都应该是字符串
data = {
"model": model,
"prompt": prompt,
@@ -182,23 +109,13 @@ class PDOpenAIGPTImageAuthToken:
"background": background,
"moderation": "low"
}
# 请求头
headers = {
"Authorization": f"Bearer {auth_token}",
"User-Agent": "ComfyUI-PD-Node/2.0"
"User-Agent": "ComfyUI-PD-Node/2.1"
}
print(f"[IMAGE_GPT_AUTHTOKEN] Sending edit request to {url}")
print(f"[IMAGE_GPT_AUTHTOKEN] Model: {model}, Size: {size}, Quality: {quality}, Background: {background}")
print(f"[IMAGE_GPT_AUTHTOKEN] Image size: {image.shape}, Mask: {'Yes' if mask is not None else 'No'}")
response = requests.post(url, headers=headers, files=files, data=data, timeout=300)
else:
# 文生图模式 - 使用 JSON
url = f"{base_url}/proxy/openai/images/generations"
payload = {
"model": model,
"prompt": prompt,
@@ -206,108 +123,63 @@ class PDOpenAIGPTImageAuthToken:
"size": size,
"quality": quality,
"background": background,
"moderation": "low",
"response_format": "url"
"moderation": "low"
}
headers = {
"Authorization": f"Bearer {auth_token}",
"Content-Type": "application/json",
"User-Agent": "ComfyUI-PD-Node/2.0"
"User-Agent": "ComfyUI-PD-Node/2.1"
}
print(f"[IMAGE_GPT_AUTHTOKEN] Sending generation request to {url}")
print(f"[IMAGE_GPT_AUTHTOKEN] Model: {model}, Size: {size}, Quality: {quality}, Background: {background}")
print(f"[IMAGE_GPT_AUTHTOKEN] Prompt length: {len(prompt)} chars")
response = requests.post(url, headers=headers, json=payload, timeout=300)
# 检查响应
if response.status_code != 200:
error_msg = f"API Error {response.status_code}: {response.text[:200]}"
print(f"[IMAGE_GPT_AUTHTOKEN] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
return (empty_image(), error_msg)
result = response.json()
data_list = result.get("data", [])
if not data_list:
error_msg = "Error: No image data in response"
print(f"[IMAGE_GPT_AUTHTOKEN] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
# 提取图片
image_tensors = []
for item in data_list:
if "url" in item:
img_tensor = load_image_to_tensor(item["url"])
image_tensors.append(img_tensor)
elif "b64_json" in item:
img_data = base64.b64decode(item["b64_json"])
img_tensor = load_image_to_tensor(img_data)
image_tensors.append(img_tensor)
image_tensors = decode_response_images(result)
if not image_tensors:
error_msg = "Error: Failed to process output images"
print(f"[IMAGE_GPT_AUTHTOKEN] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
return (empty_image(), error_msg)
final_image = torch.cat(image_tensors, dim=0)
# 计算耗时和成本
end_time = time.time()
duration = end_time - start_time
# 估算成本
if model == "gpt-image-1":
# GPT Image 1: $10/1M input tokens, $40/1M output tokens
# 估算：约 1000 tokens input, 2000 tokens output per image
cost = (1000 * 10 + 2000 * 40) / 1_000_000
elif model == "gpt-image-1.5":
# GPT Image 1.5: $8/1M input tokens, $32/1M output tokens
cost = (1000 * 8 + 2000 * 32) / 1_000_000
else:
cost = 0.10 # 默认估算
total_cost = cost * n
duration = time.time() - start_time
min_cost, max_cost = estimate_price_range_1_or_1_5(model, quality, n)
token_cost = calculate_token_price_1_or_1_5(model, n)
info_str = (
f"Model: {model}\n"
f"Mode: {'Edit' if is_edit_mode else 'Generate'}\n"
f"Status: Success\n"
f"Time: {duration:.2f}s\n"
f"Images: {len(image_tensors)}\n"
f"Size: {size}\n"
f"Quality: {quality}\n"
f"Background: {background}\n"
f"Est. Cost: ${total_cost:.4f}"
f"模型: {model}\n"
f"显示名称: OpenAI GPT Image 1 & 1.5\n"
f"模式: {'编辑' if is_edit_mode else '生成'}\n"
f"状态: 成功\n"
f"耗时: {duration:.2f} 秒\n"
f"图片数量: {len(image_tensors)}\n"
f"尺寸: {size}\n"
f"质量: {quality}\n"
f"背景: {background}\n"
f"预估价格区间(USD): ${min_cost:.3f}-${max_cost:.3f}\n"
f"预估价格区间(RMB): {format_rmb_range(min_cost, max_cost)}\n"
f"Token 公式估算(USD): ${token_cost:.4f}\n"
f"Token 公式估算(RMB): {format_rmb_value(token_cost)}"
)
print(f"[IMAGE_GPT_AUTHTOKEN] Success! Generated {len(image_tensors)} images in {duration:.2f}s")
return (final_image, info_str)
except requests.exceptions.Timeout:
error_msg = "Error: Request timeout (300s)"
print(f"[IMAGE_GPT_AUTHTOKEN] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
return (empty_image(), "Error: Request timeout (300s)")
except requests.exceptions.ConnectionError as e:
error_msg = f"Error: Connection failed - {str(e)}"
print(f"[IMAGE_GPT_AUTHTOKEN] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
return (empty_image(), f"Error: Connection failed - {str(e)}")
except Exception as e:
error_msg = f"Error: {str(e)}"
print(f"[IMAGE_GPT_AUTHTOKEN] {error_msg}")
import traceback
traceback.print_exc()
return (torch.zeros((1, 512, 512, 3)), error_msg)
return (empty_image(), f"Error: {str(e)}")
# 节点注册
NODE_CLASS_MAPPINGS = {
"PDOpenAIGPTImageAuthToken": PDOpenAIGPTImageAuthToken
}
NODE_DISPLAY_NAME_MAPPINGS = {
"PDOpenAIGPTImageAuthToken": "PD: GPT Image (comfyui_AuthToken)"
"PDOpenAIGPTImageAuthToken": "PD: GPT Image 1 & 1.5 (comfyui_AuthToken)"
}
py/_image_gpt_common.py
+189
View File
@@ -0,0 +1,189 @@
import base64
import io
import numpy as np
import requests
import torch
from PIL import Image
GPT_IMAGE_EDIT_MAX_PIXELS = 2048 * 2048
GPT_IMAGE_2_PRESET_SIZES = [
"1024x1024",
"1536x1024",
"1024x1536",
"2048x2048",
"2048x1152",
"3840x2160",
"2160x3840",
"auto",
]
# USD/CNY reference rate used for RMB estimation in node info output.
# Based on the most recent publicly indexed April 2026 daily spot pages we could verify:
# 2026-04-03: 1 USD ≈ 6.88154 CNY
USD_TO_CNY_RATE = 6.8815
def empty_image():
return torch.zeros((1, 512, 512, 3))
def tensor_to_bytes(tensor, max_pixels=GPT_IMAGE_EDIT_MAX_PIXELS):
if tensor.ndim == 4:
tensor = tensor[0]
height, width = tensor.shape[:2]
current_pixels = height * width
if current_pixels > max_pixels:
scale = (max_pixels / current_pixels) ** 0.5
new_height = max(1, int(height * scale))
new_width = max(1, int(width * scale))
array = np.clip(255.0 * tensor.cpu().numpy(), 0, 255).astype(np.uint8)
image = Image.fromarray(array)
image = image.resize((new_width, new_height), Image.Resampling.LANCZOS)
else:
array = np.clip(255.0 * tensor.cpu().numpy(), 0, 255).astype(np.uint8)
image = Image.fromarray(array)
buffered = io.BytesIO()
image.save(buffered, format="PNG")
buffered.seek(0)
return buffered
def load_image_to_tensor(image_data):
try:
if isinstance(image_data, str) and image_data.startswith("http"):
response = requests.get(image_data, timeout=30)
response.raise_for_status()
img = Image.open(io.BytesIO(response.content))
else:
img = Image.open(io.BytesIO(image_data))
img = img.convert("RGB")
image_np = np.array(img).astype(np.float32) / 255.0
return torch.from_numpy(image_np).unsqueeze(0)
except Exception:
return empty_image()
def mask_to_png_bytes(mask):
mask_tensor = mask
if mask_tensor.ndim == 3:
mask_tensor = mask_tensor[0]
mask_h, mask_w = mask_tensor.shape
rgba = np.zeros((mask_h, mask_w, 4), dtype=np.uint8)
mask_np = mask_tensor.cpu().numpy()
alpha = ((1.0 - mask_np) * 255).astype(np.uint8)
rgba[..., 3] = alpha
mask_pil = Image.fromarray(rgba)
mask_buffer = io.BytesIO()
mask_pil.save(mask_buffer, format="PNG")
mask_buffer.seek(0)
return mask_buffer
def decode_response_images(result):
data_list = result.get("data", [])
image_tensors = []
for item in data_list:
if "url" in item:
img_tensor = load_image_to_tensor(item["url"])
image_tensors.append(img_tensor)
elif "b64_json" in item:
img_data = base64.b64decode(item["b64_json"])
img_tensor = load_image_to_tensor(img_data)
image_tensors.append(img_tensor)
return image_tensors
def estimate_price_range_1_or_1_5(model, quality, n):
ranges = {
"gpt-image-1": {
"low": (0.011, 0.016),
"medium": (0.042, 0.063),
"high": (0.167, 0.250),
},
"gpt-image-1.5": {
"low": (0.009, 0.013),
"medium": (0.034, 0.050),
"high": (0.133, 0.200),
},
}
model_ranges = ranges.get(model, ranges["gpt-image-1"])
min_cost, max_cost = model_ranges.get(quality, model_ranges["medium"])
return (min_cost * n, max_cost * n)
def calculate_token_price_1_or_1_5(model, n):
if model == "gpt-image-1":
cost = (1000 * 10 + 2000 * 40) / 1_000_000
else:
cost = (1000 * 8 + 2000 * 32) / 1_000_000
return cost * n
def estimate_price_range_2(quality, n):
ranges = {
"auto": (0.005, 0.250),
"low": (0.005, 0.030),
"medium": (0.041, 0.120),
"high": (0.165, 0.250),
}
min_cost, max_cost = ranges.get(quality, ranges["auto"])
return (min_cost * n, max_cost * n)
def calculate_token_price_2(n):
cost = (1000 * 8 + 2000 * 30) / 1_000_000
return cost * n
def resolve_gpt_image_2_size(size, custom_width, custom_height):
if custom_width > 0 and custom_height > 0:
return f"{custom_width}x{custom_height}"
return size
def usd_to_cny(usd_amount):
return usd_amount * USD_TO_CNY_RATE
def format_rmb_range(min_usd, max_usd):
return f"约￥{usd_to_cny(min_usd):.2f}-￥{usd_to_cny(max_usd):.2f}"
def format_rmb_value(usd_amount):
return f"约￥{usd_to_cny(usd_amount):.2f}"
def validate_gpt_image_2_size(size, custom_width=0, custom_height=0):
resolved_size = resolve_gpt_image_2_size(size, custom_width, custom_height)
if resolved_size == "auto":
return resolved_size
try:
width_str, height_str = resolved_size.lower().split("x")
width = int(width_str)
height = int(height_str)
except Exception as exc:
raise ValueError(f"Invalid size: {resolved_size}") from exc
if width > 3840 or height > 3840:
raise ValueError("Width and height must not exceed 3840")
if width % 16 != 0 or height % 16 != 0:
raise ValueError("Width and height must be multiples of 16")
shorter_edge = min(width, height)
longer_edge = max(width, height)
if shorter_edge == 0 or (longer_edge / shorter_edge) > 3:
raise ValueError("Aspect ratio must not exceed 3:1")
total_pixels = width * height
if total_pixels < 655360 or total_pixels > 8294400:
raise ValueError("Total pixels must be between 655360 and 8294400")
return resolved_size