style: auto-format with black + isort + ruff + prettier [skip ci-format-check]

根因：信任链t2i预热/现场兜底把所有图（含养生馆场景图/门头/商品图）的portrait_prompt
都当成人像描述送Seedream纯文生图，导致：
1. 场景图被错误生成假人像替换原图，Seedance收到的reference_image全是假的
2. 若任意一张t2i失败，原图传Seedance触发400 portrait_intercept后降级纯t2v，完全零参考

修复：
- vision/assembler.py: 每个product输出加has_person布尔字段，人物类True/商品门店other类False/旧schema动态取fj.has_person
- viral_video.py _vision_fallback: 默认has_person=False
- viral_video.py _start_trust_chain_preheat: 改接products列表，内部按has_person筛选人像位，
  结果存为稀疏列表（人像位=AI图URL，非人像位=None表示保留原图）
- viral_video.py _step_render现场t2i兜底：同逻辑只对has_person=True的图生成AI人像，构建稀疏列表
- viral_video.py pti读取：支持稀疏列表模式，None位不替换
- ai_client.py video_generation信任链替换：稀疏合并，None位保留原图URL，只替换人像位
- viral_video.py 硬约束增加：严格还原参考图片场景/商品/人物特征，不凭空生成无关内容
- 清理：3处调用方统一改为传products列表

apps/worker/worker_app/tasks/viral_video.py

@@ -92,22 +92,30 @@ def _save_job(repo, job, session):
     session.commit()
 
 
-def _start_trust_chain_preheat(job_id: str, portrait_descriptions: list[str]) -> None:
-    """#2172/#2174 后台启动信任链预热（Seedream t2i 文生图人像），不阻塞调用方。
+def _start_trust_chain_preheat(job_id: str, products: list[dict]) -> None:
+    """#2172/#2174/#2220 后台启动信任链预热（Seedream t2i 文生图人像），不阻塞调用方。
 
-    #2174 重要：改为 t2i 文生图模式——用 VLM 分析出的人物外貌描述做 prompt，不传 reference_images，
-    产物是方舟信任模型输出，Seedance 直接放行不触发肖像审核。
-    i2i（传用户照片做 reference）产物不被信任，实测仍被 400 portrait_intercept 拦截。
+    #2220 修复：只对 has_person=True 的图（真人照片）生成 AI 人像替换，
+    场景图/商品图/门店图保持原图不变，传给 Seedance 作为 reference_image 直接使用。
 
