apps/api/app/api/routes/templates_editor/clips.py

@@ -17,16 +17,21 @@ from __future__ import annotations
 
 import json
 import logging
+import random
 import re
 
 from app.auth import AuthenticatedUser, get_current_user
 from app.core.storage import get_storage_service
-from app.dependencies import get_asset_repository
+from app.dependencies import get_asset_repository, get_db_session
 from app.services.edit_plan_service import EditPlanService
 from app.services.edit_template_service import EditTemplateService
 from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy.orm import Session
 
 from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
+from packages.adapters.sqlalchemy_impl.template_repository import (
+    SQLAlchemyTemplateRepository,
+)
 from packages.domain.plan_generator_utils import _calc_random_start_time
 from packages.shared.mediakit_client import get_mediakit_client
 
@@ -355,6 +360,72 @@ def batch_delete_editor_clips(
 
 
 
+def _safe_segment_duration(value, default: float) -> float:
+    """安全地将数据库中的时长值转换为正浮点数.
+
+    处理 None、无效类型、负数、NaN 等异常情况。
+    """
+    if value is None:
+        return default
+    try:
+        result = float(value)
+    except (ValueError, TypeError):
+        return default
+    if result != result or result <= 0:  # NaN check or non-positive
+        return default
+    return result
+
+
+def _get_template_segments(
+    template_id: str,
+    tpl_svc: EditTemplateService,
+    db: Session,
+) -> list[tuple[int, float, float]]:
+    """获取模板的片段配置（顺序、最短时长、最长时长）.
+
+    优先从新模板系统（template_clip_configs）查询，
+    若不存在则回退到旧模板系统（template_segments）。
+
+    Returns:
+        [(segment_order, duration_min, duration_max), ...] 按 order 排序
+    """
+    # 优先查新模板系统
+    try:
+        clip_configs = tpl_svc.list_clip_configs(template_id)
+        if clip_configs:
+            result = []
+            for cc in clip_configs:
+                dur_min = _safe_segment_duration(
+                    cc.min_duration, _DEFAULT_EDITOR_CLIP_DURATION
+                )
+                dur_max = _safe_segment_duration(
+                    cc.max_duration or cc.min_duration,
+                    _DEFAULT_EDITOR_CLIP_DURATION,
+                )
+                dur_min, dur_max = min(dur_min, dur_max), max(dur_min, dur_max)
+                result.append((cc.order, dur_min, dur_max))
+            return sorted(result, key=lambda x: x[0])
+    except Exception:
+        logger.warning("新模板系统查询clip_configs失败，回退到旧系统", exc_info=True)
+
+    # 回退到旧模板系统（template_segments表）
+    try:
+        old_repo = SQLAlchemyTemplateRepository(db)
+        segments = old_repo.list_segments(template_id)
+        if segments:
+            result = []
+            for s in segments:
+                dur_min = _safe_segment_duration(s.duration_min, _DEFAULT_EDITOR_CLIP_DURATION)
+                dur_max = _safe_segment_duration(s.duration_max, _DEFAULT_EDITOR_CLIP_DURATION)
+                dur_min, dur_max = min(dur_min, dur_max), max(dur_min, dur_max)
+                result.append((s.segment_order, dur_min, dur_max))
+            return sorted(result, key=lambda x: x[0])
+    except Exception:
+        logger.warning("旧模板系统查询segments失败", exc_info=True)
+
+    return []
+
+
 def _recommended_time_conflicts(
     start: float,
     duration: float,
@@ -489,22 +560,36 @@ def create_clips_from_assets_editor(
     plan_id: str = Depends(get_draft_plan_id),
     services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
     asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
+    db: Session = Depends(get_db_session),
     current_user: AuthenticatedUser = Depends(get_current_user),
 ) -> ClipsFromAssetsResponse:
-    """从素材批量创建片段（支持同一素材切多个片段 + 随机起始时间 + 去重）.
+    """从素材批量创建片段（按模板segment配置创建，事务性替换）.
 
     逻辑：
-    1. 模板要求 N 个片段，必须创建 N 个（不管素材有几个）
-    2. 素材数量 < 片段数量时，同一素材轮询切多个片段
-    3. 每个片段从素材中随机选取不重复时间段
-    4. 素材时长不足 5s 时缩短 clip duration
-    5. 新片段追加到时间线末尾（order 在现有最大值基础上递增）
+    1. 从模板读取 segments，片段数量 = segment 数量（忽略前端传的 required_clips_count）
+    2. 每个片段时长在 segment 的 duration_min ~ duration_max 之间随机取值（保留一位小数）
+    3. 素材按片段顺序轮询分配，素材不够时同一素材切多个片段
+    4. 使用 replace_all_clips_transactional 原子性地清空旧片段并创建新的
+    5. MediaKit 智能选片：第一个使用某素材的片段用推荐起始时间，后续用随机
+    6. 素材时长为 0 或缺失时报 400，不创建无效片段
     """
-    _, plan_svc = services
+    tpl_svc, plan_svc = services
 
-    required_count = body.required_clips_count if body.required_clips_count is not None else len(body.asset_ids)
+    # 1. 查询模板 segments
+    segments = _get_template_segments(template_id, tpl_svc, db)
+    if not segments:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="模板没有片段配置，无法创建片段",
+        )
 
-    # 去重后批量获取素材实际时长，避免重复查询
+    if not body.asset_ids:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="素材列表为空，无法创建片段",
+        )
+
+    # 2. 获取素材实际时长（去重查询）
     unique_asset_ids = list(dict.fromkeys(body.asset_ids))
     asset_durations: dict[str, float] = {}
     for asset_id in unique_asset_ids:
@@ -512,106 +597,113 @@ def create_clips_from_assets_editor(
         if asset and hasattr(asset, "duration"):
             asset_durations[asset_id] = float(asset.duration or 0.0)
 
-    # 计算追加起始 order：当前 plan 已有片段的最大 order + 1
-    # list_clips 返回 List[EditPlanClip]
-    existing_clips_list = plan_svc.list_clips(plan_id)
-    next_order = max((c.order for c in existing_clips_list), default=-1) + 1
+    # 3. 获取 MediaKit 智能选片推荐（保持60s timeout + poll 2s + 15次）
+    mediakit_recommendations = _get_mediakit_recommendations(
+        unique_asset_ids, asset_repo
+    )
 
-    # 从已有片段中构建已使用时间段，避免跨任务重复使用同一段素材区域
+    # 4. 在内存中计算所有片段数据
+    asset_first_used: set[str] = set()
     used_segments: dict[str, list[tuple[float, float]]] = {}
-    for _clip in existing_clips_list:
-        if _clip.asset_id and _clip.start_time is not None and _clip.duration is not None:
-            used_segments.setdefault(_clip.asset_id, []).append(
-                (float(_clip.start_time), float(_clip.start_time) + float(_clip.duration))
+    clips_data: list[dict] = []
+
+    for i, (_seg_order, dur_min, dur_max) in enumerate(segments):
+        # 轮询分配素材
+        asset_id = body.asset_ids[i % len(body.asset_ids)]
+        asset_total = asset_durations.get(asset_id, 0.0)
+
+        # 素材时长为 0 或缺失时无法创建有效片段
+        if asset_total <= 0:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"素材 {asset_id} 时长信息缺失或为0，无法创建片段",
             )
 
-    # 尝试获取 MediaKit 智能选片推荐
-    mediakit_recommendations = _get_mediakit_recommendations(unique_asset_ids, asset_repo)
-
-    clips = []
-
-    for i in range(required_count):
-        # 轮询分配素材：素材不够时同一素材切多个片段
-        asset_id = body.asset_ids[i % len(body.asset_ids)]
+        # 在 segment 的 duration_min ~ duration_max 之间随机取值（保留一位小数）
+        raw_duration = random.uniform(dur_min, dur_max)
+        clip_duration = round(raw_duration, 1)
 
         # 素材时长不足时缩短 clip duration
-        asset_total = asset_durations.get(asset_id, 0.0)
-        if asset_total > 0:
-            clip_duration = min(_DEFAULT_EDITOR_CLIP_DURATION, asset_total)
-        else:
-            clip_duration = _DEFAULT_EDITOR_CLIP_DURATION
+        clip_duration = min(clip_duration, asset_total)
 
-        # 优先使用 MediaKit 推荐的起始时间，冲突时降级为随机
+        if clip_duration <= 0:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"素材 {asset_id} 时长不足，无法创建有效片段",
+            )
+
+        # 确定起始时间
+        is_first_use = asset_id not in asset_first_used
         recommended_start = mediakit_recommendations.get(asset_id)
+
         if (
-            recommended_start is not None
-            and recommended_start + clip_duration <= asset_durations.get(asset_id, float("inf"))
+            is_first_use
+            and recommended_start is not None
+            and recommended_start + clip_duration <= asset_total
             and not _recommended_time_conflicts(
                 recommended_start, clip_duration, used_segments.get(asset_id, [])
             )
         ):
             start_time = recommended_start
             logger.info(
-                "使用MediaKit推荐起始时间: asset_id=%s start_time=%.2f",
-                asset_id, start_time,
+                "使用MediaKit推荐起始时间: asset_id=%s start_time=%.2f duration=%.1f",
+                asset_id,
+                start_time,
+                clip_duration,
             )
         else:
-            if recommended_start is not None:
+            if is_first_use and recommended_start is not None:
                 logger.info(
                     "MediaKit推荐时间冲突或越界，降级为随机: asset_id=%s recommended=%.2f",
-                    asset_id, recommended_start,
+                    asset_id,
+                    recommended_start,
+                )
+            elif not is_first_use:
+                logger.info(
+                    "素材%s非首次使用，使用随机起始时间",
+                    asset_id,
                 )
-            # 随机选择起始时间，避开已使用的时间段
             start_time = _calc_random_start_time(
                 asset_id, clip_duration, asset_durations, used_segments
             )
+
         if start_time is None:
-            # 素材时长信息缺失，无法计算随机起始时间
             raise HTTPException(
                 status_code=status.HTTP_400_BAD_REQUEST,
-                detail=f"素材 {asset_id} 时长信息缺失，无法创建片段",
+                detail=f"素材 {asset_id} 时长信息缺失，无法计算起始时间",
             )
 
-        # 记录已使用时间段（用于去重）
-        used_segments.setdefault(asset_id, []).append((start_time, start_time + clip_duration))
+        # 记录已使用时间段
+        used_segments.setdefault(asset_id, []).append(
+            (start_time, start_time + clip_duration)
+        )
+        asset_first_used.add(asset_id)
 
-        try:
-            clip = plan_svc.create_clip(
-                plan_id,
-                clip_type=body.clip_type or "main",
-                order=next_order + i,
-                duration=clip_duration,
-                start_time=start_time,
-                asset_id=asset_id,
-            )
-            clips.append(clip)
-        except ValueError as exc:
-            logger.error(
-                "创建片段失败: plan_id=%s asset_id=%s order=%d error=%s",
-                plan_id,
-                asset_id,
-                next_order + i,
-                exc,
-                exc_info=True,
-            )
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail=f"创建片段失败: {exc}",
-            ) from exc
+        clips_data.append(
+            {
+                "order": i,
+                "asset_id": asset_id,
+                "start_time": start_time,
+                "duration": clip_duration,
+                "clip_type": body.clip_type or "main",
+            }
+        )
+
+    # 5. 事务性替换：清空旧片段 → 创建新片段 → 标记ready（单事务，失败自动回滚）
+    created_count = plan_svc.replace_all_clips_transactional(plan_id, clips_data)
 
     logger.info(
-        "模板编辑器从素材创建片段: template_id=%s plan_id=%s required=%d actual=%d by user=%s",
+        "from-assets按模板创建片段: template_id=%s plan_id=%s segments=%d created=%d by user=%s",
         template_id,
         plan_id,
-        required_count,
-        len(clips),
+        len(segments),
+        created_count,
         current_user.user.id,
     )
 
-    # 新创建的片段已分配素材，立即标记为 ready，否则渲染管线找不到就绪片段
-    plan_svc.mark_clips_ready(plan_id)
+    # 返回事务后查询到的 clip IDs（replace 方法不返回 ID 列表，用 created_count 构造响应）
     return ClipsFromAssetsResponse(
-        created_count=len(clips),
+        created_count=created_count,
         plan_id=plan_id,
-        clip_ids=[c.id for c in clips],
+        clip_ids=[],  # 事务方法不返回 ID；前端不需要逐个 ID
     )

apps/api/app/services/edit_plan_service.py

@@ -400,7 +400,7 @@ class EditPlanService:
                 order = clip_item.get("order") or i
                 clip = EditPlanClip.create(
                     plan_id=plan_id,
-                    clip_type="main",
+                    clip_type=clip_item.get("clip_type", "main"),
                     order=order,
                     asset_id=clip_item.get("asset_id", ""),
                     start_time=clip_item.get("start_time", 0.0),

tests/unit/test_editor_clips_random_start.py

@@ -1,14 +1,15 @@
-"""测试编辑器 from-assets 端点：同一素材切多个片段 + 随机起始 + 去重.
+"""测试编辑器 from-assets 端点：按模板segment创建片段 + 事务性替换 + 随机起始.
 
 覆盖：
-- required_clips_count 精确控制片段数量
+- 片段数量 = segment 数量（required_clips_count 被忽略）
 - 素材不足时同一素材轮询切多个片段
 - 随机 start_time + used_segments 去重
 - 素材时长不足时 clip duration 缩短
-- 向后兼容（不传 required_clips_count 时等于素材数量）
-- order 追加到时间线末尾
+- 素材时长为 0 时抛 400
+- 使用 replace_all_clips_transactional 原子性替换
+- order 从 0 开始
 - start_time=None 时抛出 400
-- create_clip 失败时抛出 400 并记录日志
+- mark_clips_ready 在事务方法内部完成
 """
 
 from __future__ import annotations
@@ -29,6 +30,20 @@ from fastapi import HTTPException
 TEST_PLAN_ID = "plan-draft-001"
 TEST_USER_ID = "user-001"
 
+# 默认测试用 segments：4 个片段，每个 3~5 秒
+DEFAULT_SEGMENTS = [(0, 3.0, 5.0), (1, 3.0, 5.0), (2, 3.0, 5.0), (3, 3.0, 5.0)]
+
+
+def _segments(count: int, dur_min: float = 3.0, dur_max: float = 5.0):
+    return [(i, dur_min, dur_max) for i in range(count)]
+
+
+def _patch_segments(segments=None):
+    return patch(
+        "app.api.routes.templates_editor.clips._get_template_segments",
+        return_value=segments if segments is not None else DEFAULT_SEGMENTS,
+    )
+
 
 def _make_auth_user():
     auth = MagicMock()
@@ -39,27 +54,6 @@ def _make_auth_user():
     return auth
 
 
-def _make_mock_clip(clip_id, order, duration, start_time=0.0, asset_id=""):
-    clip = MagicMock()
-    clip.id = clip_id
-    clip.plan_id = TEST_PLAN_ID
-    clip.clip_type = "main"
-    clip.order = order
-    clip.duration = duration
-    clip.start_time = start_time
-    clip.text_content = ""
-    clip.transition_effect = "cut"
-    clip.transition_duration = 0.0
-    clip.playback_speed = 1.0
-    clip.config = {}
-    clip.asset_id = asset_id
-    clip.status = "pending"
-    clip.template_clip_config_id = ""
-    clip.created_at = None
-    clip.updated_at = None
-    return clip
-
-
 def _make_mock_asset(asset_id, duration):
     asset = MagicMock()
     asset.id = asset_id
@@ -67,212 +61,127 @@ def _make_mock_asset(asset_id, duration):
     return asset
 
 
-def _create_clips(plan_id, clip_type, order, duration=0.0, start_time=0.0, asset_id="", **kw):
-    return _make_mock_clip(
-        clip_id=f"clip-{order}",
-        order=order,
-        duration=duration,
-        start_time=start_time,
-        asset_id=asset_id,
-    )
-
-
-def _make_plan_svc(existing_clips=None):
+def _make_plan_svc(replace_return_count=None):
     svc = MagicMock()
-    svc.get_plan_or_raise = MagicMock()
-    svc.create_clip = MagicMock(side_effect=_create_clips)
-    svc.list_clips = MagicMock(return_value=existing_clips or [])
+    # replace_all_clips_transactional 返回创建的片段数量
+    if replace_return_count is not None:
+        svc.replace_all_clips_transactional = MagicMock(return_value=replace_return_count)
+    else:
+        svc.replace_all_clips_transactional = MagicMock(return_value=0)
     return svc
 
 
-class TestEditorClipsRequiredCount:
-    """测试 required_clips_count 控制片段数量 + 同素材多片段."""
+def _get_clips_data_from_call(mock_plan_svc):
+    """从 replace_all_clips_transactional 的调用中获取 clips_data。"""
+    assert mock_plan_svc.replace_all_clips_transactional.called, "replace_all_clips_transactional 未被调用"
+    call_args = mock_plan_svc.replace_all_clips_transactional.call_args
+    # call_args = ((plan_id, clips_data), kwargs)
+    if len(call_args.args) >= 2:
+        return call_args.args[1]
+    return call_args.kwargs.get("clips_data", [])
+
+
+class TestEditorClipsBySegments:
+    """测试按 segment 数量创建片段 + 素材轮询。"""
 
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_creates_exactly_required_clips_count(self, mock_storage):
-        """required_clips_count=4 时，即使只有2个素材也创建4个片段."""
+    def test_creates_clips_matching_segment_count(self, mock_storage):
+        """4 个 segment 即使只有2个素材也创建4个片段，required_clips_count 被忽略。"""
         from app.api.routes.templates_editor.clips import (
             create_clips_from_assets_editor,
         )
         from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
 
-        mock_plan_svc = _make_plan_svc()
+        mock_plan_svc = _make_plan_svc(replace_return_count=4)
         mock_asset_repo = MagicMock()
         mock_asset_repo.get = MagicMock(side_effect=lambda aid: _make_mock_asset(aid, {"a1": 30.0, "a2": 20.0}[aid]))
 
-        body = ClipsFromAssetsRequest(asset_ids=["a1", "a2"], required_clips_count=4)
+        body = ClipsFromAssetsRequest(asset_ids=["a1", "a2"], required_clips_count=2)
 
-        result = create_clips_from_assets_editor(
-            template_id="tpl-001",
-            body=body,
-            plan_id=TEST_PLAN_ID,
-            services=(MagicMock(), mock_plan_svc),
-            asset_repo=mock_asset_repo,
-            current_user=_make_auth_user(),
-        )
+        with _patch_segments(DEFAULT_SEGMENTS):
+            result = create_clips_from_assets_editor(
+                template_id="tpl-001",
+                body=body,
+                plan_id=TEST_PLAN_ID,
+                services=(MagicMock(), mock_plan_svc),
+                asset_repo=mock_asset_repo,
+                db=MagicMock(),
+                current_user=_make_auth_user(),
+            )
 
         assert result.created_count == 4
-        assert mock_plan_svc.create_clip.call_count == 4
+        clips_data = _get_clips_data_from_call(mock_plan_svc)
+        assert len(clips_data) == 4
 
         # 验证轮询分配：a1, a2, a1, a2
-        calls = mock_plan_svc.create_clip.call_args_list
-        assert calls[0].kwargs["asset_id"] == "a1"
-        assert calls[1].kwargs["asset_id"] == "a2"
-        assert calls[2].kwargs["asset_id"] == "a1"
-        assert calls[3].kwargs["asset_id"] == "a2"
+        assert clips_data[0]["asset_id"] == "a1"
+        assert clips_data[1]["asset_id"] == "a2"
+        assert clips_data[2]["asset_id"] == "a1"
+        assert clips_data[3]["asset_id"] == "a2"
 
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_orders_append_to_existing_timeline(self, mock_storage):
-        """时间线已有2个片段时，新片段 order 应从 2 开始连续递增."""
+    def test_orders_start_at_zero(self, mock_storage):
+        """片段 order 从 0 开始递增。"""
         from app.api.routes.templates_editor.clips import (
             create_clips_from_assets_editor,
         )
         from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
 
-        # 模拟已有 order=0, order=1 的片段
-        existing = [_make_mock_clip("old-1", 0, 5.0), _make_mock_clip("old-2", 1, 5.0)]
-        mock_plan_svc = _make_plan_svc(existing_clips=existing)
-
+        mock_plan_svc = _make_plan_svc(replace_return_count=3)
         mock_asset_repo = MagicMock()
         mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
 
         body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=3)
 
-        create_clips_from_assets_editor(
-            template_id="tpl-001",
-            body=body,
-            plan_id=TEST_PLAN_ID,
-            services=(MagicMock(), mock_plan_svc),
-            asset_repo=mock_asset_repo,
-            current_user=_make_auth_user(),
-        )
-
-        calls = mock_plan_svc.create_clip.call_args_list
-        assert calls[0].kwargs["order"] == 2
-        assert calls[1].kwargs["order"] == 3
-        assert calls[2].kwargs["order"] == 4
-
-    @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_orders_start_at_zero_when_empty(self, mock_storage):
-        """空时间线时 order 从 0 开始."""
-        from app.api.routes.templates_editor.clips import (
-            create_clips_from_assets_editor,
-        )
-        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
-
-        mock_plan_svc = _make_plan_svc(existing_clips=[])
-
-        mock_asset_repo = MagicMock()
-        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
-
-        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=3)
-
-        create_clips_from_assets_editor(
-            template_id="tpl-001",
-            body=body,
-            plan_id=TEST_PLAN_ID,
-            services=(MagicMock(), mock_plan_svc),
-            asset_repo=mock_asset_repo,
-            current_user=_make_auth_user(),
-        )
-
-        calls = mock_plan_svc.create_clip.call_args_list
-        assert calls[0].kwargs["order"] == 0
-        assert calls[1].kwargs["order"] == 1
-        assert calls[2].kwargs["order"] == 2
-
-    @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_backward_compatible_default_count(self, mock_storage):
-        """不传 required_clips_count 时，片段数等于素材数（向后兼容）."""
-        from app.api.routes.templates_editor.clips import (
-            create_clips_from_assets_editor,
-        )
-        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
-
-        mock_plan_svc = _make_plan_svc()
-        mock_asset_repo = MagicMock()
-        mock_asset_repo.get = MagicMock(side_effect=lambda aid: _make_mock_asset(aid, 30.0))
-
-        body = ClipsFromAssetsRequest(asset_ids=["a1", "a2", "a3"])
-
-        result = create_clips_from_assets_editor(
-            template_id="tpl-001",
-            body=body,
-            plan_id=TEST_PLAN_ID,
-            services=(MagicMock(), mock_plan_svc),
-            asset_repo=mock_asset_repo,
-            current_user=_make_auth_user(),
-        )
-
-        assert result.created_count == 3
-        assert mock_plan_svc.create_clip.call_count == 3
-
-
-class TestEditorClipsDurationAndStartTime:
-    """测试素材时长获取、clip duration 缩短、start_time 传入."""
-
-    @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_clip_duration_shortened_for_short_assets(self, mock_storage):
-        """素材只有 3s 时 clip duration 缩短到 3.0."""
-        from app.api.routes.templates_editor.clips import (
-            create_clips_from_assets_editor,
-        )
-        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
-
-        mock_plan_svc = _make_plan_svc()
-        mock_asset_repo = MagicMock()
-        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("short", 3.0))
-
-        body = ClipsFromAssetsRequest(asset_ids=["short"])
-
-        create_clips_from_assets_editor(
-            template_id="tpl-001",
-            body=body,
-            plan_id=TEST_PLAN_ID,
-            services=(MagicMock(), mock_plan_svc),
-            asset_repo=mock_asset_repo,
-            current_user=_make_auth_user(),
-        )
-
-        calls = mock_plan_svc.create_clip.call_args_list
-        assert len(calls) == 1
-        assert calls[0].kwargs["duration"] == 3.0
-
-    @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_start_time_passed_to_create_clip(self, mock_storage):
-        """验证 _calc_random_start_time 返回值被传入 create_clip."""
-        from app.api.routes.templates_editor.clips import (
-            create_clips_from_assets_editor,
-        )
-        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
-
-        mock_plan_svc = _make_plan_svc()
-        mock_asset_repo = MagicMock()
-        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
-
-        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=2)
-
-        with patch(
-            "app.api.routes.templates_editor.clips._calc_random_start_time",
-            side_effect=[12.5, 18.0],
-        ) as mock_calc:
+        with _patch_segments(_segments(3)):
             create_clips_from_assets_editor(
                 template_id="tpl-001",
                 body=body,
                 plan_id=TEST_PLAN_ID,
                 services=(MagicMock(), mock_plan_svc),
                 asset_repo=mock_asset_repo,
+                db=MagicMock(),
                 current_user=_make_auth_user(),
             )
 
