refactor(backend): 智能剪辑后端公共逻辑抽取+解耦 (#1884)

Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>

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

@@ -58,41 +58,10 @@ def _variant_value(values: list[str], index: int, fallback: str = "") -> str:
 
 
 def _query_voice_durations(db: Session, voice_ids: list[str]) -> list[float]:
-    """批量查询配音素材时长（秒），#1749 配音时长分配用。
+    """[已下沉] 路由层兼容别名 → app.services.generation_common.query_voice_durations。"""
+    from app.services.generation_common import query_voice_durations
 
-    逐项 try/float 硬化：MagicMock/异常/缺失 → 0.0（无配音不分配，不阻断）。
-
-    #1855 P0修复：不再对 voice_ids 去重，保持与调用方传入顺序/长度一致，
-    允许同配音id多次出现时返回相同时长（支持"同配音N变体"的时长对齐）。
-    """
-    # 先去重查询（IN 查询性能优化），但最终按原始 voice_ids 顺序返回
-    raw_ids = list(voice_ids or [])
-    if not raw_ids:
-        return []
-    # 去重且保序，用于 SQL IN 查询；空字符串/None 视为无效id → 0.0
-    unique_ids: list[str] = []
-    _seen: set[str] = set()
-    for v in raw_ids:
-        if v and v not in _seen:
-            _seen.add(v)
-            unique_ids.append(v)
-    if not unique_ids:
-        return [0.0 for _ in raw_ids]
-    try:
-        from packages.adapters.sqlalchemy_impl.models import AssetModel
-
-        rows = db.query(AssetModel.id, AssetModel.duration).filter(AssetModel.id.in_(unique_ids)).all()
-        dur_map: dict[str, float] = {}
-        for row in rows:
-            try:
-                dur_map[row[0]] = float(row[1] or 0.0)
-            except (TypeError, ValueError):
-                dur_map[row[0]] = 0.0
-        # 按原始 voice_ids 顺序返回，保持长度一致；空/None/未查到 → 0.0
-        return [dur_map.get(v, 0.0) if v else 0.0 for v in raw_ids]
-    except Exception:
-        logger.warning("[生成任务] 配音时长查询失败（按无配音处理，不阻断）", exc_info=True)
-        return [0.0 for _ in raw_ids]
+    return query_voice_durations(db, voice_ids)
 
 
 def _to_generation_task_response(task) -> GenerationTaskResponse:
@@ -198,61 +167,10 @@ def _writeback_edit_plan_config(
     title_config: dict | None,
     db: Session,
 ) -> None:
-    """任务入队成功后，回写 EditPlan.config：generation_task_id + title_config。
+    """[已下沉] 路由层兼容别名 → app.services.generation_common.writeback_edit_plan_config。"""
+    from app.services.generation_common import writeback_edit_plan_config
 
-    用 merge 方式更新，不整体覆盖 config，避免丢失其他字段。
-    失败只记日志，不影响任务创建。
-    """
-    if not plan_id:
-        return
-    try:
-        from packages.adapters.sqlalchemy_impl.models import EditPlanModel
-
-        plan_model = db.query(EditPlanModel).filter(EditPlanModel.id == plan_id).first()
-        if plan_model is None:
-            logger.warning("[生成任务] 回写plan.config失败: plan不存在 plan_id=%s", plan_id)
-            return
-
-        current_config = plan_model.config if isinstance(plan_model.config, dict) else {}
-        merged = dict(current_config)
-        merged["generation_task_id"] = task_id
-
-        # 检查标题是否发生变化，如果变化则清除 cover 字段强制重新生成封面
-        if title_config:
-            old_title_config = merged.get("title_config", {}) or {}
-            old_title_text = (old_title_config.get("text") or "").strip()
-            new_title_text = (title_config.get("text") or "").strip()
-            if old_title_text != new_title_text:
-                # 标题变化，清除旧封面
-                if "cover" in merged:
-                    del merged["cover"]
-                    logger.info(
-                        "[生成任务] 标题变化，清除旧封面: plan_id=%s old_title=%s new_title=%s",
-                        plan_id,
-                        old_title_text,
-                        new_title_text,
-                    )
-            merged["title_config"] = title_config
-
-        plan_model.config = merged
-        db.commit()
-        logger.info(
-            "[生成任务] 回写plan.config成功: plan_id=%s task_id=%s keys=%s",
-            plan_id,
-            task_id,
-            list(merged.keys()),
-        )
-    except Exception as e:
-        logger.warning(
-            "[生成任务] 回写plan.config异常（不影响任务创建）: plan_id=%s error=%s",
-            plan_id,
-            e,
-            exc_info=True,
-        )
-        try:
-            db.rollback()
-        except Exception:
-            pass
+    return writeback_edit_plan_config(plan_id, task_id, title_config, db)
 
 
 def _resolve_project_and_library(
@@ -503,25 +421,14 @@ def create_generation_task(
         # 各变体配音时长（查询硬化：异常 → 0.0 不阻断）
         voice_durations = _query_voice_durations(db, variant_voices)
 
-        # 解析批量源 plan：优先前端传入；否则按 template_id + user 查最新（与单任务兜底同源）
-        batch_source_plan_id = request.source_edit_plan_id
-        if not batch_source_plan_id and request.template_id:
-            try:
-                from packages.adapters.sqlalchemy_impl.models import EditPlanModel
+        # 解析批量源 plan：优先前端传入；否则按 template_id + user 查最新（公共函数）
+        from app.services.generation_common import resolve_latest_plan_by_template
 
-                _latest = (
-                    db.query(EditPlanModel)
-                    .filter(
-                        EditPlanModel.template_id == request.template_id,
-                        EditPlanModel.created_by_user_id == user_id,
-                    )
-                    .order_by(EditPlanModel.created_at.desc())
-                    .first()
-                )
-                if _latest:
-                    batch_source_plan_id = _latest.id
-            except Exception:
-                logger.warning("[生成任务] 批量源 plan 解析失败", exc_info=True)
+        batch_source_plan_id = (
+            request.source_edit_plan_id
+            or resolve_latest_plan_by_template(db, template_id=request.template_id, user_id=user_id)
+            or ""
+        )
 
         if not batch_source_plan_id and not request.variant_plan_ids:
             # 无任何可用源 plan：批量变体无从选片，明确报错，严禁静默共用/同源
@@ -566,24 +473,10 @@ def create_generation_task(
                 ) from clone_err
             variant_plan_ids.append(_plan0.id)
 
-            # #1855 P0：批次区间避让表，从变体0实际clips构建初始值
-            def _collect_segments(pid):
-                segs = {}
-                _sk, _pg = 0, 500
-                while True:
-                    _b = _plan_svc._clip_repo.list_by_plan(pid, skip=_sk, limit=_pg)
-                    if not _b:
-                        break
-                    for _c in _b:
-                        if _c.asset_id and float(_c.duration or 0) > 0:
-                            _st = float(_c.start_time or 0.0)
-                            segs.setdefault(_c.asset_id, []).append((_st, _st + float(_c.duration)))
-                    if len(_b) < _pg:
-                        break
-                    _sk += _pg
-                return segs
+            # #1855 P0：批次区间避让表，从变体0实际clips构建初始值（公共函数）
+            from app.services.generation_common import collect_plan_segments as _collect_segments
 
-            _batch_segments = _collect_segments(_plan0.id)
+            _batch_segments = _collect_segments(_plan0.id, _plan_svc._clip_repo)
 
             # 变体 1..N-1 独立选片（传入累积batch_segments做素材区间避让）
             for task_index in range(1, count):
@@ -631,7 +524,7 @@ def create_generation_task(
 
                 # #1855 P0：把新变体的clips区间追加到batch_segments，供下一变体避让
                 try:
-                    _new_segs = _collect_segments(variant.id)
+                    _new_segs = _collect_segments(variant.id, _plan_svc._clip_repo)
                     for _aid, _ivs in _new_segs.items():
                         _batch_segments.setdefault(_aid, []).extend(_ivs)
                 except Exception:
@@ -659,24 +552,13 @@ def create_generation_task(
             )
             _single_vd: list[float] = _query_voice_durations(db, _voices)
             _single_dur = _single_vd[0] if _single_vd else 0.0
-            _single_plan = request.source_edit_plan_id
-            if not _single_plan and request.template_id:
-                try:
-                    from packages.adapters.sqlalchemy_impl.models import EditPlanModel
+            from app.services.generation_common import resolve_latest_plan_by_template
 
-                    _latest = (
-                        db.query(EditPlanModel)
-                        .filter(
-                            EditPlanModel.template_id == request.template_id,
-                            EditPlanModel.created_by_user_id == user_id,
-                        )
-                        .order_by(EditPlanModel.created_at.desc())
-                        .first()
-                    )
-                    if _latest:
-                        _single_plan = _latest.id
-                except Exception:
-                    logger.warning("[生成任务] 单任务源 plan 解析失败", exc_info=True)
+            _single_plan = (
+                request.source_edit_plan_id
+                or resolve_latest_plan_by_template(db, template_id=request.template_id, user_id=user_id)
+                or ""
+            )
             if _single_dur > 0 and _single_plan:
                 from app.services.edit_plan_service import EditPlanService
 

apps/api/app/api/routes/generation_variant_plans.py

@@ -90,25 +90,12 @@ def create_variant_plans(
     except VariantVoiceError as exc:
         raise HTTPException(status_code=400, detail=str(exc)) from exc
 
-    # 解析源 plan：显式传入优先；否则按 template_id + user 查最新
+    # 解析源 plan：显式传入优先；否则按 template_id + user 查最新（公共函数）
+    from app.services.generation_common import resolve_latest_plan_by_template
+
     source_plan_id = request.source_edit_plan_id.strip()
     if not source_plan_id and request.template_id.strip():
-        try:
-            from packages.adapters.sqlalchemy_impl.models import EditPlanModel
-
-            _latest = (
-                db.query(EditPlanModel)
-                .filter(
-                    EditPlanModel.template_id == request.template_id.strip(),
-                    EditPlanModel.created_by_user_id == user_id,
-                )
-                .order_by(EditPlanModel.created_at.desc())
-                .first()
-            )
-            if _latest:
-                source_plan_id = _latest.id
-        except Exception:
-            logger.exception("[variant-plans] 源 plan 解析失败")
+        source_plan_id = resolve_latest_plan_by_template(db, template_id=request.template_id, user_id=user_id) or ""
 
     if not source_plan_id:
         raise HTTPException(

apps/api/app/services/edit_plan_service.py

@@ -1003,25 +1003,10 @@ class EditPlanService:
                 logger.exception("变体0 配音分配失败（不阻断）: plan=%s", plan0.id)
         plan_ids.append(plan0.id)
 
-        # #1855 P0：批次内素材区间避让表——从变体0实际落库的clips构建初始值
-        def _collect_plan_segments(pid: str) -> dict[str, list[tuple[float, float]]]:
-            """分页读取 plan 所有 clips，构建 {asset_id: [(start, end), ...]} 区间表。"""
-            segs: dict[str, list[tuple[float, float]]] = {}
-            _sk2, _pg2 = 0, 500
-            while True:
-                _b2 = self._clip_repo.list_by_plan(pid, skip=_sk2, limit=_pg2)
-                if not _b2:
-                    break
-                for _c in _b2:
-                    if _c.asset_id and float(_c.duration or 0) > 0:
-                        _st = float(_c.start_time or 0.0)
-                        segs.setdefault(_c.asset_id, []).append((_st, _st + float(_c.duration)))
-                if len(_b2) < _pg2:
-                    break
-                _sk2 += _pg2
-            return segs
+        # #1855 P0：批次内素材区间避让表——从变体0实际落库的clips构建初始值（公共函数）
+        from app.services.generation_common import collect_plan_segments as _collect_plan_segments
 
-        batch_segments_acc: dict[str, list[tuple[float, float]]] = _collect_plan_segments(plan0.id)
+        batch_segments_acc: dict[str, list[tuple[float, float]]] = _collect_plan_segments(plan0.id, self._clip_repo)
 
         # 变体 1..N-1：独立选片（传入累积的 batch_segments 做区间避让）
         for i in range(1, count):
@@ -1061,7 +1046,7 @@ class EditPlanService:
 
             # #1855 P0：把当前新变体的 clips 区间追加到 batch_segments，供下一变体避让
             try:
-                _new_segs = _collect_plan_segments(variant.id)
+                _new_segs = _collect_plan_segments(variant.id, self._clip_repo)
                 for _aid, _ivs in _new_segs.items():
                     batch_segments_acc.setdefault(_aid, []).extend(_ivs)
             except Exception:

apps/api/app/services/generation_common.py

@@ -0,0 +1,175 @@
+"""智能剪辑公共服务辅助函数（从 route 层下沉）。
+
+集中管理：
+- query_voice_durations：批量查询配音素材时长
+- writeback_edit_plan_config：任务入队后回写 EditPlan.config
+- collect_plan_segments：分页读取 plan clips 构建素材区间表（变体避让用）
+- resolve_latest_plan_by_template：按 template_id + user_id 查最新 EditPlan
+
+设计原则：
+- 无副作用的纯查询 / 幂等写回；失败一律不阻断主流程（记日志 + 返回安全默认值）
+- 不依赖 FastAPI / HTTPException，便于 service 层和 worker 复用
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Optional
+
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+
+def query_voice_durations(db: Session, voice_ids: list[str]) -> list[float]:
+    """批量查询配音素材时长（秒），#1749 配音时长分配用。
+
+    逐项 try/float 硬化：MagicMock/异常/缺失 → 0.0（无配音不分配，不阻断）。
+
+    #1855 P0修复：不再对 voice_ids 去重，保持与调用方传入顺序/长度一致，
+    允许同配音id多次出现时返回相同时长（支持"同配音N变体"的时长对齐）。
+    """
+    raw_ids = list(voice_ids or [])
+    if not raw_ids:
+        return []
+    unique_ids: list[str] = []
+    _seen: set[str] = set()
+    for v in raw_ids:
+        if v and v not in _seen:
+            _seen.add(v)
+            unique_ids.append(v)
+    if not unique_ids:
+        return [0.0 for _ in raw_ids]
+    try:
+        from packages.adapters.sqlalchemy_impl.models import AssetModel
+
+        rows = db.query(AssetModel.id, AssetModel.duration).filter(AssetModel.id.in_(unique_ids)).all()
+        dur_map: dict[str, float] = {}
+        for row in rows:
+            try:
+                dur_map[row[0]] = float(row[1] or 0.0)
+            except (TypeError, ValueError):
+                dur_map[row[0]] = 0.0
+        return [dur_map.get(v, 0.0) if v else 0.0 for v in raw_ids]
+    except Exception:
+        logger.warning("[generation_common] 配音时长查询失败（按无配音处理，不阻断）", exc_info=True)
+        return [0.0 for _ in raw_ids]
+
+
+def writeback_edit_plan_config(
+    plan_id: str,
+    task_id: str,
+    title_config: dict | None,
+    db: Session,
+) -> None:
+    """任务入队成功后，回写 EditPlan.config：generation_task_id + title_config。
+
+    用 merge 方式更新，不整体覆盖 config，避免丢失其他字段。
+    失败只记日志，不影响任务创建。
+    """
+    if not plan_id:
+        return
+    try:
+        from packages.adapters.sqlalchemy_impl.models import EditPlanModel
+
+        plan_model = db.query(EditPlanModel).filter(EditPlanModel.id == plan_id).first()
+        if plan_model is None:
+            logger.warning("[generation_common] 回写plan.config失败: plan不存在 plan_id=%s", plan_id)
+            return
+
+        current_config = plan_model.config if isinstance(plan_model.config, dict) else {}
+        merged = dict(current_config)
+        merged["generation_task_id"] = task_id
+
+        if title_config:
+            old_title_config = merged.get("title_config", {}) or {}
+            old_title_text = (old_title_config.get("text") or "").strip()
+            new_title_text = (title_config.get("text") or "").strip()
+            if old_title_text != new_title_text:
+                if "cover" in merged:
+                    del merged["cover"]
+                    logger.info(
+                        "[generation_common] 标题变化，清除旧封面: plan_id=%s old_title=%s new_title=%s",
+                        plan_id,
+                        old_title_text,
+                        new_title_text,
+                    )
+            merged["title_config"] = title_config
+
+        plan_model.config = merged
+        db.commit()
+        logger.info(
+            "[generation_common] 回写plan.config成功: plan_id=%s task_id=%s keys=%s",
+            plan_id,
+            task_id,
+            list(merged.keys()),
+        )
+    except Exception as e:
+        logger.warning(
+            "[generation_common] 回写plan.config异常（不影响任务创建）: plan_id=%s error=%s",
+            plan_id,
+            e,
+            exc_info=True,
+        )
+        try:
+            db.rollback()
+        except Exception:
+            pass
+
+
+def collect_plan_segments(
+    plan_id: str,
+    clip_repo: Any,
+    *,
+    page_size: int = 500,
+) -> dict[str, list[tuple[float, float]]]:
+    """分页读取 plan 所有 clips，构建 {asset_id: [(start, end), ...]} 素材区间表。
+
+    用于 #1855 P0 批次内素材区间避让（变体间素材片段重叠控制）。
+    """
+    segs: dict[str, list[tuple[float, float]]] = {}
+    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:
+            if c.asset_id and float(c.duration or 0) > 0:
+                st = float(c.start_time or 0.0)
+                segs.setdefault(c.asset_id, []).append((st, st + float(c.duration)))
+        if len(batch) < pg:
+            break
+        sk += pg
+    return segs
+
+
+def resolve_latest_plan_by_template(
+    db: Session,
+    *,
+    template_id: str,
+    user_id: str,
+) -> Optional[str]:
+    """按 template_id + user_id 查找最新的 EditPlan.id（模板兜底用）。找不到返回 None。"""
+    if not (template_id or "").strip():
+        return None
+    try:
+        from packages.adapters.sqlalchemy_impl.models import EditPlanModel
+
+        latest = (
+            db.query(EditPlanModel)
+            .filter(
+                EditPlanModel.template_id == template_id.strip(),
+                EditPlanModel.created_by_user_id == user_id,
+            )
+            .order_by(EditPlanModel.created_at.desc())
+            .first()
+        )
+        return latest.id if latest else None
+    except Exception:
+        logger.warning(
+            "[generation_common] 按template查找最新plan失败: template=%s user=%s",
+            template_id,
+            user_id,
+            exc_info=True,
+        )
+        return None

tests/unit/test_generation_common.py

@@ -0,0 +1,259 @@
+"""generation_common 公共服务辅助函数单元测试。
+
+覆盖 query_voice_durations / writeback_edit_plan_config / collect_plan_segments /
+resolve_latest_plan_by_template 四个下沉函数的主路径、边界与容错路径。
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# query_voice_durations
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+class TestQueryVoiceDurations:
+    def _make_db_with_rows(self, rows):
+        """构造 MagicMock db，query().filter().all() 返回 rows。"""
+        db = MagicMock()
+        db.query.return_value.filter.return_value.all.return_value = list(rows)
+        return db
+
+    def test_empty_input_returns_empty_list(self):
+        from app.services.generation_common import query_voice_durations
+
+        db = MagicMock()
+        assert query_voice_durations(db, []) == []
+        assert query_voice_durations(db, None) == []
+        db.query.assert_not_called()
+
+    def test_all_empty_or_falsy_ids_returns_zero_list(self):
+        from app.services.generation_common import query_voice_durations
+
+        db = MagicMock()
+        assert query_voice_durations(db, ["", None, ""]) == [0.0, 0.0, 0.0]
+
+    def test_normal_lookup_returns_durations_in_input_order(self):
+        from app.services.generation_common import query_voice_durations
+
+        db = self._make_db_with_rows([("v1", 3.5), ("v2", 7.2)])
+        result = query_voice_durations(db, ["v1", "v2", "v-missing"])
+        assert result == [3.5, 7.2, 0.0]
+
+    def test_duplicate_ids_returns_consistent_durations_preserves_order(self):
+        """#1855：同配音 id 多次出现应返回相同时长，保持输入顺序/长度。"""
+        from app.services.generation_common import query_voice_durations
+
+        db = self._make_db_with_rows([("v1", 4.0)])
+        result = query_voice_durations(db, ["v1", "v1", "v1"])
+        assert result == [4.0, 4.0, 4.0]
+
+    def test_non_numeric_duration_coerced_to_zero(self):
+        from app.services.generation_common import query_voice_durations
+
+        db = self._make_db_with_rows([("v1", None), ("v2", "not-a-number"), ("v3", 2.0)])
+        result = query_voice_durations(db, ["v1", "v2", "v3"])
+        assert result == [0.0, 0.0, 2.0]
+
+    def test_db_exception_returns_zeros_and_logs(self, caplog):
+        from app.services.generation_common import query_voice_durations
+
+        db = MagicMock()
+        db.query.side_effect = RuntimeError("DB boom")
+        with caplog.at_level("WARNING"):
+            result = query_voice_durations(db, ["v1", "v2"])
+        assert result == [0.0, 0.0]
+        assert any("配音时长查询失败" in rec.message for rec in caplog.records)
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# writeback_edit_plan_config
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+def _make_plan_model(config=None):
+    plan = MagicMock()
+    plan.config = config if config is not None else {}
+    return plan
+
+
+class TestWritebackEditPlanConfig:
+    def test_empty_plan_id_returns_immediately(self):
+        from app.services.generation_common import writeback_edit_plan_config
+
+        db = MagicMock()
+        writeback_edit_plan_config("", "task1", None, db)
+        db.query.assert_not_called()
+
+    def test_plan_not_found_logs_and_returns(self, caplog):
+        from app.services.generation_common import writeback_edit_plan_config
+
+        db = MagicMock()
+        db.query.return_value.filter.return_value.first.return_value = None
+        with caplog.at_level("WARNING"):
+            writeback_edit_plan_config("p999", "task1", None, db)
+        db.commit.assert_not_called()
+        assert any("plan不存在" in rec.message for rec in caplog.records)
+
+    def test_writes_task_id_preserves_existing_config(self):
+        from app.services.generation_common import writeback_edit_plan_config
+
+        plan = _make_plan_model({"other": "keep-me"})
+        db = MagicMock()
+        db.query.return_value.filter.return_value.first.return_value = plan
+        writeback_edit_plan_config("p1", "task-xyz", None, db)
+        assert plan.config["generation_task_id"] == "task-xyz"
+        assert plan.config["other"] == "keep-me"
+        assert "title_config" not in plan.config
+        db.commit.assert_called_once()
+
+    def test_merges_title_config_without_title_change(self):
+        from app.services.generation_common import writeback_edit_plan_config
+
+        plan = _make_plan_model({"title_config": {"text": "old"}, "cover": "x"})
+        db = MagicMock()
+        db.query.return_value.filter.return_value.first.return_value = plan
+        writeback_edit_plan_config("p1", "t1", {"text": "old"}, db)
+        assert plan.config["title_config"] == {"text": "old"}
+        # 标题未变 → cover 保留
+        assert plan.config.get("cover") == "x"
+
+    def test_title_change_clears_cover(self):
+        from app.services.generation_common import writeback_edit_plan_config
+
+        plan = _make_plan_model({"title_config": {"text": "old"}, "cover": "x"})
+        db = MagicMock()
+        db.query.return_value.filter.return_value.first.return_value = plan
+        writeback_edit_plan_config("p1", "t1", {"text": "new-title"}, db)
+        assert "cover" not in plan.config
+        assert plan.config["title_config"] == {"text": "new-title"}
+
+    def test_config_not_dict_treated_as_empty(self):
+        from app.services.generation_common import writeback_edit_plan_config
+
+        plan = _make_plan_model(config=None)
+        db = MagicMock()
+        db.query.return_value.filter.return_value.first.return_value = plan
+        writeback_edit_plan_config("p1", "t1", {"text": "hi"}, db)
+        assert plan.config["generation_task_id"] == "t1"
+        assert plan.config["title_config"] == {"text": "hi"}
+
+    def test_exception_triggers_rollback_and_logs(self, caplog):
+        from app.services.generation_common import writeback_edit_plan_config
+
+        db = MagicMock()
+        db.query.return_value.filter.return_value.first.side_effect = RuntimeError("fail")
+        with caplog.at_level("WARNING"):
+            writeback_edit_plan_config("p1", "t1", None, db)
+        db.rollback.assert_called_once()
+        assert any("回写plan.config异常" in rec.message for rec in caplog.records)
+
+    def test_exception_with_rollback_also_failing_is_safe(self, caplog):
+        """外层异常后，db.rollback() 自己也抛异常时也不应中断（pass 兜底）。"""
+        from app.services.generation_common import writeback_edit_plan_config
+
+        db = MagicMock()
+        db.query.return_value.filter.return_value.first.side_effect = RuntimeError("fail")
+        db.rollback.side_effect = RuntimeError("rollback boom")
+        with caplog.at_level("WARNING"):
+            # 不应抛出异常
+            writeback_edit_plan_config("p1", "t1", None, db)
+        assert any("回写plan.config异常" in rec.message for rec in caplog.records)
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# collect_plan_segments
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+def _make_clip(asset_id, start, duration):
+    c = MagicMock()
+    c.asset_id = asset_id
+    c.start_time = start
+    c.duration = duration
+    return c
+
+
+class TestCollectPlanSegments:
+    def test_empty_plan_returns_empty(self):
+        from app.services.generation_common import collect_plan_segments
+
+        repo = MagicMock()
+        repo.list_by_plan.return_value = []
+        assert collect_plan_segments("p1", repo) == {}
+
+    def test_single_page_collects_segments(self):
+        from app.services.generation_common import collect_plan_segments
+
+        repo = MagicMock()
+        repo.list_by_plan.side_effect = [
+            [_make_clip("a1", 0.0, 5.0), _make_clip("a1", 10.0, 3.0), _make_clip("a2", 2.0, 4.0)],
+            [],
+        ]
+        segs = collect_plan_segments("p1", repo, page_size=500)
+        assert segs["a1"] == [(0.0, 5.0), (10.0, 13.0)]
+        assert segs["a2"] == [(2.0, 6.0)]
+
+    def test_pagination_walks_all_batches(self):
+        from app.services.generation_common import collect_plan_segments
+
+        repo = MagicMock()
+        page1 = [_make_clip("a1", 0.0, 1.0)] * 2
+        page2 = [_make_clip("a2", 0.0, 2.0)] * 2
+        page3 = [_make_clip("a3", 0.0, 1.0)]  # short final batch → stop
+        repo.list_by_plan.side_effect = [page1, page2, page3]
+        segs = collect_plan_segments("p1", repo, page_size=2)
+        assert set(segs.keys()) == {"a1", "a2", "a3"}
+        assert repo.list_by_plan.call_count == 3
+
+    def test_skips_zero_or_negative_duration_clips(self):
+        from app.services.generation_common import collect_plan_segments
+
+        repo = MagicMock()
+        repo.list_by_plan.side_effect = [
+            [_make_clip(None, 0.0, 5.0), _make_clip("a1", 0.0, 0.0), _make_clip("a1", 1.0, -1.0)],
+            [],
+        ]
+        assert collect_plan_segments("p1", repo) == {}
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# resolve_latest_plan_by_template
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+class TestResolveLatestPlanByTemplate:
+    @pytest.mark.parametrize("tid", ["", None, "   "])
+    def test_empty_template_returns_none(self, tid):
+        from app.services.generation_common import resolve_latest_plan_by_template
+
+        db = MagicMock()
+        assert resolve_latest_plan_by_template(db, template_id=tid, user_id="u1") is None
+        db.query.assert_not_called()
+
+    def test_returns_latest_plan_id(self):
+        from app.services.generation_common import resolve_latest_plan_by_template
+
+        db = MagicMock()
+        latest = MagicMock(id="plan-xyz")
+        db.query.return_value.filter.return_value.order_by.return_value.first.return_value = latest
+        assert resolve_latest_plan_by_template(db, template_id=" tpl1 ", user_id="u1") == "plan-xyz"
+
+    def test_no_plan_returns_none(self):
+        from app.services.generation_common import resolve_latest_plan_by_template
+
+        db = MagicMock()
+        db.query.return_value.filter.return_value.order_by.return_value.first.return_value = None
+        assert resolve_latest_plan_by_template(db, template_id="tpl", user_id="u") is None
+
+    def test_db_exception_returns_none_and_logs(self, caplog):
+        from app.services.generation_common import resolve_latest_plan_by_template
+
+        db = MagicMock()
+        db.query.side_effect = RuntimeError("boom")
+        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)