存量积累：生图素人感约束+评图分+幂等防重跑+审核回路

- 015-017迁移：image_candidate 文案复审/AI视觉分/重生标记
- constants C7素人感约束(反电商摆拍对齐真实笔记)+C3叠字口子
- celery visibility_timeout=2h 防长任务被误判重投重复烧钱(task75教训)
- image_scorer 评图分(只筛选+展示,真实信号才进飞轮权重)
- storyboard/image_gen/pipeline_io 生图存量
- task_actions/tasks/task_service/flywheel 审核回路+飞轮存量

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/app/services/ai_engine/constants.py

@@ -127,10 +127,27 @@ IMAGE_NEGATIVE_CONSTRAINTS = (
     "只保留浅色简洁台面或产品定制场景，主体聚焦产品本身。"
 )
 
+# ── 小红书素人感正向约束（C7：反电商摆拍，对齐真实笔记观感）──────────────
+# 追加到 base_prompt 末尾，与 IMAGE_NEGATIVE_CONSTRAINTS 互补：前者管合规，本条管"像不像小红书"
+IMAGE_XHS_STYLE_CONSTRAINTS = (
+    "【小红书素人感——必须像真人随手拍后简单排版，不是电商详情页】"
+    "①禁止纯白底影棚摆拍、居中正打光、官方产品主图那种电商感；"
+    "②要有生活气：自然光/窗边光、桌面或梳妆台真实场景、可带手持或局部环境，像朋友分享而非广告；"
+    "③构图随性不刻意对称，允许轻微景深虚化，避免过度精修的塑料光泽和磨皮假面感；"
+    "④文字排版克制像博主手作：主标题手写感或简洁无衬线，避免大字促销价签/打折标/电商角标；"
+    "⑤整体观感=真实测评/日常分享，宁可朴素也不要假亮假精致。"
+)
+
+# C3 代码叠字开关（先留口子不实现，倩倩姐2026-06-12拍板）：
+# True 时由 PIL 在模型出的干净底图上叠主标题/品牌词，彻底解决 gpt-image-2 中文乱码；
+# 当前 False=仍由模型画字（偶发错别字为已知问题）。接入点见 image_gen._gen_one 的 TODO。
+OVERLAY_TEXT_RENDER_ENABLED = False
+
 # ── 飞轮信号权重（初始默认，北哥可校准）────────────────
 FLYWHEEL_WEIGHTS = {
     "text_select": 3,
     "image_select": 3,
+    "text_edit": 5,    # 改稿=用户真动手改字=最强真实意图,与approve同级(倩倩姐2026-06-16拍板)
     "approve": 5,
     "reject_with_reason": -3,
     "regenerate": -1,

backend/app/services/ai_engine/gemini_factory.py

@@ -41,7 +41,7 @@ class AIClients:
     _pool_loop_id: int | None = field(default=None, repr=False)
     _gemini_key: str | None = field(default=None, repr=False)   # 局部变量不打印
     _model_image: str = "gpt-image-2"
-    _model_text: str = "claude-sonnet-4-5"  # apiports无gpt-4o-mini,文案用claude中文质量好
+    _model_text: str = "claude-opus-4-8"  # 最强档(倩倩姐红线):Claude系一律4.8,绝不降级
 
     def _client(self) -> httpx.AsyncClient:
         """主通道(apiports) client，按当前事件循环缓存"""
@@ -111,16 +111,44 @@ class AIClients:
         resp.raise_for_status()
         return _extract_gemini_image(resp.json())
 
+    async def _chat_post_failover(self, payload: dict, timeout: float) -> dict:
+        """
+        chat/completions 发送器，带 codeproxy 回落。
+        主通道(apiports, claude-opus-4-8)若 5xx/超时/连接错，自动切 codeproxy 重试一次。
+        ⚠ codeproxy 账号池只支持 gpt 系(gpt-5.5)，无 claude，故回落时模型换成 gpt-5.5。
+        这是强档↔强档切换：红线"Claude系4.8/GPT系5.5"本就是两个平级最强档、互为兜底，
+        非降级(降级=落 sonnet/弱档)。codeproxy 回落档由 CODEPROXY_CHAT_MODEL 配置(默认 gpt-5.5)。
+        codeproxy 未配置(_alt_token 为空)时不回落，原样抛错。
+        """
+        main_base = (os.environ.get("IMAGE_API_BASE") or os.environ.get("APIPORTS_BASE_URL") or "").rstrip("/")
+        try:
+            resp = await self._client().post(f"{main_base}/chat/completions", json=payload, timeout=timeout)
+            resp.raise_for_status()
+            return resp.json()
+        except (httpx.HTTPStatusError, httpx.TransportError, httpx.TimeoutException) as exc:
+            # 仅 5xx(服务端过载)或网络层错误才回落；4xx(参数/鉴权)回落也没用，直接抛。
+            status = getattr(getattr(exc, "response", None), "status_code", None)
+            retryable = status is None or status >= 500
+            if not (retryable and self._alt_token):
+                raise
+            alt_base = (self._alt_base or os.environ.get("CODEPROXY_BASE_URL") or "").rstrip("/")
+            alt_payload = dict(payload)
+            alt_payload["model"] = os.environ.get("CODEPROXY_CHAT_MODEL", "gpt-5.5")
+            logger.warning("apiports chat 失败(%s)，回落 codeproxy 用 %s 重试（强档兜底，非降级）",
+                           status or type(exc).__name__, alt_payload["model"])
+            client = self._client_for(alt_base, self._alt_token)
+            resp = await client.post(f"{alt_base}/chat/completions", json=alt_payload, timeout=timeout)
+            resp.raise_for_status()
+            return resp.json()
+
     async def chat_complete(self, messages: list[dict], model: str | None = None, max_tokens: int = 4096, temperature: float = 0.75) -> str:
-        """文字生成（文案生成用）"""
-        base = (os.environ.get("IMAGE_API_BASE") or os.environ.get("APIPORTS_BASE_URL") or "").rstrip("/")
+        """文字生成（文案生成用）。apiports 503 时自动回落 codeproxy。"""
         payload = {"model": model or self._model_text, "messages": messages, "max_tokens": max_tokens, "temperature": temperature}
         # 单批≤4条文案正常 40-55s 返回；apiports 网关 ~60s 上限。客户端超时设 75s：
         # 略高于网关上限即可，过长（如180s）会在 apiports 卡顿时干等，拖慢整体。
         timeout = float(os.environ.get("TEXT_LLM_TIMEOUT", "75"))
-        resp = await self._client().post(f"{base}/chat/completions", json=payload, timeout=timeout)
-        resp.raise_for_status()
-        return resp.json()["choices"][0]["message"]["content"] or ""
+        data = await self._chat_post_failover(payload, timeout)
+        return data["choices"][0]["message"]["content"] or ""
 
     async def gpt_vision_analyze(self, prompt: str, images: list[bytes], model: str | None = None) -> str:
         """
@@ -139,16 +167,15 @@ class AIClients:
             })
         used_model = model or os.environ.get("MODEL_TEXT", "claude-opus-4-8")
         messages = [{"role": "user", "content": content}]
-        base = (os.environ.get("IMAGE_API_BASE") or os.environ.get("APIPORTS_BASE_URL") or "").rstrip("/")
         payload = {
             "model": used_model,
             "messages": messages,
             "max_tokens": 2048,
             "temperature": 0.2,
         }
-        resp = await self._client().post(f"{base}/chat/completions", json=payload, timeout=90.0)
-        resp.raise_for_status()
-        return resp.json()["choices"][0]["message"]["content"] or ""
+        # 评图也走 codeproxy 回落：apiports 503 时切备用站，模型档不变（守红线）。
+        data = await self._chat_post_failover(payload, 90.0)
+        return data["choices"][0]["message"]["content"] or ""
 
     # duck-type: text_variants._call_llm 用的属性
     @property

backend/app/services/ai_engine/image_gen.py

@@ -15,6 +15,7 @@ import os
 from typing import Any, Protocol
 
 from .constants import IMAGE_RETRY_ATTEMPTS, IMAGE_RETRY_BACKOFF_BASE, IMAGE_SIZE_DEFAULT
+from .image_scorer import score_image
 from .storyboard import plan_image_set, sanitize_text
 
 logger = logging.getLogger(__name__)
@@ -126,15 +127,24 @@ async def generate_storyboard_images(
     reference_images: list[bytes] | None = None,
     analysis: dict | None = None,
     strategy: str | None = None,
+    target_role: str | None = None,
+    custom_prompt: str | None = None,
 ) -> list[dict]:
     """
     按 storyboard 逐张生图（asyncio.gather 并发），返回每张结果列表。
     strategy: None=默认叙事，'A'/'B'/'C'=三套正交叙事策略
+    target_role: 非空时只生成该 role 那一张（R2 单张重生）
+    custom_prompt: 非空时追加到每张 per_prompt 末尾（R2 人工提示词）
     每项：{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），每张差异化
@@ -156,12 +166,31 @@ async def generate_storyboard_images(
             "中文文字少而清晰，主标题+最多3个短点位；可自然用✅✨🌿💧🪞🧴📦🔍种草符号但不堆砌；"
             "不要生成App截图或笔记详情页界面。"
         )
+        # R2 人工提示词：追加到末尾权重最高，但不覆盖前面合规/真实约束
+        if custom_prompt:
+            per_prompt += f"\n运营补充要求（在不违反上述合规与真实约束前提下尽量满足）：{sanitize_text(custom_prompt, 200)}。"
         try:
             img_bytes = await generate_one_image(client, per_prompt, reference_images)
-            return {"role": item["role"], "name": item["name"], "image_bytes": img_bytes, "error": None}
+            # 注：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)}
+            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"))

backend/app/services/ai_engine/image_scorer.py

@@ -0,0 +1,76 @@
+"""
+AI 评图分 — E12 飞轮·展示层（倩倩姐2026-06-16拍板）
+
+红线（不可违反）：
+- 只做"生成时筛选 + 落库 + 前端展示 + 高分优先排序"。
+- 绝不进飞轮权重：飞轮权重只认真实信号（选了哪张/改了什么/过审与否）。
+- eval_score 全程留 NULL；AI 分单独存 ai_visual_score，不复用 eval_score。
+- 评分失败绝不阻断生图：返回 None，图照常入库展示。
+
+vision 走 GPT 现有通道最强档（claude-opus-4-8），不补 GEMINI_KEY。
+"""
+from __future__ import annotations
+import json
+import logging
+import re
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# 评图维度=纯视觉质量（不碰文字对错——那条路线已撤，见 image_gen.py 注释）
+VISION_SCORE_PROMPT = (
+    "你是小红书资深视觉运营，给下面这张种草配图打分。只看视觉质量，不纠结文字错别字。\n"
+    "评分维度（综合给一个 0-100 总分）：\n"
+    "1. 构图与美感：主体突出、留白舒服、色调高级。\n"
+    "2. 清晰度：画面锐利不糊、无明显畸变。\n"
+    "3. 小红书种草感：像真实博主拍的好图，有质感、有氛围。\n"
+    "4. 去AI感：不假、不塑料、不像廉价 AI 渲染图。\n"
+    "5. 文字排版观感：标题贴纸排版是否清爽（只看排版美观，不判错别字）。\n"
+    "打分基准：80=可直接交付的好图，60=能用但平庸，40以下=明显有问题。\n"
+    '只返回 JSON，不要任何多余文字：{"score": <0-100整数>, "note": "<20字内一句话点评>"}'
+)
+
+
+def _parse_score(raw: str) -> dict[str, Any] | None:
+    """容错解析模型返回的 JSON（剥 markdown fence / 抓第一个 {...}）。"""
+    if not raw:
+        return None
+    text = raw.strip()
+    text = re.sub(r"^```(?:json)?\s*|\s*```$", "", text, flags=re.IGNORECASE).strip()
+    try:
+        obj = json.loads(text)
+    except Exception:
+        m = re.search(r"\{.*\}", text, re.DOTALL)
+        if not m:
+            return None
+        try:
+            obj = json.loads(m.group(0))
+        except Exception:
+            return None
+    if not isinstance(obj, dict) or "score" not in obj:
+        return None
+    try:
+        score = float(obj["score"])
+    except Exception:
+        return None
+    score = max(0.0, min(100.0, score))   # 钳到 0-100
+    note = str(obj.get("note") or "").strip()[:200]
+    return {"score": score, "note": note}
+
+
+async def score_image(client: Any, image_bytes: bytes) -> dict[str, Any] | None:
+    """
+    给单张图打视觉分。返回 {"score": float, "note": str} 或 None（失败/不可解析）。
+    绝不抛异常打断生图链路——任何异常都吞掉返回 None。
+    """
+    if not image_bytes:
+        return None
+    try:
+        raw = await client.gpt_vision_analyze(VISION_SCORE_PROMPT, [image_bytes])
+    except Exception as exc:
+        logger.warning("AI 评图调用失败（不阻断生图）：%s", exc)
+        return None
+    result = _parse_score(raw)
+    if result is None:
+        logger.warning("AI 评图返回无法解析（不阻断生图）：%s", (raw or "")[:120])
+    return result

backend/app/services/ai_engine/preference_aggregator.py

@@ -48,13 +48,14 @@ def aggregate_preference_context(
 
     for e in relevant:
         sig_type = e.get("signal_type", "")
-        angle    = str(e.get("angle_label", "")).strip()
+        angle    = str(e.get("angle_label") or "").strip()
         weight   = int(e.get("signal_weight", 1))
 
-        if sig_type in ("text_select", "approve") and angle:
+        # text_edit(改稿)是最强真实信号,角度按权重计入(倩倩姐2026-06-16拍板)
+        if sig_type in ("text_select", "approve", "text_edit") and angle:
             angle_counts[angle] += weight
         elif sig_type == "reject_with_reason":
-            reason = str(e.get("reason", "")).strip()
+            reason = str(e.get("reason") or "").strip()
             if reason:
                 reject_reasons.append(reason)
 

backend/app/services/ai_engine/storyboard.py

@@ -9,7 +9,7 @@ storyboard 分镜引擎
 from __future__ import annotations
 import re
 from .constants import (
-    PAGE_ROLE_MAP, IMAGE_NEGATIVE_CONSTRAINTS,
+    PAGE_ROLE_MAP, IMAGE_NEGATIVE_CONSTRAINTS, IMAGE_XHS_STYLE_CONSTRAINTS,
     STYLE_PROMPTS, STYLE_DEFAULT, NARRATIVE_BY_COUNT, NARRATIVE_BY_STRATEGY,
 )
 # sanitize_text 移至 templates（腾行数），此处 re-export 供 image_gen 沿用 import
@@ -187,6 +187,7 @@ def plan_image_set(note: dict, product: dict, image_count: int = 3, analysis: di
         "禁止肤色变白、瑕疵消失、治疗前后等视觉暗示，允许安全的未推开/推开后质地状态对比；"
         "如果提供产品图，产品是不可修改的真实商品锚点，禁止改名、换包装、混入其他产品。"
         f"\n{IMAGE_NEGATIVE_CONSTRAINTS}"
+        f"\n{IMAGE_XHS_STYLE_CONSTRAINTS}"
     )
 
     return {