-        assert mock_calc.call_count == 2
-        calls = mock_plan_svc.create_clip.call_args_list
-        assert calls[0].kwargs["start_time"] == 12.5
-        assert calls[1].kwargs["start_time"] == 18.0
+        clips_data = _get_clips_data_from_call(mock_plan_svc)
+        assert clips_data[0]["order"] == 0
+        assert clips_data[1]["order"] == 1
+        assert clips_data[2]["order"] == 2
 
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_asset_durations_deduped(self, mock_storage):
-        """asset_ids 有重复时只查询一次素材时长."""
+    def test_uses_transactional_replace(self, mock_storage):
+        """使用 replace_all_clips_transactional 而不是分别 delete + create。"""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc(replace_return_count=2)
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
+
+        body = ClipsFromAssetsRequest(asset_ids=["a1"])
+
+        with _patch_segments(_segments(2)):
+            create_clips_from_assets_editor(
+                template_id="tpl-001",
+                body=body,
+                plan_id=TEST_PLAN_ID,
+                services=(MagicMock(), mock_plan_svc),
+                asset_repo=mock_asset_repo,
+                db=MagicMock(),
+                current_user=_make_auth_user(),
+            )
+
+        # 必须调用事务方法
+        mock_plan_svc.replace_all_clips_transactional.assert_called_once()
+        # 不应调用单独的 delete 或 create
+        assert not hasattr(mock_plan_svc, "create_clip") or not mock_plan_svc.create_clip.called
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_no_segments_raises_400(self, mock_storage):
+        """模板没有 segment 配置时返回 400。"""
         from app.api.routes.templates_editor.clips import (
             create_clips_from_assets_editor,
         )
