From d5032108b54663c492285f2f318e8243e3049368 Mon Sep 17 00:00:00 2001
From: CI Bot <ci-bot@xiaoxiajianji.com>
Date: Thu, 27 Aug 2026 15:52:58 +0800
Subject: [PATCH 1/2] =?UTF-8?q?feat(editor):=20=E6=8E=A5=E5=85=A5MediaKit?=
 =?UTF-8?q?=E6=99=BA=E8=83=BD=E9=80=89=E7=89=87=EF=BC=8C=E5=9F=BA=E4=BA=8E?=
 =?UTF-8?q?=E8=A7=86=E9=A2=91=E5=86=85=E5=AE=B9=E5=88=86=E6=9E=90=E6=8E=A8?=
 =?UTF-8?q?=E8=8D=90=E6=9C=80=E4=BD=B3=E7=89=87=E6=AE=B5=E8=B5=B7=E5=A7=8B?=
 =?UTF-8?q?=E6=97=B6=E9=97=B4?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 新增 _get_mediakit_recommendations 调用 MediaKit 视频理解分析素材
- 新增 _recommended_time_conflicts 检查推荐时间与已用时间段的冲突
- from-assets 端点优先使用 MediaKit 推荐时间，冲突时降级为随机
- 多层降级策略：MediaKit不可用/超时/解析失败/冲突 → 纯随机
- 非视频素材自动跳过，不发送给 MediaKit
- 新增 21 个单测覆盖冲突检测、解析逻辑、集成降级场景
---
 .../app/api/routes/templates_editor/clips.py  | 160 +++++-
 tests/unit/test_mediakit_smart_clips.py       | 470 ++++++++++++++++++
 2 files changed, 628 insertions(+), 2 deletions(-)
 create mode 100644 tests/unit/test_mediakit_smart_clips.py

diff --git a/apps/api/app/api/routes/templates_editor/clips.py b/apps/api/app/api/routes/templates_editor/clips.py
index 0acd62fd7..bff59a302 100755
--- a/apps/api/app/api/routes/templates_editor/clips.py
+++ b/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(
diff --git a/tests/unit/test_mediakit_smart_clips.py b/tests/unit/test_mediakit_smart_clips.py
new file mode 100644
index 000000000..7c1e1721d
--- /dev/null
+++ b/tests/unit/test_mediakit_smart_clips.py
@@ -0,0 +1,470 @@
+"""测试 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 返回。"""
+        from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
+        import json
+
+        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
-- 
2.54.0


From 6d225a89b4439f21a3a643e0542490744ba9e16e Mon Sep 17 00:00:00 2001
From: CI Bot <ci-bot@xiaoxiajianji.com>
Date: Thu, 27 Aug 2026 08:03:27 +0000
Subject: [PATCH 2/2] style: auto-format with black + isort + prettier [skip
 ci-format-check]

---
 tests/unit/test_mediakit_smart_clips.py | 60 ++++++++++++++++++-------
 1 file changed, 43 insertions(+), 17 deletions(-)

diff --git a/tests/unit/test_mediakit_smart_clips.py b/tests/unit/test_mediakit_smart_clips.py
index 7c1e1721d..7eeb94fa5 100644
--- a/tests/unit/test_mediakit_smart_clips.py
+++ b/tests/unit/test_mediakit_smart_clips.py
@@ -23,7 +23,6 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
 
 import pytest
 
-
 # ── _recommended_time_conflicts 单元测试 ─────────────────────────────────────
 
 
@@ -32,45 +31,54 @@ 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
 
@@ -86,6 +94,7 @@ class TestGetMediakitRecommendations:
     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
@@ -97,9 +106,10 @@ class TestGetMediakitRecommendations:
     @patch("app.api.routes.templates_editor.clips.get_storage_service")
     def test_parses_json_response(self, mock_storage, mock_client_fn):
         """正确解析 JSON 格式的 MediaKit 返回。"""
-        from app.api.routes.templates_editor.clips import _get_mediakit_recommendations
         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 = [
@@ -128,9 +138,7 @@ class TestGetMediakitRecommendations:
 
         mock_client = MagicMock()
         mock_client.is_available = True
-        mock_client.analyze_videos.return_value = [
-            '根据分析，recommended_start_time": 8.3，画面主体明确'
-        ]
+        mock_client.analyze_videos.return_value = ['根据分析，recommended_start_time": 8.3，画面主体明确']
         mock_client_fn.return_value = mock_client
 
         mock_asset_repo = MagicMock()
@@ -269,8 +277,11 @@ def _make_mock_asset(asset_id, duration):
 
 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,
+        clip_id=f"clip-{order}",
+        order=order,
+        duration=duration,
+        start_time=start_time,
+        asset_id=asset_id,
     )
 
 
@@ -328,9 +339,12 @@ class TestMediakitIntegrationInFromAssets:
 
         body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1)
         create_clips_from_assets_editor(
-            template_id="tmpl-1", body=body, plan_id="plan-mk1",
+            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(),
+            asset_repo=mock_asset_repo,
+            current_user=_make_auth_user(),
         )
 
         create_calls = mock_plan_svc.create_clip.call_args_list
@@ -372,9 +386,12 @@ class TestMediakitIntegrationInFromAssets:
 
         body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1)
         create_clips_from_assets_editor(
-            template_id="tmpl-1", body=body, plan_id="plan-mk2",
+            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(),
+            asset_repo=mock_asset_repo,
+            current_user=_make_auth_user(),
         )
 
         create_calls = mock_plan_svc.create_clip.call_args_list
@@ -402,9 +419,12 @@ class TestMediakitIntegrationInFromAssets:
 
         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",
+            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(),
+            asset_repo=mock_asset_repo,
+            current_user=_make_auth_user(),
         )
 
         assert result.created_count == 1
@@ -438,9 +458,12 @@ class TestMediakitIntegrationInFromAssets:
 
         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",
+            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(),
+            asset_repo=mock_asset_repo,
+            current_user=_make_auth_user(),
         )
 
         # 功能正常，降级为随机
@@ -461,9 +484,12 @@ class TestMediakitIntegrationInFromAssets:
 
         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",
+            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(),
+            asset_repo=mock_asset_repo,
+            current_user=_make_auth_user(),
         )
 
         # 功能正常，降级为随机
-- 
2.54.0