From a2b37e376d0ffc12eec95387d4227e007a7abb27 Mon Sep 17 00:00:00 2001
From: CI Bot <ci-bot@xiaoxiajianji.com>
Date: Thu, 27 Aug 2026 22:31:40 +0800
Subject: [PATCH] feat(from-assets): create clips based on template segments
 instead of required_clips_count

- Query template_segments/clip_configs by template_id, create exactly N clips matching segment count
- Each clip duration randomly chosen between segment duration_min and duration_max (1 decimal)
- Assets round-robin assigned across segments; same asset can produce multiple clips
- Delete all existing clips for the plan before creating new ones (idempotent)
- MediaKit recommendation only applied to the first clip using each asset; subsequent clips use random start times
- MediaKit timeout/poll settings unchanged (60s/2s/15)
- mark_clips_ready called immediately after creation
- Ignore required_clips_count from frontend
- Clips order starts from 0 since old clips are cleared
- Add _get_template_segments helper supporting both new (clip_configs) and old (template_segments) systems
- 28 unit tests covering segment query, clip creation, round-robin, deletion, duration bounds, MediaKit integration
---
 .../app/api/routes/templates_editor/clips.py  | 175 ++--
 tests/unit/test_mediakit_smart_clips.py       | 745 +++++++++++-------
 2 files changed, 580 insertions(+), 340 deletions(-)

diff --git a/apps/api/app/api/routes/templates_editor/clips.py b/apps/api/app/api/routes/templates_editor/clips.py
index bff59a302..a60598b95 100755
--- a/apps/api/app/api/routes/templates_editor/clips.py
+++ b/apps/api/app/api/routes/templates_editor/clips.py
@@ -21,10 +21,17 @@ 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
+import random
+
+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.template_repository import (
+    SQLAlchemyTemplateRepository,
+)
 
 from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
 from packages.domain.plan_generator_utils import _calc_random_start_time