@@ -280,26 +189,61 @@ class TestEditorClipsDurationAndStartTime:
 
         mock_plan_svc = _make_plan_svc()
         mock_asset_repo = MagicMock()
-        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
 
-        # a1 出现 3 次，但时长只应查一次
-        body = ClipsFromAssetsRequest(asset_ids=["a1", "a1", "a1"])
+        body = ClipsFromAssetsRequest(asset_ids=["a1"])
 
-        create_clips_from_assets_editor(
-            template_id="tpl-001",
-            body=body,
-            plan_id=TEST_PLAN_ID,
-            services=(MagicMock(), mock_plan_svc),
-            asset_repo=mock_asset_repo,
-            current_user=_make_auth_user(),
-        )
+        with _patch_segments([]):
+            with pytest.raises(HTTPException) as exc_info:
+                create_clips_from_assets_editor(
+                    template_id="tpl-001",
+                    body=body,
+                    plan_id=TEST_PLAN_ID,
+                    services=(MagicMock(), mock_plan_svc),
+                    asset_repo=mock_asset_repo,
+                    db=MagicMock(),
+                    current_user=_make_auth_user(),
+                )
 
-        # 去重后只调用 1 次
-        assert mock_asset_repo.get.call_count == 1
+        assert exc_info.value.status_code == 400
+        assert "片段配置" in exc_info.value.detail
+        # 不应调用替换方法
+        mock_plan_svc.replace_all_clips_transactional.assert_not_called()
+
+
+class TestEditorClipsDurationAndStartTime:
+    """测试素材时长获取、clip duration 缩短、start_time 传入。"""
 
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_used_segments_passed_to_calc(self, mock_storage):
-        """同一素材切多个片段时，used_segments 应被维护并传入."""
+    def test_clip_duration_shortened_for_short_assets(self, mock_storage):
+        """素材只有 3s 时 clip duration 缩短到不超过 3.0。"""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc(replace_return_count=1)
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("short", 3.0))
+
+        body = ClipsFromAssetsRequest(asset_ids=["short"])
+
+        with _patch_segments(_segments(1, dur_min=5.0, dur_max=10.0)):
+            create_clips_from_assets_editor(
+                template_id="tpl-001",
+                body=body,
+                plan_id=TEST_PLAN_ID,
+                services=(MagicMock(), mock_plan_svc),
+                asset_repo=mock_asset_repo,
+                db=MagicMock(),
+                current_user=_make_auth_user(),
+            )
+
+        clips_data = _get_clips_data_from_call(mock_plan_svc)
+        assert clips_data[0]["duration"] <= 3.0
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_zero_duration_asset_raises_400(self, mock_storage):
+        """素材时长为 0 时应抛出 400，而不是创建无效片段。"""
         from app.api.routes.templates_editor.clips import (
             create_clips_from_assets_editor,
         )
@@ -307,6 +251,127 @@ class TestEditorClipsDurationAndStartTime:
 
         mock_plan_svc = _make_plan_svc()
         mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("bad", 0.0))
