df1856d793
- 产品编辑入口统一走 ProductFormFull(卖点/风格/人群/品牌词全字段); 修复开任务页 <form> 套 <form> 致"编辑产品"报错、改不了、跳回首个产品 - dashboard 入口卡片对齐实际路由: 系统管理(/config) 与 工作配置(/settings) 分开; settings ?tab=products 直达改用挂载后读 URL, 消除 hydration mismatch - 新增用户管理(users API/admin service/改密页) + alembic 022/023/024 - 上线部署: Dockerfile / docker-compose.prod+https / nginx https / .env.example - A8 三套正交叙事(痛点/场景/成分背书) + beige 调色去AI化 + 飞轮 text_import 高权重信号 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
300 lines
16 KiB
Python
300 lines
16 KiB
Python
"""
|
||
生图通道 — 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 偏好排序。
|
||
|
||
🔴 产品内外一体，绝不按 role 只喂单张（倩倩姐2026-06-26拍板纠正）：
|
||
旧逻辑命中偏好场景就 return 单张 → 质地特写分镜只拿到膏体微距、没有主图，
|
||
模型缺完整瓶身锚点 → 脑补瓶身走样+乱印字（"3张好/4张每套坏一张"的根因）。
|
||
现改为每张分镜都把所有场景图一起传，只把本 role 最相关的场景排最前
|
||
（多模态对靠前的图权重更高），且主图(primary)始终在列表内做瓶身锚点。
|
||
|
||
返回 (排序后的全部参考图bytes列表, 排序说明用于日志)。
|
||
"""
|
||
if images_by_scene:
|
||
prefs = ROLE_SCENE_PREFERENCE.get(role, ["primary"])
|
||
ordered: list[bytes] = []
|
||
used_scenes: list[str] = []
|
||
# 1) 先按 role 偏好顺序排（本 role 最相关的视角靠前）
|
||
for scene in prefs:
|
||
imgs = images_by_scene.get(scene)
|
||
if imgs:
|
||
ordered.extend(imgs)
|
||
used_scenes.append(scene)
|
||
# 2) 主图始终保底进列表（即使不在偏好里，也要补做瓶身锚点）
|
||
if "primary" not in used_scenes and images_by_scene.get("primary"):
|
||
ordered.extend(images_by_scene["primary"])
|
||
used_scenes.append("primary")
|
||
# 3) 其余场景图也补进来：同一产品的所有视角都给模型看，防瓶身走样
|
||
for scene, imgs in images_by_scene.items():
|
||
if scene not in used_scenes and imgs:
|
||
ordered.extend(imgs)
|
||
used_scenes.append(scene)
|
||
if ordered:
|
||
return ordered, "+".join(used_scenes)
|
||
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 has_alt_channel(self) -> bool: ...
|
||
|
||
|
||
def _image_provider_order(has_alt: bool = True) -> list[str]:
|
||
"""从环境变量读主备顺序（扒 imageProviderOrder）。
|
||
|
||
🔴 真双通道互备（倩倩姐2026-06拍板）：codeproxy / apiports 是两个独立中转站,
|
||
各用自己的 base+key, 任一站上游挂(502/503)由另一站顶上 —— 这才是真备份。
|
||
若 .env 把主备都设成同一个站(如都 codeproxy), 去重后只剩一个 → 退化成"假备份"
|
||
(一站挂全挂)。故只要序列用到 GPT 中转站家族, 就自动补齐缺失的那个,
|
||
保证 codeproxy+apiports 双通道都在。"gpt" 是 apiports 的等价别名(同 base+主key),
|
||
归一化掉避免重复尝试同一站。codeproxy 走 /images/edits，apiports 走
|
||
/chat/completions 多模态编辑；两者都必须带产品参考图，不碰纯文生图。
|
||
|
||
🔴 has_alt=False（用户/env 都没配 codeproxy key）时，绝不把 codeproxy 补进序列——
|
||
否则每张图都白白尝试 codeproxy → 撞"绝不静默降级"红线 → 空转重试 3 次拖慢整体
|
||
（倩倩姐2026-06-30方案A）。录了 key（has_alt=True）才补齐，真双通道逻辑不变。
|
||
"""
|
||
primary = os.environ.get("IMAGE_PROVIDER_PRIMARY", "gpt").lower()
|
||
fallback = os.environ.get("IMAGE_PROVIDER_FALLBACK", "gemini").lower()
|
||
order: list[str] = []
|
||
for p in [primary, fallback]:
|
||
p = "apiports" if p == "gpt" else p
|
||
# 没配 codeproxy key 时，显式点名 codeproxy 也跳过（避免空转重试）
|
||
if p == "codeproxy" and not has_alt:
|
||
continue
|
||
if p and p not in order:
|
||
order.append(p)
|
||
if any(p in ("codeproxy", "apiports") for p in order):
|
||
for ch in ("codeproxy", "apiports"):
|
||
# codeproxy 仅在真有备用 key 时补齐
|
||
if ch == "codeproxy" and not has_alt:
|
||
continue
|
||
if ch not in order:
|
||
order.append(ch)
|
||
return order
|
||
|
||
|
||
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)
|
||
# 🔴 红线：瓶身忠于参考图原样，禁止纯文生图脑补产品（倩倩姐2026-06-15拍板）。
|
||
# ALLOW_TEXT_ONLY_IMAGE 默认 false，即使显式设为 true 也拒绝——纯文生图会让产品包装
|
||
# 完全由模型自由发挥，严重走样，不符合合规与真实双重要求。
|
||
# 若要解锁请联系产品负责人，不在代码层开放。
|
||
_env_val = os.environ.get("ALLOW_TEXT_ONLY_IMAGE", "false").lower()
|
||
if _env_val == "true":
|
||
logger.warning(
|
||
"检测到 ALLOW_TEXT_ONLY_IMAGE=true，但该选项已被锁定（红线：禁纯文生图）。"
|
||
"将强制抛出缺图异常，请为产品上传参考图。"
|
||
)
|
||
raise ValueError(
|
||
"缺少产品参考图，无法生成忠于实物的产品图，请先上传该产品的主图。"
|
||
"（红线：禁止纯文生图脑补产品包装，ALLOW_TEXT_ONLY_IMAGE 已锁定为无效）"
|
||
)
|
||
|
||
|
||
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 []
|
||
# 没配 codeproxy 备用 key 时不把它补进尝试序列（避免每张图空转重试，方案A）
|
||
has_alt = bool(getattr(client, "has_alt_channel", lambda: True)())
|
||
providers = _image_provider_order(has_alt)
|
||
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','')}。"
|
||
"参考图说明：随附的多张参考图都是【同一个产品】的不同视角（主图=产品/包装本体，"
|
||
"其余=质地/细节/场景特写）；产品外形、颜色、包装上原有文字必须100%忠于主图原样，"
|
||
"禁止重绘产品本体、禁止在包装上增删改任何文字（品牌词只放图上标题/贴纸等排版层，绝不印产品本体）。"
|
||
"排版要求：独立小红书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)
|