Initial commit from comfyui-PD_comfy api node

This commit is contained in:
pandy
2026-03-08 11:10:30 +08:00
commit e13e1aeede
17 changed files with 3849 additions and 0 deletions
+4
View File
@@ -0,0 +1,4 @@
1. 你是一个经验丰富的python开发者,擅长制作commfyui的节点,现在需要创建一个新的节点
2. 确保输入输出图像或者遮罩的张量形状分别为B H W C ,B H W
3. 需要包含导入,节点注册等构建节点所需的完整代码
BIN
View File
Binary file not shown.
+182
View File
@@ -0,0 +1,182 @@
# ComfyUI PD Comfy API Node
ComfyUI 自定义节点套件,集成 Gemini 和 GPT 图像生成 API,支持两种认证方式。
## ✨ 功能特性
- 🎨 **多模型支持**:Gemini 2.5 Flash Image、GPT Image 1/1.5
- 🔑 **双认证方式**:支持 API Key 和 Auth Token 两种调用方式
- 💰 **成本追踪**:实时显示 API 调用成本
- 🖼️ **图像处理**:支持文生图、图生图、图像编辑
- ⚙️ **灵活配置**:多种宽高比、质量、分辨率选项
## 📦 安装方法
```bash
cd ComfyUI/custom_nodes
git clone https://github.com/7BEII/comfyui-PD_comfy-api-node.git
pip install -r comfyui-PD_comfy-api-node/requirements.txt
```
重启 ComfyUI 后,在节点菜单的 `PD_Tools/Image_Generation` 分类中找到节点。
## 🔑 两种认证方式
### 方式一:API Key(推荐)
使用 ComfyUI 官方 API Key,简单直接。
**获取方法**:
1. 访问 [ComfyUI API](https://www.comfy.org/api-keys)
2. 登录并创建 API Key
3. 复制 Key 并粘贴到节点中
![API Key 节点示例](image/comfyui-node-apikey.png)
**优点**:
- ✅ 官方支持,稳定可靠
- ✅ 配置简单,一个 Key 搞定
- ✅ 支持所有模型
### 方式二:Auth Token
使用浏览器 Auth Token,适合高级用户。
**获取方法**:
1. 安装油猴脚本(Tampermonkey):具体B站有教程:小熊猫pandy
2. 安装 Token 抓取脚本
3. 访问 ComfyUI 网站并登录
4. 脚本会自动抓取 Token
5. 复制 Token 并粘贴到节点中
![Auth Token 节点示例](image/comfyui-node-auth.png)
**优点**:
- ✅ 无需申请 API Key
- ✅ 使用现有账号权限
- ✅ 适合测试和开发
## 📝 更新日志
### v2.0.0 -2026.0112
- 新增 Auth Token 认证方式
- 新增 GPT Image 节点
- 优化成本计算
- 改进错误处理
### v1.0.0
- 初始版本
- 支持 Gemini 图像生成
- 支持 API Key 认证
## 📋 节点说明
### 1. PD: Gemini Image (apikey)
使用 API Key 调用 Gemini 图像生成。
**主要参数**:
- `api_key`:ComfyUI API Key
- `prompt`:图像生成提示词
- `model`:模型选择(gemini-2.5-flash-image / gemini-2.5-flash-image-preview)
- `aspect_ratio`:宽高比(auto / 1:1 / 16:9 / 9:16 等)
- `image_ref`:参考图像(可选)
**输出**:
- `image`:生成的图像
- `info`:成本和状态信息
### 2. PD: Gemini Image (comfyui_AuthToken)
使用 Auth Token 调用 Gemini 图像生成。
**主要参数**:
- `auth_token`:ComfyUI Auth Token
- `prompt`:图像生成提示词
- `model`:模型选择
- `aspect_ratio`:宽高比
- `resolution`:分辨率(auto / 1K / 2K / 4K)
- `image_ref`:参考图像(可选)
**输出**:
- `image`:生成的图像
- `info`:成本和状态信息
### 3. PD: GPT Image (apikey)
使用 API Key 调用 OpenAI GPT 图像生成。
**主要参数**:
- `api_key`:ComfyUI API Key
- `prompt`:图像生成提示词
- `model`:模型选择(gpt-image-1 / gpt-image-1.5)
- `quality`:质量(low / medium / high)
- `size`:尺寸(auto / 1024x1024 / 1024x1536 / 1536x1024)
- `background`:背景(auto / opaque / transparent)
- `image`:输入图像(可选,用于编辑)
- `mask`:遮罩(可选,用于局部编辑)
**输出**:
- `image`:生成的图像
- `info`:成本和状态信息
**特性**:
- 支持文生图
- 支持图生图编辑
- 支持遮罩局部编辑
## 💡 使用示例
### 示例 1:使用 Gemini 生成图像(API Key)
```
1. 添加 "PD: Gemini Image (apikey)" 节点
2. 输入 API Key
3. 输入提示词:"A futuristic city with flying cars"
4. 选择宽高比:16:9
5. 连接输出到预览节点
6. 执行工作流
```
### 示例 2:使用 GPT 编辑图像
```
1. 添加 "PD: GPT Image (apikey)" 节点
2. 输入 API Key
3. 连接输入图像到 image 端口
4. 输入编辑提示词:"Add a sunset sky"
5. 设置质量为 high
6. 执行工作流
```
## ⚠️ 注意事项
1. **API Key 安全**:请妥善保管 API Key,不要分享给他人
2. **成本控制**:注意查看成本信息,控制 API 调用次数
3. **网络连接**:需要稳定的网络连接
4. **生成时间**:高质量图像可能需要较长时间
## 🐛 故障排除
### 节点未显示
- 检查是否正确安装到 `custom_nodes` 目录
- 重启 ComfyUI
- 查看控制台错误信息
### API 调用失败
- 验证 API Key 或 Auth Token 是否正确
- 检查网络连接
- 查看控制台详细错误信息
### 图像格式错误
- 确保输入图像格式为 ComfyUI 标准格式(B H W C)
- 检查图像通道数是否为 3 (RGB)
## 📄 许可证
MIT License
## 🤝 贡献
欢迎提交 Issue 和 Pull Request!
---
+74
View File
@@ -0,0 +1,74 @@
"""
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__)
if subpath is not None:
dir = os.path.join(dir, subpath)
return os.path.abspath(dir)
def safe_import_module(module_path, file_name):
"""安全导入模块,避免导入错误导致崩溃"""
try:
imported_module = importlib.import_module(module_path)
# 检查模块是否包含所需的映射
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'):
NODE_DISPLAY_NAME_MAPPINGS.update(imported_module.NODE_DISPLAY_NAME_MAPPINGS)
return True
except Exception as e:
print(f"❌ 加载模块 {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("_"):
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界面目录
WEB_DIRECTORY = "./js"
# 显示加载信息
if NODE_CLASS_MAPPINGS:
print("=" * 50)
print("🎨 Comfyui_PDuse 节点套件加载完成")
print("=" * 50)
print(f"📦 总计加载节点数量: {len(NODE_CLASS_MAPPINGS)}")
print("📋 已加载的节点:")
for name, display_name in NODE_DISPLAY_NAME_MAPPINGS.items():
print(f" • {display_name}")
print("=" * 50)
else:
print("⚠️ 警告: 没有找到任何可用的节点")
# 导出变量
__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", "WEB_DIRECTORY"]
+590
View File
@@ -0,0 +1,590 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Git 自动同步工具
用法: python gitrun.py [命令]
命令说明:
push - 强制推送本地到远程(直接覆盖,保留历史记录)
pull - 强制拉取远程到本地(直接覆盖,保留历史记录)
sync - 智能同步(先拉取合并,再推送)
log - 查看提交历史并支持快速回退版本
history - 查看完整提交历史
config - 配置远程仓库地址
重要说明:
• push/pull 都会保留完整的 Git 历史记录,可以随时回退
• push 是纯粹的本地覆盖远程,不会先拉取远程内容
• pull 是纯粹的远程覆盖本地,不会保留本地未推送的修改
• sync 才会进行智能合并
• log 命令可以查看历史并直接回退版本(输入数字选择,0或回车跳过)
• 所有历史提交都保存在 .git 目录中,永不丢失
"""
import os
import sys
import subprocess
import json
from pathlib import Path
from datetime import datetime
CONFIG_FILE = Path.home() / '.gitrun_config.json'
class GitSyncTool:
def __init__(self):
self.config = self.load_config()
self.repo_path = os.getcwd()
def load_config(self):
"""加载配置文件"""
if CONFIG_FILE.exists():
with open(CONFIG_FILE, 'r', encoding='utf-8') as f:
return json.load(f)
return {}
def save_config(self):
"""保存配置文件"""
with open(CONFIG_FILE, 'w', encoding='utf-8') as f:
json.dump(self.config, f, indent=2, ensure_ascii=False)
def run_command(self, cmd, check=True, silent=False):
"""执行命令并返回结果"""
try:
# Windows 上使用 GBK 编码,其他系统使用 UTF-8
encoding = 'gbk' if sys.platform == 'win32' else 'utf-8'
result = subprocess.run(
cmd,
shell=True,
cwd=self.repo_path,
capture_output=True,
text=True,
encoding=encoding,
errors='ignore' # 忽略无法解码的字符
)
if check and result.returncode != 0:
if not silent:
print(f"❌ 命令执行失败: {cmd}")
if result.stderr:
print(f"错误信息: {result.stderr}")
return None
return result
except Exception as e:
if not silent:
print(f"❌ 执行命令时出错: {e}")
return None
def is_git_repo(self):
"""检查当前目录是否是 Git 仓库"""
result = self.run_command("git rev-parse --is-inside-work-tree", check=False, silent=True)
return result and result.returncode == 0
def get_current_branch(self):
"""获取当前分支名"""
result = self.run_command("git branch --show-current", silent=True)
return result.stdout.strip() if result else None
def has_changes(self):
"""检查是否有未提交的更改"""
result = self.run_command("git status --porcelain", silent=True)
return bool(result and result.stdout.strip())
def get_remote_url(self):
"""获取远程仓库地址"""
result = self.run_command("git remote get-url origin", check=False, silent=True)
return result.stdout.strip() if result and result.returncode == 0 else None
def config_remote(self):
"""配置远程仓库地址"""
print("🔧 配置远程仓库地址")
# 检查是否是 Git 仓库,如果不是则初始化
if not self.is_git_repo():
print("⚠️ 当前目录不是 Git 仓库")
init = input("是否初始化为 Git 仓库? (y/n): ").strip().lower()
if init == 'y':
result = self.run_command("git init")
if result:
print("✅ Git 仓库初始化成功")
else:
print("❌ Git 仓库初始化失败")
return
else:
print("❌ 操作已取消")
return
current_url = self.get_remote_url()
if current_url:
print(f"当前远程仓库: {current_url}")
change = input("是否修改? (y/n): ").strip().lower()
if change != 'y':
return
remote_url = input("请输入远程仓库地址 (如: https://github.com/username/repo.git): ").strip()
if not remote_url:
print("❌ 远程仓库地址不能为空")
return
# 检查是否已有 origin
if current_url:
result = self.run_command(f"git remote set-url origin {remote_url}")
else:
result = self.run_command(f"git remote add origin {remote_url}")
if result:
print(f"✅ 远程仓库配置成功: {remote_url}")
# 保存到配置文件
repo_name = os.path.basename(self.repo_path)
if repo_name not in self.config:
self.config[repo_name] = {}
self.config[repo_name]['remote_url'] = remote_url
self.save_config()
# 询问是否设置默认分支
print("\n💡 提示: 首次推送前需要设置默认分支")
set_branch = input("是否设置默认分支为 main? (y/n): ").strip().lower()
if set_branch == 'y':
# 检查当前分支
current_branch = self.get_current_branch()
if not current_branch:
# 创建初始提交
print("\n📝 创建初始提交...")
self.run_command("git add .")
self.run_command('git commit -m "Initial commit" --allow-empty')
self.run_command("git branch -M main")
print("✅ 默认分支已设置为 main")
elif current_branch != 'main':
self.run_command("git branch -M main")
print("✅ 分支已重命名为 main")
else:
print("✅ 当前分支已是 main")
def commit_changes(self, message=None):
"""提交本地更改"""
if not self.has_changes():
print("ℹ️ 没有需要提交的更改")
return True
print("\n📝 检测到以下更改:")
self.run_command("git status --short")
if message is None:
message = input("\n请输入提交信息: ").strip()
if not message:
print("❌ 提交信息不能为空")
return False
# 添加所有更改
print("\n📦 正在暂存所有更改...")
result = self.run_command("git add -A")
if not result:
return False
# 提交更改
print(f"💾 正在提交: {message}")
result = self.run_command(f'git commit -m "{message}"')
if result:
print(f"✅ 提交成功")
return True
return False
def force_push(self):
"""纯粹的本地覆盖远程(不拉取,直接强制推送)"""
print("🚀 强制推送本地到远程")
if not self.is_git_repo():
print("❌ 当前目录不是 Git 仓库")
return
branch = self.get_current_branch()
if not branch:
print("❌ 无法获取当前分支")
return
print(f"📍 当前分支: {branch}")
# 检查远程仓库
remote_url = self.get_remote_url()
if not remote_url:
print("⚠️ 未配置远程仓库")
self.config_remote()
remote_url = self.get_remote_url()
if not remote_url:
return
print(f"🌐 远程仓库: {remote_url}")
# 提交本地更改
print()
if not self.commit_changes(): # 改为交互式提交,不自动
print("❌ 提交失败,推送中止")
return
# 强制推送(不拉取远程)
print(f"\n⬆️ 正在强制推送到远程 ({branch})...")
result = self.run_command(f"git push -f origin {branch}")
if result:
print(f"\n✅ 推送成功!")
else:
print("\n❌ 推送失败")
def force_pull(self):
"""纯粹的远程覆盖本地(不保留本地未推送的修改)"""
print("⬇️ 强制拉取远程到本地")
if not self.is_git_repo():
print("❌ 当前目录不是 Git 仓库")
return
branch = self.get_current_branch()
if not branch:
print("❌ 无法获取当前分支")
return
print(f"📍 当前分支: {branch}")
# 检查远程仓库
remote_url = self.get_remote_url()
if not remote_url:
print("⚠️ 未配置远程仓库")
self.config_remote()
return
print(f"🌐 远程仓库: {remote_url}")
# 检查本地更改
if self.has_changes():
print("\n⚠️ 检测到未提交的本地更改,这些更改将会被丢弃")
self.run_command("git status --short")
# 先获取远程最新信息
print(f"\n📡 正在获取远程仓库信息...")
self.run_command("git fetch origin")
# 强制重置到远程版本
print(f"⬇️ 正在强制拉取远程 ({branch})...")
self.run_command(f"git reset --hard origin/{branch}")
self.run_command("git clean -fd")
print(f"\n✅ 拉取成功!")
def smart_sync(self):
"""智能同步(先拉取合并,再推送)"""
print("🔄 智能同步")
if not self.is_git_repo():
print("❌ 当前目录不是 Git 仓库")
return
branch = self.get_current_branch()
if not branch:
print("❌ 无法获取当前分支")
return
print(f"📍 当前分支: {branch}")
# 检查远程仓库
remote_url = self.get_remote_url()
if not remote_url:
print("⚠️ 未配置远程仓库")
self.config_remote()
return
print(f"🌐 远程仓库: {remote_url}")
# 提交本地更改
print()
if self.has_changes():
if not self.commit_changes(): # 改为交互式提交
print("❌ 提交失败")
return
# 拉取远程更改
print(f"\n⬇️ 正在拉取远程更新 ({branch})...")
result = self.run_command(f"git pull origin {branch}", check=False)
if result and result.returncode != 0:
# 如果有冲突
if "conflict" in result.stderr.lower() or "conflict" in result.stdout.lower():
print("\n⚠️ 检测到合并冲突!")
print("\n冲突文件:")
self.run_command("git diff --name-only --diff-filter=U")
choice = input("\n选择操作:\n1. 保留本地版本\n2. 保留远程版本\n3. 手动解决后继续\n请选择 (1/2/3): ").strip()
if choice == '1':
self.run_command("git checkout --ours .")
self.run_command("git add -A")
self.run_command('git commit --no-edit')
print("✅ 已保留本地版本")
elif choice == '2':
self.run_command("git checkout --theirs .")
self.run_command("git add -A")
self.run_command('git commit --no-edit')
print("✅ 已保留远程版本")
else:
print("ℹ️ 请手动解决冲突后执行: git add . && git commit")
return
else:
print("❌ 拉取失败")
return
else:
print("✅ 拉取成功")
# 推送到远程
print(f"\n⬆️ 正在推送到远程 ({branch})...")
push_result = self.run_command(f"git push origin {branch}")
if push_result:
print(f"\n✅ 同步成功!")
else:
print("⚠️ 推送失败或无新内容需要推送")
def show_log(self, num=10):
"""查看提交历史并支持直接回退"""
print(f"📋 最近 {num} 次提交历史")
if not self.is_git_repo():
print("❌ 当前目录不是 Git 仓库")
return
branch = self.get_current_branch()
print(f"📍 当前分支: {branch}\n")
# 使用 git log 的简单格式,然后解析
result = self.run_command(f'git log -{num} --format="COMMIT_START%nHASH:%H%nSUBJECT:%s%nAUTHOR:%an%nDATE:%ar%nCOMMIT_END"')
if not result:
return
commits = result.stdout.strip().split('COMMIT_START')
commit_hashes = []
idx = 1
for commit in commits:
if not commit.strip():
continue
lines = commit.strip().split('\n')
commit_info = {}
for line in lines:
if ':' in line:
key, value = line.split(':', 1)
commit_info[key] = value.strip()
if 'HASH' in commit_info:
hash_full = commit_info['HASH']
hash_short = hash_full[:7]
subject = commit_info.get('SUBJECT', '')
date = commit_info.get('DATE', '')
commit_hashes.append(hash_full)
print(f"{idx}. [{hash_short}] {subject} -- {date}")
idx += 1
# 询问是否回退
print()
choice = input("是否回退版本? (输入数字1-10回退,直接回车或输入0跳过): ").strip()
if not choice or choice == '0':
print("✅ 未进行回退")
return
# 判断是数字还是哈希值
if choice.isdigit():
idx = int(choice) - 1
if 0 <= idx < len(commit_hashes):
commit_hash = commit_hashes[idx][:7]
else:
print(f"❌ 无效的数字,请输入 1-{len(commit_hashes)}")
return
else:
commit_hash = choice
# 验证提交是否存在
verify = self.run_command(f"git cat-file -t {commit_hash}", check=False, silent=True)
if not verify or verify.returncode != 0:
print(f"❌ 提交 {commit_hash} 不存在")
return
# 显示目标提交信息
print(f"\n📍 即将回退到: {commit_hash}")
# 选择回退方式
print("\n回退方式:")
print("1. 软回退 (保留工作区和暂存区的更改)")
print("2. 混合回退 (保留工作区更改,清空暂存区) [推荐]")
print("3. 硬回退 (完全回退,丢弃所有更改)")
mode_choice = input("\n请选择 (1/2/3,直接回车默认选2): ").strip()
if not mode_choice:
mode_choice = '2'
mode_map = {
'1': ('--soft', '软回退'),
'2': ('--mixed', '混合回退'),
'3': ('--hard', '硬回退')
}
if mode_choice not in mode_map:
print("❌ 无效选择")
return
mode, mode_name = mode_map[mode_choice]
# 执行回退
print(f"\n⏮️ 正在执行{mode_name}...")
result = self.run_command(f"git reset {mode} {commit_hash}")
if result:
print(f"\n✅ {mode_name}成功!")
print(f"📍 已回退到: {commit_hash}\n")
self.run_command("git status --short")
else:
print(f"\n❌ 回退失败")
def show_full_history(self):
"""查看完整提交历史"""
print("📚 完整提交历史")
if not self.is_git_repo():
print("❌ 当前目录不是 Git 仓库")
return
branch = self.get_current_branch()
print(f"📍 当前分支: {branch}\n")
# 显示完整历史(简化格式以兼容 Windows)
self.run_command("git log --graph --oneline --decorate --all")
def rollback(self):
"""回退到指定版本"""
print("⏮️ 版本回退")
if not self.is_git_repo():
print("❌ 当前目录不是 Git 仓库")
return
# 显示最近10次提交
print("📋 最近10次提交:\n")
result = self.run_command("git log -10 --pretty=format:%H|%s|%an|%ar")
if not result:
return
lines = result.stdout.strip().split('\n')
commits = []
for i, line in enumerate(lines, 1):
if '|' in line:
hash_full, subject, author, date = line.split('|', 3)
hash_short = hash_full[:7]
commits.append(hash_full)
print(f"{i}. [{hash_short}] {subject}")
print(f" 作者: {author} | 时间: {date}\n")
print()
# 获取用户输入
choice = input("请输入要回退到的版本 (输入数字1-10,或输入提交哈希值,q 取消): ").strip()
if not choice or choice.lower() == 'q':
print("❌ 操作已取消")
return
# 判断是数字还是哈希值
if choice.isdigit():
idx = int(choice) - 1
if 0 <= idx < len(commits):
commit_hash = commits[idx][:7]
else:
print(f"❌ 无效的数字,请输入 1-{len(commits)}")
return
else:
commit_hash = choice
# 验证提交是否存在
verify = self.run_command(f"git cat-file -t {commit_hash}", check=False, silent=True)
if not verify or verify.returncode != 0:
print(f"❌ 提交 {commit_hash} 不存在")
return
# 显示目标提交信息
print(f"\n📍 即将回退到:")
self.run_command(f"git log -1 {commit_hash} --pretty=format:[%h] %s%n作者: %an | 时间: %ar")
# 选择回退方式
print("\n\n回退方式:")
print("1. 软回退 (保留工作区和暂存区的更改)")
print("2. 混合回退 (保留工作区更改,清空暂存区) [推荐]")
print("3. 硬回退 (完全回退,丢弃所有更改)")
mode_choice = input("\n请选择 (1/2/3,输入 q 取消): ").strip()
if mode_choice.lower() == 'q':
print("❌ 操作已取消")
return
mode_map = {
'1': ('--soft', '软回退'),
'2': ('--mixed', '混合回退'),
'3': ('--hard', '硬回退')
}
if mode_choice not in mode_map:
print("❌ 无效选择")
return
mode, mode_name = mode_map[mode_choice]
# 执行回退
print(f"\n⏮️ 正在执行{mode_name}...")
result = self.run_command(f"git reset {mode} {commit_hash}")
if result:
print(f"\n✅ {mode_name}成功!")
print(f"📍 已回退到: {commit_hash}\n")
self.run_command("git status --short")
else:
print(f"\n❌ 回退失败")
def main():
tool = GitSyncTool()
if len(sys.argv) < 2:
print(__doc__)
return
command = sys.argv[1].lower()
if command == 'push':
tool.force_push()
elif command == 'pull':
tool.force_pull()
elif command == 'sync':
tool.smart_sync()
elif command == 'log':
tool.show_log(5)
elif command == 'history':
tool.show_full_history()
elif command == 'rollback':
tool.rollback()
elif command == 'config':
tool.config_remote()
else:
print(f"❌ 未知命令: {command}")
print(__doc__)
if __name__ == '__main__':
main()
Binary file not shown.

After

Width:  |  Height:  |  Size: 77 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 113 KiB

+312
View File
@@ -0,0 +1,312 @@
"""
PD: OpenAI GPT Image Node (ComfyUI API Key)
使用 ComfyUI API Key 调用 ComfyUI 代理 API
参数和官方节点完全一样
"""
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))
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,
"placeholder": "Your ComfyUI API Key"
}),
"prompt": ("STRING", {
"multiline": True,
"default": "",
"tooltip": "Text prompt for GPT Image"
}),
"seed": ("INT", {
"default": 0,
"min": 0,
"max": 2**31 - 1,
"step": 1,
"tooltip": "not implemented yet in backend"
}),
"quality": (["low", "medium", "high"], {
"default": "low",
"tooltip": "Image quality, affects cost and generation time"
}),
"background": (["auto", "opaque", "transparent"], {
"default": "auto",
"tooltip": "Return image with or without background"
}),
"size": (["auto", "1024x1024", "1024x1536", "1536x1024"], {
"default": "auto",
"tooltip": "Image size"
}),
"n": ("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)"
}),
"model": (["gpt-image-1", "gpt-image-1.5"], {
"default": "gpt-image-1"
}),
},
"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, seed, quality, background, size, n, unique_id, image=None, mask=None, model="gpt-image-1"):
"""生成或编辑图片"""
# 验证 API Key
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()
# 判断模式:编辑 vs 生成
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"
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
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')
# 表单数据 - multipart/form-data 模式下所有值都应该是字符串
data = {
"model": model,
"prompt": prompt,
"n": str(n),
"size": size,
"quality": quality,
"background": background,
"moderation": "low"
}
# 请求头
headers = {
"X-API-KEY": api_key,
"Accept": "application/json",
"User-Agent": "ComfyUI-PD-Node/2.0"
}
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,
"n": n,
"size": size,
"quality": quality,
"background": background,
"response_format": "url"
}
headers = {
"X-API-KEY": api_key,
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "ComfyUI-PD-Node/2.0"
}
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)
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)
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)
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
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}"
)
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)
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)
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)
# 节点注册
NODE_CLASS_MAPPINGS = {
"PDOpenAIGPTImageAPIKey": PDOpenAIGPTImageAPIKey
}
NODE_DISPLAY_NAME_MAPPINGS = {
"PDOpenAIGPTImageAPIKey": "PD: GPT Image (apikey)"
}
+313
View File
@@ -0,0 +1,313 @@
"""
PD: OpenAI GPT Image Node (ComfyUI Auth Token)
基于 gemini_image_api 的 Auth Token 认证方式
支持文生图和图片编辑
"""
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_AUTHTOKEN] Error loading image: {e}")
return torch.zeros((1, 512, 512, 3))
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,
"placeholder": "Paste your Auth Token here"
}),
"prompt": ("STRING", {
"multiline": True,
"default": "",
"tooltip": "Describe what you want to generate or edit"
}),
"model": (["gpt-image-1", "gpt-image-1.5"], {
"default": "gpt-image-1",
"tooltip": "GPT Image model version"
}),
"quality": (["low", "medium", "high"], {
"default": "low",
"tooltip": "Image quality, affects cost and generation time"
}),
"background": (["auto", "opaque", "transparent"], {
"default": "auto",
"tooltip": "Background transparency (GPT Image only)"
}),
"size": (["auto", "1024x1024", "1024x1536", "1536x1024"], {
"default": "auto",
"tooltip": "Image size"
}),
"n": ("INT", {
"default": 1,
"min": 1,
"max": 8,
"step": 1,
"tooltip": "Number of 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, 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
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
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')
# 表单数据 - multipart/form-data 模式下所有值都应该是字符串
data = {
"model": model,
"prompt": prompt,
"n": str(n),
"size": size,
"quality": quality,
"background": background,
"moderation": "low"
}
# 请求头
headers = {
"Authorization": f"Bearer {auth_token}",
"User-Agent": "ComfyUI-PD-Node/2.0"
}
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,
"n": n,
"size": size,
"quality": quality,
"background": background,
"moderation": "low",
"response_format": "url"
}
headers = {
"Authorization": f"Bearer {auth_token}",
"Content-Type": "application/json",
"User-Agent": "ComfyUI-PD-Node/2.0"
}
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)
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)
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)
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
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}"
)
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)
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)
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)
# 节点注册
NODE_CLASS_MAPPINGS = {
"PDOpenAIGPTImageAuthToken": PDOpenAIGPTImageAuthToken
}
NODE_DISPLAY_NAME_MAPPINGS = {
"PDOpenAIGPTImageAuthToken": "PD: GPT Image (comfyui_AuthToken)"
}
+279
View File
@@ -0,0 +1,279 @@
"""
PD: Nano Banana 2 (ComfyUI Key)
基于 Google Gemini 3.1 Flash Image 模型的图像生成/编辑节点
支持进度反馈、Token 成本计算、及其独立输出接口
"""
import torch
import base64
import json
import time
import uuid
import requests
import numpy as np
import io
from PIL import Image
from typing import Optional, List
from server import PromptServer
import comfy.utils
# 默认系统提示词
GEMINI_IMAGE_SYS_PROMPT = (
"You are an expert image-generation engine. You must ALWAYS produce an image.\n"
"Interpret all user input—regardless of "
"format, intent, or abstraction—as literal visual directives for image composition.\n"
"If a prompt is conversational or lacks specific visual details, "
"you must creatively invent a concrete visual scenario that depicts the concept.\n"
"Prioritize generating the visual representation above any text, formatting, or conversational requests."
)
def tensor_to_base64(tensor):
"""将 Tensor 转换为 base64 字符串"""
if tensor.ndim == 4:
tensor = tensor[0]
array = np.clip(255. * tensor.cpu().numpy(), 0, 255).astype(np.uint8)
image = Image.fromarray(array)
buffered = io.BytesIO()
image.save(buffered, format="PNG")
return base64.b64encode(buffered.getvalue()).decode("utf-8")
def bytes_to_tensor(data):
"""将图片字节数据转换为 Tensor"""
try:
img = Image.open(io.BytesIO(data)).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"[PD_NB2] Image decode error: {e}")
return torch.zeros((1, 512, 512, 3))
class PDGeminiNanoBanana2:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"comfy_api_key": ("STRING", {
"default": "",
"multiline": False,
"placeholder": "Paste your ComfyUI API Key here"
}),
"prompt": ("STRING", {
"multiline": True,
"default": "A futuristic cityscape with flying cars and neon lights",
}),
"model": (["Nano Banana 2 (Gemini 3.1 Flash Image)"], {
"default": "Nano Banana 2 (Gemini 3.1 Flash Image)"
}),
"seed": ("INT", {
"default": 42,
"min": 0,
"max": 0x7FFFFFFF,
}),
"aspect_ratio": (["auto", "1:1", "2:3", "3:2", "3:4", "4:3", "4:5", "5:4", "9:16", "16:9", "21:9"], {
"default": "auto",
}),
"resolution": (["1K", "2K", "4K"], {
"default": "1K",
}),
"response_modalities": (["IMAGE", "IMAGE+TEXT"], {
"default": "IMAGE",
}),
"thinking_level": (["MINIMAL", "HIGH"], {
"default": "MINIMAL"
}),
},
"optional": {
"images": ("IMAGE", ),
"files": ("GEMINI_INPUT_FILES", ),
"system_prompt": ("STRING", {
"multiline": True,
"default": GEMINI_IMAGE_SYS_PROMPT,
}),
},
"hidden": {
"unique_id": "UNIQUE_ID",
}
}
RETURN_TYPES = ("IMAGE", "STRING", "STRING")
RETURN_NAMES = ("image", "info", "price")
FUNCTION = "generate_image"
CATEGORY = "PD_Tools/Image_Generation"
def generate_image(self, comfy_api_key, prompt, model, seed, aspect_ratio, resolution, response_modalities, thinking_level, unique_id, images=None, files=None, system_prompt=""):
pbar = comfy.utils.ProgressBar(100)
pbar.update(10)
if not comfy_api_key or not comfy_api_key.strip():
return (torch.zeros((1, 512, 512, 3)), "Error: Missing ComfyUI API Key", "$0.00")
api_key = comfy_api_key.strip()
api_model = "gemini-3.1-flash-image-preview"
# 发送进度文字到前端
self._send_status(unique_id, "Preparing tokens...")
# 构建请求内容
parts = [{"text": prompt}]
if images is not None:
for i in range(min(images.shape[0], 14)):
b64_str = tensor_to_base64(images[i])
parts.append({
"inlineData": {
"mimeType": "image/png",
"data": b64_str
}
})
# Files 输入处理
if files:
for file_part in files:
if hasattr(file_part, "model_dump"):
parts.append(file_part.model_dump(exclude_none=True))
else:
parts.append(file_part)
# 核心参数修正:匹配官方 Pydantic 模型的 camelCase 结构
# 且不发送 seed (官方 GeminiNanoBanana2.execute 并没有发送 seed 到 API)
image_config = {
"imageSize": resolution,
"imageOutputOptions": {"mimeType": "image/png"}
}
if aspect_ratio != "auto":
image_config["aspectRatio"] = aspect_ratio
generation_config = {
"responseModalities": (["IMAGE"] if response_modalities == "IMAGE" else ["TEXT", "IMAGE"]),
"imageConfig": image_config,
"thinkingConfig": {"thinkingLevel": thinking_level}
}
payload = {
"contents": [{
"role": "user",
"parts": parts
}],
"generationConfig": generation_config
}
if system_prompt and system_prompt.strip():
payload["systemInstruction"] = {
"parts": [{"text": system_prompt.strip()}]
}
headers = {
"X-API-KEY": api_key,
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "ComfyUI-PD-Node/2.0"
}
base_url = f"https://api.comfy.org/proxy/vertexai/gemini/{api_model}"
self._send_status(unique_id, "Sending request to Google...")
pbar.update(30)
start_time = time.time()
try:
# 打印发送的 Payload 调试 (可选)
# print(f"[PD_NB2] Payload: {json.dumps(payload)}")
response = requests.post(base_url, headers=headers, json=payload, timeout=120)
pbar.update(80)
if response.status_code != 200:
error_body = response.text
error_msg = f"API Error {response.status_code}: {error_body}"
self._send_status(unique_id, "Error occurred")
return (torch.zeros((1, 512, 512, 3)), error_msg, "N/A")
result = response.json()
image_tensors = []
text_parts = []
self._send_status(unique_id, "Processing response...")
candidates = result.get("candidates", [])
for cand in candidates:
content = cand.get("content", {})
parts_out = content.get("parts", [])
for part in parts_out:
inline_data = part.get("inline_data") or part.get("inlineData")
if inline_data:
data_b64 = inline_data.get("data")
if data_b64:
raw_data = base64.b64decode(data_b64)
image_tensors.append(bytes_to_tensor(raw_data))
if "text" in part and part["text"]:
text_parts.append(part["text"])
if not image_tensors:
return (torch.zeros((1, 512, 512, 3)), "Error: No images found in response", "N/A")
final_image = torch.cat(image_tensors, dim=0)
final_text = "\n".join(text_parts) if text_parts else "No description"
# 计算成本
cost_info = self._calculate_cost(result, api_model, time.time() - start_time)
pbar.update(100)
self._send_status(unique_id, "Success")
return (final_image, final_text, cost_info)
except Exception as e:
self._send_status(unique_id, "Exception caught")
return (torch.zeros((1, 512, 512, 3)), f"Error: {str(e)}", "N/A")
def _send_status(self, node_id, text):
"""发送状态文字到节点下方显示"""
try:
PromptServer.instance.send_sync("progress_text", {"text": text, "node": node_id})
except:
pass
def _calculate_cost(self, result, model, duration):
"""基于 API 响应计算成本"""
usage = result.get("usageMetadata", {})
prompt_tokens = usage.get("promptTokenCount", 0)
candidates_tokens = usage.get("candidatesTokenCount", 0)
details = usage.get("candidatesTokensDetails", [])
thoughts_tokens = usage.get("thoughtsTokenCount", 0)
# 价格配置 (3.1 Flash Image Preview)
prices = {"input": 0.5, "output_text": 3.0, "output_image": 60.0}
text_tokens = thoughts_tokens
img_tokens = 0
for d in details:
modality = d.get("modality")
count = d.get("tokenCount", 0)
if modality == "IMAGE":
img_tokens += count
else:
text_tokens += count
if not details and candidates_tokens > 0:
text_tokens += candidates_tokens
input_cost = (prompt_tokens / 1_000_000) * prices["input"]
text_cost = (text_tokens / 1_000_000) * prices["output_text"]
img_cost = (img_tokens / 1_000_000) * prices["output_image"]
total_cost = input_cost + text_cost + img_cost
info = [
f"Model: {model}",
f"Status: Success",
f"Time: {duration:.2f}s",
f"Tokens: In:{prompt_tokens:,} | Out Text:{text_tokens:,} | Out Img:{img_tokens:,}",
f"Est. Cost: ${total_cost:.6f}"
]
return "\n".join(info)
# 注册
NODE_CLASS_MAPPINGS = {"PDGeminiNanoBanana2": PDGeminiNanoBanana2}
NODE_DISPLAY_NAME_MAPPINGS = {"PDGeminiNanoBanana2": "PD: Nano Banana 2 (ComfyUI Key)"}
+284
View File
@@ -0,0 +1,284 @@
"""
PD: Gemini Image Generation Node (ComfyUI API Key)
使用 ComfyUI API Key 调用 ComfyUI 代理 API
参数和官方节点完全一样
"""
import torch
import requests
import time
import io
import base64
import numpy as np
from PIL import Image
# 默认系统提示词(来自 ComfyUI 官方 API 节点)
GEMINI_IMAGE_SYS_PROMPT = (
"You are an expert image-generation engine. You must ALWAYS produce an image.\n"
"Interpret all user input—regardless of "
"format, intent, or abstraction—as literal visual directives for image composition.\n"
"If a prompt is conversational or lacks specific visual details, "
"you must creatively invent a concrete visual scenario that depicts the concept.\n"
"Prioritize generating the visual representation above any text, formatting, or conversational requests."
)
def tensor_to_base64(tensor):
"""将 Tensor 转换为 base64 字符串"""
if tensor.ndim == 4:
tensor = tensor[0]
array = np.clip(255. * tensor.cpu().numpy(), 0, 255).astype(np.uint8)
image = Image.fromarray(array)
buffered = io.BytesIO()
image.save(buffered, format="PNG")
return base64.b64encode(buffered.getvalue()).decode("utf-8")
def bytes_to_tensor(data):
"""将图片字节数据转换为 Tensor"""
try:
img = Image.open(io.BytesIO(data)).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"[GEMINI_APIKEY] Image decode error: {e}")
return torch.zeros((1, 512, 512, 3))
class PDGeminiImageGenAPIKey:
"""
PD: Gemini Image Generation (API Key)
使用 ComfyUI API Key 调用 ComfyUI 代理 API
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"api_key": ("STRING", {
"default": "",
"multiline": False,
"placeholder": "Your ComfyUI API Key"
}),
"prompt": ("STRING", {
"multiline": True,
"default": "A futuristic city with flying cars",
"tooltip": "Describe what you want to generate"
}),
"model": ([
"gemini-2.5-flash-image",
"gemini-2.5-flash-image-preview"
], {
"default": "gemini-2.5-flash-image"
}),
"aspect_ratio": ([
"auto", "1:1", "2:3", "3:2", "3:4", "4:3", "4:5", "5:4", "9:16", "16:9", "21:9"
], {
"default": "auto"
}),
"response_modalities": (["IMAGE+TEXT", "IMAGE"], {
"default": "IMAGE+TEXT",
"tooltip": "Choose 'IMAGE' for image-only output, or 'IMAGE+TEXT' for both"
}),
"seed": ("INT", {
"default": 42,
"min": 0,
"max": 0xFFFFFFFFFFFFFFFF
}),
},
"optional": {
"image_ref": ("IMAGE", {
"tooltip": "Reference image(s) for image-to-image generation"
}),
"system_prompt": ("STRING", {
"multiline": True,
"default": GEMINI_IMAGE_SYS_PROMPT,
"tooltip": "Optional system instructions"
}),
},
"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, aspect_ratio, response_modalities, seed, unique_id, image_ref=None, system_prompt=""):
"""生成图片"""
# 验证 API Key
if not api_key or not api_key.strip():
error_msg = "Error: ComfyUI API Key is required"
print(f"[GEMINI_APIKEY] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
# 清理 API Key
api_key = api_key.strip()
# ComfyUI 代理 API URL
base_url = f"https://api.comfy.org/proxy/vertexai/gemini/{model}"
# 构建请求内容
parts = []
# 添加用户提示词
parts.append({"text": prompt})
# 添加参考图片(如果有)
if image_ref is not None:
for i in range(image_ref.shape[0]):
b64_str = tensor_to_base64(image_ref[i])
parts.append({
"inlineData": {
"mimeType": "image/png",
"data": b64_str
}
})
print(f"[GEMINI_APIKEY] Added reference image {i+1}")
# 生成配置
generation_config = {
"responseModalities": ["IMAGE"] if response_modalities == "IMAGE" else ["TEXT", "IMAGE"]
}
# 图片配置
if aspect_ratio != "auto":
generation_config["imageConfig"] = {
"aspectRatio": aspect_ratio
}
# 系统指令
system_instruction = None
if system_prompt and system_prompt.strip():
system_instruction = {
"parts": [{"text": system_prompt.strip()}]
}
# 完整请求体
payload = {
"contents": [{
"role": "user",
"parts": parts
}],
"generationConfig": generation_config
}
if system_instruction:
payload["systemInstruction"] = system_instruction
# 请求头
headers = {
"X-API-KEY": api_key,
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "ComfyUI-PD-Node/2.0"
}
start_time = time.time()
try:
print(f"[GEMINI_APIKEY] Sending request to {model}")
print(f"[GEMINI_APIKEY] Aspect Ratio: {aspect_ratio}")
response = requests.post(base_url, headers=headers, json=payload, timeout=120)
# 检查响应
if response.status_code != 200:
error_msg = f"API Error {response.status_code}: {response.text[:200]}"
print(f"[GEMINI_APIKEY] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
result = response.json()
# 提取图片和文本
image_tensors = []
text_parts = []
candidates = result.get("candidates", [])
if not candidates:
error_msg = "Error: No candidates in response"
print(f"[GEMINI_APIKEY] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
for cand in candidates:
content = cand.get("content", {})
parts = content.get("parts", [])
for part in parts:
# 提取图片
inline_data = part.get("inline_data") or part.get("inlineData")
if inline_data:
mime = inline_data.get("mime_type") or inline_data.get("mimeType")
data_b64 = inline_data.get("data")
if mime in ["image/png", "image/jpeg"] and data_b64:
raw_data = base64.b64decode(data_b64)
image_tensors.append(bytes_to_tensor(raw_data))
print(f"[GEMINI_APIKEY] Extracted image ({mime})")
# 提取文本
if "text" in part and part["text"]:
text_parts.append(part["text"])
if not image_tensors:
error_msg = f"Error: No images found in response"
print(f"[GEMINI_APIKEY] {error_msg}")
print(f"[GEMINI_APIKEY] Response: {str(result)[:500]}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
final_image = torch.cat(image_tensors, dim=0)
final_text = "\n".join(text_parts) if text_parts else "No text output"
# 计算耗时和成本
end_time = time.time()
duration = end_time - start_time
# 估算成本
cost_per_img = 0.04
total_cost = cost_per_img * len(image_tensors)
info_str = (
f"Model: {model}\n"
f"Status: Success\n"
f"Time: {duration:.2f}s\n"
f"Images: {len(image_tensors)}\n"
f"Aspect Ratio: {aspect_ratio}\n"
f"Est. Cost: ${total_cost:.4f}\n"
f"---\n"
f"{final_text}"
)
print(f"[GEMINI_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 (120s)"
print(f"[GEMINI_APIKEY] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
except requests.exceptions.ConnectionError as e:
error_msg = f"Error: Connection failed - {str(e)}"
print(f"[GEMINI_APIKEY] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
except Exception as e:
error_msg = f"Error: {str(e)}"
print(f"[GEMINI_APIKEY] {error_msg}")
import traceback
traceback.print_exc()
return (torch.zeros((1, 512, 512, 3)), error_msg)
# 节点注册
NODE_CLASS_MAPPINGS = {
"PDGeminiImageGenAPIKey": PDGeminiImageGenAPIKey
}
NODE_DISPLAY_NAME_MAPPINGS = {
"PDGeminiImageGenAPIKey": "PD: Gemini Image (apikey)"
}
+281
View File
@@ -0,0 +1,281 @@
"""
PD: Gemini Image Generation Node (ComfyUI Auth Token)
基于 gemini_image_api 的 Auth Token 认证方式
支持文生图和图生图
"""
import torch
import requests
import time
import io
import base64
import numpy as np
from PIL import Image
# 默认系统提示词(来自 ComfyUI 官方 API 节点)
GEMINI_IMAGE_SYS_PROMPT = (
"You are an expert image-generation engine. You must ALWAYS produce an image.\n"
"Interpret all user input—regardless of "
"format, intent, or abstraction—as literal visual directives for image composition.\n"
"If a prompt is conversational or lacks specific visual details, "
"you must creatively invent a concrete visual scenario that depicts the concept.\n"
"Prioritize generating the visual representation above any text, formatting, or conversational requests."
)
def tensor_to_base64(tensor):
"""将 Tensor 转换为 base64 字符串"""
if tensor.ndim == 4:
tensor = tensor[0]
array = np.clip(255. * tensor.cpu().numpy(), 0, 255).astype(np.uint8)
image = Image.fromarray(array)
buffered = io.BytesIO()
image.save(buffered, format="PNG")
return base64.b64encode(buffered.getvalue()).decode("utf-8")
def bytes_to_tensor(data):
"""将图片字节数据转换为 Tensor"""
try:
img = Image.open(io.BytesIO(data)).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"[GEMINI_AUTHTOKEN] Image decode error: {e}")
return torch.zeros((1, 512, 512, 3))
class PDGeminiImageGenAuthToken:
"""
PD: Gemini Image Generation (Auth Token)
使用 ComfyUI Auth Token 调用 Gemini Image API
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"auth_token": ("STRING", {
"default": "",
"multiline": False,
"placeholder": "Paste your Auth Token here"
}),
"prompt": ("STRING", {
"multiline": True,
"default": "A futuristic city with flying cars",
"tooltip": "Describe what you want to generate"
}),
"model": ([
"gemini-2.5-flash-image",
"gemini-2.5-flash-image-preview"
], {
"default": "gemini-2.5-flash-image"
}),
"aspect_ratio": ([
"auto", "1:1", "16:9", "9:16", "4:3", "3:4", "21:9", "9:21"
], {
"default": "auto"
}),
"resolution": (["auto", "1K", "2K", "4K"], {
"default": "1K"
}),
"seed": ("INT", {
"default": 0,
"min": 0,
"max": 0xFFFFFFFFFFFFFFFF
}),
},
"optional": {
"image_ref": ("IMAGE", {
"tooltip": "Reference image(s) for image-to-image generation"
}),
"system_prompt": ("STRING", {
"multiline": True,
"default": GEMINI_IMAGE_SYS_PROMPT,
"tooltip": "Optional system instructions"
}),
},
"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, aspect_ratio, resolution, seed, unique_id, image_ref=None, system_prompt=""):
"""生成图片"""
print(f"[GEMINI_AUTHTOKEN] ========== 开始执行 ==========")
print(f"[GEMINI_AUTHTOKEN] Prompt: {prompt[:50]}...")
print(f"[GEMINI_AUTHTOKEN] Model: {model}")
# 验证 Token
if not auth_token or not auth_token.strip():
error_msg = "Error: Auth Token is required"
print(f"[GEMINI_AUTHTOKEN] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
# 清理 Token
auth_token = auth_token.strip()
if auth_token.lower().startswith("bearer "):
auth_token = auth_token[7:].strip()
# 构建 API URL (不需要添加 :generateContent 后缀,API 会自动处理)
base_url = f"https://api.comfy.org/proxy/vertexai/gemini/{model}"
# 构建请求内容
parts = []
# 添加系统提示词(如果有)
if system_prompt and system_prompt.strip():
parts.append({"text": system_prompt.strip()})
# 添加用户提示词
parts.append({"text": prompt})
# 添加参考图片(如果有)
if image_ref is not None:
for i in range(image_ref.shape[0]):
b64_str = tensor_to_base64(image_ref[i])
parts.append({
"inlineData": {
"mimeType": "image/png",
"data": b64_str
}
})
print(f"[GEMINI_AUTHTOKEN] Added reference image {i+1}")
# 生成配置
generation_config = {
"responseModalities": ["IMAGE"]
}
# 图片配置
image_config = {}
if aspect_ratio != "auto":
image_config["aspectRatio"] = aspect_ratio
if resolution != "auto":
image_config["resolution"] = resolution
if image_config:
generation_config["imageConfig"] = image_config
# 完整请求体
payload = {
"contents": [{
"role": "user",
"parts": parts
}],
"generationConfig": generation_config
}
# 请求头
headers = {
"Authorization": f"Bearer {auth_token}",
"Content-Type": "application/json",
"User-Agent": "ComfyUI-PD-Node/2.0"
}
start_time = time.time()
try:
print(f"[GEMINI_AUTHTOKEN] Sending request to {model}")
print(f"[GEMINI_AUTHTOKEN] Aspect Ratio: {aspect_ratio}, Resolution: {resolution}")
response = requests.post(base_url, headers=headers, json=payload, timeout=120)
# 检查响应
if response.status_code != 200:
error_msg = f"API Error {response.status_code}: {response.text[:200]}"
print(f"[GEMINI_AUTHTOKEN] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
result = response.json()
# 提取图片
image_tensors = []
candidates = result.get("candidates", [])
if not candidates:
error_msg = "Error: No candidates in response"
print(f"[GEMINI_AUTHTOKEN] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
for cand in candidates:
content = cand.get("content", {})
parts = content.get("parts", [])
for part in parts:
# 支持两种字段名格式
inline_data = part.get("inline_data") or part.get("inlineData")
if inline_data:
mime = inline_data.get("mime_type") or inline_data.get("mimeType")
data_b64 = inline_data.get("data")
if mime in ["image/png", "image/jpeg"] and data_b64:
raw_data = base64.b64decode(data_b64)
image_tensors.append(bytes_to_tensor(raw_data))
print(f"[GEMINI_AUTHTOKEN] Extracted image ({mime})")
if not image_tensors:
error_msg = f"Error: No images found in response"
print(f"[GEMINI_AUTHTOKEN] {error_msg}")
print(f"[GEMINI_AUTHTOKEN] Response: {str(result)[:500]}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
final_image = torch.cat(image_tensors, dim=0)
# 计算耗时和成本
end_time = time.time()
duration = end_time - start_time
# 估算成本(根据模型)
cost_per_img = 0.04
if "pro" in model.lower():
cost_per_img = 0.08
total_cost = cost_per_img * len(image_tensors)
info_str = (
f"Model: {model}\n"
f"Status: Success\n"
f"Time: {duration:.2f}s\n"
f"Images: {len(image_tensors)}\n"
f"Aspect Ratio: {aspect_ratio}\n"
f"Resolution: {resolution}\n"
f"Est. Cost: ${total_cost:.4f}"
)
print(f"[GEMINI_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 (120s)"
print(f"[GEMINI_AUTHTOKEN] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
except requests.exceptions.ConnectionError as e:
error_msg = f"Error: Connection failed - {str(e)}"
print(f"[GEMINI_AUTHTOKEN] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
except Exception as e:
error_msg = f"Error: {str(e)}"
print(f"[GEMINI_AUTHTOKEN] {error_msg}")
import traceback
traceback.print_exc()
return (torch.zeros((1, 512, 512, 3)), error_msg)
# 节点注册
NODE_CLASS_MAPPINGS = {
"PDGeminiImageGenAuthToken": PDGeminiImageGenAuthToken
}
NODE_DISPLAY_NAME_MAPPINGS = {
"PDGeminiImageGenAuthToken": "PD: Gemini Image (comfyui_AuthToken)"
}
+294
View File
@@ -0,0 +1,294 @@
"""
PD: Gemini Pro Image Generation Node (ComfyUI API Key)
Nano Banana Pro - 使用 ComfyUI API Key 调用 ComfyUI 代理 API
支持高分辨率图像生成(1K/2K/4K)
"""
import torch
import requests
import time
import io
import base64
import numpy as np
from PIL import Image
# 默认系统提示词(来自 ComfyUI 官方 API 节点)
GEMINI_IMAGE_SYS_PROMPT = (
"You are an expert image-generation engine. You must ALWAYS produce an image.\n"
"Interpret all user input—regardless of "
"format, intent, or abstraction—as literal visual directives for image composition.\n"
"If a prompt is conversational or lacks specific visual details, "
"you must creatively invent a concrete visual scenario that depicts the concept.\n"
"Prioritize generating the visual representation above any text, formatting, or conversational requests."
)
def tensor_to_base64(tensor):
"""将 Tensor 转换为 base64 字符串"""
if tensor.ndim == 4:
tensor = tensor[0]
array = np.clip(255. * tensor.cpu().numpy(), 0, 255).astype(np.uint8)
image = Image.fromarray(array)
buffered = io.BytesIO()
image.save(buffered, format="PNG")
return base64.b64encode(buffered.getvalue()).decode("utf-8")
def bytes_to_tensor(data):
"""将图片字节数据转换为 Tensor"""
try:
img = Image.open(io.BytesIO(data)).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"[GEMINI_PRO_APIKEY] Image decode error: {e}")
return torch.zeros((1, 512, 512, 3))
class PDGeminiProImageGenAPIKey:
"""
PD: Gemini Pro Image Generation (API Key)
Nano Banana Pro - 使用 ComfyUI API Key 调用 ComfyUI 代理 API
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"api_key": ("STRING", {
"default": "",
"multiline": False,
"placeholder": "Your ComfyUI API Key"
}),
"prompt": ("STRING", {
"multiline": True,
"default": "A futuristic city with flying cars",
"tooltip": "Text prompt describing the image to generate or the edits to apply"
}),
"model": (["gemini-3-pro-image-preview"], {
"default": "gemini-3-pro-image-preview"
}),
"aspect_ratio": ([
"auto", "1:1", "2:3", "3:2", "3:4", "4:3", "4:5", "5:4", "9:16", "16:9", "21:9"
], {
"default": "auto",
"tooltip": "If set to 'auto', matches your input image's aspect ratio"
}),
"resolution": (["1K", "2K", "4K"], {
"default": "1K",
"tooltip": "Target output resolution. For 2K/4K the native Gemini upscaler is used"
}),
"response_modalities": (["IMAGE+TEXT", "IMAGE"], {
"default": "IMAGE+TEXT",
"tooltip": "Choose 'IMAGE' for image-only output, or 'IMAGE+TEXT' for both"
}),
"seed": ("INT", {
"default": 42,
"min": 0,
"max": 0xFFFFFFFFFFFFFFFF
}),
},
"optional": {
"images": ("IMAGE", {
"tooltip": "Optional reference image(s) (up to 14)"
}),
"system_prompt": ("STRING", {
"multiline": True,
"default": GEMINI_IMAGE_SYS_PROMPT,
"tooltip": "Foundational instructions that dictate an AI's behavior"
}),
},
"hidden": {
"unique_id": "UNIQUE_ID",
}
}
RETURN_TYPES = ("IMAGE", "STRING")
RETURN_NAMES = ("image", "text")
FUNCTION = "generate_image"
CATEGORY = "PD_Tools/Image_Generation"
def generate_image(self, api_key, prompt, model, aspect_ratio, resolution, response_modalities, seed, unique_id, images=None, system_prompt=""):
"""生成图片"""
# 验证 API Key
if not api_key or not api_key.strip():
error_msg = "Error: ComfyUI API Key is required"
print(f"[GEMINI_PRO_APIKEY] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
# 清理 API Key
api_key = api_key.strip()
# ComfyUI 代理 API URL
base_url = f"https://api.comfy.org/proxy/vertexai/gemini/{model}"
# 构建请求内容
parts = []
# 添加用户提示词
parts.append({"text": prompt})
# 添加参考图片(如果有,最多14张)
if images is not None:
num_images = min(images.shape[0], 14)
for i in range(num_images):
b64_str = tensor_to_base64(images[i])
parts.append({
"inlineData": {
"mimeType": "image/png",
"data": b64_str
}
})
print(f"[GEMINI_PRO_APIKEY] Added reference image {i+1}/{num_images}")
if images.shape[0] > 14:
print(f"[GEMINI_PRO_APIKEY] Warning: Only first 14 images will be used (provided {images.shape[0]})")
# 生成配置
generation_config = {
"responseModalities": ["IMAGE"] if response_modalities == "IMAGE" else ["TEXT", "IMAGE"]
}
# 图片配置
image_config = {
"imageSize": resolution
}
if aspect_ratio != "auto":
image_config["aspectRatio"] = aspect_ratio
generation_config["imageConfig"] = image_config
# 系统指令
system_instruction = None
if system_prompt and system_prompt.strip():
system_instruction = {
"parts": [{"text": system_prompt.strip()}]
}
# 完整请求体
payload = {
"contents": [{
"role": "user",
"parts": parts
}],
"generationConfig": generation_config
}
if system_instruction:
payload["systemInstruction"] = system_instruction
# 请求头
headers = {
"X-API-KEY": api_key,
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "ComfyUI-PD-Node/2.0"
}
start_time = time.time()
try:
print(f"[GEMINI_PRO_APIKEY] Sending request to {model}")
print(f"[GEMINI_PRO_APIKEY] Aspect Ratio: {aspect_ratio}, Resolution: {resolution}")
response = requests.post(base_url, headers=headers, json=payload, timeout=180)
# 检查响应
if response.status_code != 200:
error_msg = f"API Error {response.status_code}: {response.text[:200]}"
print(f"[GEMINI_PRO_APIKEY] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
result = response.json()
# 提取图片和文本
image_tensors = []
text_parts = []
candidates = result.get("candidates", [])
if not candidates:
error_msg = "Error: No candidates in response"
print(f"[GEMINI_PRO_APIKEY] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
for cand in candidates:
content = cand.get("content", {})
parts = content.get("parts", [])
for part in parts:
# 提取图片
inline_data = part.get("inline_data") or part.get("inlineData")
if inline_data:
mime = inline_data.get("mime_type") or inline_data.get("mimeType")
data_b64 = inline_data.get("data")
if mime in ["image/png", "image/jpeg"] and data_b64:
raw_data = base64.b64decode(data_b64)
image_tensors.append(bytes_to_tensor(raw_data))
print(f"[GEMINI_PRO_APIKEY] Extracted image ({mime})")
# 提取文本
if "text" in part and part["text"]:
text_parts.append(part["text"])
if not image_tensors:
error_msg = f"Error: No images found in response"
print(f"[GEMINI_PRO_APIKEY] {error_msg}")
print(f"[GEMINI_PRO_APIKEY] Response: {str(result)[:500]}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
final_image = torch.cat(image_tensors, dim=0)
final_text = "\n".join(text_parts) if text_parts else "No text output"
# 计算耗时和成本
end_time = time.time()
duration = end_time - start_time
# 估算成本(Gemini 3 Pro Image)
cost_per_img = 0.12 # 估算
total_cost = cost_per_img * len(image_tensors)
info_str = (
f"Model: {model}\n"
f"Status: Success\n"
f"Time: {duration:.2f}s\n"
f"Images: {len(image_tensors)}\n"
f"Aspect Ratio: {aspect_ratio}\n"
f"Resolution: {resolution}\n"
f"Est. Cost: ${total_cost:.4f}\n"
f"---\n"
f"{final_text}"
)
print(f"[GEMINI_PRO_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 (180s)"
print(f"[GEMINI_PRO_APIKEY] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
except requests.exceptions.ConnectionError as e:
error_msg = f"Error: Connection failed - {str(e)}"
print(f"[GEMINI_PRO_APIKEY] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
except Exception as e:
error_msg = f"Error: {str(e)}"
print(f"[GEMINI_PRO_APIKEY] {error_msg}")
import traceback
traceback.print_exc()
return (torch.zeros((1, 512, 512, 3)), error_msg)
# 节点注册
NODE_CLASS_MAPPINGS = {
"PDGeminiProImageGenAPIKey": PDGeminiProImageGenAPIKey
}
NODE_DISPLAY_NAME_MAPPINGS = {
"PDGeminiProImageGenAPIKey": "PD: Gemini Pro Image (apikey)"
}
+300
View File
@@ -0,0 +1,300 @@
"""
PD: Gemini Pro Image Generation Node (ComfyUI Auth Token)
Nano Banana Pro - 基于 Gemini 3 Pro Image
支持高分辨率图像生成(1K/2K/4K)
"""
import torch
import requests
import time
import io
import base64
import numpy as np
from PIL import Image
# 默认系统提示词(来自 ComfyUI 官方 API 节点)
GEMINI_IMAGE_SYS_PROMPT = (
"You are an expert image-generation engine. You must ALWAYS produce an image.\n"
"Interpret all user input—regardless of "
"format, intent, or abstraction—as literal visual directives for image composition.\n"
"If a prompt is conversational or lacks specific visual details, "
"you must creatively invent a concrete visual scenario that depicts the concept.\n"
"Prioritize generating the visual representation above any text, formatting, or conversational requests."
)
def tensor_to_base64(tensor):
"""将 Tensor 转换为 base64 字符串"""
if tensor.ndim == 4:
tensor = tensor[0]
array = np.clip(255. * tensor.cpu().numpy(), 0, 255).astype(np.uint8)
image = Image.fromarray(array)
buffered = io.BytesIO()
image.save(buffered, format="PNG")
return base64.b64encode(buffered.getvalue()).decode("utf-8")
def bytes_to_tensor(data):
"""将图片字节数据转换为 Tensor"""
try:
img = Image.open(io.BytesIO(data)).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"[GEMINI_PRO_AUTHTOKEN] Image decode error: {e}")
return torch.zeros((1, 512, 512, 3))
class PDGeminiProImageGenAuthToken:
"""
PD: Gemini Pro Image Generation (Auth Token)
Nano Banana Pro - 使用 ComfyUI Auth Token 调用 Gemini 3 Pro Image API
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"auth_token": ("STRING", {
"default": "",
"multiline": False,
"placeholder": "Paste your Auth Token here"
}),
"prompt": ("STRING", {
"multiline": True,
"default": "A futuristic city with flying cars",
"tooltip": "Text prompt describing the image to generate or the edits to apply"
}),
"model": (["gemini-3-pro-image-preview"], {
"default": "gemini-3-pro-image-preview"
}),
"aspect_ratio": ([
"auto", "1:1", "2:3", "3:2", "3:4", "4:3", "4:5", "5:4", "9:16", "16:9", "21:9"
], {
"default": "auto",
"tooltip": "If set to 'auto', matches your input image's aspect ratio"
}),
"resolution": (["1K", "2K", "4K"], {
"default": "1K",
"tooltip": "Target output resolution. For 2K/4K the native Gemini upscaler is used"
}),
"response_modalities": (["IMAGE+TEXT", "IMAGE"], {
"default": "IMAGE+TEXT",
"tooltip": "Choose 'IMAGE' for image-only output, or 'IMAGE+TEXT' for both"
}),
"seed": ("INT", {
"default": 42,
"min": 0,
"max": 0xFFFFFFFFFFFFFFFF
}),
},
"optional": {
"images": ("IMAGE", {
"tooltip": "Optional reference image(s) (up to 14)"
}),
"system_prompt": ("STRING", {
"multiline": True,
"default": GEMINI_IMAGE_SYS_PROMPT,
"tooltip": "Foundational instructions that dictate an AI's behavior"
}),
},
"hidden": {
"unique_id": "UNIQUE_ID",
}
}
RETURN_TYPES = ("IMAGE", "STRING")
RETURN_NAMES = ("image", "text")
FUNCTION = "generate_image"
CATEGORY = "PD_Tools/Image_Generation"
def generate_image(self, auth_token, prompt, model, aspect_ratio, resolution, response_modalities, seed, unique_id, images=None, system_prompt=""):
"""生成图片"""
print(f"[GEMINI_PRO_AUTHTOKEN] ========== 开始执行 ==========")
print(f"[GEMINI_PRO_AUTHTOKEN] Prompt: {prompt[:50]}...")
print(f"[GEMINI_PRO_AUTHTOKEN] Model: {model}")
# 验证 Token
if not auth_token or not auth_token.strip():
error_msg = "Error: Auth Token is required"
print(f"[GEMINI_PRO_AUTHTOKEN] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
# 清理 Token
auth_token = auth_token.strip()
if auth_token.lower().startswith("bearer "):
auth_token = auth_token[7:].strip()
# 构建 API URL
base_url = f"https://api.comfy.org/proxy/vertexai/gemini/{model}"
# 构建请求内容
parts = []
# 添加用户提示词
parts.append({"text": prompt})
# 添加参考图片(如果有,最多14张)
if images is not None:
num_images = min(images.shape[0], 14)
for i in range(num_images):
b64_str = tensor_to_base64(images[i])
parts.append({
"inlineData": {
"mimeType": "image/png",
"data": b64_str
}
})
print(f"[GEMINI_PRO_AUTHTOKEN] Added reference image {i+1}/{num_images}")
if images.shape[0] > 14:
print(f"[GEMINI_PRO_AUTHTOKEN] Warning: Only first 14 images will be used (provided {images.shape[0]})")
# 生成配置
generation_config = {
"responseModalities": ["IMAGE"] if response_modalities == "IMAGE" else ["TEXT", "IMAGE"]
}
# 图片配置
image_config = {
"imageSize": resolution
}
if aspect_ratio != "auto":
image_config["aspectRatio"] = aspect_ratio
generation_config["imageConfig"] = image_config
# 系统指令
system_instruction = None
if system_prompt and system_prompt.strip():
system_instruction = {
"parts": [{"text": system_prompt.strip()}]
}
# 完整请求体
payload = {
"contents": [{
"role": "user",
"parts": parts
}],
"generationConfig": generation_config
}
if system_instruction:
payload["systemInstruction"] = system_instruction
# 请求头
headers = {
"Authorization": f"Bearer {auth_token}",
"Content-Type": "application/json",
"User-Agent": "ComfyUI-PD-Node/2.0"
}
start_time = time.time()
try:
print(f"[GEMINI_PRO_AUTHTOKEN] Sending request to {model}")
print(f"[GEMINI_PRO_AUTHTOKEN] Aspect Ratio: {aspect_ratio}, Resolution: {resolution}")
response = requests.post(base_url, headers=headers, json=payload, timeout=180)
# 检查响应
if response.status_code != 200:
error_msg = f"API Error {response.status_code}: {response.text[:200]}"
print(f"[GEMINI_PRO_AUTHTOKEN] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
result = response.json()
# 提取图片和文本
image_tensors = []
text_parts = []
candidates = result.get("candidates", [])
if not candidates:
error_msg = "Error: No candidates in response"
print(f"[GEMINI_PRO_AUTHTOKEN] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
for cand in candidates:
content = cand.get("content", {})
parts = content.get("parts", [])
for part in parts:
# 提取图片
inline_data = part.get("inline_data") or part.get("inlineData")
if inline_data:
mime = inline_data.get("mime_type") or inline_data.get("mimeType")
data_b64 = inline_data.get("data")
if mime in ["image/png", "image/jpeg"] and data_b64:
raw_data = base64.b64decode(data_b64)
image_tensors.append(bytes_to_tensor(raw_data))
print(f"[GEMINI_PRO_AUTHTOKEN] Extracted image ({mime})")
# 提取文本
if "text" in part and part["text"]:
text_parts.append(part["text"])
if not image_tensors:
error_msg = f"Error: No images found in response"
print(f"[GEMINI_PRO_AUTHTOKEN] {error_msg}")
print(f"[GEMINI_PRO_AUTHTOKEN] Response: {str(result)[:500]}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
final_image = torch.cat(image_tensors, dim=0)
final_text = "\n".join(text_parts) if text_parts else "No text output"
# 计算耗时和成本
end_time = time.time()
duration = end_time - start_time
# 估算成本(Gemini 3 Pro Image)
# $2/1M input tokens, $12/1M output text tokens, $120/1M output image tokens
cost_per_img = 0.12 # 估算
total_cost = cost_per_img * len(image_tensors)
info_str = (
f"Model: {model}\n"
f"Status: Success\n"
f"Time: {duration:.2f}s\n"
f"Images: {len(image_tensors)}\n"
f"Aspect Ratio: {aspect_ratio}\n"
f"Resolution: {resolution}\n"
f"Est. Cost: ${total_cost:.4f}\n"
f"---\n"
f"{final_text}"
)
print(f"[GEMINI_PRO_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 (180s)"
print(f"[GEMINI_PRO_AUTHTOKEN] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
except requests.exceptions.ConnectionError as e:
error_msg = f"Error: Connection failed - {str(e)}"
print(f"[GEMINI_PRO_AUTHTOKEN] {error_msg}")
return (torch.zeros((1, 512, 512, 3)), error_msg)
except Exception as e:
error_msg = f"Error: {str(e)}"
print(f"[GEMINI_PRO_AUTHTOKEN] {error_msg}")
import traceback
traceback.print_exc()
return (torch.zeros((1, 512, 512, 3)), error_msg)
# 节点注册
NODE_CLASS_MAPPINGS = {
"PDGeminiProImageGenAuthToken": PDGeminiProImageGenAuthToken
}
NODE_DISPLAY_NAME_MAPPINGS = {
"PDGeminiProImageGenAuthToken": "PD: Gemini Pro Image (comfyui_AuthToken)"
}
+78
View File
@@ -0,0 +1,78 @@
import torch
from comfy_api_nodes.apis import GeminiPart, GeminiInlineData, GeminiMimeType
from comfy_api_nodes.util.conversions import tensor_to_base64_string
class PD_ComfyPlusImage:
"""
将最多9个尺寸不一的图片按顺序整合打包为 GEMINI_INPUT_FILES,
可以直接连接给支持 files 的大模型节点 (如 Nano Banana Pro 等),
从而完美突破 ComfyUI Batch 必须相同尺寸的限制。
"""
@classmethod
def INPUT_TYPES(cls):
return {
# 第1张图片强制必填
"required": {
"image1": ("IMAGE", {"tooltip": "必须输入的第1张图片"}),
},
# 第 2-9 张及 previous_files 作为可选填
"optional": {
"image2": ("IMAGE", {"tooltip": "可选输入的第2张图片"}),
"image3": ("IMAGE", {"tooltip": "可选输入的第3张图片"}),
"image4": ("IMAGE", {"tooltip": "可选输入的第4张图片"}),
"image5": ("IMAGE", {"tooltip": "可选输入的第5张图片"}),
"image6": ("IMAGE", {"tooltip": "可选输入的第6张图片"}),
"image7": ("IMAGE", {"tooltip": "可选输入的第7张图片"}),
"image8": ("IMAGE", {"tooltip": "可选输入的第8张图片"}),
"image9": ("IMAGE", {"tooltip": "可选输入的第9张图片"}),
"previous_files": ("GEMINI_INPUT_FILES", {"tooltip": "如超过9张图需继续打包,可连接上一个该节点输出的files"}),
}
}
RETURN_TYPES = ("GEMINI_INPUT_FILES",)
RETURN_NAMES = ("files",)
FUNCTION = "pack_images"
CATEGORY = "PD_Tools"
def pack_images(self, image1, image2=None, image3=None, image4=None,
image5=None, image6=None, image7=None, image8=None,
image9=None, previous_files=None):
files_list = []
# 如果需要无缝串联别的节点输出的 files ,优先加进去
if previous_files is not None:
files_list.extend(previous_files)
images = [image1, image2, image3, image4, image5, image6, image7, image8, image9]
for img in images:
if img is not None:
# 兼容万一单条线里输入了 Batch(N) 图片的特殊情况,分别解开处理
for i in range(img.shape[0]):
try:
# 转换成 Base64
img_b64 = tensor_to_base64_string(img[i].unsqueeze(0))
# 构造为 Gemini 认证的 Part
new_part = GeminiPart(
inlineData=GeminiInlineData(
mimeType=GeminiMimeType.image_png,
data=img_b64
)
)
# 转成字典防止 Pydantic 跨模块传递对象时校验失败
files_list.append(new_part.model_dump(exclude_none=True))
except Exception as e:
print(f"[Error] PD_comfyplus_image pack failed for an image: {e}")
return (files_list,)
# 注册节点
NODE_CLASS_MAPPINGS = {
"PD_comfyplus_image": PD_ComfyPlusImage
}
NODE_DISPLAY_NAME_MAPPINGS = {
"PD_comfyplus_image": "PD_comfyplus_image (Multi-Size Pack)"
}
+352
View File
@@ -0,0 +1,352 @@
# PD ComfyUI API Nodes
ComfyUI 自定义节点集合,用于调用各种 AI 图像生成 API(GPT Image、Gemini Image、Flux Kontext)。
## 📦 节点列表
### ComfyUI AuthToken 版本(通过 ComfyUI 代理)
ComfyUI AuthToken 版本(通过 ComfyUI 代理)
- PD: GPT Image (comfyui_AuthToken)
- PD: Gemini Image (comfyui_AuthToken)
- PD: Flux Kontext (comfyui_AuthToken)
### 直接 API Key 版本直接调用官方comfyui的key
API Key 版本:
- 只保留 ComfyUI API Key 输入
- 参数和调用逻辑跟 ComfyUI 官方自带的 API 节点一样 - 只是在调用时可以手动填入 key
| 节点名称 | 文件 | 说明 |
|---------|------|------|
| **PD: GPT Image (apikey)** | `IMAGE_GPT_apikey.py` | OpenAI GPT Image 生成/编辑 |
| **PD: Gemini Image (apikey)** | `IMGE_GeminiNode_apikey.py` | Google Gemini 图像生成 |
| **PD: Gemini Pro Image (apikey)** | `IMGE_GeminiPro_apikey.py` | Google Gemini 3 Pro 高分辨率图像生成 |
| **PD: Flux Kontext (apikey)** | `IMGE_kontext_apikey.py` | Flux Kontext 图像编辑 |
---
## 🔑 两种版本的区别
### ComfyUI AuthToken 版本 (comfyui_AuthToken)
**认证方式**:使用 ComfyUI Auth Token,pandy自制油猴插件获取。
**认证头格式**:`Authorization: Bearer {auth_token}`
**使用示例**:
```
auth_token: eyJhbGciOiJSUzI1NiIsImtpZCI6ImEzOGVhNmEw...
```
### API Key 版本 (apikey)
**认证方式**:使用 ComfyUI 官方 API Key
**认证头格式**:`X-API-KEY: {api_key}` (与 ComfyUI 官方 API 节点完全一致)
**使用示例**:
```
api_key: comfy_xxxxxxxxxxxxxxxxxxxxx
```
**特点**:
- 参数和调用逻辑跟 ComfyUI 官方自带的 API 节点一样
- 只是在调用时可以手动填入 key
- 直接使用官方账号的 key 申请下来填入使用
## 📖 节点详细说明
### 1. GPT Image 节点
**功能**:
- 文生图(Text-to-Image)
- 图片编辑(Image Editing)
- 图片修复(Inpainting with Mask)
**支持模型**:
**AuthToken 版本**:
- `gpt-image-1`:OpenAI GPT Image 1(推荐)
- `gpt-image-1.5`:OpenAI GPT Image 1.5(更高质量)
### 3. Gemini Pro Image 节点(Nano Banana Pro)
**功能**:
- 高分辨率图像生成(1K/2K/4K)
- 文生图(Text-to-Image)
- 图生图(Image-to-Image)
- 支持多张参考图片(最多14张)
- 同时输出图像和文本说明
**支持模型**:
- `gemini-3-pro-image-preview`:Gemini 3 Pro Image(最高质量)
**参数说明**:
- `prompt`:文本提示词,描述要生成的图像或编辑指令
- `model`:gemini-3-pro-image-preview
- `aspect_ratio`:宽高比
- auto:自动匹配输入图片比例,无图片时默认 16:9
- 1:1, 2:3, 3:2, 3:4, 4:3, 4:5, 5:4, 9:16, 16:9, 21:9
- `resolution`:目标输出分辨率
- 1K:标准分辨率
- 2K:使用原生 Gemini 放大器
- 4K:使用原生 Gemini 放大器(最高质量)
- `response_modalities`:输出模式
- IMAGE+TEXT:同时返回图像和文本说明
- IMAGE:仅返回图像
- `images`(可选):参考图片(最多14张,使用 Batch Images 节点)
- `system_prompt`(可选):系统提示词,默认使用官方优化的图像生成提示词
**输出**:
- `image`:生成的图像
- `text`:文本说明(包含模型的思考过程和描述)
**成本估算**:
- Gemini 3 Pro Image: 约 $0.12/张
- 基于 token 使用量:$2/1M input + $12/1M output text + $120/1M output image
**特殊说明(apikey 版本)**:
- 需要提供 `project_id`(Google Cloud 项目 ID)
- 需要提供 `location`(区域,默认 us-central1)
- 超时时间:180秒(因为高分辨率生成需要更长时间)
**使用建议**:
- 适合需要高质量、高分辨率图像的场景
- 2K/4K 分辨率会使用原生放大器,质量更好但成本更高
- IMAGE+TEXT 模式可以看到模型的思考过程,有助于理解生成结果
- 可以提供多张参考图片来引导生成风格
---
### 4. Flux Kontext 节点
**功能**:
- 图片编辑(Image Editing)
- 基于上下文的智能修改
**支持模型**:
- `flux-kontext-pro`:快速版本
- `flux-kontext-max`:高质量版本
**参数说明**:
- `prompt`:编辑指令
- `aspect_ratio`:宽高比
- `guidance`:提示词遵循度(0.1-99.0)
- `steps`:生成步数(1-150)
- `prompt_upsampling`:AI 增强提示词
- `input_image`(可选):输入图片
**成本估算**:
- Kontext Pro: $0.05/张
- Kontext Max: $0.10/张
**特点**:
- 异步生成,需要轮询结果
- 最长等待时间:10 分钟
- 轮询间隔:2 秒
---
## 🚀 使用建议
### 选择 ComfyUI AuthToken 版本,如果你:
- ✅ 已经有 ComfyUI Auth Token
- ✅ 想要统一管理多个 AI 模型
- ✅ 喜欢通过 ComfyUI 生态系统
- ✅ 不想配置多个平台的 API Key
### 选择直接 API Key 版本,如果你:
- ✅ 已经有各个平台的 API Key
- ✅ 想要直接控制 API 调用
- ✅ 不想依赖 ComfyUI 代理服务
- ✅ 需要更灵活的配置(如 Gemini 的 project_id)
---
## 📝 使用示例
### 示例 1:使用 ComfyUI AuthToken 生成图片(GPT Image)
```
节点:PD: GPT Image (comfyui_AuthToken)
参数:
- auth_token: eyJhbGciOiJSUzI1NiIsImtpZCI6ImEzOGVhNmEw...
- prompt: A futuristic city with flying cars at sunset
- model: gpt-image-1
- quality: medium
- background: auto
- size: 1024x1024
- n: 1
```
### 示例 2:使用直接 API Key 生成图片(DALL-E)
```
节点:PD: GPT Image (apikey)
参数:
- api_key: sk-proj-abc123...
- prompt: A futuristic city with flying cars at sunset
- model: dall-e-3
- quality: hd
- size: 1024x1792
```
### 示例 3:使用 Gemini 生成图片(apikey 版本)
```
节点:PD: Gemini Image (apikey)
参数:
- api_key: AIzaSyD...
- project_id: my-project-123
- location: us-central1
- prompt: A beautiful landscape with mountains
- model: gemini-2.5-flash-image
- aspect_ratio: 16:9
- resolution: 2K
```
### 示例 4:使用 Flux Kontext 编辑图片
```
节点:PD: Flux Kontext (comfyui_AuthToken)
参数:
- auth_token: eyJhbGciOiJSUzI1NiIsImtpZCI6ImEzOGVhNmEw...
- prompt: Change the sky to sunset colors
- model: flux-kontext-pro
- aspect_ratio: 16:9
- guidance: 3.0
- steps: 50
- input_image: [连接输入图片]
```
---
## ⚠️ 注意事项
1. **API Key 安全**:
- 不要在公开的工作流中暴露 API Key
- 建议使用环境变量或配置文件管理
2. **成本控制**:
- 所有节点都会显示预估成本
- 实际成本以各平台账单为准
- 建议设置 API 使用限额
3. **超时设置**:
- GPT/Gemini: 120 秒超时
- Flux Kontext: 10 分钟超时(包含轮询)
4. **错误处理**:
- 所有节点都有详细的错误日志
- 检查 ComfyUI 控制台查看详细信息
5. **网络要求**:
- 需要稳定的网络连接
- 某些地区可能需要代理访问
---
## 🔧 故障排除
### 问题 1:认证失败(401 错误)
**ComfyUI AuthToken 版本**:
- 检查 Auth Token 是否正确
- Token 可能已过期,重新获取
- 认证头格式:`Authorization: Bearer {auth_token}`
**直接 API Key 版本**:
- 检查 API Key 格式是否正确
- 确认 API Key 有效且有余额
- 认证头格式:`X-API-KEY: {api_key}` (与 ComfyUI 官方 API 节点一致)
- Gemini 需要确认 Project ID 和 Location 正确
### 问题 2:请求超时
- 检查网络连接
- 尝试增加超时时间
- Flux Kontext 可能需要更长时间,属于正常现象
### 问题 3:图片无法生成 / 返回黑图
**症状**:节点返回 512x512 的黑色图片
**可能原因**:
1. **API 端点错误**(已修复):
- 旧版本的 Gemini AuthToken 节点使用了错误的端点格式
- 错误:`https://api.comfy.org/proxy/vertexai/gemini/{model}:generateContent`
- 正确:`https://api.comfy.org/proxy/vertexai/gemini/{model}`
- **解决方案**:更新到最新版本(v2.1+)
2. **模型名称不支持**:
- 检查控制台是否有 "Unsupported model" 错误
- 确认使用的模型在支持列表中
- Gemini AuthToken 版本支持:`gemini-2.5-flash-image`, `gemini-2.5-flash-image-preview`
3. **提示词违规**:
- 检查提示词是否符合内容政策
- 查看控制台日志获取详细错误信息
4. **API 配额不足**:
- 确认账户有足够的配额/余额
- 检查 API 使用限制
**调试步骤**:
1. 打开 ComfyUI 控制台
2. 查找 `[GEMINI_AUTHTOKEN]` 或 `[GEMINI_APIKEY]` 开头的日志
3. 检查是否有 API Error 信息
4. 根据错误信息采取相应措施
### 问题 4:认证失败(Gemini 节点)
- 确认已启用 Vertex AI API
- 确认 Project ID 正确
- 确认 Location 支持 Gemini 模型
- 检查 API Key 权限
---
## 📚 相关链接
- [ComfyUI 官网](https://www.comfy.org)
- [OpenAI Platform](https://platform.openai.com)
- [Google Cloud Console](https://console.cloud.google.com)
- [Black Forest Labs](https://blackforestlabs.ai)
- [ComfyUI GitHub](https://github.com/comfyanonymous/ComfyUI)
---
## 📄 许可证
本项目遵循 MIT 许可证。
---
## 🤝 贡献
欢迎提交 Issue 和 Pull Request!
---
## 📧 联系方式
如有问题或建议,请通过 GitHub Issues 联系。
---
**最后更新**:2025-01-12
**版本**:2.5
**更新日志**:
- 2025-01-12 v2.5: 修复 API Key 版本认证头问题(401 错误),使用 `X-API-KEY` 与官方一致
- 2025-01-12 v2.4: 新增 Gemini Pro Image (Nano Banana Pro) 节点,支持高分辨率生成
- 2025-01-12 v2.3: 更新 GPT Image AuthToken 节点参数,匹配官方 ComfyUI API 节点
- 2025-01-12 v2.2: 添加官方 ComfyUI API 节点的默认 system prompt
- 2025-01-12 v2.1: 修复 Gemini AuthToken 版本黑图问题,更新模型列表
- 2025-01-11 v2.0: 初始版本,支持 AuthToken 和 API Key 两种认证方式
+506
View File
@@ -0,0 +1,506 @@
// ==UserScript==
// @name Pandy-ComfyUI Token Grabber
// @namespace http://tampermonkey.net/
// @version 3.5
// @description 直接从 ComfyUI 提取 auth_token_comfy_org 并同步到 Pandy
// @author You
// @match *://localhost:8188/*
// @match *://127.0.0.1:8188/*
// @match *://127.0.0.1:8180/*
// @grant GM_setValue
// @grant GM_getValue
// @grant GM_setClipboard
// @grant GM_xmlhttpRequest
// @grant unsafeWindow
// @connect 127.0.0.1
// @connect localhost
// @run-at document-idle
// ==/UserScript==
(function () {
'use strict';
const STORAGE_KEY = 'comfyui_auth_token';
const POS_KEY = 'token_grabber_pos';
const COLLAPSED_KEY = 'token_grabber_collapsed';
let foundToken = '';
let isCollapsed = GM_getValue(COLLAPSED_KEY, false);
let savedPos = GM_getValue(POS_KEY, { x: null, y: 10 });
// 确保位置在可视范围内
function clampPosition() {
const panel = document.getElementById('token-grabber-panel');
if (!panel) return;
const rect = panel.getBoundingClientRect();
let needUpdate = false;
// 如果面板超出右边界
if (rect.right > window.innerWidth) {
panel.style.left = 'auto';
panel.style.right = '10px';
needUpdate = true;
}
// 如果面板超出下边界
if (rect.bottom > window.innerHeight) {
panel.style.top = Math.max(10, window.innerHeight - rect.height - 10) + 'px';
needUpdate = true;
}
// 如果面板超出左边界
if (rect.left < 0) {
panel.style.left = '10px';
panel.style.right = 'auto';
needUpdate = true;
}
// 如果面板超出上边界
if (rect.top < 0) {
panel.style.top = '10px';
needUpdate = true;
}
}
// 搜索 token
function searchToken() {
// 1. localStorage
for (let i = 0; i < localStorage.length; i++) {
const key = localStorage.key(i);
const value = localStorage.getItem(key);
if (value && value.includes('eyJ')) {
const token = extractJWT(value);
if (token && isValidJWT(token)) return token;
}
}
// 2. sessionStorage
for (let i = 0; i < sessionStorage.length; i++) {
const key = sessionStorage.key(i);
const value = sessionStorage.getItem(key);
if (value && value.includes('eyJ')) {
const token = extractJWT(value);
if (token && isValidJWT(token)) return token;
}
}
// 3. window 对象
const props = ['app', 'api', 'comfyAPI'];
for (const prop of props) {
if (window[prop]) {
const token = deepSearch(window[prop], 'auth_token_comfy_org', 5);
if (token) return token;
}
}
return null;
}
function extractJWT(str) {
const match = str.match(/eyJ[A-Za-z0-9_-]+\.eyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+/);
return match ? match[0] : null;
}
function isValidJWT(token) {
try {
const parts = token.split('.');
if (parts.length !== 3) return false;
let payload = parts[1];
payload += '='.repeat((4 - payload.length % 4) % 4);
const info = JSON.parse(atob(payload));
return info.exp !== undefined;
} catch (e) { return false; }
}
function deepSearch(obj, key, depth) {
if (depth <= 0 || !obj || typeof obj !== 'object') return null;
try {
if (obj[key]) return obj[key];
for (const k in obj) {
const result = deepSearch(obj[k], key, depth - 1);
if (result) return result;
}
} catch (e) { }
return null;
}
function parseToken(token) {
try {
const parts = token.split('.');
let payload = parts[1];
payload += '='.repeat((4 - payload.length % 4) % 4);
const info = JSON.parse(atob(payload));
const remaining = Math.floor((info.exp * 1000 - Date.now()) / 60000);
return { valid: remaining > 0, remaining };
} catch (e) { return { valid: false, remaining: 0 }; }
}
function updatePanel() {
const statusEl = document.getElementById('tg-status');
const previewEl = document.getElementById('tg-preview');
const expiryEl = document.getElementById('tg-expiry');
const iconStatus = document.getElementById('tg-icon-status');
if (!foundToken) {
if (statusEl) {
statusEl.textContent = '❌ 未找到';
statusEl.className = 'status expired';
}
if (previewEl) previewEl.textContent = '请确保已登录';
if (iconStatus) iconStatus.style.background = '#f66';
return;
}
const info = parseToken(foundToken);
if (info.valid) {
if (statusEl) {
statusEl.textContent = '✅ Token 已获取';
statusEl.className = 'status captured';
}
if (expiryEl) expiryEl.innerHTML = `⏱️ 剩余 <b>${info.remaining}</b> 分钟`;
if (iconStatus) iconStatus.style.background = '#6f6';
} else {
if (statusEl) {
statusEl.textContent = '❌ Token 已过期';
statusEl.className = 'status expired';
}
if (expiryEl) expiryEl.innerHTML = '请重新登录';
if (iconStatus) iconStatus.style.background = '#f66';
}
if (previewEl) previewEl.textContent = foundToken.substring(0, 50) + '...';
}
function copyToken() {
if (foundToken) {
GM_setClipboard(foundToken);
const btn = document.getElementById('tg-copy-btn');
if (btn) {
btn.textContent = '✅ 已复制!';
setTimeout(() => btn.textContent = '📋 复制 Token', 1500);
}
} else {
alert('未找到 Token');
}
}
function refreshToken() {
foundToken = searchToken() || '';
if (foundToken) GM_setValue(STORAGE_KEY, foundToken);
updatePanel();
}
function toggleCollapse() {
isCollapsed = !isCollapsed;
GM_setValue(COLLAPSED_KEY, isCollapsed);
const panel = document.getElementById('token-grabber-panel');
if (isCollapsed) {
panel.classList.add('collapsed');
} else {
panel.classList.remove('collapsed');
}
}
// 同步到 Pandy Image 页面(使用 GM_xmlhttpRequest 绕过跨域)
async function syncToPandy() {
if (!foundToken) {
alert('未找到 Token');
return;
}
const btn = document.getElementById('tg-sync-btn');
if (btn) btn.textContent = '⏳ 同步中...';
const PANDY_BASE = 'http://127.0.0.1:8180';
// 使用 GM_xmlhttpRequest 绕过跨域限制
// 1. 先获取现有配置
GM_xmlhttpRequest({
method: 'GET',
url: `${PANDY_BASE}/config?name=settings`,
onload: function (response) {
let config = {};
try {
if (response.status === 200) {
config = JSON.parse(response.responseText);
}
} catch (e) { }
// 2. 合并 token
config.token = foundToken;
// 3. POST 回去
GM_xmlhttpRequest({
method: 'POST',
url: `${PANDY_BASE}/config?name=settings`,
headers: { 'Content-Type': 'application/json' },
data: JSON.stringify(config),
onload: function (resp) {
if (resp.status === 200) {
if (btn) {
btn.textContent = '✅ 已同步!';
setTimeout(() => btn.textContent = '🚀 同步到 Pandy', 2000);
}
console.log('🍌 Token 已同步到 Pandy');
} else {
fallbackCopy(btn);
}
},
onerror: function () {
fallbackCopy(btn);
}
});
},
onerror: function () {
fallbackCopy(btn);
}
});
}
function fallbackCopy(btn) {
GM_setClipboard(foundToken);
if (btn) {
btn.textContent = '❌ 服务未启动,已复制';
setTimeout(() => btn.textContent = '🚀 同步到 Pandy', 3000);
}
console.log('🍌 同步失败,请确保 Pandy 服务器已启动 (python server.py)');
}
unsafeWindow.copyToken = copyToken;
unsafeWindow.refreshToken = refreshToken;
unsafeWindow.toggleCollapse = toggleCollapse;
unsafeWindow.syncToPandy = syncToPandy;
function createPanel() {
const panel = document.createElement('div');
panel.id = 'token-grabber-panel';
if (isCollapsed) panel.classList.add('collapsed');
// 设置位置
if (savedPos.x !== null) {
panel.style.left = savedPos.x + 'px';
panel.style.right = 'auto';
}
panel.style.top = savedPos.y + 'px';
panel.innerHTML = `
<style>
#token-grabber-panel {
position: fixed;
top: 10px;
right: 10px;
background: #1a1a2e;
border: 2px solid #f0c040;
border-radius: 10px;
padding: 12px 15px;
z-index: 999999;
font-family: sans-serif;
font-size: 13px;
color: #fff;
width: 260px;
box-shadow: 0 5px 20px rgba(0,0,0,0.5);
transition: width 0.2s, padding 0.2s, border-radius 0.2s;
cursor: default;
}
#token-grabber-panel .header {
display: flex;
align-items: center;
gap: 8px;
cursor: move;
user-select: none;
padding-bottom: 10px;
border-bottom: 1px solid #333;
margin-bottom: 10px;
}
#token-grabber-panel .icon {
font-size: 20px;
}
#token-grabber-panel .title {
flex: 1;
font-weight: bold;
color: #f0c040;
font-size: 14px;
}
#token-grabber-panel .icon-status {
width: 10px;
height: 10px;
border-radius: 50%;
background: #666;
}
#token-grabber-panel .collapse-btn {
cursor: pointer;
color: #888;
font-size: 16px;
padding: 2px 6px;
border-radius: 4px;
}
#token-grabber-panel .collapse-btn:hover {
background: #333;
color: #fff;
}
#token-grabber-panel .content {
transition: opacity 0.2s;
}
#token-grabber-panel .status {
padding: 6px 10px;
border-radius: 6px;
margin: 8px 0;
font-weight: bold;
font-size: 12px;
}
#token-grabber-panel .status.captured { background: rgba(50,200,50,0.2); color: #6f6; }
#token-grabber-panel .status.expired { background: rgba(200,50,50,0.2); color: #f66; }
#token-grabber-panel .token-preview {
font-family: monospace;
font-size: 9px;
color: #666;
word-break: break-all;
background: #111;
padding: 6px;
border-radius: 4px;
margin: 8px 0;
}
#token-grabber-panel #tg-expiry {
color: #6f6;
font-size: 12px;
margin: 6px 0;
}
#token-grabber-panel button {
background: #f0c040;
color: #000;
border: none;
padding: 8px;
border-radius: 6px;
cursor: pointer;
font-weight: bold;
width: 100%;
margin-top: 4px;
font-size: 12px;
}
#token-grabber-panel button:hover { background: #e6a000; }
#token-grabber-panel button.secondary {
background: #333;
color: #aaa;
}
/* 收缩状态 */
#token-grabber-panel.collapsed {
width: auto;
padding: 8px 12px;
border-radius: 25px;
}
#token-grabber-panel.collapsed .header {
padding-bottom: 0;
border-bottom: none;
margin-bottom: 0;
}
#token-grabber-panel.collapsed .content {
display: none;
}
#token-grabber-panel.collapsed .title {
display: none;
}
</style>
<div class="header" id="tg-header">
<span class="icon">🍌</span>
<span class="title">Token Grabber</span>
<span class="icon-status" id="tg-icon-status"></span>
<span class="collapse-btn" onclick="toggleCollapse()" title="收缩/展开">▼</span>
</div>
<div class="content">
<div class="status" id="tg-status">🔍 搜索中...</div>
<div class="token-preview" id="tg-preview">正在搜索...</div>
<div id="tg-expiry"></div>
<button id="tg-copy-btn" onclick="copyToken()">📋 复制 Token</button>
<button id="tg-sync-btn" onclick="syncToPandy()">🚀 同步到 Pandy</button>
<button class="secondary" onclick="refreshToken()">🔄 刷新</button>
</div>
`;
document.body.appendChild(panel);
// 拖拽功能
const header = document.getElementById('tg-header');
let isDragging = false;
let startX, startY, startLeft, startTop;
header.addEventListener('mousedown', (e) => {
if (e.target.classList.contains('collapse-btn')) return;
isDragging = true;
const rect = panel.getBoundingClientRect();
startX = e.clientX;
startY = e.clientY;
startLeft = rect.left;
startTop = rect.top;
panel.style.right = 'auto';
panel.style.left = startLeft + 'px';
e.preventDefault();
});
document.addEventListener('mousemove', (e) => {
if (!isDragging) return;
const dx = e.clientX - startX;
const dy = e.clientY - startY;
let newLeft = startLeft + dx;
let newTop = startTop + dy;
// 边界限制
newLeft = Math.max(0, Math.min(newLeft, window.innerWidth - panel.offsetWidth));
newTop = Math.max(0, Math.min(newTop, window.innerHeight - panel.offsetHeight));
panel.style.left = newLeft + 'px';
panel.style.top = newTop + 'px';
});
document.addEventListener('mouseup', () => {
if (isDragging) {
isDragging = false;
// 保存位置
const rect = panel.getBoundingClientRect();
savedPos = { x: rect.left, y: rect.top };
GM_setValue(POS_KEY, savedPos);
}
});
}
function init() {
console.log('🍌 Token Grabber: init() 开始执行');
console.log('🍌 document.body:', document.body);
if (!document.body) {
console.log('🍌 body 不存在,延迟重试...');
setTimeout(init, 500);
return;
}
createPanel();
console.log('🍌 面板已创建');
// 确保面板在可视范围内
setTimeout(clampPosition, 100);
// 监听窗口大小变化
window.addEventListener('resize', clampPosition);
setTimeout(() => {
foundToken = searchToken() || GM_getValue(STORAGE_KEY, '');
if (foundToken) GM_setValue(STORAGE_KEY, foundToken);
updatePanel();
console.log('🍌 Token 搜索完成:', foundToken ? '找到' : '未找到');
}, 1500);
setInterval(() => {
const newToken = searchToken();
if (newToken && newToken !== foundToken) {
foundToken = newToken;
GM_setValue(STORAGE_KEY, foundToken);
updatePanel();
}
}, 10000);
}
// 多种方式确保执行
console.log('🍌 Token Grabber: 脚本已加载, readyState:', document.readyState);
if (document.readyState === 'complete') {
init();
} else if (document.readyState === 'interactive') {
setTimeout(init, 100);
} else {
window.addEventListener('load', init);
document.addEventListener('DOMContentLoaded', () => setTimeout(init, 100));
}
})();