from runtime import Args
from typings.nanobanana_apiyi.nanobanana_apiyi import Input, Output
import requests
import base64
import io
import oss2
import uuid
import re
from datetime import datetime
# 阿里雲 OSS 配置
ACCESS_KEY_ID = "" #填入自己的阿里雲 Access Key ID
ACCESS_KEY_SECRET = "" #填入自己的阿里雲 Access Key Secret
BUCKET_NAME = "" #填入自己的阿里雲 OSS Bucket 名稱
ENDPOINT = "oss-cn-beijing.aliyuncs.com" #填入自己的阿里雲 OSS Endpoint，例如 "oss-cn-beijing.aliyuncs.com"
# 解析度超時時間
TIMEOUT = {
"1K": 360, # 快速預覽
"2K": 600, # 推薦使用
"4K": 1200, # 超高畫質
}
def upload_base64_to_oss(image_base64: str) -> str:
"""
將 base64 圖片上傳到阿里雲 OSS 並返回連結
支援帶 data:image/...;base64, 字首 和 純 base64 兩種情況
"""
# 去掉 data:image/...;base64, 字首
base64_str = re.sub(r"^data:image/[^;]+;base64,", "", image_base64)
image_data = base64.b64decode(base64_str)
image_io = io.BytesIO(image_data)
auth = oss2.Auth(ACCESS_KEY_ID, ACCESS_KEY_SECRET)
bucket = oss2.Bucket(auth, ENDPOINT, BUCKET_NAME)
object_name = f"coze/generated_{uuid.uuid4().hex}.png"
bucket.put_object(object_name, image_io)
return f"https://{BUCKET_NAME}.{ENDPOINT}/{object_name}"
# ==============================
# 工具函式：根據 URL 猜測 MIME 型別
# ==============================
def guess_mime_from_url(url: str) -> str:
url_lower = url.lower()
if url_lower.endswith(".png"):
return "image/png"
if url_lower.endswith(".jpg") or url_lower.endswith(".jpeg"):
return "image/jpeg"
if url_lower.endswith(".webp"):
return "image/webp"
if url_lower.endswith(".gif"):
return "image/gif"
# 預設
return "image/png"
# ==============================
# 核心：生成 / 編輯圖片
# ==============================
def generate_image(prompt: str, aspect_ratio: str, resolution: str, apikey:str,apiurl:str,image_urls=None):
"""
生成 / 編輯圖片的核心函式
- 如果 image_urls 為空：純文生圖
- 如果 image_urls 不為空：把 URL 指向的圖片下載下來，按 inline_data 方式傳給 API，實現改圖
"""
# 組裝 parts
parts = []
# 1. 如果有圖片 URL，則按 apiyi 改圖 demo 的方式構造 inline_data
if image_urls:
for url in image_urls:
try:
resp = requests.get(url, timeout=180)
if resp.status_code != 200:
return {
"success": False,
"error": f"圖片上傳階段，獲取圖片失敗（{url}）HTTP {resp.status_code}"
}
image_bytes = resp.content
image_base64 = base64.b64encode(image_bytes).decode("utf-8")
mime_type = guess_mime_from_url(url)
parts.append({
"inline_data": {
"mime_type": mime_type,
"data": image_base64
}
})
except Exception as e:
return {
"success": False,
"error": f"圖片上傳階段，獲取圖片失敗（{url}）: {e}"
}
# 2. 文字部分（編輯指令或文生圖提示詞）
# 注意：這裡不再把圖片 URL 塞進 prompt 裡，僅用純文字描述
if prompt:
parts.append({"text": prompt})
else:
# 沒有文字時給一個預設提示（可按需要修改）
parts.append({"text": "根據圖片進行合理的編輯生成。"})
# 3. 構造請求 payload（和官方改圖 demo 一致的結構）
payload = {
"contents": [
{
"parts": parts
}
],
"generationConfig": {
"responseModalities": ["IMAGE"],
"imageConfig": {
"aspectRatio": aspect_ratio,
"image_size": resolution
}
}
}
headers = {
"Authorization": f"Bearer {apikey}",
"Content-Type": "application/json"
}
try:
response = requests.post(apiurl, headers=headers, json=payload, timeout=TIMEOUT[resolution])
# HTTP 非200
if response.status_code != 200:
return {"success": False, "error": f"HTTP {response.status_code}: {response.text}"}
# JSON 解析
try:
data = response.json()
except ValueError:
return {"success": False, "error": "響應不是有效JSON", "response": (response.text or "")[:500]}
# 1️⃣ 最高優先順序：candidatesTokenCount
usage = data.get("usageMetadata") or {}
if usage.get("candidatesTokenCount") == 0:
return {
"success": False,
"errorType": "ZERO_CANDIDATES_TOKEN",
"error": "❌ 內容稽核失敗\n您的請求在內容稽核階段被拒絕，請修改提示詞或圖片",
"response": data
}
# 2️⃣ candidates 檢查
candidates = data.get("candidates")
if not isinstance(candidates, list) or len(candidates) == 0:
return {
"success": False,
"errorType": "NO_CANDIDATES",
"error": "系統出錯，請稍後重試",
"response": data
}
candidate = candidates[0] if isinstance(candidates[0], dict) else None
if candidate is None:
return {
"success": False,
"errorType": "NO_CANDIDATES",
"error": "系統出錯，請稍後重試（candidates[0]結構異常）",
"response": data
}
# 3️⃣ finishReason
finish_reason = candidate.get("finishReason")
if isinstance(finish_reason, str) and finish_reason != "STOP":
reason_map = {
"PROHIBITED_CONTENT": "內容違反安全策略，已被拒絕處理",
"SAFETY": "內容觸發了安全過濾器",
"RECITATION": "內容可能涉及版權問題",
"MAX_TOKENS": "內容長度超出限制",
}
return {
"success": False,
"errorType": "FINISH_REASON",
"finishReason": finish_reason,
"error": reason_map.get(finish_reason, f"請求被拒絕：{finish_reason}"),
"response": data
}
# 4️⃣ content.parts
content = candidate.get("content") or {}
parts = content.get("parts")
if not isinstance(parts, list) or len(parts) == 0:
return {
"success": False,
"errorType": "NO_PARTS",
"error": "生成失敗，請重試（content.parts為空）",
"response": data
}
# 5️⃣ 提取圖片和文本（更精準：識別 inlineData 存在但 data 為空）
images = []
texts = []
saw_inline_but_empty = False
for i, part in enumerate(parts):
if not isinstance(part, dict):
continue
# 收集 text（即使有 thoughtSignature，也照收）
t = part.get("text")
if isinstance(t, str) and t.strip() and not t.startswith("data:image/"):
texts.append(t.strip())
# 相容 inlineData / inline_data
inline = None
if isinstance(part.get("inlineData"), dict):
inline = part["inlineData"]
elif isinstance(part.get("inline_data"), dict):
inline = part["inline_data"]
if inline is not None:
b64 = inline.get("data")
if not isinstance(b64, str) or not b64.strip():
saw_inline_but_empty = True
continue
images.append(b64.strip())
# ✅ 更精準：inlineData 存在但全都沒 data
if not images and saw_inline_but_empty:
return {
"success": False,
"errorType": "INLINE_DATA_EMPTY",
"error": "生成失敗：檢測到 inlineData 但圖片資料為空（inlineData.data為空）",
"response": data
}
# 6️⃣ 有圖片：成功（保持你原來的返回結構）
if images:
return {"success": True, "image_data": images[0]}
# 7️⃣ 無圖片：有文本 -> TEXT_RESPONSE
if texts:
text_content = "\n".join(texts)
# —— 可選：不做函式，直接就地識別型別（想更簡單可刪掉這段 detectedType）——
low = text_content.lower()
detected = "general"
if any(k in low for k in ["watermark", "remove watermark", "去水印", "移除水印", "刪除水印"]):
detected = "拒絕處理水印任務"
elif any(k in low for k in ["faceswap", "face swap", "換臉", "deepfake"]):
detected = "拒絕處理換臉任務"
elif any(k in low for k in ["sexually", "explicit", "porn", "nude", "nsfw", "色情", "不雅", "裸"]):
detected = "拒絕色情任務"
elif any(str(y) in low for y in range(2026, 2101)):
detected = "拒絕超過知識庫範圍任務"
return {
"success": False,
"errorType": "TEXT_RESPONSE",
"error": detected, # 你文件要求：直接展示 API text
"response": data
}
# ✅ 更精準：parts 有結構但既無圖也無文本
return {
"success": False,
"error": "生成失敗：parts存在但未找到圖片資料或文本說明",
"response": data
}
except requests.exceptions.Timeout:
return {"success": False, "error": f"圖片生成請求超時（超過 {TIMEOUT[resolution]} 秒）"}
except Exception as e:
return {"success": False, "error": f"圖片生成請求失敗: {str(e)}"}
# ==============================
# Coze Node 入口
# ==============================
def handler(args: Args[Input]) -> Output:
"""
Coze / NanobananaPro 節點入口
- args.input.cleantext: 使用者文字提示詞
- args.input.fileurls: 使用者上傳圖片的 URL 列表（用於改圖）
- args.input.aspect_ratio: 寬高比，如 "1:1" / "9:16"
- args.input.resolution: 解析度，如 "1K" / "2K" / "4K"
"""
API_URL = "https://api.apiyi.com/v1beta/models/gemini-3-pro-image-preview:generateContent"
API_KEY = args.input.apikey
cleanttext = args.input.cleantext or ""
fileurls = args.input.fileurls or []
aspectratio = args.input.aspect_ratio
resolution = args.input.resolution
# - 圖片通過 image_urls 傳入 generate_image，走 inline_data 改圖邏輯
prompt = cleanttext.strip()
# 呼叫 Gemini 3 Pro 生成 / 編輯圖片
# 如果 fileurls 不為空，會按"改圖"模式呼叫
result = generate_image(prompt, aspectratio, resolution, API_KEY,API_URL,image_urls=fileurls if fileurls else None)
if result["success"]:
image_base64 = result["image_data"]
oss_url = upload_base64_to_oss(image_base64)
return {"analysis": "圖片生成成功", "url": oss_url, "error": None}
else:
return {"analysis": "圖片生成失敗", "url": None, "error": result["error"]}
