561 lines
22 KiB
Python
561 lines
22 KiB
Python
import json
|
|
import requests
|
|
from typing import List, Dict, Any
|
|
import torch
|
|
import numpy as np
|
|
from PIL import Image
|
|
import io
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
|
|
class ADIC_COMMON_API:
|
|
"""ADIC Common API"""
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
return {
|
|
"required": {
|
|
"params": ("STRING", {"forceInput": True}),
|
|
"app_name": ("STRING", {"default": "NHCI"}),
|
|
"api_key": ("STRING", {"default": ""}),
|
|
"api_endpoint": ("STRING", {"default": "/open/api/v1/ai/imageTranslate"}),
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("STRING",)
|
|
RETURN_NAMES = ("response",)
|
|
|
|
FUNCTION = "translate"
|
|
|
|
OUTPUT_NODE = True
|
|
|
|
CATEGORY = "Malette"
|
|
|
|
def translate(self, params, app_name, api_key, api_endpoint):
|
|
try:
|
|
# 验证必填参数
|
|
if not api_key or api_key.strip() == "":
|
|
raise ValueError("API密钥不能为空")
|
|
|
|
if not api_endpoint or api_endpoint.strip() == "":
|
|
raise ValueError("API端点不能为空")
|
|
|
|
# 解析params参数
|
|
try:
|
|
if isinstance(params, str):
|
|
request_body = json.loads(params)
|
|
else:
|
|
request_body = params
|
|
|
|
|
|
except json.JSONDecodeError as e:
|
|
raise ValueError(f"params参数JSON解析失败: {str(e)}")
|
|
|
|
# 设置请求头
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Accept": "*/*",
|
|
"Authorization": f"Bearer {api_key.strip()}",
|
|
"X-App-Name": app_name.strip()
|
|
}
|
|
|
|
print(f"[ADIC_COMMON_API] 请求参数: {json.dumps(request_body, ensure_ascii=False)}")
|
|
|
|
# 发送POST请求
|
|
response = requests.post(
|
|
api_endpoint.strip(),
|
|
headers=headers,
|
|
json=request_body,
|
|
timeout=600,
|
|
verify=False # 忽略SSL证书验证
|
|
)
|
|
|
|
# 检查响应状态
|
|
response.raise_for_status()
|
|
|
|
# 解析响应
|
|
response_data = response.json()
|
|
|
|
# 检查业务状态
|
|
if not response_data.get("success", False):
|
|
error_msg = response_data.get("message", "API调用失败")
|
|
print(f"[ADIC_COMMON_API] 业务错误: {error_msg}")
|
|
return (json.dumps({
|
|
"error": error_msg,
|
|
"code": response_data.get("code", -1),
|
|
"debugInfo": response_data.get("debugInfo")
|
|
}, ensure_ascii=False, indent=2),)
|
|
|
|
# 返回成功结果
|
|
result_json = json.dumps(response_data, ensure_ascii=False, indent=2)
|
|
print(f"[ADIC_COMMON_API] 请求结果: {result_json}")
|
|
|
|
return (result_json,)
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
error_msg = f"网络请求失败: {str(e)}"
|
|
print(f"[ADIC_COMMON_API] {error_msg}")
|
|
return (json.dumps({"error": error_msg}, ensure_ascii=False),)
|
|
|
|
except json.JSONDecodeError as e:
|
|
error_msg = f"JSON 解析失败: {str(e)}"
|
|
print(f"[ADIC_COMMON_API] {error_msg}")
|
|
return (json.dumps({"error": error_msg}, ensure_ascii=False),)
|
|
|
|
except Exception as e:
|
|
error_msg = f"未知错误: {str(e)}"
|
|
print(f"[ADIC_COMMON_API] {error_msg}")
|
|
return (json.dumps({"error": error_msg}, ensure_ascii=False),)
|
|
|
|
|
|
class ImageTranslateAPI:
|
|
"""阿里云图片翻译API节点"""
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
return {
|
|
"required": {
|
|
"params": ("STRING", {"forceInput": True}),
|
|
"app_name": ("STRING", {"default": "NHCI"}),
|
|
"api_key": ("STRING", {"default": ""}),
|
|
"api_endpoint": ("STRING", {"default": "/open/api/v1/ai/imageTranslate"}),
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("STRING",)
|
|
RETURN_NAMES = ("response",)
|
|
|
|
FUNCTION = "translate"
|
|
|
|
OUTPUT_NODE = True
|
|
|
|
CATEGORY = "Malette"
|
|
|
|
def translate(self, params, app_name, api_key, api_endpoint):
|
|
try:
|
|
# 验证必填参数
|
|
if not api_key or api_key.strip() == "":
|
|
raise ValueError("API密钥不能为空")
|
|
|
|
if not api_endpoint or api_endpoint.strip() == "":
|
|
raise ValueError("API端点不能为空")
|
|
|
|
# 解析params参数
|
|
try:
|
|
if isinstance(params, str):
|
|
request_body = json.loads(params)
|
|
else:
|
|
request_body = params
|
|
|
|
except json.JSONDecodeError as e:
|
|
raise ValueError(f"params参数JSON解析失败: {str(e)}")
|
|
|
|
# 设置请求头
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Accept": "*/*",
|
|
"Authorization": f"Bearer {api_key.strip()}",
|
|
"X-App-Name": app_name.strip()
|
|
}
|
|
|
|
print(f"[ImageTranslateAPI] 请求参数: {json.dumps(request_body, ensure_ascii=False)}")
|
|
|
|
# 发送POST请求
|
|
response = requests.post(
|
|
api_endpoint.strip(),
|
|
headers=headers,
|
|
json=request_body,
|
|
timeout=600,
|
|
verify=False # 忽略SSL证书验证
|
|
)
|
|
|
|
# 检查响应状态
|
|
response.raise_for_status()
|
|
|
|
# 解析响应
|
|
response_data = response.json()
|
|
|
|
# 检查业务状态
|
|
if not response_data.get("success", False):
|
|
error_msg = response_data.get("message", "API调用失败")
|
|
print(f"[ImageTranslateAPI] 业务错误: {error_msg}")
|
|
return (json.dumps({
|
|
"error": error_msg,
|
|
"code": response_data.get("code", -1),
|
|
"debugInfo": response_data.get("debugInfo")
|
|
}, ensure_ascii=False, indent=2),)
|
|
|
|
# 返回成功结果
|
|
result_json = json.dumps(response_data, ensure_ascii=False, indent=2)
|
|
print(f"[ImageTranslateAPI] 翻译完成，处理了 {len(response_data.get('data', []))} 张图片")
|
|
print(f"[ImageTranslateAPI] 翻译结果: {result_json}")
|
|
|
|
return (result_json,)
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
error_msg = f"网络请求失败: {str(e)}"
|
|
print(f"[ImageTranslateAPI] {error_msg}")
|
|
return (json.dumps({"error": error_msg}, ensure_ascii=False),)
|
|
|
|
except json.JSONDecodeError as e:
|
|
error_msg = f"JSON 解析失败: {str(e)}"
|
|
print(f"[ImageTranslateAPI] {error_msg}")
|
|
return (json.dumps({"error": error_msg}, ensure_ascii=False),)
|
|
|
|
except Exception as e:
|
|
error_msg = f"未知错误: {str(e)}"
|
|
print(f"[ImageTranslateAPI] {error_msg}")
|
|
return (json.dumps({"error": error_msg}, ensure_ascii=False),)
|
|
|
|
|
|
class ImageTranslateParamsBuilder:
|
|
"""图片翻译参数构建器节点"""
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
return {
|
|
"required": {
|
|
"url1": ("STRING", {"default": ""}),
|
|
},
|
|
"optional": {
|
|
"source_language1": (["", "zh", "en", "tr"], {"default": ""}),
|
|
"target_language1": (["", "es", "fr", "pt", "ko", "en"], {"default": "en"}),
|
|
"url2": ("STRING", {"default": ""}),
|
|
"source_language2": (["", "zh", "en", "tr"], {"default": ""}),
|
|
"target_language2": (["", "es", "fr", "pt", "ko", "en"], {"default": "en"}),
|
|
"url3": ("STRING", {"default": ""}),
|
|
"source_language3": (["", "zh", "en", "tr"], {"default": ""}),
|
|
"target_language3": (["", "es", "fr", "pt", "ko", "en"], {"default": "en"}),
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("STRING",)
|
|
RETURN_NAMES = ("params_json",)
|
|
|
|
FUNCTION = "build"
|
|
|
|
OUTPUT_NODE = False
|
|
|
|
CATEGORY = "Malette"
|
|
|
|
def build(self, url1, source_language1="", target_language1="en",
|
|
url2="", source_language2="", target_language2="en",
|
|
url3="", source_language3="", target_language3="en"):
|
|
"""构建符合API要求的params JSON字符串"""
|
|
params_array = []
|
|
|
|
# 处理所有URL和对应的语言设置
|
|
url_configs = [
|
|
(url1, source_language1, target_language1),
|
|
(url2, source_language2, target_language2),
|
|
(url3, source_language3, target_language3),
|
|
]
|
|
|
|
for url, source_lang, target_lang in url_configs:
|
|
url = url.strip()
|
|
if not url:
|
|
continue
|
|
|
|
# 验证URL格式
|
|
if not (url.startswith('http://') or url.startswith('https://')):
|
|
print(f"[ImageTranslateParamsBuilder] 警告: URL格式不正确: {url}")
|
|
continue
|
|
|
|
param = {"url": url}
|
|
|
|
# 添加源语言（如果指定了）
|
|
if source_lang and source_lang.strip():
|
|
param["sourceLanguage"] = source_lang.strip()
|
|
|
|
# 添加目标语言（如果指定了）
|
|
if target_lang and target_lang.strip():
|
|
param["targetLanguage"] = target_lang.strip()
|
|
|
|
params_array.append(param)
|
|
|
|
if not params_array:
|
|
print("[ImageTranslateParamsBuilder] 警告: 没有有效的URL")
|
|
return (json.dumps({"params": []}),)
|
|
|
|
result = {"params": params_array}
|
|
result_json = json.dumps(result, ensure_ascii=False, indent=2)
|
|
print(f"[ImageTranslateParamsBuilder] 构建了 {len(params_array)} 个翻译参数")
|
|
|
|
return (result_json,)
|
|
|
|
|
|
class ImageTranslateResultExtractor:
|
|
"""图片翻译结果提取器节点"""
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
return {
|
|
"required": {
|
|
"api_response": ("STRING", {"forceInput": True}),
|
|
},
|
|
"optional": {
|
|
"extract_type": (["all", "result_urls", "result_jsons", "source_info"], {"default": "result_urls"}),
|
|
"index": ("INT", {"default": 0, "min": 0, "max": 100}),
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("STRING",)
|
|
RETURN_NAMES = ("extracted_data",)
|
|
|
|
FUNCTION = "extract"
|
|
|
|
OUTPUT_NODE = False
|
|
|
|
CATEGORY = "Malette"
|
|
|
|
def extract(self, api_response, extract_type="result_urls", index=0):
|
|
"""从API响应中提取特定数据"""
|
|
try:
|
|
# 解析API响应
|
|
response_data = json.loads(api_response)
|
|
|
|
# 检查是否有错误
|
|
if "error" in response_data:
|
|
return (api_response,)
|
|
|
|
# 获取数据列表
|
|
data_list = response_data.get("data", [])
|
|
if not data_list:
|
|
return (json.dumps([], ensure_ascii=False),)
|
|
|
|
if extract_type == "all":
|
|
# 返回完整的数据列表
|
|
result = json.dumps(data_list, ensure_ascii=False, indent=2)
|
|
|
|
elif extract_type == "result_urls":
|
|
# 提取所有结果图片URL
|
|
urls = [item.get("resultUrl", "") for item in data_list if item.get("resultUrl")]
|
|
result = json.dumps(urls, ensure_ascii=False)
|
|
|
|
elif extract_type == "result_jsons":
|
|
# 提取所有结果JSON
|
|
jsons = [item.get("resultJson", "") for item in data_list if item.get("resultJson")]
|
|
result = json.dumps(jsons, ensure_ascii=False)
|
|
|
|
elif extract_type == "source_info":
|
|
# 提取源信息（URL和语言）
|
|
source_info = []
|
|
for item in data_list:
|
|
info = {
|
|
"url": item.get("url", ""),
|
|
"sourceLanguage": item.get("sourceLanguage", ""),
|
|
"targetLanguage": item.get("targetLanguage", "")
|
|
}
|
|
source_info.append(info)
|
|
result = json.dumps(source_info, ensure_ascii=False, indent=2)
|
|
|
|
else:
|
|
result = json.dumps(data_list, ensure_ascii=False, indent=2)
|
|
|
|
print(f"[ImageTranslateResultExtractor] 提取类型: {extract_type}, 数据数量: {len(data_list)}")
|
|
return (result,)
|
|
|
|
except json.JSONDecodeError as e:
|
|
error_msg = f"JSON 解析失败: {str(e)}"
|
|
print(f"[ImageTranslateResultExtractor] {error_msg}")
|
|
return (json.dumps({"error": error_msg}, ensure_ascii=False),)
|
|
|
|
except Exception as e:
|
|
error_msg = f"提取过程出错: {str(e)}"
|
|
print(f"[ImageTranslateResultExtractor] {error_msg}")
|
|
return (json.dumps({"error": error_msg}, ensure_ascii=False),)
|
|
|
|
|
|
class LoadImagesFromUrls:
|
|
"""从URL列表加载图片节点"""
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
return {
|
|
"required": {
|
|
"urls": ("STRING", {"forceInput": True, "multiline": True}),
|
|
},
|
|
"optional": {
|
|
"input_format": (["json_array", "newline_separated", "auto"], {"default": "auto"}),
|
|
"max_images": ("INT", {"default": 10, "min": 1, "max": 50}),
|
|
"timeout": ("INT", {"default": 30, "min": 5, "max": 300}),
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("IMAGE", "STRING", "INT")
|
|
RETURN_NAMES = ("images", "loaded_urls", "count")
|
|
|
|
OUTPUT_IS_LIST = (True, False, False)
|
|
|
|
FUNCTION = "load_images"
|
|
|
|
OUTPUT_NODE = False
|
|
|
|
CATEGORY = "Malette"
|
|
|
|
def load_images(self, urls, input_format="auto", max_images=10, timeout=30):
|
|
"""从URL列表加载图片"""
|
|
try:
|
|
# 解析URL列表
|
|
url_list = self._parse_urls(urls, input_format)
|
|
|
|
if not url_list:
|
|
print("[LoadImagesFromUrls] 警告: 没有找到有效的URL")
|
|
# 返回空图片数据
|
|
empty_image = torch.zeros((1, 64, 64, 3), dtype=torch.float32)
|
|
return (empty_image, json.dumps([]), 0)
|
|
|
|
# 限制图片数量
|
|
url_list = url_list[:max_images]
|
|
|
|
loaded_images = []
|
|
loaded_urls = []
|
|
|
|
print(f"[LoadImagesFromUrls] 开始加载 {len(url_list)} 张图片")
|
|
|
|
for i, url in enumerate(url_list):
|
|
try:
|
|
print(f"[LoadImagesFromUrls] 正在加载第 {i+1}/{len(url_list)} 张图片: {url}")
|
|
|
|
# 验证URL格式
|
|
if not self._is_valid_url(url):
|
|
print(f"[LoadImagesFromUrls] 跳过无效URL: {url}")
|
|
continue
|
|
|
|
# 下载图片
|
|
response = requests.get(
|
|
url,
|
|
timeout=timeout,
|
|
headers={
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
|
|
},
|
|
verify=False
|
|
)
|
|
response.raise_for_status()
|
|
|
|
# 检查内容类型
|
|
content_type = response.headers.get('content-type', '')
|
|
if not content_type.startswith('image/'):
|
|
print(f"[LoadImagesFromUrls] 跳过非图片内容: {url}, content-type: {content_type}")
|
|
continue
|
|
|
|
# 将响应内容转换为PIL图片
|
|
image = Image.open(io.BytesIO(response.content))
|
|
|
|
# 确保图片是RGB模式
|
|
if image.mode != 'RGB':
|
|
image = image.convert('RGB')
|
|
|
|
# 转换为numpy数组
|
|
image_np = np.array(image).astype(np.float32) / 255.0
|
|
|
|
# 转换为torch张量 (H, W, C)
|
|
image_tensor = torch.from_numpy(image_np)
|
|
|
|
loaded_images.append(image_tensor)
|
|
loaded_urls.append(url)
|
|
|
|
print(f"[LoadImagesFromUrls] 成功加载图片: {image.size}, 模式: {image.mode}")
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"[LoadImagesFromUrls] 下载图片失败 {url}: {str(e)}")
|
|
continue
|
|
except Exception as e:
|
|
print(f"[LoadImagesFromUrls] 处理图片失败 {url}: {str(e)}")
|
|
continue
|
|
|
|
if not loaded_images:
|
|
print("[LoadImagesFromUrls] 警告: 没有成功加载任何图片")
|
|
# 返回空图片列表
|
|
empty_image = torch.zeros((64, 64, 3), dtype=torch.float32)
|
|
return ([empty_image], json.dumps([]), 0)
|
|
|
|
# 保持原始尺寸，将每张图片作为独立的张量返回
|
|
# 为每张图片添加batch维度，但保持原始尺寸
|
|
output_images = []
|
|
for i, img_tensor in enumerate(loaded_images):
|
|
# 图片格式为 (H, W, C)，添加batch维度变为 (1, H, W, C)
|
|
batched_img = img_tensor.unsqueeze(0)
|
|
output_images.append(batched_img)
|
|
print(f" 图片 {i+1}: {img_tensor.shape[1]}x{img_tensor.shape[0]} (WxH)")
|
|
|
|
print(f"[LoadImagesFromUrls] 成功加载 {len(loaded_images)} 张图片，保持原始尺寸")
|
|
|
|
return (output_images, json.dumps(loaded_urls, ensure_ascii=False), len(loaded_images))
|
|
|
|
except Exception as e:
|
|
error_msg = f"加载图片过程出错: {str(e)}"
|
|
print(f"[LoadImagesFromUrls] {error_msg}")
|
|
# 返回错误时的默认数据
|
|
empty_image = torch.zeros((1, 64, 64, 3), dtype=torch.float32)
|
|
return ([empty_image], json.dumps({"error": error_msg}, ensure_ascii=False), 0)
|
|
|
|
def _parse_urls(self, urls_input, input_format):
|
|
"""解析URL输入"""
|
|
urls_input = urls_input.strip()
|
|
if not urls_input:
|
|
return []
|
|
|
|
url_list = []
|
|
|
|
if input_format == "auto":
|
|
# 自动检测格式
|
|
try:
|
|
# 尝试解析为JSON
|
|
parsed = json.loads(urls_input)
|
|
if isinstance(parsed, list):
|
|
url_list = [str(url).strip() for url in parsed if str(url).strip()]
|
|
print("[LoadImagesFromUrls] 检测到JSON数组格式")
|
|
else:
|
|
raise ValueError("JSON不是数组格式")
|
|
except:
|
|
# 解析为换行分隔
|
|
url_list = [line.strip() for line in urls_input.split('\n') if line.strip()]
|
|
print("[LoadImagesFromUrls] 检测到换行分隔格式")
|
|
|
|
elif input_format == "json_array":
|
|
try:
|
|
parsed = json.loads(urls_input)
|
|
if isinstance(parsed, list):
|
|
url_list = [str(url).strip() for url in parsed if str(url).strip()]
|
|
else:
|
|
raise ValueError("JSON格式错误：不是数组")
|
|
except Exception as e:
|
|
print(f"[LoadImagesFromUrls] JSON解析失败: {str(e)}")
|
|
return []
|
|
|
|
elif input_format == "newline_separated":
|
|
url_list = [line.strip() for line in urls_input.split('\n') if line.strip()]
|
|
|
|
# 过滤有效的URL
|
|
valid_urls = [url for url in url_list if self._is_valid_url(url)]
|
|
|
|
print(f"[LoadImagesFromUrls] 解析得到 {len(valid_urls)} 个有效URL")
|
|
return valid_urls
|
|
|
|
def _is_valid_url(self, url):
|
|
"""验证URL格式"""
|
|
try:
|
|
result = urlparse(url)
|
|
return all([result.scheme in ['http', 'https'], result.netloc])
|
|
except:
|
|
return False
|
|
|
|
|
|
# 节点映射
|
|
NODE_CLASS_MAPPINGS = {
|
|
"ImageTranslateAPI": ImageTranslateAPI,
|
|
"ImageTranslateParamsBuilder": ImageTranslateParamsBuilder,
|
|
"ImageTranslateResultExtractor": ImageTranslateResultExtractor,
|
|
"ADIC_COMMON_API": ADIC_COMMON_API,
|
|
"LoadImagesFromUrls": LoadImagesFromUrls
|
|
}
|
|
|
|
NODE_DISPLAY_NAME_MAPPINGS = {
|
|
"ImageTranslateAPI": "图片翻译 API",
|
|
"ImageTranslateParamsBuilder": "图片翻译参数构建器",
|
|
"ImageTranslateResultExtractor": "图片翻译结果提取器",
|
|
"ADIC_COMMON_API": "ADIC Common API",
|
|
"LoadImagesFromUrls": "从URL列表加载图片"
|
|
} |