-    预热成功后把结果写入 job.pre_trusted_images，阶段3 渲染直接使用，省掉串行等待。
-    预热失败静默（pre_trusted_images 保持 None），阶段3 会走 #2166 自动降级纯 t2v。
+    预热结果写入 job.pre_trusted_images：与 products 等长的稀疏列表，
+    人像位是 AI 图 URL，非人像位是 None（表示保留原图）。
     """
-    # 过滤有效描述：非空且不是"无人像"
-    _valid = [
-        d for d in (portrait_descriptions or []) if d and isinstance(d, str) and "无人像" not in d and len(d) >= 10
-    ]
+    # 构建人像位索引映射：person_indices[k] = products中第k个人像的位置
+    _person_indices: list[int] = []
+    _valid: list[str] = []
+    for _i, _p in enumerate(products or []):
+        if not isinstance(_p, dict):
+            continue
+        if not _p.get("has_person", False):
+            continue
+        _d = (_p.get("portrait_prompt") or "").strip()
+        if not _d or "无人像" in _d or len(_d) < 10:
+            continue
+        _person_indices.append(_i)
+        _valid.append(_d)
     if not _valid:
-        logger.info("[trust-chain][preheat] 无有效人物描述（可能是纯商品图），跳过预热 job=%s", job_id)
+        logger.info("[trust-chain][preheat] 无有效人物描述（可能是纯商品/场景图），跳过预热 job=%s", job_id)
         return
     # 判断是否是 doubao provider（DashScope/Wan 不需要信任链）
     try:
@@ -133,21 +141,33 @@ def _start_trust_chain_preheat(job_id: str, portrait_descriptions: list[str]) ->
 
             logger.info("[trust-chain][preheat] 后台t2i预热启动 job=%s n=%d", job_id, len(_valid))
             result = preheat_trust_chain(_valid, timeout=120)
-            if result and len(result) >= 1:
+            if result and len(result) == len(_valid):
                 sess2, repo2, job2 = _get_repo_and_job(job_id)
                 try:
-                    job2.pre_trusted_images = result
+                    # #2220: 构建与 products 等长的稀疏列表，人像位放AI图URL，非人像位None
+                    _imgs2 = job2.images or []
+                    _sparse: list[str | None] = [None] * max(len(_imgs2), len(products or []))
+                    for _k, _url in enumerate(result):
+                        if _k < len(_person_indices):
+                            _sparse[_person_indices[_k]] = _url
+                    job2.pre_trusted_images = _sparse
                     repo2.update(job2)
                     sess2.commit()
                     logger.info(
-                        "[trust-chain][preheat] t2i预热完成并持久化 job=%s n=%d",
+                        "[trust-chain][preheat] t2i预热完成并持久化 job=%s n_person=%d total=%d",
                         job_id,
                         len(result),
+                        len(_sparse),
                     )
                 finally:
                     sess2.close()
             else:
-                logger.info("[trust-chain][preheat] 预热失败 job=%s，阶段3现场跑兜底", job_id)
+                logger.info(
+                    "[trust-chain][preheat] 预热失败或数量不匹配 job=%s got=%s expect=%d，阶段3现场跑兜底",
+                    job_id,
+                    len(result) if result else 0,
+                    len(_valid),
+                )
         except Exception as e:
             logger.warning("[trust-chain][preheat] 预热异常 job=%s err=%s", job_id, e, exc_info=True)
 
@@ -268,6 +288,7 @@ def _recover_stale_jobs() -> int:
 
 _DEFAULT_HARD_CONSTRAINTS = [
     "无字幕、无水印、无任何自动生成文字、无 logo",
+    "严格还原参考图片中的真实场景、门店环境、商品陈列、人物外貌服装特征，不得凭空生成与参考图无关的人物、场景或物品",
     "同一人物全程保持一致的五官、发型、服装、身材，不得换脸或变形",
     "口播语音必须在指定时长内自然念完，语速自然，口型与语音同步",
     "画面流畅无闪烁、无多余肢体、无扭曲变形、无穿模",
@@ -323,6 +344,7 @@ def _vision_fallback(idx: int, reason: str, extra: dict | None = None) -> dict:
         "scene": "通用",
         "portrait_prompt": "无人像",
         "summary": "",
+        "has_person": False,
         "_source": reason,
     }
     if extra:
@@ -1245,9 +1267,12 @@ def _step_render(job: ViralVideoJob, copy_result: dict, tts_audio_url: str | Non
         if u not in all_portrait_urls:
             all_portrait_urls.append(u)
     pti = getattr(job, "pre_trusted_images", None)
-    if pti and len(pti) == len(all_portrait_urls):
-        pre_trusted = list(pti)
-        logger.info("[爆款视频] 使用信任链预热结果 n=%d，跳过现场 Seedream AI 化", len(pre_trusted))
+    # #2220: 稀疏列表模式（与images等长，None表示该位置保留原图）
+    _n_total = len(all_portrait_urls)
+    if pti and isinstance(pti, list) and len(pti) >= _n_total and _n_total > 0:
+        pre_trusted = list(pti[:_n_total])
+        _n_trusted = sum(1 for _x in pre_trusted if _x)
+        logger.info("[爆款视频] 使用信任链预热结果 person=%d total=%d，跳过现场Seedream AI化", _n_trusted, _n_total)
     elif all_portrait_urls and _mcfg.get("provider", "doubao") == "doubao":
         # #2183: 真·现场跑信任链——同步调用 Seedream t2i，拿到 AI 人像 URL 后再传 Seedance
         logger.info(
@@ -1260,32 +1285,39 @@ def _step_render(job: ViralVideoJob, copy_result: dict, tts_audio_url: str | Non
 
             _ia = getattr(job, "image_analysis", None) or {}
             _prods = (_ia.get("products") if isinstance(_ia, dict) else None) or []
-            _pdescs = []
-            if _prods:
-                _pdescs = [(pp.get("portrait_prompt") or "无人像") for pp in _prods]
-            elif isinstance(_ia, dict):
-                _pp0 = _ia.get("portrait_prompt") or "无人像"
-                if _pp0 and _pp0 != "无人像":
-                    _pdescs = [_pp0]
-            _valid = [d for d in _pdescs if d and isinstance(d, str) and "无人像" not in d and len(d) >= 10]
-            if _valid:
+            _live_person_idx: list[int] = []
+            _live_pdescs: list[str] = []
+            for _i, _pp in enumerate(_prods):
+                if not isinstance(_pp, dict) or not _pp.get("has_person", False):
+                    continue
+                _d = (_pp.get("portrait_prompt") or "").strip()
+                if not _d or "无人像" in _d or len(_d) < 10:
+                    continue
+                _live_person_idx.append(_i)
+                _live_pdescs.append(_d)
+            if _live_pdescs:
                 _t0 = time.time()
-                _live_urls = preheat_trust_chain(_valid, timeout=120)
-                if _live_urls and len(_live_urls) == len(all_portrait_urls):
-                    pre_trusted = list(_live_urls)
+                _live_urls = preheat_trust_chain(_live_pdescs, timeout=120)
+                if _live_urls and len(_live_urls) == len(_live_pdescs):
+                    # #2220: 构建稀疏列表，人像位替换AI图，非人像位保留None(ai_client里用原图)
+                    pre_trusted = [None] * len(all_portrait_urls)
+                    for _k, _u in enumerate(_live_urls):
+                        if _k < len(_live_person_idx):
+                            pre_trusted[_live_person_idx[_k]] = _u
                     logger.info(
-                        "[爆款视频] 现场信任链t2i完成 %d张 耗时%.1fs，将用AI人像传Seedance",
-                        len(pre_trusted),
+                        "[爆款视频] 现场信任链t2i完成 %d张人像AI化 耗时%.1fs（共%d张图，其余保留原图）",
+                        len(_live_urls),
                         time.time() - _t0,
+                        len(all_portrait_urls),
                     )
                 else:
                     logger.warning(
-                        "[爆款视频] 现场信任链t2i返回不匹配 urls=%s n_portraits=%d，回退原图+400降级纯t2v",
+                        "[爆款视频] 现场信任链t2i返回不匹配 urls=%s n_person=%d，人像位原图传Seedance（可能触发400拦截）",
                         _live_urls,
-                        len(all_portrait_urls),
+                        len(_live_pdescs),
                     )
             else:
-                logger.info("[爆款视频] 无有效人物描述（可能是商品图），无需现场跑信任链")
+                logger.info("[爆款视频] 无有效人物描述（商品/场景图），无需AI化，直接传原图给Seedance")
         except Exception as _te:
             logger.warning("[爆款视频] 现场跑信任链异常: %s，回退原图+400降级纯t2v", _te, exc_info=True)
 
@@ -1408,13 +1440,8 @@ def run_viral_video_pipeline(self: Task, job_id: str) -> dict:
         if job.images:
             try:
                 _products = (image_analysis or {}).get("products", []) or []
-                _portrait_descs = [(p.get("portrait_prompt") or "无人像") for p in _products] if _products else []
-                # 兼容单图结果格式（非products列表）
-                if not _portrait_descs and isinstance(image_analysis, dict):
-                    _pp = image_analysis.get("portrait_prompt") or "无人像"
-                    if _pp and _pp != "无人像":
-                        _portrait_descs = [_pp]
-                _start_trust_chain_preheat(job.id, _portrait_descs)
+                # #2220: 直接传 products 列表，由 _start_trust_chain_preheat 内部按 has_person 筛选
+                _start_trust_chain_preheat(job.id, _products)
             except Exception as _e:
                 logger.warning("[爆款视频][阶段1] 启动信任链t2i预热失败: %s", _e)
         _save_job(repo, job, session)
@@ -1586,12 +1613,8 @@ def run_viral_video_analyze(self: Task, job_id: str) -> dict:
         if job.images:
             try:
                 _products = (image_analysis or {}).get("products", []) or []
-                _portrait_descs = [(p.get("portrait_prompt") or "无人像") for p in _products] if _products else []
-                if not _portrait_descs and isinstance(image_analysis, dict):
-                    _pp = image_analysis.get("portrait_prompt") or "无人像"
-                    if _pp and _pp != "无人像":
-                        _portrait_descs = [_pp]
-                _start_trust_chain_preheat(job.id, _portrait_descs)
+                # #2220: 直接传 products 列表，由 _start_trust_chain_preheat 内部按 has_person 筛选
+                _start_trust_chain_preheat(job.id, _products)
             except Exception as _e:
                 logger.warning("[爆款视频][阶段1] 启动信任链t2i预热失败: %s", _e)
         _save_job(repo, job, session)

apps/worker/worker_app/tasks/vision/assembler.py

@@ -498,6 +498,7 @@ def _assemble_v4(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
             "portrait_prompt": portrait_prompt[:300],
             "summary": summary[:50],
             "_source": "v2_fast_json_v5",
+            "has_person": True,
         }
 
     # 商品类
@@ -558,6 +559,7 @@ def _assemble_v4(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
             "mood": mood,
             "portrait_prompt": portrait_prompt[:200],
             "summary": str(summary)[:60],
+            "has_person": False,
             "_source": "v2_fast_json_v4",
         }
 
@@ -606,6 +608,7 @@ def _assemble_v4(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
             "mood": atmosphere or mood,
             "portrait_prompt": portrait_prompt[:200],
             "summary": summary[:40],
+            "has_person": False,
             "_source": "v2_fast_json_v4",
         }
 
@@ -623,6 +626,7 @@ def _assemble_v4(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
         "mood": mood,
         "portrait_prompt": f"{scene}，{mood}氛围，{desc}"[:200],
         "summary": desc[:40],
+        "has_person": False,
         "_source": "v2_fast_json_v4_other",
     }
 
@@ -660,4 +664,5 @@ def _assemble_old(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
         "portrait_prompt": portrait_prompt,
         "summary": summary,
         "_source": "v2_fast_json",
+        "has_person": bool(fj.get("has_person", False)),
     }

packages/shared/ai_client.py

@@ -590,28 +590,32 @@ class DoubaoClient:
         # 信任链只作用于 doubao provider；DashScope(Wan) 保持原行为。
         trust_chain_applied = False
         if provider == "doubao" and getattr(self, "trust_chain_enabled", True) and pre_trusted_images:
+            # #2220: 稀疏列表模式——pre_trusted_images 与 raw_portrait_urls 等长，
+            # None 位保留原图，非 None 位用 AI 人像替换。
             raw_portrait_urls: list[str] = []
             if image_url:
                 raw_portrait_urls.append(image_url)
             for u in ref_imgs:
                 if u not in raw_portrait_urls:
                     raw_portrait_urls.append(u)
-            trusted_urls: list[str] = []
-            if len(pre_trusted_images) >= 1:
-                trusted_urls = list(pre_trusted_images)
+            _n_trusted = sum(1 for _x in pre_trusted_images if _x)
+            if _n_trusted >= 1 and len(pre_trusted_images) >= len(raw_portrait_urls):
+                merged: list[str] = []
+                for _i, _orig in enumerate(raw_portrait_urls):
+                    _ai = pre_trusted_images[_i] if _i < len(pre_trusted_images) else None
+                    merged.append(str(_ai) if _ai else _orig)
                 trust_chain_applied = True
                 logger.info(
-                    "[trust-chain] 使用预热t2i结果 %d 张，替换原参考图走 reference_image 模式（原n=%d）",
-                    len(trusted_urls),
+                    "[trust-chain] 稀疏替换 %d/%d 张为AI人像（场景/商品图保留原图），走reference_image模式",
+                    _n_trusted,
                     len(raw_portrait_urls),
                 )
-            if trust_chain_applied and trusted_urls:
-                # 替换：原 image_url 用第一张 AI 图，ref_imgs 用剩余
-                if image_url and trusted_urls:
-                    image_url = trusted_urls[0]
-                    ref_imgs = trusted_urls[1:] if len(trusted_urls) > 1 else []
+                # 替换：image_url 用第一张（可能是AI或原图），ref_imgs 用剩余
+                if image_url and merged:
+                    image_url = merged[0]
+                    ref_imgs = merged[1:] if len(merged) > 1 else []
                 else:
-                    ref_imgs = trusted_urls
+                    ref_imgs = merged
         # ─────────────────────────────────────────────────────────────────
 
         # 判断任务模式：