""" 生图通道 — gpt-image-2 主（edits 带产品图） / Gemini 备 + 重试退避 扒自：worker/src/image.js generateOneImage / requestProviderImage / imageProviderOrder 新增：asyncio 重试退避（上线版缺的，banana 有 _retry 思路） 铁律： - IMAGE_PROVIDER_PRIMARY/FALLBACK 走环境变量，不写死 - GPT 主通道必须有产品参考图，无图报错（禁纯文生图防产品跑偏） - key 不在本模块，由 worker 传入构造好的 async HTTP client """ from __future__ import annotations import asyncio import logging import os from typing import Any, Protocol from .constants import IMAGE_RETRY_ATTEMPTS, IMAGE_RETRY_BACKOFF_BASE, IMAGE_SIZE_DEFAULT, ROLE_SCENE_PREFERENCE from .image_scorer import score_image from .storyboard import plan_image_set, sanitize_text logger = logging.getLogger(__name__) def _pick_reference_for_role( role: str, images_by_scene: dict[str, list[bytes]] | None, fallback: list[bytes] | None, ) -> tuple[list[bytes] | None, str]: """R5多图：按分镜 role 选该场景的产品图。取不到回落主图。 返回 (参考图bytes列表, 命中scene标签用于日志)。 """ if images_by_scene: for scene in ROLE_SCENE_PREFERENCE.get(role, ["primary"]): imgs = images_by_scene.get(scene) if imgs: return imgs, scene # 偏好全落空：用任意可用图兜底（仍优先 primary） if images_by_scene.get("primary"): return images_by_scene["primary"], "primary" for scene, imgs in images_by_scene.items(): if imgs: return imgs, f"{scene}(兜底)" return fallback, "fallback" class ImageClient(Protocol): """worker 注入的图片生成客户端协议（隔离 key 细节）""" async def gpt_edits( self, prompt: str, reference_images: list[bytes], size: str, provider: str | None = None ) -> bytes: ... async def gpt_generate(self, prompt: str, size: str, provider: str | None = None) -> bytes: ... async def gemini_generate( self, prompt: str, reference_images: list[bytes], model: str ) -> bytes: ... def _image_provider_order() -> list[str]: """从环境变量读主备顺序（扒 imageProviderOrder）""" primary = os.environ.get("IMAGE_PROVIDER_PRIMARY", "gpt").lower() fallback = os.environ.get("IMAGE_PROVIDER_FALLBACK", "gemini").lower() seen: list[str] = [] for p in [primary, fallback]: if p and p not in seen: seen.append(p) return seen def _gemini_models() -> list[str]: """Gemini fallback 模型列表（多模型依次重试）""" env_val = os.environ.get("GEMINI_IMAGE_MODELS", "gemini-2.0-flash-preview-image-generation,imagen-3.0-generate-002") return [m.strip() for m in env_val.split(",") if m.strip()] async def _retry(coro_fn, attempts: int = IMAGE_RETRY_ATTEMPTS, backoff: float = IMAGE_RETRY_BACKOFF_BASE) -> Any: """指数退避重试（扒 banana _retry 思路）""" last_exc: Exception | None = None for i in range(attempts): try: return await coro_fn() except Exception as exc: last_exc = exc if i < attempts - 1: wait = backoff ** i logger.warning("生图失败第%d次，%.1fs后重试：%s", i + 1, wait, exc) await asyncio.sleep(wait) raise RuntimeError(f"重试{attempts}次均失败") from last_exc async def _request_gpt(client: ImageClient, prompt: str, reference_images: list[bytes], provider: str | None = None) -> bytes: if reference_images: return await client.gpt_edits(prompt, reference_images, IMAGE_SIZE_DEFAULT, provider) # 无产品参考图时降级为纯文生图（需 ALLOW_TEXT_ONLY_IMAGE=true 或 M2阶段） allow_text_only = os.environ.get("ALLOW_TEXT_ONLY_IMAGE", "true").lower() == "true" if allow_text_only: logger.warning("无产品参考图，降级为纯文生图（可能产品跑偏，建议前端上传参考图）") return await client.gpt_generate(prompt, IMAGE_SIZE_DEFAULT, provider) raise ValueError("GPT 主通道缺产品图：禁止纯文生图以免产品跑偏（设 ALLOW_TEXT_ONLY_IMAGE=true 可解锁）") async def _request_gemini(client: ImageClient, prompt: str, reference_images: list[bytes]) -> bytes: errors: list[str] = [] for model in _gemini_models(): try: return await client.gemini_generate(prompt, reference_images, model) except Exception as exc: errors.append(f"{model}: {exc}") raise RuntimeError("Gemini 全部模型失败：" + "；".join(errors)) async def generate_one_image( client: ImageClient, prompt: str, reference_images: list[bytes] | None = None, ) -> bytes: """ 主入口：按主备顺序依次尝试，每个 provider 内部有重试退避。 返回图片 bytes（PNG/JPEG）。 """ refs = reference_images or [] providers = _image_provider_order() errors: list[str] = [] for provider in providers: try: # apiports/codeproxy/openai 都是 OpenAI 兼容中转站，走 gpt 协议， # 但传 provider 进去 → client 按 provider 切到对应中转站的 base+key。 # 这才是真主备：apiports 503 → codeproxy 用独立 base+key 顶上。 if provider in ("apiports", "codeproxy", "openai"): img = await _retry(lambda p=provider: _request_gpt(client, prompt, refs, p)) elif provider == "gpt": img = await _retry(lambda: _request_gpt(client, prompt, refs, None)) elif provider == "gemini": img = await _retry(lambda: _request_gemini(client, prompt, refs)) else: raise ValueError(f"未知图片通道：{provider}") return img except Exception as exc: errors.append(f"{provider}: {exc}") logger.warning("图片通道 %s 失败，尝试下一个：%s", provider, exc) raise RuntimeError("所有图片通道均失败：" + "；".join(errors)) async def generate_storyboard_images( client: ImageClient, note: dict, product: dict, image_count: int = 3, reference_images: list[bytes] | None = None, analysis: dict | None = None, strategy: str | None = None, target_role: str | None = None, custom_prompt: str | None = None, images_by_scene: dict[str, list[bytes]] | None = None, flywheel_fragment: str | None = None, ) -> list[dict]: """ 按 storyboard 逐张生图（asyncio.gather 并发），返回每张结果列表。 strategy: None=默认叙事，'A'/'B'/'C'=三套正交叙事策略 target_role: 非空时只生成该 role 那一张（R2 单张重生） custom_prompt: 非空时追加到每张 per_prompt 末尾（R2 人工提示词） images_by_scene: R5多图，{scene: [bytes]}，按分镜 role 选对应场景图； 为空则全分镜共用 reference_images（向后兼容）。 flywheel_fragment: R7 飞轮偏好片段（最近选图/打回真实信号聚合），注入图片 排版偏好；仅影响文字角度/版式取向，绝不改瓶身（合规红线）。 每项：{role, name, image_bytes, error} """ plan = plan_image_set(note, product, image_count, analysis, strategy=strategy) storyboard = plan["storyboard"] base_prompt = plan["base_prompt"] # R2 单张重生：只保留目标 role 的分镜（匹配不到则原样全跑，避免空结果） if target_role: _filtered = [it for it in storyboard if it.get("role") == target_role] if _filtered: storyboard = _filtered async def _gen_one(item: dict) -> dict: # 逐图 prompt 9 字段（扒 promptFromStoryboard:323-334），每张差异化 brand_line = "" if item.get("brand_keyword"): brand_line = f"品牌词=「{item['brand_keyword']}」，{item.get('brand_keyword_rule','自然植入')}。" per_prompt = ( f"{base_prompt}\n" f"本张名称={item['name']}。" f"本张目标={item.get('goal') or item['focus']}。" f"图上主文字=「{sanitize_text(item.get('overlay_text',''), 20)}」。" f"使用卖点={item.get('selling_point','')}。" f"文案依据={item.get('source_basis','')}。" f"画面主体={item.get('visual_strategy','')}。" f"素材使用={item.get('asset_use','')}。" f"{brand_line}" f"禁止事项={item.get('forbidden','')}。" "排版要求：独立小红书3:4图文海报，画面完整；标题只出现一次，不与其他页重复；" "中文文字少而清晰，主标题+最多3个短点位；可自然用✅✨🌿💧🪞🧴📦🔍种草符号但不堆砌；" "不要生成App截图或笔记详情页界面。" ) # R2 人工提示词：追加到末尾权重最高，但不覆盖前面合规/真实约束 if custom_prompt: per_prompt += f"\n运营补充要求（在不违反上述合规与真实约束前提下尽量满足）：{sanitize_text(custom_prompt, 200)}。" # R7 飞轮偏好：仅作用于文字角度/版式取向参考，绝不改瓶身（合规+真实红线） if flywheel_fragment: per_prompt += ( f"\n历史偏好参考（仅影响标题文字角度与排版取向，不得据此改动产品瓶身）：" f"{sanitize_text(flywheel_fragment, 300)}。" ) try: # R5多图：按本张分镜 role 选对应场景产品图；无多图则共用 reference_images ref_for_item, _scene_hit = _pick_reference_for_role( item["role"], images_by_scene, reference_images ) if images_by_scene: logger.info("分镜 %s 选用产品图场景=%s", item["role"], _scene_hit) img_bytes = await generate_one_image(client, per_prompt, ref_for_item) # 注：gpt-image-2 渲染中文偶发错别字(约1/6)。vision/OCR 文字校验闸门实测 # 不可靠(漏报形近字+幻觉误伤品牌词),倩倩姐2026-06-16拍板先撤,纯生图, # 错别字作已知问题记录,后续迭代再处理。详见记忆 clover-image-text-check-shelved。 # # C3 canvas 叠字口子(倩倩姐2026-06-12拍板"只留口子不实现")： # 当 OVERLAY_TEXT_RENDER_ENABLED=True 时，此处由 PIL 在模型出的干净底图上 # 叠 item['overlay_text']/brand_keyword(字体资源+排版坐标后续补)，彻底解决中文乱码。 # TODO(C3-overlay): from .constants import OVERLAY_TEXT_RENDER_ENABLED # if OVERLAY_TEXT_RENDER_ENABLED: img_bytes = overlay_text_on_image(img_bytes, item) # E12 AI评图分：只做展示+排序,绝不进飞轮权重,失败返 None 不阻断(倩倩姐2026-06-16)。 visual = await score_image(client, img_bytes) return {"role": item["role"], "name": item["name"], "image_bytes": img_bytes, "error": None, "text_review": None, "ai_visual_score": (visual or {}).get("score"), "ai_visual_note": (visual or {}).get("note")} except Exception as exc: logger.error("分镜 %s 生图失败: %s", item["role"], exc) return {"role": item["role"], "name": item["name"], "image_bytes": None, "error": str(exc), "text_review": None, "ai_visual_score": None, "ai_visual_note": None} # 限并发：apiports 图片接口有 QPS 限制，6 张全并发会撞 429/503 concurrency = int(os.environ.get("IMAGE_CONCURRENCY", "2")) sem = asyncio.Semaphore(max(1, concurrency)) async def _gen_guarded(item: dict) -> dict: async with sem: return await _gen_one(item) results = await asyncio.gather(*(_gen_guarded(item) for item in storyboard)) return list(results)