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

@@ -15,7 +15,9 @@
 
 from __future__ import annotations
 
+import json
 import logging
+import re
 
 from app.auth import AuthenticatedUser, get_current_user
 from app.core.storage import get_storage_service
@@ -26,6 +28,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query, status
 
 from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
 from packages.domain.plan_generator_utils import _calc_random_start_time
+from packages.shared.mediakit_client import get_mediakit_client
 
 from .dependencies import get_draft_plan_id, get_editor_services
 from .schemas import (
@@ -351,6 +354,134 @@ def batch_delete_editor_clips(
     return ClipBatchDeleteResponse(deleted_count=deleted, plan_id=plan_id)
 
 
+
+def _recommended_time_conflicts(
+    start: float,
+    duration: float,
+    used: list[tuple[float, float]],
+) -> bool:
+    """检查推荐起始时间是否与已使用时间段冲突."""
+    end = start + duration
+    for used_start, used_end in used:
+        if start < used_end and end > used_start:
+            return True
+    return False
+
+
+def _get_mediakit_recommendations(
+    asset_ids: list[str],
+    asset_repo,
+) -> dict[str, float]:
+    """调用 MediaKit 视频理解，获取智能选片推荐起始时间.
+
+    尝试让 MediaKit 分析视频内容，返回每个素材的推荐起始时间。
+    任何异常都优雅降级，返回空字典（调用方降级到随机选择）。
+    """
+    try:
+        client = get_mediakit_client()
+        if not client.is_available:
+            logger.info("MediaKit 未配置，使用随机起始时间")
+            return {}
+
+        storage = get_storage_service()
+
+        video_urls: list[str] = []
+        valid_asset_ids: list[str] = []
+        for asset_id in asset_ids[:10]:
+            asset = asset_repo.get(asset_id)
+            if not asset or not getattr(asset, "storage_key", None):
+                continue
+            mime = getattr(asset, "mime_type", "")
+            if not mime.startswith("video/"):
+                continue
+            try:
+                url = storage.get_download_url(asset.storage_key)
+                if url:
+                    video_urls.append(url)
+                    valid_asset_ids.append(asset_id)
+            except Exception as e:
+                logger.warning("获取素材URL失败: asset_id=%s error=%s", asset_id, e)
+
+        if not video_urls:
+            return {}
+
+        prompt = (
+            "请分析每段视频，找出最精彩的5秒片段应该从哪个时间点开始。"
+            "考虑因素：画面清晰度、主体是否明确、是否有明显的动作或场景变化。"
+            '请严格以JSON数组格式返回，不要包含其他文字：'
+            '[{"asset_id": "素材ID", "recommended_start_time": 12.5, "reason": "原因"}]'
+        )
+
+        contents = client.analyze_videos(
+            video_urls=video_urls,
+            prompt=prompt,
+            level="Economy",
+            poll_interval=2.0,
+            max_poll_attempts=15,
+        )
+
+        if not contents:
+            logger.info("MediaKit 分析无结果，降级为随机选择")
+            return {}
+
+        # 按索引映射结果：contents[i] 对应 valid_asset_ids[i]
+        recommendations: dict[str, float] = {}
+        for idx, content_text in enumerate(contents):
+            if idx >= len(valid_asset_ids):
+                break
+            asset_id = valid_asset_ids[idx]
+            if not content_text:
+                continue
+
+            # 尝试从文本中提取 JSON
+            parsed = False
+            # 尝试直接解析
+            try:
+                data = json.loads(content_text.strip())
+                if isinstance(data, list) and data:
+                    for item in data:
+                        if isinstance(item, dict) and "recommended_start_time" in item:
+                            recommendations[asset_id] = float(item["recommended_start_time"])
+                            parsed = True
+                            break
+            except (json.JSONDecodeError, ValueError, TypeError):
+                pass
+
+            # 尝试从 markdown 代码块中提取 JSON
+            if not parsed:
+                json_match = re.search(r"\[\s*(\{.*?\})\s*\]", content_text, re.DOTALL)
+                if json_match:
+                    try:
+                        item = json.loads(json_match.group(1))
+                        if isinstance(item, dict) and "recommended_start_time" in item:
+                            recommendations[asset_id] = float(item["recommended_start_time"])
+                            parsed = True
+                    except (json.JSONDecodeError, ValueError, TypeError):
+                        pass
+
+            # 尝试正则提取
+            if not parsed:
+                time_match = re.search(
+                    r'recommended_start_time["\s:]+([\d.]+)', content_text
+                )
+                if time_match:
+                    try:
+                        recommendations[asset_id] = float(time_match.group(1))
+                    except (ValueError, TypeError):
+                        pass
+
+        if recommendations:
+            logger.info("MediaKit 智能选片推荐: %s", recommendations)
+        else:
+            logger.info("MediaKit 结果解析失败，降级为随机选择")
+
+        return recommendations
+
+    except Exception as e:
+        logger.warning("MediaKit 智能选片异常，降级为随机选择: %s", e)
+        return {}
+
+
 @router.post("/clips/from-assets", response_model=ClipsFromAssetsResponse)
 def create_clips_from_assets_editor(
     template_id: str,
@@ -394,6 +525,9 @@ def create_clips_from_assets_editor(
                 (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):
@@ -407,8 +541,30 @@ def create_clips_from_assets_editor(
         else:
             clip_duration = _DEFAULT_EDITOR_CLIP_DURATION
 
-        # 计算随机 start_time，避开已使用的时间段
-        start_time = _calc_random_start_time(asset_id, clip_duration, asset_durations, used_segments)
+        # 优先使用 MediaKit 推荐的起始时间，冲突时降级为随机
+        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"))
+            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,
+            )
+        else:
+            if recommended_start is not None:
+                logger.info(
+                    "MediaKit推荐时间冲突或越界，降级为随机: asset_id=%s recommended=%.2f",
+                    asset_id, recommended_start,
+                )
+            # 随机选择起始时间，避开已使用的时间段
+            start_time = _calc_random_start_time(
+                asset_id, clip_duration, asset_durations, used_segments
+            )
         if start_time is None:
             # 素材时长信息缺失，无法计算随机起始时间
             raise HTTPException(

tests/unit/test_mediakit_smart_clips.py

@@ -0,0 +1,496 @@
+"""测试 MediaKit 智能选片集成。
+
+覆盖：
+- _recommended_time_conflicts 冲突检测
+- _get_mediakit_recommendations 解析与降级
+- from-assets 端点：推荐时间优先使用
+- from-assets 端点：推荐时间冲突时降级随机
+- from-assets 端点：MediaKit 不可用时降级随机
+- from-assets 端点：MediaKit 返回不可解析内容时降级随机
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
+os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+import pytest
+
+# ── _recommended_time_conflicts 单元测试 ─────────────────────────────────────
+
+
+class TestRecommendedTimeConflicts:
+    """测试推荐时间与已使用时间段的冲突检测。"""
+
+    def test_no_conflict_when_empty(self):
+        from app.api.routes.templates_editor.clips import _recommended_time_conflicts
+
+        assert _recommended_time_conflicts(5.0, 5.0, []) is False
+
+    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
+
+    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
+
+    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
+
+    def test_conflict_adjacent_not_conflict(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
+
+    def test_conflict_with_multiple_segments(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
+
+
+# ── _get_mediakit_recommendations 单元测试 ───────────────────────────────────
+
+
+class TestGetMediakitRecommendations:
+    """测试 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 不可用时返回空字典。"""
+        from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
+
+        mock_client = MagicMock()
+        mock_client.is_available = False
+        mock_client_fn.return_value = mock_client
+
+        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_parses_json_response(self, mock_storage, mock_client_fn):
+        """正确解析 JSON 格式的 MediaKit 返回。"""
+        import json
+
+        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_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
+
+        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
+
+        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
+
+        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 时返回空字典。"""
+        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_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
+
+        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_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")
+
+        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_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()
+        mock_client.is_available = True
+        mock_client_fn.return_value = mock_client
+
+        mock_asset_repo = MagicMock()
+        mock_asset = MagicMock()
+        mock_asset.storage_key = "images/test.jpg"
+        mock_asset.mime_type = "image/jpeg"
+        mock_asset_repo.get.return_value = mock_asset
+
+        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()
+
+
+# ── from-assets 端点集成测试 ────────────────────────────────────────────────
+
+
+def _make_auth_user():
+    auth = MagicMock()
+    auth.user.id = "user-001"
+    auth.user.email = "test@example.com"
+    auth.user.display_name = "test"
+    auth.user_id = "user-001"
+    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_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):
+    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 [])
+    return svc
+
+
+class TestMediakitIntegrationInFromAssets:
+    """测试 from-assets 端点中 MediaKit 推荐的集成使用。"""
+
+    @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 推荐时间被优先使用。"""
+        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 = [
+            '[{"asset_id": "a1", "recommended_start_time": 15.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
+
+        mock_plan_svc = _make_plan_svc()
+        mock_asset_repo = MagicMock()
+
+        # 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)
+        create_clips_from_assets_editor(
+            template_id="tmpl-1",
+            body=body,
+            plan_id="plan-mk1",
+            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")
+        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