+
+        body = ClipsFromAssetsRequest(asset_ids=["bad"])
+
+        with _patch_segments(_segments(1)):
+            with pytest.raises(HTTPException) as exc_info:
+                create_clips_from_assets_editor(
+                    template_id="tpl-001",
+                    body=body,
+                    plan_id=TEST_PLAN_ID,
+                    services=(MagicMock(), mock_plan_svc),
+                    asset_repo=mock_asset_repo,
+                    db=MagicMock(),
+                    current_user=_make_auth_user(),
+                )
+
+        assert exc_info.value.status_code == 400
+        assert "时长" in exc_info.value.detail
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_missing_duration_asset_raises_400(self, mock_storage):
+        """素材时长缺失（asset_repo.get 返回 None）时抛出 400。"""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc()
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(return_value=None)
+
+        body = ClipsFromAssetsRequest(asset_ids=["missing"])
+
+        with _patch_segments(_segments(1)):
+            with pytest.raises(HTTPException) as exc_info:
+                create_clips_from_assets_editor(
+                    template_id="tpl-001",
+                    body=body,
+                    plan_id=TEST_PLAN_ID,
+                    services=(MagicMock(), mock_plan_svc),
+                    asset_repo=mock_asset_repo,
+                    db=MagicMock(),
+                    current_user=_make_auth_user(),
+                )
+
+        assert exc_info.value.status_code == 400
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_start_time_passed_to_create(self, mock_storage):
+        """_calc_random_start_time 返回值被传入 clips_data。"""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc(replace_return_count=2)
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(side_effect=lambda aid: _make_mock_asset(aid, 30.0))
+
+        body = ClipsFromAssetsRequest(asset_ids=["a1", "a2"], required_clips_count=2)
+
+        with (
+            _patch_segments(_segments(2)),
+            patch(
+                "app.api.routes.templates_editor.clips._calc_random_start_time",
+                side_effect=[12.5, 18.0],
+            ) as mock_calc,
+        ):
+            create_clips_from_assets_editor(
+                template_id="tpl-001",
+                body=body,
+                plan_id=TEST_PLAN_ID,
+                services=(MagicMock(), mock_plan_svc),
+                asset_repo=mock_asset_repo,
+                db=MagicMock(),
+                current_user=_make_auth_user(),
+            )
+
+        assert mock_calc.call_count == 2
+        clips_data = _get_clips_data_from_call(mock_plan_svc)
+        assert clips_data[0]["start_time"] == 12.5
+        assert clips_data[1]["start_time"] == 18.0
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_asset_durations_deduped(self, mock_storage):
+        """asset_ids 有重复时只查询一次素材时长。"""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc(replace_return_count=3)
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
+
+        body = ClipsFromAssetsRequest(asset_ids=["a1", "a1", "a1"])
+
+        with _patch_segments(_segments(3)):
+            create_clips_from_assets_editor(
+                template_id="tpl-001",
+                body=body,
+                plan_id=TEST_PLAN_ID,
+                services=(MagicMock(), mock_plan_svc),
+                asset_repo=mock_asset_repo,
+                db=MagicMock(),
+                current_user=_make_auth_user(),
+            )
+
+        # 去重后只调用 1 次获取素材时长
+        assert mock_asset_repo.get.call_count == 1
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_used_segments_maintained_across_clips(self, mock_storage):
+        """同一素材切多个片段时，used_segments 应被维护。"""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc(replace_return_count=3)
+        mock_asset_repo = MagicMock()
         mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
 
         body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=3)
@@ -317,9 +382,12 @@ class TestEditorClipsDurationAndStartTime:
             captured_used_segments.append({aid: list(segs) for aid, segs in (used_segments or {}).items()})
             return (len(captured_used_segments) - 1) * 5.0
 
-        with patch(
-            "app.api.routes.templates_editor.clips._calc_random_start_time",
-            side_effect=fake_calc,
+        with (
+            _patch_segments(_segments(3)),
+            patch(
+                "app.api.routes.templates_editor.clips._calc_random_start_time",
+                side_effect=fake_calc,
+            ),
         ):
             create_clips_from_assets_editor(
                 template_id="tpl-001",
@@ -327,215 +395,80 @@ class TestEditorClipsDurationAndStartTime:
                 plan_id=TEST_PLAN_ID,
                 services=(MagicMock(), mock_plan_svc),
                 asset_repo=mock_asset_repo,
+                db=MagicMock(),
                 current_user=_make_auth_user(),
             )
 
+        # 第一次没有已使用时间段
         assert captured_used_segments[0] == {}
-        assert captured_used_segments[1] == {"a1": [(0.0, 5.0)]}
-        assert captured_used_segments[2] == {"a1": [(0.0, 5.0), (5.0, 10.0)]}
+        # 第二次有第一次的记录
+        assert len(captured_used_segments[1]["a1"]) == 1
+        # 第三次有前两次的记录
+        assert len(captured_used_segments[2]["a1"]) == 2
 
 
 class TestEditorClipsErrorHandling:
-    """测试异常处理."""
+    """测试异常处理。"""
 
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
     def test_none_start_time_raises_400(self, mock_storage):
-        """_calc_random_start_time 返回 None 时应抛出 HTTPException 400."""
+        """_calc_random_start_time 返回 None 时应抛出 HTTPException 400。"""
         from app.api.routes.templates_editor.clips import (
             create_clips_from_assets_editor,
         )
         from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
 
         mock_plan_svc = _make_plan_svc()
-        # asset_repo.get 返回 None → asset_durations 为空 → _calc_random_start_time 返回 None
         mock_asset_repo = MagicMock()
-        mock_asset_repo.get = MagicMock(return_value=None)
+        # 素材有 duration 但 random 返回 None
+        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
 
-        body = ClipsFromAssetsRequest(asset_ids=["missing-asset"])
+        body = ClipsFromAssetsRequest(asset_ids=["a1"])
 
-        with pytest.raises(HTTPException) as exc_info:
-            create_clips_from_assets_editor(
-                template_id="tpl-001",
-                body=body,
-                plan_id=TEST_PLAN_ID,
-                services=(MagicMock(), mock_plan_svc),
-                asset_repo=mock_asset_repo,
-                current_user=_make_auth_user(),
-            )
+        with (
+            _patch_segments(_segments(1)),
+            patch(
+                "app.api.routes.templates_editor.clips._calc_random_start_time",
+                return_value=None,
+            ),
+        ):
+            with pytest.raises(HTTPException) as exc_info:
+                create_clips_from_assets_editor(
+                    template_id="tpl-001",
+                    body=body,
+                    plan_id=TEST_PLAN_ID,
+                    services=(MagicMock(), mock_plan_svc),
+                    asset_repo=mock_asset_repo,
+                    db=MagicMock(),
+                    current_user=_make_auth_user(),
+                )
 
         assert exc_info.value.status_code == 400
-        assert "时长信息缺失" in exc_info.value.detail
+        assert "时长" in exc_info.value.detail
 
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_create_clip_value_error_raises_400(self, mock_storage):
-        """create_clip 抛出 ValueError 时应转为 HTTPException 400."""
+    def test_transactional_replace_exception_propagates(self, mock_storage):
+        """replace_all_clips_transactional 抛异常时应向上传播（事务已回滚）。"""
         from app.api.routes.templates_editor.clips import (
             create_clips_from_assets_editor,
         )
         from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
 
         mock_plan_svc = _make_plan_svc()
-        mock_plan_svc.create_clip = MagicMock(side_effect=ValueError("计划不存在"))
-
+        mock_plan_svc.replace_all_clips_transactional = MagicMock(side_effect=RuntimeError("DB connection lost"))
         mock_asset_repo = MagicMock()
         mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
 
         body = ClipsFromAssetsRequest(asset_ids=["a1"])
 
-        with pytest.raises(HTTPException) as exc_info:
-            create_clips_from_assets_editor(
-                template_id="tpl-001",
-                body=body,
-                plan_id=TEST_PLAN_ID,
-                services=(MagicMock(), mock_plan_svc),
-                asset_repo=mock_asset_repo,
-                current_user=_make_auth_user(),
-            )
-
-        assert exc_info.value.status_code == 400
-        assert "创建片段失败" in exc_info.value.detail
-
-
-class TestMarkClipsReadyAfterCreation:
-    """验证 from-assets 创建片段后立即调用 mark_clips_ready，确保渲染管线能找到就绪片段。"""
-
-    @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_mark_clips_ready_called_after_creation(self, _mock_storage):
-        """创建片段后必须调用 plan_svc.mark_clips_ready(plan_id)。"""
-        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
-        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
-
-        mock_plan_svc = _make_plan_svc()
-        mock_asset_repo = MagicMock()
-        mock_asset_repo.get = MagicMock(side_effect=lambda aid: _make_mock_asset(aid, {"a1": 30.0}[aid]))
-
-        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=2)
-        create_clips_from_assets_editor(
-            template_id="tmpl-1",
-            body=body,
-            plan_id="plan-xyz",
-            services=(MagicMock(), mock_plan_svc),
-            asset_repo=mock_asset_repo,
-            current_user=_make_auth_user(),
-        )
-
-        # 关键断言：mark_clips_ready 必须被调用，且传入正确的 plan_id
-        mock_plan_svc.mark_clips_ready.assert_called_once_with("plan-xyz")
-
-
-class TestCrossTaskSegmentDedup:
-    """验证 from-assets 创建片段时，used_segments 从已有片段构建，实现跨任务去重。"""
-
-    @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_used_segments_populated_from_existing_clips(self, _mock_storage):
-        """已有片段的 asset_id/start_time/duration 必须被纳入 used_segments，新片段避开已用区间。"""
-        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
-        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
-
-        # 模拟已有片段：asset "a1" 在 0~5s 已使用
-        existing = [
-            _make_mock_clip("c1", order=0, duration=5.0, start_time=0.0, asset_id="a1"),
-        ]
-        mock_plan_svc = _make_plan_svc(existing_clips=existing)
-        mock_asset_repo = MagicMock()
-        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
-
-        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1)
-        create_clips_from_assets_editor(
-            template_id="tmpl-1",
-            body=body,
-            plan_id="plan-dedup",
-            services=(MagicMock(), mock_plan_svc),
-            asset_repo=mock_asset_repo,
-            current_user=_make_auth_user(),
-        )
-
-        # 验证：新创建的 clip 的 start_time 不应与已有片段 [0, 5] 重叠
-        # create_clip 被调用时传入的 start_time 应该 >= 5 或 < 0 (不可能)
-        # 实际上 _calc_random_start_time 会避开 [0, 5]，所以 start_time 应该 > 5
-        create_calls = mock_plan_svc.create_clip.call_args_list
-        assert len(create_calls) == 1
-        new_start_time = create_calls[0].kwargs.get("start_time") or create_calls[0][1].get("start_time")
-        # 新片段不应从 0 开始（因为 0~5 已被占用）
-        # 注意：_calc_random_start_time 有随机性，但在 30s 素材中避开 [0,5] 后随机到 0~5 的概率极低
-        # 我们用一个宽松断言：start_time 应该是一个有效值
-        assert new_start_time is not None
-
-    @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_multiple_existing_clips_build_used_segments(self, _mock_storage):
-        """多个已有片段的时间段都应被收集到 used_segments 中。"""
-        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
-        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
-
-        # 模拟已有片段：asset "a1" 在 [0,5] 和 [10,15] 已使用
-        existing = [
-            _make_mock_clip("c1", order=0, duration=5.0, start_time=0.0, asset_id="a1"),
-            _make_mock_clip("c2", order=1, duration=5.0, start_time=10.0, asset_id="a1"),
-        ]
-        mock_plan_svc = _make_plan_svc(existing_clips=existing)
-        mock_asset_repo = MagicMock()
-        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
-
-        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1)
-        create_clips_from_assets_editor(
-            template_id="tmpl-1",
-            body=body,
-            plan_id="plan-dedup2",
-            services=(MagicMock(), mock_plan_svc),
-            asset_repo=mock_asset_repo,
-            current_user=_make_auth_user(),
-        )
-
-        create_calls = mock_plan_svc.create_clip.call_args_list
-        assert len(create_calls) == 1
-        new_start_time = create_calls[0].kwargs.get("start_time") or create_calls[0][1].get("start_time")
-        assert new_start_time is not None
-
-    @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_existing_clips_without_asset_id_ignored(self, _mock_storage):
-        """没有 asset_id 的已有片段不影响 used_segments。"""
-        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
-        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
-
-        # 模拟已有片段：一个没有 asset_id 的片段
-        existing = [
-            _make_mock_clip("c1", order=0, duration=5.0, start_time=0.0, asset_id=""),
-        ]
-        mock_plan_svc = _make_plan_svc(existing_clips=existing)
-        mock_asset_repo = MagicMock()
-        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
-
-        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1)
-        result = create_clips_from_assets_editor(
-            template_id="tmpl-1",
-            body=body,
-            plan_id="plan-dedup3",
-            services=(MagicMock(), mock_plan_svc),
-            asset_repo=mock_asset_repo,
-            current_user=_make_auth_user(),
-        )
-
-        # 应正常创建，不受空 asset_id 片段影响
-        assert result.created_count == 1
-
-    @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_no_existing_clips_works_same_as_before(self, _mock_storage):
-        """没有已有片段时，行为与修复前一致（used_segments 为空）。"""
-        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
-        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
-
-        mock_plan_svc = _make_plan_svc(existing_clips=[])
-        mock_asset_repo = MagicMock()
-        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
-
-        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1)
-        result = create_clips_from_assets_editor(
-            template_id="tmpl-1",
-            body=body,
-            plan_id="plan-dedup4",
-            services=(MagicMock(), mock_plan_svc),
-            asset_repo=mock_asset_repo,
-            current_user=_make_auth_user(),
-        )
-
-        assert result.created_count == 1
+        with _patch_segments(_segments(1)):
+            with pytest.raises(RuntimeError, match="DB connection lost"):
+                create_clips_from_assets_editor(
+                    template_id="tpl-001",
+                    body=body,
+                    plan_id=TEST_PLAN_ID,
+                    services=(MagicMock(), mock_plan_svc),
+                    asset_repo=mock_asset_repo,
+                    db=MagicMock(),
+                    current_user=_make_auth_user(),
+                )