@@ -355,6 +362,59 @@ def batch_delete_editor_clips(
 
 
 
+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:
+            return sorted(
+                [
+                    (
+                        cc.order,
+                        float(cc.min_duration or _DEFAULT_EDITOR_CLIP_DURATION),
+                        float(
+                            cc.max_duration
+                            or cc.min_duration
+                            or _DEFAULT_EDITOR_CLIP_DURATION
+                        ),
+                    )
+                    for cc in clip_configs
+                ],
+                key=lambda x: x[0],
+            )
+    except Exception:
+        logger.debug("新模板系统查询clip_configs失败，回退到旧系统", exc_info=True)
+
+    # 回退到旧模板系统（template_segments表）
+    try:
+        old_repo = SQLAlchemyTemplateRepository(db)
+        segments = old_repo.list_segments(template_id)
+        if segments:
+            return sorted(
+                [
+                    (s.segment_order, float(s.duration_min), float(s.duration_max))
+                    for s in segments
+                ],
+                key=lambda x: x[0],
+            )
+    except Exception:
+        logger.debug("旧模板系统查询segments失败", exc_info=True)
+
+    return []
+
+
 def _recommended_time_conflicts(
     start: float,
     duration: float,
@@ -489,22 +549,38 @@ 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. 先清空该 plan 下所有旧片段，再创建新的（避免重复）
+    5. MediaKit 智能选片：第一个使用某素材的片段用推荐起始时间，后续用随机
+    6. 片段创建后立即 mark_clips_ready
     """
-    _, 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="模板没有片段配置，无法创建片段",
+        )
 
-    # 去重后批量获取素材实际时长，避免重复查询
+    # 2. 先清空旧片段
+    deleted_count = plan_svc.delete_all_clips(plan_id)
+    logger.info(
+        "from-assets 清空旧片段: plan_id=%s deleted=%d",
+        plan_id,
+        deleted_count,
+    )
+
+    # 3. 获取素材实际时长
     unique_asset_ids = list(dict.fromkeys(body.asset_ids))
     asset_durations: dict[str, float] = {}
     for asset_id in unique_asset_ids:
@@ -512,74 +588,83 @@ 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
+    # 4. 获取 MediaKit 智能选片推荐（保持60s timeout + poll 2s + 15次）
+    mediakit_recommendations = _get_mediakit_recommendations(
+        unique_asset_ids, asset_repo
+    )
 
-    # 从已有片段中构建已使用时间段，避免跨任务重复使用同一段素材区域
+    # 5. 跟踪素材使用情况
+    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))
-            )
-
-    # 尝试获取 MediaKit 智能选片推荐
-    mediakit_recommendations = _get_mediakit_recommendations(unique_asset_ids, asset_repo)
 
     clips = []
 
-    for i in range(required_count):
-        # 轮询分配素材：素材不够时同一素材切多个片段
+    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)
+
+        # 在 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 推荐的起始时间，冲突时降级为随机
+        # 确定起始时间
+        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 asset_total > 0
+            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} 时长信息缺失，无法创建片段",
             )
 
-        # 记录已使用时间段（用于去重）
-        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,
+                order=i,
                 duration=clip_duration,
                 start_time=start_time,
                 asset_id=asset_id,
@@ -590,7 +675,7 @@ def create_clips_from_assets_editor(
                 "创建片段失败: plan_id=%s asset_id=%s order=%d error=%s",
                 plan_id,
                 asset_id,
-                next_order + i,
+                i,
                 exc,
                 exc_info=True,
             )
@@ -600,15 +685,15 @@ def create_clips_from_assets_editor(
             ) from exc
 
     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 actual=%d by user=%s",
         template_id,
         plan_id,
-        required_count,
+        len(segments),
         len(clips),
         current_user.user.id,
     )
 
-    # 新创建的片段已分配素材，立即标记为 ready，否则渲染管线找不到就绪片段
+    # 片段已分配素材，立即标记为 ready
     plan_svc.mark_clips_ready(plan_id)
     return ClipsFromAssetsResponse(
         created_count=len(clips),
diff --git a/tests/unit/test_mediakit_smart_clips.py b/tests/unit/test_mediakit_smart_clips.py
index 7eeb94fa5..dc0bfd2a0 100644
--- a/tests/unit/test_mediakit_smart_clips.py
+++ b/tests/unit/test_mediakit_smart_clips.py
@@ -1,12 +1,15 @@
-"""测试 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 __future__ import annotations
@@ -23,6 +26,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
 
 import pytest
 
+
 # ── _recommended_time_conflicts 单元测试 ─────────────────────────────────────
 
 
@@ -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,96 @@ 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 端点集成测试 ────────────────────────────────────────────────
 
 
@@ -268,38 +335,281 @@ def _make_mock_clip(clip_id, order, duration, start_time=0.0, asset_id=""):
     return clip
 
 
-def _make_mock_asset(asset_id, duration):
-    asset = MagicMock()
-    asset.id = asset_id
-    asset.duration = duration
-    return asset
+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 _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():
     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 [])
+    svc.create_clip = MagicMock(
+        side_effect=lambda plan_id, clip_type, order, **kw: _make_mock_clip(
+            clip_id=f"clip-{order}",
+            order=order,
+            duration=kw.get("duration", 5.0),
+            start_time=kw.get("start_time", 0.0),
+            asset_id=kw.get("asset_id", ""),
+        )
+    )
+    svc.delete_all_clips = MagicMock(return_value=0)
+    svc.mark_clips_ready = 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
+
+
+class TestFromAssetsByTemplateSegments:
+    """测试 from-assets 按模板 segment 创建片段。"""
+
+    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()
+
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0)
+
+        # 传了 required_clips_count=2，但应该创建4个片段
+        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
+        assert mock_plan_svc.create_clip.call_count == 4
+
+    def test_deletes_old_clips_first(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, 4.0, 8.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("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.delete_all_clips.assert_called_once_with("plan-1")
+
+    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()
+
+        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(),
+        )
+
+        calls = mock_plan_svc.create_clip.call_args_list
+        durations = [c.kwargs.get("duration", 0) for c in calls]
+        assert 3.0 <= durations[0] <= 5.0
+        assert 4.0 <= durations[1] <= 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()
+
+        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(),
+        )
+
+        calls = mock_plan_svc.create_clip.call_args_list
+        asset_ids = [c.kwargs.get("asset_id", "") for c in calls]
+        # 4 segments, 2 assets: a1, a2, a1, a2
+        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()
+
+        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(),
+        )
+
+        calls = mock_plan_svc.create_clip.call_args_list
+        orders = [c.kwargs.get("order", -1) for c in calls]
+        assert orders == [0, 1, 2]
+
+    def test_no_segments_raises_400(self):
+        """模板没有 segment 配置时返回 400。"""
+        from fastapi import HTTPException
+
+        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        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
+
+    def test_marks_clips_ready_after_creation(self):
+        """片段创建后调用 mark_clips_ready。"""
+        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)]
+        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("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.mark_clips_ready.assert_called_once_with("plan-1")
+
+    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)]  # 模板要求10-20秒
+        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("a1", 5.0)  # 素材只有5秒
+
+        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(),
+        )
+
+        calls = mock_plan_svc.create_clip.call_args_list
+        duration = calls[0].kwargs.get("duration", 0)
+        assert duration <= 5.0
 
     @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 +621,31 @@ 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
 
+        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()
+
         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
-
-    @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
-
-        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)
-
-        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 = MagicMock()
-        mock_asset_repo.get.return_value = rich_asset
-
-        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(),
-        )
-
-        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
-
-    @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
-
-        mock_client = MagicMock()
-        mock_client.is_available = False
-        mock_client_fn.return_value = mock_client
-
-        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-mk3",
-            services=(MagicMock(), mock_plan_svc),
-            asset_repo=mock_asset_repo,
-            current_user=_make_auth_user(),
-        )
-
-        assert result.created_count == 1
-        # analyze_videos 不应被调用
-        mock_client.analyze_videos.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_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
-
-        mock_client = MagicMock()
-        mock_client.is_available = True
-        mock_client.analyze_videos.return_value = ["这段视频内容很精彩，有很多好看的画面"]
-        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.return_value = mock_storage_svc
-
-        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
-
-        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
+        calls = mock_plan_svc.create_clip.call_args_list
+        # 第一个片段应使用推荐时间 15.0
+        start_0 = calls[0].kwargs.get("start_time")
+        assert start_0 == 15.0
+        # 第二个片段（同一素材）不应使用推荐时间
+        start_1 = calls[1].kwargs.get("start_time")
+        assert start_1 != 15.0