feat(#1970): 批量变体跨变体原子片段级硬避让

- reselect_plan_for_variant 新增 batch_used_atom_ids 参数
- generation_common 新增 collect_plan_atom_clip_ids
- 批量生成循环累积各变体已用 atom_clip_id 传入下一变体

apps/api/app/api/routes/generation_tasks.py

@@ -477,8 +477,15 @@ def create_generation_task(
 
             # #1855 P0：批次区间避让表，从变体0实际clips构建初始值（公共函数）
             from app.services.generation_common import collect_plan_segments as _collect_segments
+            from app.services.generation_common import (
+                collect_plan_atom_clip_ids as _collect_atom_ids,
+            )
 
             _batch_segments = _collect_segments(_plan0.id, _plan_svc._clip_repo)
+            # #1970：批次内原子片段硬避让集合
+            _batch_atom_ids: list[str] = _collect_atom_ids(
+                _plan0.id, _plan_svc._clip_repo
+            )
 
             # 变体 1..N-1 独立选片（传入累积batch_segments做素材区间避让）
             for task_index in range(1, count):
@@ -493,6 +500,7 @@ def create_generation_task(
                             name_suffix=f"批量{task_index + 1}",
                             voice_duration=voice_durations[task_index] if task_index < len(voice_durations) else 0.0,
                             batch_segments=_batch_segments,
+                            batch_used_atom_ids=_batch_atom_ids,
                         )
                         break
                     except ValueError as ve:
@@ -529,6 +537,10 @@ def create_generation_task(
                     _new_segs = _collect_segments(variant.id, _plan_svc._clip_repo)
                     for _aid, _ivs in _new_segs.items():
                         _batch_segments.setdefault(_aid, []).extend(_ivs)
+                    # #1970：同步累积原子片段ID
+                    _batch_atom_ids.extend(
+                        _collect_atom_ids(variant.id, _plan_svc._clip_repo)
+                    )
                 except Exception:
                     logger.exception("[生成任务] 变体%d 区间收集失败（不阻断）", task_index)
 

apps/api/app/services/edit_plan_service.py

@@ -475,6 +475,7 @@ class EditPlanService:
         voice_duration: float = 0.0,
         rng=None,
         batch_segments: dict[str, list[tuple[float, float]]] | None = None,
+        batch_used_atom_ids: set[str] | list[str] | None = None,
     ) -> EditPlan:
         """为批量变体生成独立 plan：完整重跑单视频选片流程（#1743）。
 
@@ -653,7 +654,9 @@ class EditPlanService:
                     source_clips_data,
                     atom_candidates,
                     historical_atom_ids=historical_atom_ids,
-                    batch_used_atom_ids=None,
+                    batch_used_atom_ids=(
+                        set(batch_used_atom_ids) if batch_used_atom_ids else None
+                    ),
                     rng=rng,
                 )
         except Exception:

apps/api/app/services/generation_common.py

@@ -157,6 +157,33 @@ def collect_plan_segments(
     return segs
 
 
+def collect_plan_atom_clip_ids(
+    plan_id: str,
+    clip_repo: Any,
+    *,
+    page_size: int = 500,
+) -> list[str]:
+    """分页读取 plan 所有 clips，收集已选用的原子片段 ID（#1970）。
+
+    用于批量变体间原子片段级硬避让：同一原子片段在同批次内只用一次。
+    旧路径 clips 的 atom_clip_id 为空串，自动忽略。
+    """
+    ids: list[str] = []
+    sk, pg = 0, page_size
+    while True:
+        batch = clip_repo.list_by_plan(plan_id, skip=sk, limit=pg)
+        if not batch:
+            break
+        for c in batch:
+            acid = getattr(c, "atom_clip_id", "") or ""
+            if acid:
+                ids.append(acid)
+        if len(batch) < pg:
+            break
+        sk += pg
+    return ids
+
+
 def resolve_latest_plan_by_template(
     db: Session,
     *,

tests/unit/test_generation_common.py

@@ -297,3 +297,46 @@ class TestResolveLatestPlanByTemplate:
         with caplog.at_level("WARNING"):
             assert resolve_latest_plan_by_template(db, template_id="tpl", user_id="u") is None
         assert any("查找最新plan失败" in rec.message for rec in caplog.records)
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# collect_plan_atom_clip_ids (#1970)
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+def _make_atom_clip(atom_clip_id):
+    c = MagicMock()
+    c.atom_clip_id = atom_clip_id
+    return c
+
+
+class TestCollectPlanAtomClipIds:
+    def test_empty_plan_returns_empty(self):
+        from app.services.generation_common import collect_plan_atom_clip_ids
+
+        repo = MagicMock()
+        repo.list_by_plan.return_value = []
+        assert collect_plan_atom_clip_ids("p1", repo) == []
+
+    def test_collects_non_empty_ids_and_ignores_blank(self):
+        from app.services.generation_common import collect_plan_atom_clip_ids
+
+        repo = MagicMock()
+        repo.list_by_plan.side_effect = [
+            [
+                _make_atom_clip("atom-1"),
+                _make_atom_clip(""),
+                _make_atom_clip("atom-2"),
+            ],
+            [],
+        ]
+        assert collect_plan_atom_clip_ids("p1", repo) == ["atom-1", "atom-2"]
+
+    def test_missing_attribute_treated_as_blank(self):
+        from app.services.generation_common import collect_plan_atom_clip_ids
+
+        legacy = MagicMock()
+        del legacy.atom_clip_id  # 旧对象无该属性
+        repo = MagicMock()
+        repo.list_by_plan.side_effect = [[legacy, _make_atom_clip("atom-9")], []]
+        assert collect_plan_atom_clip_ids("p1", repo) == ["atom-9"]