Files
pod_trend_agent/graph/backends/openai_image_backend.py
3218485270 3dc594cf57 v92-v105 多国适配 + 模板导出增强 + 生图可靠性优化
- 新增韩国(KR)适配:countries/pinterest 种子词、K-pop 提示词、UI 国家列表
- 爬虫跨运行持久化已采集 URL(.collected_urls.json),同关键词重复搜索采新图
- 模板导出增强:SPU 商品属性列名前缀剥离匹配、尺码下拉框 INDIRECT 动态引用、
  SPU 字段映射(袖长/门襟/胸垫等)、季节/印花图案女装映射、建议售价统一必填、
  SKU 分类/数量/单位直接填默认值
- 自定义模式:多模态提示词独立(custom_analyze_*)、生图提示词模板可配置
- Pinterest:空选品守卫、连续空分析保护(max_empty_analyze)、flat_prompt 配置覆盖
- 生图可靠性:开始/完成进度日志、异步任务轮询超时 60s→300s
2026-08-31 18:35:51 +08:00

319 lines
16 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""OpenAI 图像后端(生成设计稿 + 三图合成)。
支持 OpenAI images/generations(文生图)与 images/editsimg2img 多参考图),
兼容两类网关返回:
1) 同步:{"data": [{"b64_json" | "url"}]}
2) 异步任务(ai-media.vip 等重负载自动转异步):
提交返回 202 + {"object":"image.task","task_id","poll_url","poll_after_ms"}
需轮询 GET {base}/images/tasks/{task_id} 直到 succeeded,再从成功响应取图。
所有请求跳过环境代理(NO_PROXY),适配用户挂 VPN 时直连国内/自建网关。
"""
import base64
import json
import time
from pathlib import Path
from typing import Optional
import requests
import os
# 模型调用一律直连:用户常开 VPN(系统代理),网关多为国内/自建,走代理会被拦截或变慢。
# 环境变量级 NO_PROXY 双保险(requests/urllib3 均读取),Google 采集(pytrends)不受影响。
os.environ.setdefault('NO_PROXY', '*')
os.environ.setdefault('no_proxy', '*')
from .base import ImageBackend
# 直连策略:忽略环境代理(用户挂 VPN 时代理会拦截国内/自建网关的请求)
NO_PROXY = {"http": None, "https": None}
_FINAL_STATUS = ("succeeded", "completed", "done")
_FAIL_STATUS = ("failed", "error")
def _retry_after(resp, attempt: int) -> int:
"""429 限流等待秒数:优先取网关 Retry-After 头,否则指数退避(3/6/9s)。"""
ra = resp.headers.get("Retry-After")
try:
if ra is not None and ra.isdigit():
return min(int(ra), 30)
except Exception: # noqa: BLE001
pass
return 3 * (attempt + 1)
def _shrink_blob_to_2mb(img_path: str, blob: bytes, max_bytes: int = 2 * 1024 * 1024) -> bytes:
"""把大图压缩到 <max_bytes(默认 2MB)再上传:
- 尺寸过大先缩放(合成输入 1600x2200 内足够);
- 有透明通道 → PNG 优化;否则 → JPEG 循环降质。
失败时返回原 blob(不阻塞流程)。"""
try:
import io
from PIL import Image
img = Image.open(io.BytesIO(blob))
max_w, max_h = 1600, 2200
if img.width > max_w or img.height > max_h:
img.thumbnail((max_w, max_h))
has_alpha = img.mode in ("RGBA", "LA")
if not has_alpha:
img = img.convert("RGB")
for q in (88, 70, 50, 35):
buf = io.BytesIO()
if has_alpha:
img.save(buf, format="PNG", optimize=True)
else:
img.save(buf, format="JPEG", quality=q)
if buf.tell() <= max_bytes:
return buf.getvalue()
# 保底:最小质量 PNG/JPEG
buf = io.BytesIO()
if has_alpha:
img.save(buf, format="PNG", optimize=True)
else:
img.save(buf, format="JPEG", quality=30)
return buf.getvalue()
except Exception as e: # noqa: BLE001
print(f"[img] 图片压缩失败(用原图): {e}")
return blob
def _save_from_response(j: dict, out_path: str, started: float = None) -> str:
"""从同步/异步最终响应提取图片(data[].b64_json 或 url)并保存。"""
data_item = (j.get("data") or [{}])[0]
b64 = data_item.get("b64_json")
if b64:
Path(out_path).parent.mkdir(parents=True, exist_ok=True)
Path(out_path).write_bytes(base64.b64decode(b64))
else:
url = data_item.get("url")
if not url:
raise RuntimeError(f"图像 API 返回无 b64_json/url: {str(j)[:200]}")
img_resp = requests.get(url, proxies=NO_PROXY, timeout=120)
img_resp.raise_for_status()
Path(out_path).parent.mkdir(parents=True, exist_ok=True)
Path(out_path).write_bytes(img_resp.content)
if started is not None:
print(f"[img] 已生成(耗时 {time.time() - started:.0f}s: {out_path}")
return out_path
def _wait_task(base_url: str, headers: dict, task_id: str, poll_after_ms: int,
timeout: int = 300) -> dict:
"""轮询异步图像任务直到完成;uncertain(结果暂时不确定)继续等,不重复提交。
timeout 默认 300s(与提交请求 timeout 一致):网关异步任务实际需 30s~6 分钟
(设计稿/三图合成),60s 轮询超时会把仍在处理中的任务误判为失败而重复提交,
白白消耗配额和时间。超时即抛异常由调用方重试同步提交。"""
deadline = time.time() + timeout
last = ""
while time.time() < deadline:
time.sleep(max(int(poll_after_ms or 2000), 2000) / 1000.0)
try:
resp = requests.get(f"{base_url}/images/tasks/{task_id}", headers=headers,
timeout=60, proxies=NO_PROXY)
if resp.status_code >= 400:
continue
j = resp.json()
except Exception as e: # noqa: BLE001
print(f"[img] 任务轮询异常(重试): {e}")
continue
st = j.get("status")
if st != last:
print(f"[img] 异步任务 {task_id[:20]}… status={st}")
last = st
if st in _FINAL_STATUS:
return j
if st in _FAIL_STATUS:
raise RuntimeError(f"图像异步任务失败: {j.get('error')} {j.get('error_code')}")
# queued / running / uncertain → 继续等
raise RuntimeError(f"图像异步任务超时({timeout}s")
def _resolve_task_or_sync(j: dict, base_url: str, headers: dict, out_path: str,
started: float = None) -> str:
"""提交后的统一处理:异步任务则轮询,然后提取图片保存。"""
if j.get("object") == "image.task" or j.get("task_id") or j.get("id", "").startswith("imgtask"):
tid = j.get("task_id") or j.get("id") or ""
if not tid:
raise RuntimeError(f"异步任务无 task_id: {str(j)[:200]}")
j = _wait_task(base_url, headers, tid, int(j.get("poll_after_ms") or 2000))
return _save_from_response(j, out_path, started=started)
class OpenAIImageBackend(ImageBackend):
name = "openai"
def __init__(self):
self._cfg: dict = {}
def bind_config(self, cfg: dict):
self._cfg = cfg or {}
def _base_url(self) -> str:
return str(self._cfg.get("base_url") or "https://api.openai.com/v1").rstrip("/")
def print(self, prompt: str, base_image: str, out_path: str, negative: str = "",
extra_images=None, size: str = "", seed: Optional[int] = None) -> str:
"""img2img 编辑:参考图 base_image+ 可选 extra_images 多参考图)按 prompt 生成新图。
- 平铺服装图:base_image=平铺衣服底图(图3)extra_images=[印花设计稿(图2)]
- 三图模特合成:base_image=模特图(图1)extra_images=[印花设计稿(图2), 平铺底图(图3)]
提交顺序即图1→图2→图3,与提示词中的图片角色一一对应。
size: 显式尺寸覆盖(如 "1536x2048");留空用配置 size(默认 1024x1024)。
seed: 随机种子(None=不传,网关随机;固定值=可复现,网关支持才生效)。
"""
cfg = self._cfg
api_key = cfg.get("api_key", "")
model = cfg.get("model", "gpt-image-1")
if not api_key:
raise RuntimeError("compose.api_key 未配置,无法调用 OpenAI 图像后端。")
full_prompt = prompt + (f"\nNegative: {negative}" if negative else "")
base_url = self._base_url()
headers = {"Authorization": f"Bearer {api_key}"}
# 必须把文件读成 bytes 再提交(句柄随 with 关闭会导致 "read of closed file");
# 输入图(模特/设计/底图)先压缩校验:>2MB 压缩到 <2MB 再上传,避免大图拖垮网关导致超时
file_payloads = []
for img_path in [base_image] + list(extra_images or []):
blob = Path(img_path).read_bytes()
if len(blob) > 2 * 1024 * 1024:
shrunk = _shrink_blob_to_2mb(img_path, blob)
print(f"[img] 输入图压缩: {Path(img_path).name} {len(blob)//1024}KB → {len(shrunk)//1024}KB<2MB")
blob = shrunk
file_payloads.append((Path(img_path).name, blob))
files = [("image", (name, blob, "image/png")) for name, blob in file_payloads]
data = {
"prompt": full_prompt,
"n": 1,
"size": size or cfg.get("size", "1024x1024"),
"model": model,
}
# execution_mode 仅对明确支持的网关(如 ai-media)传;yunfei 等标准 OpenAI 网关
# 不认识该参数,硬传会导致空响应,故默认不传,仅 config 显式配置时才带上
em = str(cfg.get("execution_mode") or "").strip()
if em:
data["execution_mode"] = em
if seed is not None:
data["seed"] = seed
started = time.time()
print(f"[img] 开始图生图({model}{data['size']},参考图 {len(file_payloads)} 张)→ {Path(out_path).name}")
# 提交重试:异步路径不稳定 → 失败重试同步提交(最多 3 次);
# 内容政策拦截(content_policy_violation)多为网关误判 → 等待后重试
last_err: Optional[str] = None
for attempt in range(3):
resp = requests.post(f"{base_url}/images/edits", headers=headers, files=files,
data=data, timeout=300, proxies=NO_PROXY)
if resp.status_code == 429:
# 限流:尊重网关负载退避等待再重试,避免硬撞雪崩
wait = _retry_after(resp, attempt)
print(f"[img] 图像 API 429 限流,等待 {wait}s 重试 {attempt + 1}/3")
time.sleep(wait)
continue
if resp.status_code >= 500:
# 5xx 服务端错误(500/502/503/504 超时等):网关临时故障/过载,退避后重试
body = resp.text or ""
last_err = f"图像 API {resp.status_code}: {body[:300]}"
wait = 3 * (attempt + 1)
print(f"[img] 图像 API {resp.status_code}(网关临时故障/超时),等待 {wait}s 重试 {attempt + 1}/3")
time.sleep(wait)
continue
if resp.status_code >= 400:
body = resp.text or ""
if "content_policy" in body and attempt < 2:
print(f"[img] 内容政策拦截(可能误判),等待后重试 {attempt + 2}/3")
time.sleep(2)
continue
if attempt == 0:
data.pop("execution_mode", None) # 网关不认识该参数(官方 OpenAI)→ 去掉重试
continue
raise RuntimeError(f"图像 API {resp.status_code}: {body[:300]}")
try:
return _resolve_task_or_sync(resp.json(), base_url, headers, out_path,
started=started)
except json.JSONDecodeError as e:
# 空/非 JSON 响应:多为网关过载返回空 body → 退避后重试,
# 避免即时重压触发网关 429(用户实测空响应风暴 → 429)
last_err = str(e)
wait = 3 * (attempt + 1)
print(f"[img] 网关空响应(JSON 解析失败),等待 {wait}s 重试提交 {attempt + 1}/3: {e}")
time.sleep(wait)
except Exception as e: # noqa: BLE001
last_err = str(e)
print(f"[img] 第 {attempt + 1} 次提交异步失败,等待后重试同步提交: {e}")
time.sleep(2 * (attempt + 1))
raise RuntimeError(f"图像合成多次提交均失败: {last_err}")
def generate(self, prompt: str, out_path: str, negative: str = "", size: str = "",
seed: Optional[int] = None) -> str:
"""纯文生图:生成白底纯印花设计稿(standalone pure print design)。
size: 显式尺寸覆盖(印花设计统一 1024x1024);留空用配置 size。
seed: 随机种子(None=不传,网关随机;固定值=可复现,网关支持才生效)。"""
cfg = self._cfg
api_key = cfg.get("api_key", "")
model = cfg.get("model", "gpt-image-1")
if not api_key:
raise RuntimeError("compose.api_key 未配置,无法调用 OpenAI 图像后端。")
full_prompt = prompt + (f"\nNegative: {negative}" if negative else "")
base_url = self._base_url()
headers = {"Authorization": f"Bearer {api_key}"}
data = {
"prompt": full_prompt,
"n": 1,
"size": size or cfg.get("size", "1024x1024"),
"model": model,
"response_format": "b64_json",
}
# execution_mode 仅对明确支持的网关(如 ai-media)传;yunfei 等标准 OpenAI 网关
# 不认识该参数,硬传会导致空响应,故默认不传,仅 config 显式配置时才带上
em = str(cfg.get("execution_mode") or "").strip()
if em:
data["execution_mode"] = em
if seed is not None:
data["seed"] = seed
started = time.time()
print(f"[img] 开始文生图({model}{data['size']})→ {Path(out_path).name}")
last_err: Optional[str] = None
for attempt in range(3):
resp = requests.post(f"{base_url}/images/generations", headers=headers, json=data,
timeout=300, proxies=NO_PROXY)
if resp.status_code == 429:
# 限流:尊重网关负载退避等待再重试,避免硬撞雪崩
wait = _retry_after(resp, attempt)
print(f"[img] 图像 API 429 限流,等待 {wait}s 重试 {attempt + 1}/3")
time.sleep(wait)
continue
if resp.status_code >= 500:
# 5xx 服务端错误(500/502/503/504 超时等):网关临时故障/过载,退避后重试
body = resp.text or ""
last_err = f"图像 API {resp.status_code}: {body[:300]}"
wait = 3 * (attempt + 1)
print(f"[img] 图像 API {resp.status_code}(网关临时故障/超时),等待 {wait}s 重试 {attempt + 1}/3")
time.sleep(wait)
continue
if resp.status_code >= 400:
body = resp.text or ""
if "content_policy" in body and attempt < 2:
print(f"[img] 内容政策拦截(可能误判),等待后重试 {attempt + 2}/3")
time.sleep(2)
continue
if attempt == 0:
data.pop("execution_mode", None) # 网关不认识该参数(官方 OpenAI)→ 去掉重试
continue
raise RuntimeError(f"图像 API {resp.status_code}: {body[:300]}")
try:
return _resolve_task_or_sync(resp.json(), base_url, headers, out_path,
started=started)
except json.JSONDecodeError as e:
# 空/非 JSON 响应:多为网关过载返回空 body → 退避后重试,
# 避免即时重压触发网关 429(用户实测空响应风暴 → 429)
last_err = str(e)
wait = 3 * (attempt + 1)
print(f"[img] 网关空响应(JSON 解析失败),等待 {wait}s 重试生成 {attempt + 1}/3: {e}")
time.sleep(wait)
except Exception as e: # noqa: BLE001
last_err = str(e)
print(f"[img] 第 {attempt + 1} 次生成异步失败,等待后重试同步提交: {e}")
time.sleep(2 * (attempt + 1))
raise RuntimeError(f"图像生成多次提交均失败: {last_err}")