tests/unit/test_mediakit_smart_clips.py

@@ -1,12 +1,16 @@
-"""测试 MediaKit 智能选片集成。
+"""测试 MediaKit 智能选片 + from-assets 按模板 segment 创建片段。
 
 覆盖：
 - _recommended_time_conflicts 冲突检测
 - _get_mediakit_recommendations 解析与降级
-- from-assets 端点：推荐时间优先使用
-- from-assets 端点：推荐时间冲突时降级随机
-- from-assets 端点：MediaKit 不可用时降级随机
-- from-assets 端点：MediaKit 返回不可解析内容时降级随机
+- _get_template_segments 查询模板片段配置
+- from-assets 端点：按模板 segment 数量和时长创建片段
+- from-assets 端点：事务性原子替换
+- from-assets 端点：素材轮询分配
+- from-assets 端点：MediaKit 推荐时间首片段使用
+- from-assets 端点：同一素材多片段时后续用随机
+- from-assets 端点：无 segment 配置时报错
+- from-assets 端点：素材时长为0时报400
 """
 
 from __future__ import annotations
@@ -37,62 +41,82 @@ class TestRecommendedTimeConflicts:
     def test_no_conflict_when_before(self):
         from app.api.routes.templates_editor.clips import _recommended_time_conflicts
 
-        # 推荐 [5, 10]，已用 [15, 20]
         assert _recommended_time_conflicts(5.0, 5.0, [(15.0, 20.0)]) is False
 
     def test_no_conflict_when_after(self):
         from app.api.routes.templates_editor.clips import _recommended_time_conflicts
 
-        # 推荐 [20, 25]，已用 [0, 10]
         assert _recommended_time_conflicts(20.0, 5.0, [(0.0, 10.0)]) is False
 
     def test_conflict_overlap_start(self):
         from app.api.routes.templates_editor.clips import _recommended_time_conflicts
 
-        # 推荐 [8, 13]，已用 [10, 20]
-        assert _recommended_time_conflicts(8.0, 5.0, [(10.0, 20.0)]) is True
+        # 推荐 [5, 10]，已用 [0, 7]
+        assert _recommended_time_conflicts(5.0, 5.0, [(0.0, 7.0)]) is True
 
     def test_conflict_overlap_end(self):
         from app.api.routes.templates_editor.clips import _recommended_time_conflicts
 
-        # 推荐 [15, 20]，已用 [10, 18]
-        assert _recommended_time_conflicts(15.0, 5.0, [(10.0, 18.0)]) is True
+        # 推荐 [5, 10]，已用 [8, 15]
+        assert _recommended_time_conflicts(5.0, 5.0, [(8.0, 15.0)]) is True
 
     def test_conflict_contained(self):
         from app.api.routes.templates_editor.clips import _recommended_time_conflicts
 
-        # 推荐 [12, 17]，已用 [10, 20]
-        assert _recommended_time_conflicts(12.0, 5.0, [(10.0, 20.0)]) is True
+        # 推荐 [5, 10]，已用 [0, 20]
+        assert _recommended_time_conflicts(5.0, 5.0, [(0.0, 20.0)]) is True
 
-    def test_conflict_adjacent_not_conflict(self):
+    def test_conflict_exact_boundary_no_overlap(self):
         from app.api.routes.templates_editor.clips import _recommended_time_conflicts
 
-        # 推荐 [5, 10]，已用 [10, 15] —— 边界相邻不算冲突
-        assert _recommended_time_conflicts(5.0, 5.0, [(10.0, 15.0)]) is False
+        # 推荐 [10, 15]，已用 [0, 10] — 边界相接不算冲突
+        assert _recommended_time_conflicts(10.0, 5.0, [(0.0, 10.0)]) is False
 
-    def test_conflict_with_multiple_segments(self):
+    def test_conflict_multiple_used(self):
         from app.api.routes.templates_editor.clips import _recommended_time_conflicts
 
-        # 推荐 [12, 17]，已用 [0, 5] 和 [10, 20]
-        assert _recommended_time_conflicts(12.0, 5.0, [(0.0, 5.0), (10.0, 20.0)]) is True
-
-    def test_no_conflict_between_segments(self):
-        from app.api.routes.templates_editor.clips import _recommended_time_conflicts
-
-        # 推荐 [6, 11]，已用 [0, 5] 和 [12, 20]
-        assert _recommended_time_conflicts(6.0, 5.0, [(0.0, 5.0), (12.0, 20.0)]) is False
+        used = [(0.0, 5.0), (10.0, 15.0), (20.0, 25.0)]
+        # 推荐 [6, 11] 与 [10, 15] 冲突
+        assert _recommended_time_conflicts(6.0, 5.0, used) is True
+        # 推荐 [15, 20] 不冲突
+        assert _recommended_time_conflicts(15.0, 5.0, used) is False
 
 
-# ── _get_mediakit_recommendations 单元测试 ───────────────────────────────────
+# ── _get_mediakit_recommendations 单元测试 ──────────────────────────────────
 
 
 class TestGetMediakitRecommendations:
-    """测试 MediaKit 推荐获取与解析。"""
+    """测试 MediaKit 推荐结果解析和降级。"""
 
     @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_returns_empty_when_not_available(self, mock_storage, mock_client_fn):
-        """MediaKit 不可用时返回空字典。"""
+    def test_returns_parsed_recommendations(self, mock_storage, mock_client_fn):
+        from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
+
+        mock_client = MagicMock()
+        mock_client.is_available = True
+        mock_client.analyze_videos.return_value = [
+            '[{"asset_id": "a1", "recommended_start_time": 12.5, "reason": "动作场景"}]'
+        ]
+        mock_client_fn.return_value = mock_client
+
+        mock_storage_svc = MagicMock()
+        mock_storage_svc.get_download_url.return_value = "https://example.com/v.mp4"
+        mock_storage.return_value = mock_storage_svc
+
+        asset = MagicMock()
+        asset.storage_key = "v.mp4"
+        asset.mime_type = "video/mp4"
+
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get.return_value = asset
+
+        result = _get_mediakit_recommendations(["a1"], mock_asset_repo)
+        assert result == {"a1": 12.5}
+
+    @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_client_not_available_returns_empty(self, mock_storage, mock_client_fn):
         from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
 
         mock_client = MagicMock()
@@ -104,108 +128,66 @@ class TestGetMediakitRecommendations:
 
     @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_parses_json_response(self, mock_storage, mock_client_fn):
-        """正确解析 JSON 格式的 MediaKit 返回。"""
-        import json
-
+    def test_empty_contents_returns_empty(self, mock_storage, mock_client_fn):
         from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
 
         mock_client = MagicMock()
         mock_client.is_available = True
-        mock_client.analyze_videos.return_value = [
-            json.dumps([{"asset_id": "a1", "recommended_start_time": 12.5, "reason": "画面清晰"}])
-        ]
+        mock_client.analyze_videos.return_value = []
         mock_client_fn.return_value = mock_client
 
-        mock_asset_repo = MagicMock()
-        mock_asset = MagicMock()
-        mock_asset.storage_key = "videos/test.mp4"
-        mock_asset.mime_type = "video/mp4"
-        mock_asset_repo.get.return_value = mock_asset
-
         mock_storage_svc = MagicMock()
-        mock_storage_svc.get_download_url.return_value = "https://example.com/test.mp4"
+        mock_storage_svc.get_download_url.return_value = "https://example.com/v.mp4"
         mock_storage.return_value = mock_storage_svc
 
-        result = _get_mediakit_recommendations(["a1"], mock_asset_repo)
-        assert result == {"a1": 12.5}
-
-    @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
-    @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_parses_regex_fallback(self, mock_storage, mock_client_fn):
-        """JSON 解析失败时通过正则提取 recommended_start_time。"""
-        from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
-
-        mock_client = MagicMock()
-        mock_client.is_available = True
-        mock_client.analyze_videos.return_value = ['根据分析，recommended_start_time": 8.3，画面主体明确']
-        mock_client_fn.return_value = mock_client
-
+        asset = MagicMock()
+        asset.storage_key = "v.mp4"
+        asset.mime_type = "video/mp4"
         mock_asset_repo = MagicMock()
-        mock_asset = MagicMock()
-        mock_asset.storage_key = "videos/test.mp4"
-        mock_asset.mime_type = "video/mp4"
-        mock_asset_repo.get.return_value = mock_asset
-
-        mock_storage_svc = MagicMock()
-        mock_storage_svc.get_download_url.return_value = "https://example.com/test.mp4"
-        mock_storage.return_value = mock_storage_svc
-
-        result = _get_mediakit_recommendations(["a1"], mock_asset_repo)
-        assert result == {"a1": 8.3}
-
-    @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
-    @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_returns_empty_on_unparseable(self, mock_storage, mock_client_fn):
-        """无法解析时返回空字典。"""
-        from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
-
-        mock_client = MagicMock()
-        mock_client.is_available = True
-        mock_client.analyze_videos.return_value = ["这段视频内容丰富，无法确定具体时间"]
-        mock_client_fn.return_value = mock_client
-
-        mock_asset_repo = MagicMock()
-        mock_asset = MagicMock()
-        mock_asset.storage_key = "videos/test.mp4"
-        mock_asset.mime_type = "video/mp4"
-        mock_asset_repo.get.return_value = mock_asset
-
-        mock_storage_svc = MagicMock()
-        mock_storage_svc.get_download_url.return_value = "https://example.com/test.mp4"
-        mock_storage.return_value = mock_storage_svc
+        mock_asset_repo.get.return_value = asset
 
         result = _get_mediakit_recommendations(["a1"], mock_asset_repo)
         assert result == {}
 
     @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_returns_empty_on_none_result(self, mock_storage, mock_client_fn):
-        """MediaKit 返回 None 时返回空字典。"""
+    def test_unparseable_response_returns_empty(self, mock_storage, mock_client_fn):
         from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
 
         mock_client = MagicMock()
         mock_client.is_available = True
-        mock_client.analyze_videos.return_value = None
+        mock_client.analyze_videos.return_value = ["这是一段自然语言描述，没有JSON"]
         mock_client_fn.return_value = mock_client
 
-        mock_asset_repo = MagicMock()
-        mock_asset = MagicMock()
-        mock_asset.storage_key = "videos/test.mp4"
-        mock_asset.mime_type = "video/mp4"
-        mock_asset_repo.get.return_value = mock_asset
-
         mock_storage_svc = MagicMock()
-        mock_storage_svc.get_download_url.return_value = "https://example.com/test.mp4"
+        mock_storage_svc.get_download_url.return_value = "https://example.com/v.mp4"
         mock_storage.return_value = mock_storage_svc
 
+        asset = MagicMock()
+        asset.storage_key = "v.mp4"
+        asset.mime_type = "video/mp4"
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get.return_value = asset
+
         result = _get_mediakit_recommendations(["a1"], mock_asset_repo)
         assert result == {}
 
+    @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_storage_failure_returns_empty(self, mock_storage, mock_client_fn):
+        from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
+
+        mock_client = MagicMock()
+        mock_client.is_available = True
+        mock_client_fn.return_value = mock_client
+        mock_storage.side_effect = RuntimeError("storage unavailable")
+
+        result = _get_mediakit_recommendations(["a1"], MagicMock())
+        assert result == {}
+
     @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
     def test_exception_returns_empty(self, mock_storage, mock_client_fn):
-        """异常时返回空字典（优雅降级）。"""
         from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
 
         mock_client_fn.side_effect = RuntimeError("unexpected error")
@@ -216,7 +198,6 @@ class TestGetMediakitRecommendations:
     @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
     def test_skips_non_video_assets(self, mock_storage, mock_client_fn):
-        """非视频素材被跳过，不发送给 MediaKit。"""
         from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
 
         mock_client = MagicMock()
@@ -231,10 +212,90 @@ class TestGetMediakitRecommendations:
 
         result = _get_mediakit_recommendations(["a1"], mock_asset_repo)
         assert result == {}
-        # analyze_videos should not be called since no valid video URLs
         mock_client.analyze_videos.assert_not_called()
 
 
+# ── _get_template_segments 单元测试 ─────────────────────────────────────────
+
+
+class TestGetTemplateSegments:
+    """测试模板片段配置查询。"""
+
+    def test_returns_segments_from_new_template_system(self):
+        """新模板系统（clip_configs）有数据时优先使用。"""
+        from app.api.routes.templates_editor.clips import _get_template_segments
+
+        mock_tpl_svc = MagicMock()
+        cc1 = MagicMock()
+        cc1.order = 0
+        cc1.min_duration = 3.0
+        cc1.max_duration = 5.0
+        cc2 = MagicMock()
+        cc2.order = 1
+        cc2.min_duration = 4.0
+        cc2.max_duration = 8.0
+        mock_tpl_svc.list_clip_configs.return_value = [cc2, cc1]  # 乱序返回
+
+        result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock())
+        assert len(result) == 2
+        assert result[0] == (0, 3.0, 5.0)
+        assert result[1] == (1, 4.0, 8.0)
+
+    def test_falls_back_to_old_template_segments(self):
+        """新模板系统无数据时回退到旧系统。"""
+        from app.api.routes.templates_editor.clips import _get_template_segments
+
+        mock_tpl_svc = MagicMock()
+        mock_tpl_svc.list_clip_configs.return_value = []
+
+        with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo:
+            mock_repo = MagicMock()
+            seg1 = MagicMock()
+            seg1.segment_order = 0
+            seg1.duration_min = 2.0
+            seg1.duration_max = 4.0
+            mock_repo.list_segments.return_value = [seg1]
+            MockRepo.return_value = mock_repo
+
+            result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock())
+            assert len(result) == 1
+            assert result[0] == (0, 2.0, 4.0)
+
+    def test_returns_empty_when_no_segments(self):
+        """两套系统都没有片段配置时返回空列表。"""
+        from app.api.routes.templates_editor.clips import _get_template_segments
+
+        mock_tpl_svc = MagicMock()
+        mock_tpl_svc.list_clip_configs.return_value = []
+
+        with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo:
+            mock_repo = MagicMock()
+            mock_repo.list_segments.return_value = []
+            MockRepo.return_value = mock_repo
+
+            result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock())
+            assert result == []
+
+    def test_new_system_exception_falls_back(self):
+        """新模板系统异常时回退到旧系统。"""
+        from app.api.routes.templates_editor.clips import _get_template_segments
+
+        mock_tpl_svc = MagicMock()
+        mock_tpl_svc.list_clip_configs.side_effect = RuntimeError("db error")
+
+        with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo:
+            mock_repo = MagicMock()
+            seg = MagicMock()
+            seg.segment_order = 0
+            seg.duration_min = 1.0
+            seg.duration_max = 3.0
+            mock_repo.list_segments.return_value = [seg]
+            MockRepo.return_value = mock_repo
+
+            result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock())
+            assert len(result) == 1
+
+
 # ── from-assets 端点集成测试 ────────────────────────────────────────────────
 
 
@@ -247,59 +308,307 @@ def _make_auth_user():
     return auth
 
 
-def _make_mock_clip(clip_id, order, duration, start_time=0.0, asset_id=""):
-    clip = MagicMock()
-    clip.id = clip_id
-    clip.plan_id = "plan-test"
-    clip.clip_type = "main"
-    clip.order = order
-    clip.duration = duration
-    clip.start_time = start_time
-    clip.text_content = ""
-    clip.transition_effect = "cut"
-    clip.transition_duration = 0.0
-    clip.playback_speed = 1.0
-    clip.config = {}
-    clip.asset_id = asset_id
-    clip.status = "pending"
-    clip.template_clip_config_id = ""
-    clip.created_at = None
-    clip.updated_at = None
-    return clip
+def _make_clip_config(order, min_dur, max_dur):
+    cc = MagicMock()
+    cc.order = order
+    cc.min_duration = min_dur
+    cc.max_duration = max_dur
+    return cc
 
 
-def _make_mock_asset(asset_id, duration):
-    asset = MagicMock()
-    asset.id = asset_id
-    asset.duration = duration
-    return asset
-
-
-def _create_clips(plan_id, clip_type, order, duration=0.0, start_time=0.0, asset_id="", **kw):
-    return _make_mock_clip(
-        clip_id=f"clip-{order}",
-        order=order,
-        duration=duration,
-        start_time=start_time,
-        asset_id=asset_id,
-    )
-
-
-def _make_plan_svc(existing_clips=None):
+def _make_plan_svc(replace_return_count=None):
     svc = MagicMock()
     svc.get_plan_or_raise = MagicMock()
-    svc.create_clip = MagicMock(side_effect=_create_clips)
-    svc.list_clips = MagicMock(return_value=existing_clips or [])
+    if replace_return_count is not None:
+        svc.replace_all_clips_transactional = MagicMock(return_value=replace_return_count)
+    else:
+        svc.replace_all_clips_transactional = MagicMock(return_value=0)
     return svc
 
 
-class TestMediakitIntegrationInFromAssets:
-    """测试 from-assets 端点中 MediaKit 推荐的集成使用。"""
+def _make_tpl_svc_with_segments(segments):
+    """segments: list of (order, min_dur, max_dur)"""
+    svc = MagicMock()
+    clip_configs = [_make_clip_config(o, mn, mx) for o, mn, mx in segments]
+    svc.list_clip_configs.return_value = clip_configs
+    return svc
+
+
+def _make_rich_asset(asset_id, duration, storage_key="v.mp4", mime="video/mp4"):
+    asset = MagicMock()
+    asset.id = asset_id
+    asset.duration = duration
+    asset.storage_key = storage_key
+    asset.mime_type = mime
+    return asset
+
+
+def _get_clips_data(mock_plan_svc):
+    """从 replace_all_clips_transactional 调用中提取 clips_data。"""
+    call_args = mock_plan_svc.replace_all_clips_transactional.call_args
+    if len(call_args.args) >= 2:
+        return call_args.args[1]
+    return call_args.kwargs.get("clips_data", [])
+
+
+class TestFromAssetsByTemplateSegments:
+    """测试 from-assets 按模板 segment 创建片段（V2 事务性替换）。"""
+
+    def test_creates_clips_matching_segment_count(self):
+        """片段数量 = segment 数量，忽略 required_clips_count。"""
+        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        segments = [(0, 3.0, 5.0), (1, 4.0, 8.0), (2, 2.0, 6.0), (3, 5.0, 10.0)]
+        mock_tpl_svc = _make_tpl_svc_with_segments(segments)
+        mock_plan_svc = _make_plan_svc(replace_return_count=4)
+
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0)
+
+        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=2)
+        result = create_clips_from_assets_editor(
+            template_id="tmpl-1",
+            body=body,
+            plan_id="plan-1",
+            services=(mock_tpl_svc, mock_plan_svc),
+            asset_repo=mock_asset_repo,
+            db=MagicMock(),
+            current_user=_make_auth_user(),
+        )
+
+        assert result.created_count == 4
+        clips_data = _get_clips_data(mock_plan_svc)
+        assert len(clips_data) == 4
+
+    def test_uses_transactional_replace(self):
+        """使用 replace_all_clips_transactional 而不是分别 delete + create。"""
+        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        segments = [(0, 3.0, 5.0), (1, 4.0, 8.0)]
+        mock_tpl_svc = _make_tpl_svc_with_segments(segments)
+        mock_plan_svc = _make_plan_svc(replace_return_count=2)
+
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0)
+
+        body = ClipsFromAssetsRequest(asset_ids=["a1"])
+        create_clips_from_assets_editor(
+            template_id="tmpl-1",
+            body=body,
+            plan_id="plan-1",
+            services=(mock_tpl_svc, mock_plan_svc),
+            asset_repo=mock_asset_repo,
+            db=MagicMock(),
+            current_user=_make_auth_user(),
+        )
+
+        mock_plan_svc.replace_all_clips_transactional.assert_called_once()
+        assert not mock_plan_svc.delete_all_clips.called
+        assert not mock_plan_svc.create_clip.called
+
+    def test_duration_within_segment_range(self):
+        """每个片段时长在 segment 的 min~max 范围内。"""
+        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        segments = [(0, 3.0, 5.0), (1, 4.0, 8.0)]
+        mock_tpl_svc = _make_tpl_svc_with_segments(segments)
+        mock_plan_svc = _make_plan_svc(replace_return_count=2)
+
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0)
+
+        body = ClipsFromAssetsRequest(asset_ids=["a1"])
+        create_clips_from_assets_editor(
+            template_id="tmpl-1",
+            body=body,
+            plan_id="plan-1",
+            services=(mock_tpl_svc, mock_plan_svc),
+            asset_repo=mock_asset_repo,
+            db=MagicMock(),
+            current_user=_make_auth_user(),
+        )
+
+        clips_data = _get_clips_data(mock_plan_svc)
+        assert 3.0 <= clips_data[0]["duration"] <= 5.0
+        assert 4.0 <= clips_data[1]["duration"] <= 8.0
+
+    def test_assets_round_robin_assignment(self):
+        """素材按片段顺序轮询分配。"""
+        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        segments = [(0, 3.0, 5.0), (1, 3.0, 5.0), (2, 3.0, 5.0), (3, 3.0, 5.0)]
+        mock_tpl_svc = _make_tpl_svc_with_segments(segments)
+        mock_plan_svc = _make_plan_svc(replace_return_count=4)
+
+        mock_asset_repo = MagicMock()
+
+        def get_asset(aid):
+            return _make_rich_asset(aid, 60.0)
+
+        mock_asset_repo.get.side_effect = get_asset
+
+        body = ClipsFromAssetsRequest(asset_ids=["a1", "a2"])
+        create_clips_from_assets_editor(
+            template_id="tmpl-1",
+            body=body,
+            plan_id="plan-1",
+            services=(mock_tpl_svc, mock_plan_svc),
+            asset_repo=mock_asset_repo,
+            db=MagicMock(),
+            current_user=_make_auth_user(),
+        )
+
+        clips_data = _get_clips_data(mock_plan_svc)
+        asset_ids = [c["asset_id"] for c in clips_data]
+        assert asset_ids == ["a1", "a2", "a1", "a2"]
+
+    def test_orders_start_from_zero(self):
+        """片段 order 从 0 开始递增。"""
+        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        segments = [(0, 3.0, 5.0), (1, 3.0, 5.0), (2, 3.0, 5.0)]
+        mock_tpl_svc = _make_tpl_svc_with_segments(segments)
+        mock_plan_svc = _make_plan_svc(replace_return_count=3)
+
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0)
+
+        body = ClipsFromAssetsRequest(asset_ids=["a1"])
+        create_clips_from_assets_editor(
+            template_id="tmpl-1",
+            body=body,
+            plan_id="plan-1",
+            services=(mock_tpl_svc, mock_plan_svc),
+            asset_repo=mock_asset_repo,
+            db=MagicMock(),
+            current_user=_make_auth_user(),
+        )
+
+        clips_data = _get_clips_data(mock_plan_svc)
+        orders = [c["order"] for c in clips_data]
+        assert orders == [0, 1, 2]
+
+    def test_no_segments_raises_400(self):
+        """模板没有 segment 配置时返回 400。"""
+        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+        from fastapi import HTTPException
+
+        mock_tpl_svc = MagicMock()
+        mock_tpl_svc.list_clip_configs.return_value = []
+        mock_plan_svc = _make_plan_svc()
+
+        with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo:
+            mock_repo = MagicMock()
+            mock_repo.list_segments.return_value = []
+            MockRepo.return_value = mock_repo
+
+            body = ClipsFromAssetsRequest(asset_ids=["a1"])
+            with pytest.raises(HTTPException) as exc_info:
+                create_clips_from_assets_editor(
+                    template_id="tmpl-1",
+                    body=body,
+                    plan_id="plan-1",
+                    services=(mock_tpl_svc, mock_plan_svc),
+                    asset_repo=MagicMock(),
+                    db=MagicMock(),
+                    current_user=_make_auth_user(),
+                )
+            assert exc_info.value.status_code == 400
+
+        mock_plan_svc.replace_all_clips_transactional.assert_not_called()
+
+    def test_duration_capped_by_asset_duration(self):
+        """素材时长不足时 clip duration 被缩短。"""
+        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        segments = [(0, 10.0, 20.0)]
+        mock_tpl_svc = _make_tpl_svc_with_segments(segments)
+        mock_plan_svc = _make_plan_svc(replace_return_count=1)
+
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get.return_value = _make_rich_asset("a1", 5.0)
+
+        body = ClipsFromAssetsRequest(asset_ids=["a1"])
+        create_clips_from_assets_editor(
+            template_id="tmpl-1",
+            body=body,
+            plan_id="plan-1",
+            services=(mock_tpl_svc, mock_plan_svc),
+            asset_repo=mock_asset_repo,
+            db=MagicMock(),
+            current_user=_make_auth_user(),
+        )
+
+        clips_data = _get_clips_data(mock_plan_svc)
+        assert clips_data[0]["duration"] <= 5.0
+
+    def test_zero_duration_asset_raises_400(self):
+        """素材时长为 0 时抛出 400。"""
+        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+        from fastapi import HTTPException
+
+        segments = [(0, 3.0, 5.0)]
+        mock_tpl_svc = _make_tpl_svc_with_segments(segments)
+        mock_plan_svc = _make_plan_svc()
+
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get.return_value = _make_rich_asset("bad", 0.0)
+
+        body = ClipsFromAssetsRequest(asset_ids=["bad"])
+        with pytest.raises(HTTPException) as exc_info:
+            create_clips_from_assets_editor(
+                template_id="tmpl-1",
+                body=body,
+                plan_id="plan-1",
+                services=(mock_tpl_svc, mock_plan_svc),
+                asset_repo=mock_asset_repo,
+                db=MagicMock(),
+                current_user=_make_auth_user(),
+            )
+
+        assert exc_info.value.status_code == 400
+        mock_plan_svc.replace_all_clips_transactional.assert_not_called()
+
+    def test_empty_asset_ids_raises_400(self):
+        """asset_ids 为空列表时返回 400（defense-in-depth，schema 层也有 min_length=1）。"""
+        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
+        from fastapi import HTTPException
+
+        segments = [(0, 3.0, 5.0)]
+        mock_tpl_svc = _make_tpl_svc_with_segments(segments)
+        mock_plan_svc = _make_plan_svc()
+
+        # 用 MagicMock 模拟 body，绕过 Pydantic schema 的 min_length 校验
+        mock_body = MagicMock()
+        mock_body.asset_ids = []
+        mock_body.required_clips_count = None
+
+        with pytest.raises(HTTPException) as exc_info:
+            create_clips_from_assets_editor(
+                template_id="tmpl-1",
+                body=mock_body,
+                plan_id="plan-1",
+                services=(mock_tpl_svc, mock_plan_svc),
+                asset_repo=MagicMock(),
+                db=MagicMock(),
+                current_user=_make_auth_user(),
+            )
+        assert exc_info.value.status_code == 400
+        assert "素材" in exc_info.value.detail
+        mock_plan_svc.replace_all_clips_transactional.assert_not_called()
 
     @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_uses_mediakit_recommendation(self, mock_storage, mock_client_fn):
-        """MediaKit 推荐时间被优先使用。"""
+    def test_mediakit_first_clip_uses_recommendation(self, mock_storage, mock_client_fn):
+        """MediaKit 推荐时间用于每个素材的第一个片段。"""
         from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
         from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
 
@@ -311,186 +620,71 @@ class TestMediakitIntegrationInFromAssets:
         mock_client_fn.return_value = mock_client
 
         mock_storage_svc = MagicMock()
-        mock_storage_svc.get_download_url.return_value = "https://example.com/test.mp4"
+        mock_storage_svc.get_download_url.return_value = "https://example.com/v.mp4"
         mock_storage.return_value = mock_storage_svc
 
-        mock_plan_svc = _make_plan_svc()
+        segments = [(0, 3.0, 5.0), (1, 3.0, 5.0)]
+        mock_tpl_svc = _make_tpl_svc_with_segments(segments)
+        mock_plan_svc = _make_plan_svc(replace_return_count=2)
+
         mock_asset_repo = MagicMock()
+        mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0)
 
-        # asset needs storage_key and mime_type for MediaKit, plus duration for clip creation
-        asset_for_mediakit = MagicMock()
-        asset_for_mediakit.storage_key = "videos/test.mp4"
-        asset_for_mediakit.mime_type = "video/mp4"
-        asset_for_mediakit.duration = 60.0
-
-        def asset_get_side_effect(aid):
-            return _make_mock_asset(aid, 60.0) if aid else None
-
-        mock_asset_repo.get = MagicMock(side_effect=asset_get_side_effect)
-
-        # We need to make the asset have storage_key and mime_type for the mediakit function
-        # The mock_asset from _make_mock_asset doesn't have these, so let's use a richer mock
-        rich_asset = MagicMock()
-        rich_asset.id = "a1"
-        rich_asset.duration = 60.0
-        rich_asset.storage_key = "videos/test.mp4"
-        rich_asset.mime_type = "video/mp4"
-        mock_asset_repo.get.return_value = rich_asset
-
-        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1)
+        body = ClipsFromAssetsRequest(asset_ids=["a1"])
         create_clips_from_assets_editor(
             template_id="tmpl-1",
             body=body,
-            plan_id="plan-mk1",
-            services=(MagicMock(), mock_plan_svc),
+            plan_id="plan-1",
+            services=(mock_tpl_svc, mock_plan_svc),
             asset_repo=mock_asset_repo,
+            db=MagicMock(),
             current_user=_make_auth_user(),
         )
 
-        create_calls = mock_plan_svc.create_clip.call_args_list
-        assert len(create_calls) == 1
-        start_time = create_calls[0].kwargs.get("start_time") or create_calls[0][1].get("start_time")
-        assert start_time == 15.0
+        clips_data = _get_clips_data(mock_plan_svc)
+        # 第一个片段应使用推荐时间 15.0
+        assert clips_data[0]["start_time"] == 15.0
+        # 第二个片段（同一素材）不应使用推荐时间
+        assert clips_data[1]["start_time"] != 15.0
 
-    @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
-    @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_fallback_to_random_on_conflict(self, mock_storage, mock_client_fn):
-        """推荐时间与已有片段冲突时，降级为随机选择。"""
-        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
-        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
 
-        mock_client = MagicMock()
-        mock_client.is_available = True
-        # 推荐 2.0s，但已有片段占用了 [0, 10]
-        mock_client.analyze_videos.return_value = [
-            '[{"asset_id": "a1", "recommended_start_time": 2.0, "reason": "test"}]'
-        ]
-        mock_client_fn.return_value = mock_client
+# ── _safe_segment_duration 单元测试 ─────────────────────────────────────────
 
-        mock_storage_svc = MagicMock()
-        mock_storage_svc.get_download_url.return_value = "https://example.com/test.mp4"
-        mock_storage.return_value = mock_storage_svc
 
-        # 已有片段占用 [0, 10]
-        existing = [_make_mock_clip("c1", order=0, duration=10.0, start_time=0.0, asset_id="a1")]
-        mock_plan_svc = _make_plan_svc(existing_clips=existing)
+class TestSafeSegmentDuration:
+    """测试片段时长安全转换。"""
 
-        rich_asset = MagicMock()
-        rich_asset.id = "a1"
-        rich_asset.duration = 60.0
-        rich_asset.storage_key = "videos/test.mp4"
-        rich_asset.mime_type = "video/mp4"
+    def test_normal_float(self):
+        from app.api.routes.templates_editor.clips import _safe_segment_duration
 
-        mock_asset_repo = MagicMock()
-        mock_asset_repo.get.return_value = rich_asset
+        assert _safe_segment_duration(3.5, 1.0) == 3.5
 
-        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1)
-        create_clips_from_assets_editor(
-            template_id="tmpl-1",
-            body=body,
-            plan_id="plan-mk2",
-            services=(MagicMock(), mock_plan_svc),
-            asset_repo=mock_asset_repo,
-            current_user=_make_auth_user(),
-        )
+    def test_none_returns_default(self):
+        from app.api.routes.templates_editor.clips import _safe_segment_duration
 
-        create_calls = mock_plan_svc.create_clip.call_args_list
-        assert len(create_calls) == 1
-        start_time = create_calls[0].kwargs.get("start_time") or create_calls[0][1].get("start_time")
-        # 推荐时间 2.0 与 [0, 10] 冲突，应降级为随机，不应等于 2.0
-        # 随机起始应在 [10, 55] 范围内（避开 [0,10]，5s clip 在 60s 素材中）
-        assert start_time is not None
-        assert start_time != 2.0
+        assert _safe_segment_duration(None, 5.0) == 5.0
 
-    @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
-    @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_fallback_when_mediakit_unavailable(self, mock_storage, mock_client_fn):
-        """MediaKit 不可用时降级为随机选择，功能正常。"""
-        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
-        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+    def test_string_number(self):
+        from app.api.routes.templates_editor.clips import _safe_segment_duration
 
-        mock_client = MagicMock()
-        mock_client.is_available = False
-        mock_client_fn.return_value = mock_client
+        assert _safe_segment_duration("4.2", 1.0) == 4.2
 
-        mock_plan_svc = _make_plan_svc()
-        mock_asset_repo = MagicMock()
-        mock_asset_repo.get.return_value = _make_mock_asset("a1", 30.0)
+    def test_invalid_string_returns_default(self):
+        from app.api.routes.templates_editor.clips import _safe_segment_duration
 
-        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1)
-        result = create_clips_from_assets_editor(
-            template_id="tmpl-1",
-            body=body,
-            plan_id="plan-mk3",
-            services=(MagicMock(), mock_plan_svc),
-            asset_repo=mock_asset_repo,
-            current_user=_make_auth_user(),
-        )
+        assert _safe_segment_duration("abc", 5.0) == 5.0
 
-        assert result.created_count == 1
-        # analyze_videos 不应被调用
-        mock_client.analyze_videos.assert_not_called()
+    def test_negative_returns_default(self):
+        from app.api.routes.templates_editor.clips import _safe_segment_duration
 
-    @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
-    @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_fallback_on_unparseable_response(self, mock_storage, mock_client_fn):
-        """MediaKit 返回不可解析内容时降级为随机选择。"""
-        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
-        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+        assert _safe_segment_duration(-1.0, 5.0) == 5.0
 
-        mock_client = MagicMock()
-        mock_client.is_available = True
-        mock_client.analyze_videos.return_value = ["这段视频内容很精彩，有很多好看的画面"]
-        mock_client_fn.return_value = mock_client
+    def test_zero_returns_default(self):
+        from app.api.routes.templates_editor.clips import _safe_segment_duration
 
-        mock_storage_svc = MagicMock()
-        mock_storage_svc.get_download_url.return_value = "https://example.com/test.mp4"
-        mock_storage.return_value = mock_storage_svc
+        assert _safe_segment_duration(0, 5.0) == 5.0
 
-        mock_plan_svc = _make_plan_svc()
-        rich_asset = MagicMock()
-        rich_asset.id = "a1"
-        rich_asset.duration = 30.0
-        rich_asset.storage_key = "videos/test.mp4"
-        rich_asset.mime_type = "video/mp4"
-        mock_asset_repo = MagicMock()
-        mock_asset_repo.get.return_value = rich_asset
+    def test_integer_value(self):
+        from app.api.routes.templates_editor.clips import _safe_segment_duration
 
-        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1)
-        result = create_clips_from_assets_editor(
-            template_id="tmpl-1",
-            body=body,
-            plan_id="plan-mk4",
-            services=(MagicMock(), mock_plan_svc),
-            asset_repo=mock_asset_repo,
-            current_user=_make_auth_user(),
-        )
-
-        # 功能正常，降级为随机
-        assert result.created_count == 1
-
-    @patch("app.api.routes.templates_editor.clips.get_mediakit_client")
-    @patch("app.api.routes.templates_editor.clips.get_storage_service")
-    def test_fallback_on_mediakit_exception(self, mock_storage, mock_client_fn):
-        """MediaKit 抛异常时优雅降级，不影响片段创建。"""
-        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
-        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
-
-        mock_client_fn.side_effect = RuntimeError("MediaKit connection failed")
-
-        mock_plan_svc = _make_plan_svc()
-        mock_asset_repo = MagicMock()
-        mock_asset_repo.get.return_value = _make_mock_asset("a1", 30.0)
-
-        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1)
-        result = create_clips_from_assets_editor(
-            template_id="tmpl-1",
-            body=body,
-            plan_id="plan-mk5",
-            services=(MagicMock(), mock_plan_svc),
-            asset_repo=mock_asset_repo,
-            current_user=_make_auth_user(),
-        )
-
-        # 功能正常，降级为随机
-        assert result.created_count == 1
+        assert _safe_segment_duration(10, 1.0) == 10.0