fix: harden online validation flows

backend/app/services/ai_engine/fission_prompt.py

@@ -165,6 +165,20 @@ def build_fission_prompt(
         kw_parts = [x for x in ([name] + sp_words + ([audience] if audience != "未提供" else [])) if x and x != "未提供"]
         kw_line = "、".join(kw_parts) if kw_parts else ""
 
+    benchmark_refs = prod.get("benchmark_refs") or []
+    benchmark_lines: list[str] = []
+    for idx, ref in enumerate(benchmark_refs[:3], start=1):
+        if not isinstance(ref, dict):
+            continue
+        compact = []
+        for key in ("title_style", "cover_hook", "pain_point", "scene", "selling_angle", "visual_structure"):
+            val = ref.get(key)
+            if val:
+                compact.append(f"{key}:{val}")
+        if compact:
+            benchmark_lines.append(f"标杆配方{idx}：" + "；".join(compact))
+    benchmark_block = "\n".join(benchmark_lines) if benchmark_lines else "未选择或未分析完成"
+
     # D3修复：按套序循环分配 A/B/C 叙事主线
     narrative_keys = list(TEXT_NARRATIVE_BY_STRATEGY.keys())  # ["A","B","C"]
     narrative_lines = []
@@ -182,6 +196,7 @@ def build_fission_prompt(
 产品卖点：{'、'.join(points) or '未提供'}
 目标人群：{audience}
 关键词：{kw_line}
+标杆8维配方参考：{benchmark_block}
 裂变维度：{'、'.join(dims)}
 爆款参考度：{strategy['level_label']}。{strategy['prompt_rule']}
 生成数量：{note_count}套

backend/app/services/fission_pipeline.py

@@ -130,7 +130,7 @@ def execute_fission_pipeline(db: Session, clients, fission_id: int, source_task_
     from app.models.fission import FissionTask, FissionNote
     from app.models.task import GenerationTask
     from app.models.product import Product
-    from app.workers.pipeline_steps import build_product_dict
+    from app.workers.pipeline_steps import build_product_dict, load_benchmark_features
     from app.services.ai_engine.fission_prompt import build_fission_prompt
 
     ft = db.query(FissionTask).filter(FissionTask.id == fission_id).first()
@@ -150,6 +150,7 @@ def execute_fission_pipeline(db: Session, clients, fission_id: int, source_task_
         return {"fission_id": fission_id, "status": "failed", "reason": "源任务或产品缺失"}
 
     product = build_product_dict(product_row)
+    product["benchmark_refs"] = load_benchmark_features(db, src, ft.workspace_id)
     image_count = max(1, src.image_count or 3)
     banned = []  # 违禁词分级表后续接入；评分器内置 BANNED_WORDS_DEFAULT 已兜底
 

backend/app/services/task_service.py

@@ -87,6 +87,29 @@ def create_generation_task(
     # 第2环：关联标杆笔记ID存库（JSON list）。pipeline 据此读 features_json 注入文案 prompt。
     import json
     _bids = getattr(body, "benchmark_ids", None) or []
+    if _bids:
+        from app.models.product import BenchmarkNote
+
+        ids = [int(i) for i in _bids]
+        rows = (
+            db.query(BenchmarkNote)
+            .filter(
+                BenchmarkNote.id.in_(ids),
+                BenchmarkNote.workspace_id == current_user.workspace_id,
+                BenchmarkNote.product_id == body.product_id,
+            )
+            .all()
+        )
+        by_id = {int(b.id): b for b in rows}
+        missing = [i for i in ids if i not in by_id]
+        if missing:
+            raise_business(f"标杆笔记不存在或不属于当前产品：{missing[:3]}")
+        not_ready = [
+            i for i in ids
+            if by_id[i].analyze_status != "done" or by_id[i].analysis_source == "fallback"
+        ]
+        if not_ready:
+            raise_business("所选标杆尚未分析完成，请先到系统管理里点击「分析标杆」")
     benchmark_ids_json = json.dumps([int(i) for i in _bids]) if _bids else None
 
     task = GenerationTask(