diff --git a/apps/api/app/api/routes/templates_editor/clips.py b/apps/api/app/api/routes/templates_editor/clips.py index bff59a302..2d670f0dc 100755 --- a/apps/api/app/api/routes/templates_editor/clips.py +++ b/apps/api/app/api/routes/templates_editor/clips.py @@ -17,16 +17,21 @@ from __future__ import annotations import json import logging +import random import re from app.auth import AuthenticatedUser, get_current_user from app.core.storage import get_storage_service -from app.dependencies import get_asset_repository +from app.dependencies import get_asset_repository, get_db_session from app.services.edit_plan_service import EditPlanService from app.services.edit_template_service import EditTemplateService from fastapi import APIRouter, Depends, HTTPException, Query, status +from sqlalchemy.orm import Session from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository +from packages.adapters.sqlalchemy_impl.template_repository import ( + SQLAlchemyTemplateRepository, +) from packages.domain.plan_generator_utils import _calc_random_start_time from packages.shared.mediakit_client import get_mediakit_client @@ -355,6 +360,72 @@ def batch_delete_editor_clips( +def _safe_segment_duration(value, default: float) -> float: + """安全地将数据库中的时长值转换为正浮点数. + + 处理 None、无效类型、负数、NaN 等异常情况。 + """ + if value is None: + return default + try: + result = float(value) + except (ValueError, TypeError): + return default + if result != result or result <= 0: # NaN check or non-positive + return default + return result + + +def _get_template_segments( + template_id: str, + tpl_svc: EditTemplateService, + db: Session, +) -> list[tuple[int, float, float]]: + """获取模板的片段配置（顺序、最短时长、最长时长）. + + 优先从新模板系统（template_clip_configs）查询， + 若不存在则回退到旧模板系统（template_segments）。 + + Returns: + [(segment_order, duration_min, duration_max), ...] 按 order 排序 + """ + # 优先查新模板系统 + try: + clip_configs = tpl_svc.list_clip_configs(template_id) + if clip_configs: + result = [] + for cc in clip_configs: + dur_min = _safe_segment_duration( + cc.min_duration, _DEFAULT_EDITOR_CLIP_DURATION + ) + dur_max = _safe_segment_duration( + cc.max_duration or cc.min_duration, + _DEFAULT_EDITOR_CLIP_DURATION, + ) + dur_min, dur_max = min(dur_min, dur_max), max(dur_min, dur_max) + result.append((cc.order, dur_min, dur_max)) + return sorted(result, key=lambda x: x[0]) + except Exception: + logger.warning("新模板系统查询clip_configs失败，回退到旧系统", exc_info=True) + + # 回退到旧模板系统（template_segments表） + try: + old_repo = SQLAlchemyTemplateRepository(db) + segments = old_repo.list_segments(template_id) + if segments: + result = [] + for s in segments: + dur_min = _safe_segment_duration(s.duration_min, _DEFAULT_EDITOR_CLIP_DURATION) + dur_max = _safe_segment_duration(s.duration_max, _DEFAULT_EDITOR_CLIP_DURATION) + dur_min, dur_max = min(dur_min, dur_max), max(dur_min, dur_max) + result.append((s.segment_order, dur_min, dur_max)) + return sorted(result, key=lambda x: x[0]) + except Exception: + logger.warning("旧模板系统查询segments失败", exc_info=True) + + return [] + + def _recommended_time_conflicts( start: float, duration: float, @@ -489,22 +560,36 @@ def create_clips_from_assets_editor( plan_id: str = Depends(get_draft_plan_id), services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository), + db: Session = Depends(get_db_session), current_user: AuthenticatedUser = Depends(get_current_user), ) -> ClipsFromAssetsResponse: - """从素材批量创建片段（支持同一素材切多个片段 + 随机起始时间 + 去重）. + """从素材批量创建片段（按模板segment配置创建，事务性替换）. 逻辑： - 1. 模板要求 N 个片段，必须创建 N 个（不管素材有几个） - 2. 素材数量 < 片段数量时，同一素材轮询切多个片段 - 3. 每个片段从素材中随机选取不重复时间段 - 4. 素材时长不足 5s 时缩短 clip duration - 5. 新片段追加到时间线末尾（order 在现有最大值基础上递增） + 1. 从模板读取 segments，片段数量 = segment 数量（忽略前端传的 required_clips_count） + 2. 每个片段时长在 segment 的 duration_min ~ duration_max 之间随机取值（保留一位小数） + 3. 素材按片段顺序轮询分配，素材不够时同一素材切多个片段 + 4. 使用 replace_all_clips_transactional 原子性地清空旧片段并创建新的 + 5. MediaKit 智能选片：第一个使用某素材的片段用推荐起始时间，后续用随机 + 6. 素材时长为 0 或缺失时报 400，不创建无效片段 """ - _, plan_svc = services + tpl_svc, plan_svc = services - required_count = body.required_clips_count if body.required_clips_count is not None else len(body.asset_ids) + # 1. 查询模板 segments + segments = _get_template_segments(template_id, tpl_svc, db) + if not segments: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="模板没有片段配置，无法创建片段", + ) - # 去重后批量获取素材实际时长，避免重复查询 + if not body.asset_ids: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="素材列表为空，无法创建片段", + ) + + # 2. 获取素材实际时长（去重查询） unique_asset_ids = list(dict.fromkeys(body.asset_ids)) asset_durations: dict[str, float] = {} for asset_id in unique_asset_ids: @@ -512,106 +597,113 @@ def create_clips_from_assets_editor( if asset and hasattr(asset, "duration"): asset_durations[asset_id] = float(asset.duration or 0.0) - # 计算追加起始 order：当前 plan 已有片段的最大 order + 1 - # list_clips 返回 List[EditPlanClip] - existing_clips_list = plan_svc.list_clips(plan_id) - next_order = max((c.order for c in existing_clips_list), default=-1) + 1 + # 3. 获取 MediaKit 智能选片推荐（保持60s timeout + poll 2s + 15次） + mediakit_recommendations = _get_mediakit_recommendations( + unique_asset_ids, asset_repo + ) - # 从已有片段中构建已使用时间段，避免跨任务重复使用同一段素材区域 + # 4. 在内存中计算所有片段数据 + asset_first_used: set[str] = set() used_segments: dict[str, list[tuple[float, float]]] = {} - for _clip in existing_clips_list: - if _clip.asset_id and _clip.start_time is not None and _clip.duration is not None: - used_segments.setdefault(_clip.asset_id, []).append( - (float(_clip.start_time), float(_clip.start_time) + float(_clip.duration)) + clips_data: list[dict] = [] + + for i, (_seg_order, dur_min, dur_max) in enumerate(segments): + # 轮询分配素材 + asset_id = body.asset_ids[i % len(body.asset_ids)] + asset_total = asset_durations.get(asset_id, 0.0) + + # 素材时长为 0 或缺失时无法创建有效片段 + if asset_total <= 0: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"素材 {asset_id} 时长信息缺失或为0，无法创建片段", ) - # 尝试获取 MediaKit 智能选片推荐 - mediakit_recommendations = _get_mediakit_recommendations(unique_asset_ids, asset_repo) - - clips = [] - - for i in range(required_count): - # 轮询分配素材：素材不够时同一素材切多个片段 - asset_id = body.asset_ids[i % len(body.asset_ids)] + # 在 segment 的 duration_min ~ duration_max 之间随机取值（保留一位小数） + raw_duration = random.uniform(dur_min, dur_max) + clip_duration = round(raw_duration, 1) # 素材时长不足时缩短 clip duration - asset_total = asset_durations.get(asset_id, 0.0) - if asset_total > 0: - clip_duration = min(_DEFAULT_EDITOR_CLIP_DURATION, asset_total) - else: - clip_duration = _DEFAULT_EDITOR_CLIP_DURATION + clip_duration = min(clip_duration, asset_total) - # 优先使用 MediaKit 推荐的起始时间，冲突时降级为随机 + if clip_duration <= 0: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"素材 {asset_id} 时长不足，无法创建有效片段", + ) + + # 确定起始时间 + is_first_use = asset_id not in asset_first_used recommended_start = mediakit_recommendations.get(asset_id) + if ( - recommended_start is not None - and recommended_start + clip_duration <= asset_durations.get(asset_id, float("inf")) + is_first_use + and recommended_start is not None + and recommended_start + clip_duration <= asset_total and not _recommended_time_conflicts( recommended_start, clip_duration, used_segments.get(asset_id, []) ) ): start_time = recommended_start logger.info( - "使用MediaKit推荐起始时间: asset_id=%s start_time=%.2f", - asset_id, start_time, + "使用MediaKit推荐起始时间: asset_id=%s start_time=%.2f duration=%.1f", + asset_id, + start_time, + clip_duration, ) else: - if recommended_start is not None: + if is_first_use and recommended_start is not None: logger.info( "MediaKit推荐时间冲突或越界，降级为随机: asset_id=%s recommended=%.2f", - asset_id, recommended_start, + asset_id, + recommended_start, + ) + elif not is_first_use: + logger.info( + "素材%s非首次使用，使用随机起始时间", + asset_id, ) - # 随机选择起始时间，避开已使用的时间段 start_time = _calc_random_start_time( asset_id, clip_duration, asset_durations, used_segments ) + if start_time is None: - # 素材时长信息缺失，无法计算随机起始时间 raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, - detail=f"素材 {asset_id} 时长信息缺失，无法创建片段", + detail=f"素材 {asset_id} 时长信息缺失，无法计算起始时间", ) - # 记录已使用时间段（用于去重） - used_segments.setdefault(asset_id, []).append((start_time, start_time + clip_duration)) + # 记录已使用时间段 + used_segments.setdefault(asset_id, []).append( + (start_time, start_time + clip_duration) + ) + asset_first_used.add(asset_id) - try: - clip = plan_svc.create_clip( - plan_id, - clip_type=body.clip_type or "main", - order=next_order + i, - duration=clip_duration, - start_time=start_time, - asset_id=asset_id, - ) - clips.append(clip) - except ValueError as exc: - logger.error( - "创建片段失败: plan_id=%s asset_id=%s order=%d error=%s", - plan_id, - asset_id, - next_order + i, - exc, - exc_info=True, - ) - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=f"创建片段失败: {exc}", - ) from exc + clips_data.append( + { + "order": i, + "asset_id": asset_id, + "start_time": start_time, + "duration": clip_duration, + "clip_type": body.clip_type or "main", + } + ) + + # 5. 事务性替换：清空旧片段 → 创建新片段 → 标记ready（单事务，失败自动回滚） + created_count = plan_svc.replace_all_clips_transactional(plan_id, clips_data) logger.info( - "模板编辑器从素材创建片段: template_id=%s plan_id=%s required=%d actual=%d by user=%s", + "from-assets按模板创建片段: template_id=%s plan_id=%s segments=%d created=%d by user=%s", template_id, plan_id, - required_count, - len(clips), + len(segments), + created_count, current_user.user.id, ) - # 新创建的片段已分配素材，立即标记为 ready，否则渲染管线找不到就绪片段 - plan_svc.mark_clips_ready(plan_id) + # 返回事务后查询到的 clip IDs（replace 方法不返回 ID 列表，用 created_count 构造响应） return ClipsFromAssetsResponse( - created_count=len(clips), + created_count=created_count, plan_id=plan_id, - clip_ids=[c.id for c in clips], + clip_ids=[], # 事务方法不返回 ID；前端不需要逐个 ID ) diff --git a/apps/api/app/services/edit_plan_service.py b/apps/api/app/services/edit_plan_service.py index 926a7e47f..a87060234 100755 --- a/apps/api/app/services/edit_plan_service.py +++ b/apps/api/app/services/edit_plan_service.py @@ -400,7 +400,7 @@ class EditPlanService: order = clip_item.get("order") or i clip = EditPlanClip.create( plan_id=plan_id, - clip_type="main", + clip_type=clip_item.get("clip_type", "main"), order=order, asset_id=clip_item.get("asset_id", ""), start_time=clip_item.get("start_time", 0.0), diff --git a/tests/unit/test_editor_clips_random_start.py b/tests/unit/test_editor_clips_random_start.py index f9a27aaaf..a9471aaf6 100644 --- a/tests/unit/test_editor_clips_random_start.py +++ b/tests/unit/test_editor_clips_random_start.py @@ -1,14 +1,15 @@ -"""测试编辑器 from-assets 端点：同一素材切多个片段 + 随机起始 + 去重. +"""测试编辑器 from-assets 端点：按模板segment创建片段 + 事务性替换 + 随机起始. 覆盖： -- required_clips_count 精确控制片段数量 +- 片段数量 = segment 数量（required_clips_count 被忽略） - 素材不足时同一素材轮询切多个片段 - 随机 start_time + used_segments 去重 - 素材时长不足时 clip duration 缩短 -- 向后兼容（不传 required_clips_count 时等于素材数量） -- order 追加到时间线末尾 +- 素材时长为 0 时抛 400 +- 使用 replace_all_clips_transactional 原子性替换 +- order 从 0 开始 - start_time=None 时抛出 400 -- create_clip 失败时抛出 400 并记录日志 +- mark_clips_ready 在事务方法内部完成 """ from __future__ import annotations @@ -29,6 +30,20 @@ from fastapi import HTTPException TEST_PLAN_ID = "plan-draft-001" TEST_USER_ID = "user-001" +# 默认测试用 segments：4 个片段，每个 3~5 秒 +DEFAULT_SEGMENTS = [(0, 3.0, 5.0), (1, 3.0, 5.0), (2, 3.0, 5.0), (3, 3.0, 5.0)] + + +def _segments(count: int, dur_min: float = 3.0, dur_max: float = 5.0): + return [(i, dur_min, dur_max) for i in range(count)] + + +def _patch_segments(segments=None): + return patch( + "app.api.routes.templates_editor.clips._get_template_segments", + return_value=segments if segments is not None else DEFAULT_SEGMENTS, + ) + def _make_auth_user(): auth = MagicMock() @@ -39,27 +54,6 @@ def _make_auth_user(): return auth -def _make_mock_clip(clip_id, order, duration, start_time=0.0, asset_id=""): - clip = MagicMock() - clip.id = clip_id - clip.plan_id = TEST_PLAN_ID - clip.clip_type = "main" - clip.order = order - clip.duration = duration - clip.start_time = start_time - clip.text_content = "" - clip.transition_effect = "cut" - clip.transition_duration = 0.0 - clip.playback_speed = 1.0 - clip.config = {} - clip.asset_id = asset_id - clip.status = "pending" - clip.template_clip_config_id = "" - clip.created_at = None - clip.updated_at = None - return clip - - def _make_mock_asset(asset_id, duration): asset = MagicMock() asset.id = asset_id @@ -67,212 +61,127 @@ def _make_mock_asset(asset_id, duration): return asset -def _create_clips(plan_id, clip_type, order, duration=0.0, start_time=0.0, asset_id="", **kw): - return _make_mock_clip( - clip_id=f"clip-{order}", - order=order, - duration=duration, - start_time=start_time, - asset_id=asset_id, - ) - - -def _make_plan_svc(existing_clips=None): +def _make_plan_svc(replace_return_count=None): svc = MagicMock() - svc.get_plan_or_raise = MagicMock() - svc.create_clip = MagicMock(side_effect=_create_clips) - svc.list_clips = MagicMock(return_value=existing_clips or []) + # replace_all_clips_transactional 返回创建的片段数量 + if replace_return_count is not None: + svc.replace_all_clips_transactional = MagicMock(return_value=replace_return_count) + else: + svc.replace_all_clips_transactional = MagicMock(return_value=0) return svc -class TestEditorClipsRequiredCount: - """测试 required_clips_count 控制片段数量 + 同素材多片段.""" +def _get_clips_data_from_call(mock_plan_svc): + """从 replace_all_clips_transactional 的调用中获取 clips_data。""" + assert mock_plan_svc.replace_all_clips_transactional.called, "replace_all_clips_transactional 未被调用" + call_args = mock_plan_svc.replace_all_clips_transactional.call_args + # call_args = ((plan_id, clips_data), kwargs) + if len(call_args.args) >= 2: + return call_args.args[1] + return call_args.kwargs.get("clips_data", []) + + +class TestEditorClipsBySegments: + """测试按 segment 数量创建片段 + 素材轮询。""" @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_creates_exactly_required_clips_count(self, mock_storage): - """required_clips_count=4 时，即使只有2个素材也创建4个片段.""" + def test_creates_clips_matching_segment_count(self, mock_storage): + """4 个 segment 即使只有2个素材也创建4个片段，required_clips_count 被忽略。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest - mock_plan_svc = _make_plan_svc() + mock_plan_svc = _make_plan_svc(replace_return_count=4) mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock(side_effect=lambda aid: _make_mock_asset(aid, {"a1": 30.0, "a2": 20.0}[aid])) - body = ClipsFromAssetsRequest(asset_ids=["a1", "a2"], required_clips_count=4) + body = ClipsFromAssetsRequest(asset_ids=["a1", "a2"], required_clips_count=2) - result = create_clips_from_assets_editor( - template_id="tpl-001", - body=body, - plan_id=TEST_PLAN_ID, - services=(MagicMock(), mock_plan_svc), - asset_repo=mock_asset_repo, - current_user=_make_auth_user(), - ) + with _patch_segments(DEFAULT_SEGMENTS): + result = create_clips_from_assets_editor( + template_id="tpl-001", + body=body, + plan_id=TEST_PLAN_ID, + services=(MagicMock(), mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) assert result.created_count == 4 - assert mock_plan_svc.create_clip.call_count == 4 + clips_data = _get_clips_data_from_call(mock_plan_svc) + assert len(clips_data) == 4 # 验证轮询分配：a1, a2, a1, a2 - calls = mock_plan_svc.create_clip.call_args_list - assert calls[0].kwargs["asset_id"] == "a1" - assert calls[1].kwargs["asset_id"] == "a2" - assert calls[2].kwargs["asset_id"] == "a1" - assert calls[3].kwargs["asset_id"] == "a2" + assert clips_data[0]["asset_id"] == "a1" + assert clips_data[1]["asset_id"] == "a2" + assert clips_data[2]["asset_id"] == "a1" + assert clips_data[3]["asset_id"] == "a2" @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_orders_append_to_existing_timeline(self, mock_storage): - """时间线已有2个片段时，新片段 order 应从 2 开始连续递增.""" + def test_orders_start_at_zero(self, mock_storage): + """片段 order 从 0 开始递增。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest - # 模拟已有 order=0, order=1 的片段 - existing = [_make_mock_clip("old-1", 0, 5.0), _make_mock_clip("old-2", 1, 5.0)] - mock_plan_svc = _make_plan_svc(existing_clips=existing) - + mock_plan_svc = _make_plan_svc(replace_return_count=3) mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=3) - create_clips_from_assets_editor( - template_id="tpl-001", - body=body, - plan_id=TEST_PLAN_ID, - services=(MagicMock(), mock_plan_svc), - asset_repo=mock_asset_repo, - current_user=_make_auth_user(), - ) - - calls = mock_plan_svc.create_clip.call_args_list - assert calls[0].kwargs["order"] == 2 - assert calls[1].kwargs["order"] == 3 - assert calls[2].kwargs["order"] == 4 - - @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_orders_start_at_zero_when_empty(self, mock_storage): - """空时间线时 order 从 0 开始.""" - from app.api.routes.templates_editor.clips import ( - create_clips_from_assets_editor, - ) - from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest - - mock_plan_svc = _make_plan_svc(existing_clips=[]) - - mock_asset_repo = MagicMock() - mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) - - body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=3) - - create_clips_from_assets_editor( - template_id="tpl-001", - body=body, - plan_id=TEST_PLAN_ID, - services=(MagicMock(), mock_plan_svc), - asset_repo=mock_asset_repo, - current_user=_make_auth_user(), - ) - - calls = mock_plan_svc.create_clip.call_args_list - assert calls[0].kwargs["order"] == 0 - assert calls[1].kwargs["order"] == 1 - assert calls[2].kwargs["order"] == 2 - - @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_backward_compatible_default_count(self, mock_storage): - """不传 required_clips_count 时，片段数等于素材数（向后兼容）.""" - from app.api.routes.templates_editor.clips import ( - create_clips_from_assets_editor, - ) - from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest - - mock_plan_svc = _make_plan_svc() - mock_asset_repo = MagicMock() - mock_asset_repo.get = MagicMock(side_effect=lambda aid: _make_mock_asset(aid, 30.0)) - - body = ClipsFromAssetsRequest(asset_ids=["a1", "a2", "a3"]) - - result = create_clips_from_assets_editor( - template_id="tpl-001", - body=body, - plan_id=TEST_PLAN_ID, - services=(MagicMock(), mock_plan_svc), - asset_repo=mock_asset_repo, - current_user=_make_auth_user(), - ) - - assert result.created_count == 3 - assert mock_plan_svc.create_clip.call_count == 3 - - -class TestEditorClipsDurationAndStartTime: - """测试素材时长获取、clip duration 缩短、start_time 传入.""" - - @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_clip_duration_shortened_for_short_assets(self, mock_storage): - """素材只有 3s 时 clip duration 缩短到 3.0.""" - from app.api.routes.templates_editor.clips import ( - create_clips_from_assets_editor, - ) - from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest - - mock_plan_svc = _make_plan_svc() - mock_asset_repo = MagicMock() - mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("short", 3.0)) - - body = ClipsFromAssetsRequest(asset_ids=["short"]) - - create_clips_from_assets_editor( - template_id="tpl-001", - body=body, - plan_id=TEST_PLAN_ID, - services=(MagicMock(), mock_plan_svc), - asset_repo=mock_asset_repo, - current_user=_make_auth_user(), - ) - - calls = mock_plan_svc.create_clip.call_args_list - assert len(calls) == 1 - assert calls[0].kwargs["duration"] == 3.0 - - @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_start_time_passed_to_create_clip(self, mock_storage): - """验证 _calc_random_start_time 返回值被传入 create_clip.""" - from app.api.routes.templates_editor.clips import ( - create_clips_from_assets_editor, - ) - from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest - - mock_plan_svc = _make_plan_svc() - mock_asset_repo = MagicMock() - mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) - - body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=2) - - with patch( - "app.api.routes.templates_editor.clips._calc_random_start_time", - side_effect=[12.5, 18.0], - ) as mock_calc: + with _patch_segments(_segments(3)): create_clips_from_assets_editor( template_id="tpl-001", body=body, plan_id=TEST_PLAN_ID, services=(MagicMock(), mock_plan_svc), asset_repo=mock_asset_repo, + db=MagicMock(), current_user=_make_auth_user(), ) - assert mock_calc.call_count == 2 - calls = mock_plan_svc.create_clip.call_args_list - assert calls[0].kwargs["start_time"] == 12.5 - assert calls[1].kwargs["start_time"] == 18.0 + clips_data = _get_clips_data_from_call(mock_plan_svc) + assert clips_data[0]["order"] == 0 + assert clips_data[1]["order"] == 1 + assert clips_data[2]["order"] == 2 @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_asset_durations_deduped(self, mock_storage): - """asset_ids 有重复时只查询一次素材时长.""" + def test_uses_transactional_replace(self, mock_storage): + """使用 replace_all_clips_transactional 而不是分别 delete + create。""" + from app.api.routes.templates_editor.clips import ( + create_clips_from_assets_editor, + ) + from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + + mock_plan_svc = _make_plan_svc(replace_return_count=2) + mock_asset_repo = MagicMock() + mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) + + body = ClipsFromAssetsRequest(asset_ids=["a1"]) + + with _patch_segments(_segments(2)): + create_clips_from_assets_editor( + template_id="tpl-001", + body=body, + plan_id=TEST_PLAN_ID, + services=(MagicMock(), mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) + + # 必须调用事务方法 + mock_plan_svc.replace_all_clips_transactional.assert_called_once() + # 不应调用单独的 delete 或 create + assert not hasattr(mock_plan_svc, "create_clip") or not mock_plan_svc.create_clip.called + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_no_segments_raises_400(self, mock_storage): + """模板没有 segment 配置时返回 400。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) @@ -280,26 +189,61 @@ class TestEditorClipsDurationAndStartTime: mock_plan_svc = _make_plan_svc() mock_asset_repo = MagicMock() - mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) - # a1 出现 3 次，但时长只应查一次 - body = ClipsFromAssetsRequest(asset_ids=["a1", "a1", "a1"]) + body = ClipsFromAssetsRequest(asset_ids=["a1"]) - create_clips_from_assets_editor( - template_id="tpl-001", - body=body, - plan_id=TEST_PLAN_ID, - services=(MagicMock(), mock_plan_svc), - asset_repo=mock_asset_repo, - current_user=_make_auth_user(), - ) + with _patch_segments([]): + with pytest.raises(HTTPException) as exc_info: + create_clips_from_assets_editor( + template_id="tpl-001", + body=body, + plan_id=TEST_PLAN_ID, + services=(MagicMock(), mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) - # 去重后只调用 1 次 - assert mock_asset_repo.get.call_count == 1 + assert exc_info.value.status_code == 400 + assert "片段配置" in exc_info.value.detail + # 不应调用替换方法 + mock_plan_svc.replace_all_clips_transactional.assert_not_called() + + +class TestEditorClipsDurationAndStartTime: + """测试素材时长获取、clip duration 缩短、start_time 传入。""" @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_used_segments_passed_to_calc(self, mock_storage): - """同一素材切多个片段时，used_segments 应被维护并传入.""" + def test_clip_duration_shortened_for_short_assets(self, mock_storage): + """素材只有 3s 时 clip duration 缩短到不超过 3.0。""" + from app.api.routes.templates_editor.clips import ( + create_clips_from_assets_editor, + ) + from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + + mock_plan_svc = _make_plan_svc(replace_return_count=1) + mock_asset_repo = MagicMock() + mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("short", 3.0)) + + body = ClipsFromAssetsRequest(asset_ids=["short"]) + + with _patch_segments(_segments(1, dur_min=5.0, dur_max=10.0)): + create_clips_from_assets_editor( + template_id="tpl-001", + body=body, + plan_id=TEST_PLAN_ID, + services=(MagicMock(), mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) + + clips_data = _get_clips_data_from_call(mock_plan_svc) + assert clips_data[0]["duration"] <= 3.0 + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_zero_duration_asset_raises_400(self, mock_storage): + """素材时长为 0 时应抛出 400，而不是创建无效片段。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) @@ -307,6 +251,127 @@ class TestEditorClipsDurationAndStartTime: mock_plan_svc = _make_plan_svc() mock_asset_repo = MagicMock() + mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("bad", 0.0)) + + body = ClipsFromAssetsRequest(asset_ids=["bad"]) + + with _patch_segments(_segments(1)): + with pytest.raises(HTTPException) as exc_info: + create_clips_from_assets_editor( + template_id="tpl-001", + body=body, + plan_id=TEST_PLAN_ID, + services=(MagicMock(), mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) + + assert exc_info.value.status_code == 400 + assert "时长" in exc_info.value.detail + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_missing_duration_asset_raises_400(self, mock_storage): + """素材时长缺失（asset_repo.get 返回 None）时抛出 400。""" + from app.api.routes.templates_editor.clips import ( + create_clips_from_assets_editor, + ) + from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + + mock_plan_svc = _make_plan_svc() + mock_asset_repo = MagicMock() + mock_asset_repo.get = MagicMock(return_value=None) + + body = ClipsFromAssetsRequest(asset_ids=["missing"]) + + with _patch_segments(_segments(1)): + with pytest.raises(HTTPException) as exc_info: + create_clips_from_assets_editor( + template_id="tpl-001", + body=body, + plan_id=TEST_PLAN_ID, + services=(MagicMock(), mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) + + assert exc_info.value.status_code == 400 + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_start_time_passed_to_create(self, mock_storage): + """_calc_random_start_time 返回值被传入 clips_data。""" + from app.api.routes.templates_editor.clips import ( + create_clips_from_assets_editor, + ) + from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + + mock_plan_svc = _make_plan_svc(replace_return_count=2) + mock_asset_repo = MagicMock() + mock_asset_repo.get = MagicMock(side_effect=lambda aid: _make_mock_asset(aid, 30.0)) + + body = ClipsFromAssetsRequest(asset_ids=["a1", "a2"], required_clips_count=2) + + with ( + _patch_segments(_segments(2)), + patch( + "app.api.routes.templates_editor.clips._calc_random_start_time", + side_effect=[12.5, 18.0], + ) as mock_calc, + ): + create_clips_from_assets_editor( + template_id="tpl-001", + body=body, + plan_id=TEST_PLAN_ID, + services=(MagicMock(), mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) + + assert mock_calc.call_count == 2 + clips_data = _get_clips_data_from_call(mock_plan_svc) + assert clips_data[0]["start_time"] == 12.5 + assert clips_data[1]["start_time"] == 18.0 + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_asset_durations_deduped(self, mock_storage): + """asset_ids 有重复时只查询一次素材时长。""" + from app.api.routes.templates_editor.clips import ( + create_clips_from_assets_editor, + ) + from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + + mock_plan_svc = _make_plan_svc(replace_return_count=3) + mock_asset_repo = MagicMock() + mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) + + body = ClipsFromAssetsRequest(asset_ids=["a1", "a1", "a1"]) + + with _patch_segments(_segments(3)): + create_clips_from_assets_editor( + template_id="tpl-001", + body=body, + plan_id=TEST_PLAN_ID, + services=(MagicMock(), mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) + + # 去重后只调用 1 次获取素材时长 + assert mock_asset_repo.get.call_count == 1 + + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_used_segments_maintained_across_clips(self, mock_storage): + """同一素材切多个片段时，used_segments 应被维护。""" + from app.api.routes.templates_editor.clips import ( + create_clips_from_assets_editor, + ) + from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + + mock_plan_svc = _make_plan_svc(replace_return_count=3) + mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=3) @@ -317,9 +382,12 @@ class TestEditorClipsDurationAndStartTime: captured_used_segments.append({aid: list(segs) for aid, segs in (used_segments or {}).items()}) return (len(captured_used_segments) - 1) * 5.0 - with patch( - "app.api.routes.templates_editor.clips._calc_random_start_time", - side_effect=fake_calc, + with ( + _patch_segments(_segments(3)), + patch( + "app.api.routes.templates_editor.clips._calc_random_start_time", + side_effect=fake_calc, + ), ): create_clips_from_assets_editor( template_id="tpl-001", @@ -327,215 +395,80 @@ class TestEditorClipsDurationAndStartTime: plan_id=TEST_PLAN_ID, services=(MagicMock(), mock_plan_svc), asset_repo=mock_asset_repo, + db=MagicMock(), current_user=_make_auth_user(), ) + # 第一次没有已使用时间段 assert captured_used_segments[0] == {} - assert captured_used_segments[1] == {"a1": [(0.0, 5.0)]} - assert captured_used_segments[2] == {"a1": [(0.0, 5.0), (5.0, 10.0)]} + # 第二次有第一次的记录 + assert len(captured_used_segments[1]["a1"]) == 1 + # 第三次有前两次的记录 + assert len(captured_used_segments[2]["a1"]) == 2 class TestEditorClipsErrorHandling: - """测试异常处理.""" + """测试异常处理。""" @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_none_start_time_raises_400(self, mock_storage): - """_calc_random_start_time 返回 None 时应抛出 HTTPException 400.""" + """_calc_random_start_time 返回 None 时应抛出 HTTPException 400。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest mock_plan_svc = _make_plan_svc() - # asset_repo.get 返回 None → asset_durations 为空 → _calc_random_start_time 返回 None mock_asset_repo = MagicMock() - mock_asset_repo.get = MagicMock(return_value=None) + # 素材有 duration 但 random 返回 None + mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) - body = ClipsFromAssetsRequest(asset_ids=["missing-asset"]) + body = ClipsFromAssetsRequest(asset_ids=["a1"]) - with pytest.raises(HTTPException) as exc_info: - create_clips_from_assets_editor( - template_id="tpl-001", - body=body, - plan_id=TEST_PLAN_ID, - services=(MagicMock(), mock_plan_svc), - asset_repo=mock_asset_repo, - current_user=_make_auth_user(), - ) + with ( + _patch_segments(_segments(1)), + patch( + "app.api.routes.templates_editor.clips._calc_random_start_time", + return_value=None, + ), + ): + with pytest.raises(HTTPException) as exc_info: + create_clips_from_assets_editor( + template_id="tpl-001", + body=body, + plan_id=TEST_PLAN_ID, + services=(MagicMock(), mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) assert exc_info.value.status_code == 400 - assert "时长信息缺失" in exc_info.value.detail + assert "时长" in exc_info.value.detail @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_create_clip_value_error_raises_400(self, mock_storage): - """create_clip 抛出 ValueError 时应转为 HTTPException 400.""" + def test_transactional_replace_exception_propagates(self, mock_storage): + """replace_all_clips_transactional 抛异常时应向上传播（事务已回滚）。""" from app.api.routes.templates_editor.clips import ( create_clips_from_assets_editor, ) from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest mock_plan_svc = _make_plan_svc() - mock_plan_svc.create_clip = MagicMock(side_effect=ValueError("计划不存在")) - + mock_plan_svc.replace_all_clips_transactional = MagicMock(side_effect=RuntimeError("DB connection lost")) mock_asset_repo = MagicMock() mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) body = ClipsFromAssetsRequest(asset_ids=["a1"]) - with pytest.raises(HTTPException) as exc_info: - create_clips_from_assets_editor( - template_id="tpl-001", - body=body, - plan_id=TEST_PLAN_ID, - services=(MagicMock(), mock_plan_svc), - asset_repo=mock_asset_repo, - current_user=_make_auth_user(), - ) - - assert exc_info.value.status_code == 400 - assert "创建片段失败" in exc_info.value.detail - - -class TestMarkClipsReadyAfterCreation: - """验证 from-assets 创建片段后立即调用 mark_clips_ready，确保渲染管线能找到就绪片段。""" - - @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_mark_clips_ready_called_after_creation(self, _mock_storage): - """创建片段后必须调用 plan_svc.mark_clips_ready(plan_id)。""" - from app.api.routes.templates_editor.clips import create_clips_from_assets_editor - from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest - - mock_plan_svc = _make_plan_svc() - mock_asset_repo = MagicMock() - mock_asset_repo.get = MagicMock(side_effect=lambda aid: _make_mock_asset(aid, {"a1": 30.0}[aid])) - - body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=2) - create_clips_from_assets_editor( - template_id="tmpl-1", - body=body, - plan_id="plan-xyz", - services=(MagicMock(), mock_plan_svc), - asset_repo=mock_asset_repo, - current_user=_make_auth_user(), - ) - - # 关键断言：mark_clips_ready 必须被调用，且传入正确的 plan_id - mock_plan_svc.mark_clips_ready.assert_called_once_with("plan-xyz") - - -class TestCrossTaskSegmentDedup: - """验证 from-assets 创建片段时，used_segments 从已有片段构建，实现跨任务去重。""" - - @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_used_segments_populated_from_existing_clips(self, _mock_storage): - """已有片段的 asset_id/start_time/duration 必须被纳入 used_segments，新片段避开已用区间。""" - from app.api.routes.templates_editor.clips import create_clips_from_assets_editor - from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest - - # 模拟已有片段：asset "a1" 在 0~5s 已使用 - existing = [ - _make_mock_clip("c1", order=0, duration=5.0, start_time=0.0, asset_id="a1"), - ] - mock_plan_svc = _make_plan_svc(existing_clips=existing) - mock_asset_repo = MagicMock() - mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) - - body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1) - create_clips_from_assets_editor( - template_id="tmpl-1", - body=body, - plan_id="plan-dedup", - services=(MagicMock(), mock_plan_svc), - asset_repo=mock_asset_repo, - current_user=_make_auth_user(), - ) - - # 验证：新创建的 clip 的 start_time 不应与已有片段 [0, 5] 重叠 - # create_clip 被调用时传入的 start_time 应该 >= 5 或 < 0 (不可能) - # 实际上 _calc_random_start_time 会避开 [0, 5]，所以 start_time 应该 > 5 - create_calls = mock_plan_svc.create_clip.call_args_list - assert len(create_calls) == 1 - new_start_time = create_calls[0].kwargs.get("start_time") or create_calls[0][1].get("start_time") - # 新片段不应从 0 开始（因为 0~5 已被占用） - # 注意：_calc_random_start_time 有随机性，但在 30s 素材中避开 [0,5] 后随机到 0~5 的概率极低 - # 我们用一个宽松断言：start_time 应该是一个有效值 - assert new_start_time is not None - - @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_multiple_existing_clips_build_used_segments(self, _mock_storage): - """多个已有片段的时间段都应被收集到 used_segments 中。""" - from app.api.routes.templates_editor.clips import create_clips_from_assets_editor - from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest - - # 模拟已有片段：asset "a1" 在 [0,5] 和 [10,15] 已使用 - existing = [ - _make_mock_clip("c1", order=0, duration=5.0, start_time=0.0, asset_id="a1"), - _make_mock_clip("c2", order=1, duration=5.0, start_time=10.0, asset_id="a1"), - ] - mock_plan_svc = _make_plan_svc(existing_clips=existing) - mock_asset_repo = MagicMock() - mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) - - body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1) - create_clips_from_assets_editor( - template_id="tmpl-1", - body=body, - plan_id="plan-dedup2", - services=(MagicMock(), mock_plan_svc), - asset_repo=mock_asset_repo, - current_user=_make_auth_user(), - ) - - create_calls = mock_plan_svc.create_clip.call_args_list - assert len(create_calls) == 1 - new_start_time = create_calls[0].kwargs.get("start_time") or create_calls[0][1].get("start_time") - assert new_start_time is not None - - @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_existing_clips_without_asset_id_ignored(self, _mock_storage): - """没有 asset_id 的已有片段不影响 used_segments。""" - from app.api.routes.templates_editor.clips import create_clips_from_assets_editor - from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest - - # 模拟已有片段：一个没有 asset_id 的片段 - existing = [ - _make_mock_clip("c1", order=0, duration=5.0, start_time=0.0, asset_id=""), - ] - mock_plan_svc = _make_plan_svc(existing_clips=existing) - mock_asset_repo = MagicMock() - mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) - - body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1) - result = create_clips_from_assets_editor( - template_id="tmpl-1", - body=body, - plan_id="plan-dedup3", - services=(MagicMock(), mock_plan_svc), - asset_repo=mock_asset_repo, - current_user=_make_auth_user(), - ) - - # 应正常创建，不受空 asset_id 片段影响 - assert result.created_count == 1 - - @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_no_existing_clips_works_same_as_before(self, _mock_storage): - """没有已有片段时，行为与修复前一致（used_segments 为空）。""" - from app.api.routes.templates_editor.clips import create_clips_from_assets_editor - from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest - - mock_plan_svc = _make_plan_svc(existing_clips=[]) - mock_asset_repo = MagicMock() - mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0)) - - body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1) - result = create_clips_from_assets_editor( - template_id="tmpl-1", - body=body, - plan_id="plan-dedup4", - services=(MagicMock(), mock_plan_svc), - asset_repo=mock_asset_repo, - current_user=_make_auth_user(), - ) - - assert result.created_count == 1 + with _patch_segments(_segments(1)): + with pytest.raises(RuntimeError, match="DB connection lost"): + create_clips_from_assets_editor( + template_id="tpl-001", + body=body, + plan_id=TEST_PLAN_ID, + services=(MagicMock(), mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) diff --git a/tests/unit/test_mediakit_smart_clips.py b/tests/unit/test_mediakit_smart_clips.py index 7eeb94fa5..69c1351c9 100644 --- a/tests/unit/test_mediakit_smart_clips.py +++ b/tests/unit/test_mediakit_smart_clips.py @@ -1,12 +1,16 @@ -"""测试 MediaKit 智能选片集成。 +"""测试 MediaKit 智能选片 + from-assets 按模板 segment 创建片段。 覆盖： - _recommended_time_conflicts 冲突检测 - _get_mediakit_recommendations 解析与降级 -- from-assets 端点：推荐时间优先使用 -- from-assets 端点：推荐时间冲突时降级随机 -- from-assets 端点：MediaKit 不可用时降级随机 -- from-assets 端点：MediaKit 返回不可解析内容时降级随机 +- _get_template_segments 查询模板片段配置 +- from-assets 端点：按模板 segment 数量和时长创建片段 +- from-assets 端点：事务性原子替换 +- from-assets 端点：素材轮询分配 +- from-assets 端点：MediaKit 推荐时间首片段使用 +- from-assets 端点：同一素材多片段时后续用随机 +- from-assets 端点：无 segment 配置时报错 +- from-assets 端点：素材时长为0时报400 """ from __future__ import annotations @@ -37,62 +41,82 @@ class TestRecommendedTimeConflicts: def test_no_conflict_when_before(self): from app.api.routes.templates_editor.clips import _recommended_time_conflicts - # 推荐 [5, 10]，已用 [15, 20] assert _recommended_time_conflicts(5.0, 5.0, [(15.0, 20.0)]) is False def test_no_conflict_when_after(self): from app.api.routes.templates_editor.clips import _recommended_time_conflicts - # 推荐 [20, 25]，已用 [0, 10] assert _recommended_time_conflicts(20.0, 5.0, [(0.0, 10.0)]) is False def test_conflict_overlap_start(self): from app.api.routes.templates_editor.clips import _recommended_time_conflicts - # 推荐 [8, 13]，已用 [10, 20] - assert _recommended_time_conflicts(8.0, 5.0, [(10.0, 20.0)]) is True + # 推荐 [5, 10]，已用 [0, 7] + assert _recommended_time_conflicts(5.0, 5.0, [(0.0, 7.0)]) is True def test_conflict_overlap_end(self): from app.api.routes.templates_editor.clips import _recommended_time_conflicts - # 推荐 [15, 20]，已用 [10, 18] - assert _recommended_time_conflicts(15.0, 5.0, [(10.0, 18.0)]) is True + # 推荐 [5, 10]，已用 [8, 15] + assert _recommended_time_conflicts(5.0, 5.0, [(8.0, 15.0)]) is True def test_conflict_contained(self): from app.api.routes.templates_editor.clips import _recommended_time_conflicts - # 推荐 [12, 17]，已用 [10, 20] - assert _recommended_time_conflicts(12.0, 5.0, [(10.0, 20.0)]) is True + # 推荐 [5, 10]，已用 [0, 20] + assert _recommended_time_conflicts(5.0, 5.0, [(0.0, 20.0)]) is True - def test_conflict_adjacent_not_conflict(self): + def test_conflict_exact_boundary_no_overlap(self): from app.api.routes.templates_editor.clips import _recommended_time_conflicts - # 推荐 [5, 10]，已用 [10, 15] —— 边界相邻不算冲突 - assert _recommended_time_conflicts(5.0, 5.0, [(10.0, 15.0)]) is False + # 推荐 [10, 15]，已用 [0, 10] — 边界相接不算冲突 + assert _recommended_time_conflicts(10.0, 5.0, [(0.0, 10.0)]) is False - def test_conflict_with_multiple_segments(self): + def test_conflict_multiple_used(self): from app.api.routes.templates_editor.clips import _recommended_time_conflicts - # 推荐 [12, 17]，已用 [0, 5] 和 [10, 20] - assert _recommended_time_conflicts(12.0, 5.0, [(0.0, 5.0), (10.0, 20.0)]) is True - - def test_no_conflict_between_segments(self): - from app.api.routes.templates_editor.clips import _recommended_time_conflicts - - # 推荐 [6, 11]，已用 [0, 5] 和 [12, 20] - assert _recommended_time_conflicts(6.0, 5.0, [(0.0, 5.0), (12.0, 20.0)]) is False + used = [(0.0, 5.0), (10.0, 15.0), (20.0, 25.0)] + # 推荐 [6, 11] 与 [10, 15] 冲突 + assert _recommended_time_conflicts(6.0, 5.0, used) is True + # 推荐 [15, 20] 不冲突 + assert _recommended_time_conflicts(15.0, 5.0, used) is False -# ── _get_mediakit_recommendations 单元测试 ─────────────────────────────────── +# ── _get_mediakit_recommendations 单元测试 ────────────────────────────────── class TestGetMediakitRecommendations: - """测试 MediaKit 推荐获取与解析。""" + """测试 MediaKit 推荐结果解析和降级。""" @patch("app.api.routes.templates_editor.clips.get_mediakit_client") @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_returns_empty_when_not_available(self, mock_storage, mock_client_fn): - """MediaKit 不可用时返回空字典。""" + def test_returns_parsed_recommendations(self, mock_storage, mock_client_fn): + from app.api.routes.templates_editor.clips import _get_mediakit_recommendations + + mock_client = MagicMock() + mock_client.is_available = True + mock_client.analyze_videos.return_value = [ + '[{"asset_id": "a1", "recommended_start_time": 12.5, "reason": "动作场景"}]' + ] + mock_client_fn.return_value = mock_client + + mock_storage_svc = MagicMock() + mock_storage_svc.get_download_url.return_value = "https://example.com/v.mp4" + mock_storage.return_value = mock_storage_svc + + asset = MagicMock() + asset.storage_key = "v.mp4" + asset.mime_type = "video/mp4" + + mock_asset_repo = MagicMock() + mock_asset_repo.get.return_value = asset + + result = _get_mediakit_recommendations(["a1"], mock_asset_repo) + assert result == {"a1": 12.5} + + @patch("app.api.routes.templates_editor.clips.get_mediakit_client") + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_client_not_available_returns_empty(self, mock_storage, mock_client_fn): from app.api.routes.templates_editor.clips import _get_mediakit_recommendations mock_client = MagicMock() @@ -104,108 +128,66 @@ class TestGetMediakitRecommendations: @patch("app.api.routes.templates_editor.clips.get_mediakit_client") @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_parses_json_response(self, mock_storage, mock_client_fn): - """正确解析 JSON 格式的 MediaKit 返回。""" - import json - + def test_empty_contents_returns_empty(self, mock_storage, mock_client_fn): from app.api.routes.templates_editor.clips import _get_mediakit_recommendations mock_client = MagicMock() mock_client.is_available = True - mock_client.analyze_videos.return_value = [ - json.dumps([{"asset_id": "a1", "recommended_start_time": 12.5, "reason": "画面清晰"}]) - ] + mock_client.analyze_videos.return_value = [] mock_client_fn.return_value = mock_client - mock_asset_repo = MagicMock() - mock_asset = MagicMock() - mock_asset.storage_key = "videos/test.mp4" - mock_asset.mime_type = "video/mp4" - mock_asset_repo.get.return_value = mock_asset - mock_storage_svc = MagicMock() - mock_storage_svc.get_download_url.return_value = "https://example.com/test.mp4" + mock_storage_svc.get_download_url.return_value = "https://example.com/v.mp4" mock_storage.return_value = mock_storage_svc - result = _get_mediakit_recommendations(["a1"], mock_asset_repo) - assert result == {"a1": 12.5} - - @patch("app.api.routes.templates_editor.clips.get_mediakit_client") - @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_parses_regex_fallback(self, mock_storage, mock_client_fn): - """JSON 解析失败时通过正则提取 recommended_start_time。""" - from app.api.routes.templates_editor.clips import _get_mediakit_recommendations - - mock_client = MagicMock() - mock_client.is_available = True - mock_client.analyze_videos.return_value = ['根据分析，recommended_start_time": 8.3，画面主体明确'] - mock_client_fn.return_value = mock_client - + asset = MagicMock() + asset.storage_key = "v.mp4" + asset.mime_type = "video/mp4" mock_asset_repo = MagicMock() - mock_asset = MagicMock() - mock_asset.storage_key = "videos/test.mp4" - mock_asset.mime_type = "video/mp4" - mock_asset_repo.get.return_value = mock_asset - - mock_storage_svc = MagicMock() - mock_storage_svc.get_download_url.return_value = "https://example.com/test.mp4" - mock_storage.return_value = mock_storage_svc - - result = _get_mediakit_recommendations(["a1"], mock_asset_repo) - assert result == {"a1": 8.3} - - @patch("app.api.routes.templates_editor.clips.get_mediakit_client") - @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_returns_empty_on_unparseable(self, mock_storage, mock_client_fn): - """无法解析时返回空字典。""" - from app.api.routes.templates_editor.clips import _get_mediakit_recommendations - - mock_client = MagicMock() - mock_client.is_available = True - mock_client.analyze_videos.return_value = ["这段视频内容丰富，无法确定具体时间"] - mock_client_fn.return_value = mock_client - - mock_asset_repo = MagicMock() - mock_asset = MagicMock() - mock_asset.storage_key = "videos/test.mp4" - mock_asset.mime_type = "video/mp4" - mock_asset_repo.get.return_value = mock_asset - - mock_storage_svc = MagicMock() - mock_storage_svc.get_download_url.return_value = "https://example.com/test.mp4" - mock_storage.return_value = mock_storage_svc + mock_asset_repo.get.return_value = asset result = _get_mediakit_recommendations(["a1"], mock_asset_repo) assert result == {} @patch("app.api.routes.templates_editor.clips.get_mediakit_client") @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_returns_empty_on_none_result(self, mock_storage, mock_client_fn): - """MediaKit 返回 None 时返回空字典。""" + def test_unparseable_response_returns_empty(self, mock_storage, mock_client_fn): from app.api.routes.templates_editor.clips import _get_mediakit_recommendations mock_client = MagicMock() mock_client.is_available = True - mock_client.analyze_videos.return_value = None + mock_client.analyze_videos.return_value = ["这是一段自然语言描述，没有JSON"] mock_client_fn.return_value = mock_client - mock_asset_repo = MagicMock() - mock_asset = MagicMock() - mock_asset.storage_key = "videos/test.mp4" - mock_asset.mime_type = "video/mp4" - mock_asset_repo.get.return_value = mock_asset - mock_storage_svc = MagicMock() - mock_storage_svc.get_download_url.return_value = "https://example.com/test.mp4" + mock_storage_svc.get_download_url.return_value = "https://example.com/v.mp4" mock_storage.return_value = mock_storage_svc + asset = MagicMock() + asset.storage_key = "v.mp4" + asset.mime_type = "video/mp4" + mock_asset_repo = MagicMock() + mock_asset_repo.get.return_value = asset + result = _get_mediakit_recommendations(["a1"], mock_asset_repo) assert result == {} + @patch("app.api.routes.templates_editor.clips.get_mediakit_client") + @patch("app.api.routes.templates_editor.clips.get_storage_service") + def test_storage_failure_returns_empty(self, mock_storage, mock_client_fn): + from app.api.routes.templates_editor.clips import _get_mediakit_recommendations + + mock_client = MagicMock() + mock_client.is_available = True + mock_client_fn.return_value = mock_client + mock_storage.side_effect = RuntimeError("storage unavailable") + + result = _get_mediakit_recommendations(["a1"], MagicMock()) + assert result == {} + @patch("app.api.routes.templates_editor.clips.get_mediakit_client") @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_exception_returns_empty(self, mock_storage, mock_client_fn): - """异常时返回空字典（优雅降级）。""" from app.api.routes.templates_editor.clips import _get_mediakit_recommendations mock_client_fn.side_effect = RuntimeError("unexpected error") @@ -216,7 +198,6 @@ class TestGetMediakitRecommendations: @patch("app.api.routes.templates_editor.clips.get_mediakit_client") @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_skips_non_video_assets(self, mock_storage, mock_client_fn): - """非视频素材被跳过，不发送给 MediaKit。""" from app.api.routes.templates_editor.clips import _get_mediakit_recommendations mock_client = MagicMock() @@ -231,10 +212,90 @@ class TestGetMediakitRecommendations: result = _get_mediakit_recommendations(["a1"], mock_asset_repo) assert result == {} - # analyze_videos should not be called since no valid video URLs mock_client.analyze_videos.assert_not_called() +# ── _get_template_segments 单元测试 ───────────────────────────────────────── + + +class TestGetTemplateSegments: + """测试模板片段配置查询。""" + + def test_returns_segments_from_new_template_system(self): + """新模板系统（clip_configs）有数据时优先使用。""" + from app.api.routes.templates_editor.clips import _get_template_segments + + mock_tpl_svc = MagicMock() + cc1 = MagicMock() + cc1.order = 0 + cc1.min_duration = 3.0 + cc1.max_duration = 5.0 + cc2 = MagicMock() + cc2.order = 1 + cc2.min_duration = 4.0 + cc2.max_duration = 8.0 + mock_tpl_svc.list_clip_configs.return_value = [cc2, cc1] # 乱序返回 + + result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock()) + assert len(result) == 2 + assert result[0] == (0, 3.0, 5.0) + assert result[1] == (1, 4.0, 8.0) + + def test_falls_back_to_old_template_segments(self): + """新模板系统无数据时回退到旧系统。""" + from app.api.routes.templates_editor.clips import _get_template_segments + + mock_tpl_svc = MagicMock() + mock_tpl_svc.list_clip_configs.return_value = [] + + with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo: + mock_repo = MagicMock() + seg1 = MagicMock() + seg1.segment_order = 0 + seg1.duration_min = 2.0 + seg1.duration_max = 4.0 + mock_repo.list_segments.return_value = [seg1] + MockRepo.return_value = mock_repo + + result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock()) + assert len(result) == 1 + assert result[0] == (0, 2.0, 4.0) + + def test_returns_empty_when_no_segments(self): + """两套系统都没有片段配置时返回空列表。""" + from app.api.routes.templates_editor.clips import _get_template_segments + + mock_tpl_svc = MagicMock() + mock_tpl_svc.list_clip_configs.return_value = [] + + with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo: + mock_repo = MagicMock() + mock_repo.list_segments.return_value = [] + MockRepo.return_value = mock_repo + + result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock()) + assert result == [] + + def test_new_system_exception_falls_back(self): + """新模板系统异常时回退到旧系统。""" + from app.api.routes.templates_editor.clips import _get_template_segments + + mock_tpl_svc = MagicMock() + mock_tpl_svc.list_clip_configs.side_effect = RuntimeError("db error") + + with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo: + mock_repo = MagicMock() + seg = MagicMock() + seg.segment_order = 0 + seg.duration_min = 1.0 + seg.duration_max = 3.0 + mock_repo.list_segments.return_value = [seg] + MockRepo.return_value = mock_repo + + result = _get_template_segments("tmpl-1", mock_tpl_svc, MagicMock()) + assert len(result) == 1 + + # ── from-assets 端点集成测试 ──────────────────────────────────────────────── @@ -247,59 +308,307 @@ def _make_auth_user(): return auth -def _make_mock_clip(clip_id, order, duration, start_time=0.0, asset_id=""): - clip = MagicMock() - clip.id = clip_id - clip.plan_id = "plan-test" - clip.clip_type = "main" - clip.order = order - clip.duration = duration - clip.start_time = start_time - clip.text_content = "" - clip.transition_effect = "cut" - clip.transition_duration = 0.0 - clip.playback_speed = 1.0 - clip.config = {} - clip.asset_id = asset_id - clip.status = "pending" - clip.template_clip_config_id = "" - clip.created_at = None - clip.updated_at = None - return clip +def _make_clip_config(order, min_dur, max_dur): + cc = MagicMock() + cc.order = order + cc.min_duration = min_dur + cc.max_duration = max_dur + return cc -def _make_mock_asset(asset_id, duration): - asset = MagicMock() - asset.id = asset_id - asset.duration = duration - return asset - - -def _create_clips(plan_id, clip_type, order, duration=0.0, start_time=0.0, asset_id="", **kw): - return _make_mock_clip( - clip_id=f"clip-{order}", - order=order, - duration=duration, - start_time=start_time, - asset_id=asset_id, - ) - - -def _make_plan_svc(existing_clips=None): +def _make_plan_svc(replace_return_count=None): svc = MagicMock() svc.get_plan_or_raise = MagicMock() - svc.create_clip = MagicMock(side_effect=_create_clips) - svc.list_clips = MagicMock(return_value=existing_clips or []) + if replace_return_count is not None: + svc.replace_all_clips_transactional = MagicMock(return_value=replace_return_count) + else: + svc.replace_all_clips_transactional = MagicMock(return_value=0) return svc -class TestMediakitIntegrationInFromAssets: - """测试 from-assets 端点中 MediaKit 推荐的集成使用。""" +def _make_tpl_svc_with_segments(segments): + """segments: list of (order, min_dur, max_dur)""" + svc = MagicMock() + clip_configs = [_make_clip_config(o, mn, mx) for o, mn, mx in segments] + svc.list_clip_configs.return_value = clip_configs + return svc + + +def _make_rich_asset(asset_id, duration, storage_key="v.mp4", mime="video/mp4"): + asset = MagicMock() + asset.id = asset_id + asset.duration = duration + asset.storage_key = storage_key + asset.mime_type = mime + return asset + + +def _get_clips_data(mock_plan_svc): + """从 replace_all_clips_transactional 调用中提取 clips_data。""" + call_args = mock_plan_svc.replace_all_clips_transactional.call_args + if len(call_args.args) >= 2: + return call_args.args[1] + return call_args.kwargs.get("clips_data", []) + + +class TestFromAssetsByTemplateSegments: + """测试 from-assets 按模板 segment 创建片段（V2 事务性替换）。""" + + def test_creates_clips_matching_segment_count(self): + """片段数量 = segment 数量，忽略 required_clips_count。""" + from app.api.routes.templates_editor.clips import create_clips_from_assets_editor + from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + + segments = [(0, 3.0, 5.0), (1, 4.0, 8.0), (2, 2.0, 6.0), (3, 5.0, 10.0)] + mock_tpl_svc = _make_tpl_svc_with_segments(segments) + mock_plan_svc = _make_plan_svc(replace_return_count=4) + + mock_asset_repo = MagicMock() + mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0) + + body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=2) + result = create_clips_from_assets_editor( + template_id="tmpl-1", + body=body, + plan_id="plan-1", + services=(mock_tpl_svc, mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) + + assert result.created_count == 4 + clips_data = _get_clips_data(mock_plan_svc) + assert len(clips_data) == 4 + + def test_uses_transactional_replace(self): + """使用 replace_all_clips_transactional 而不是分别 delete + create。""" + from app.api.routes.templates_editor.clips import create_clips_from_assets_editor + from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + + segments = [(0, 3.0, 5.0), (1, 4.0, 8.0)] + mock_tpl_svc = _make_tpl_svc_with_segments(segments) + mock_plan_svc = _make_plan_svc(replace_return_count=2) + + mock_asset_repo = MagicMock() + mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0) + + body = ClipsFromAssetsRequest(asset_ids=["a1"]) + create_clips_from_assets_editor( + template_id="tmpl-1", + body=body, + plan_id="plan-1", + services=(mock_tpl_svc, mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) + + mock_plan_svc.replace_all_clips_transactional.assert_called_once() + assert not mock_plan_svc.delete_all_clips.called + assert not mock_plan_svc.create_clip.called + + def test_duration_within_segment_range(self): + """每个片段时长在 segment 的 min~max 范围内。""" + from app.api.routes.templates_editor.clips import create_clips_from_assets_editor + from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + + segments = [(0, 3.0, 5.0), (1, 4.0, 8.0)] + mock_tpl_svc = _make_tpl_svc_with_segments(segments) + mock_plan_svc = _make_plan_svc(replace_return_count=2) + + mock_asset_repo = MagicMock() + mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0) + + body = ClipsFromAssetsRequest(asset_ids=["a1"]) + create_clips_from_assets_editor( + template_id="tmpl-1", + body=body, + plan_id="plan-1", + services=(mock_tpl_svc, mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) + + clips_data = _get_clips_data(mock_plan_svc) + assert 3.0 <= clips_data[0]["duration"] <= 5.0 + assert 4.0 <= clips_data[1]["duration"] <= 8.0 + + def test_assets_round_robin_assignment(self): + """素材按片段顺序轮询分配。""" + from app.api.routes.templates_editor.clips import create_clips_from_assets_editor + from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + + segments = [(0, 3.0, 5.0), (1, 3.0, 5.0), (2, 3.0, 5.0), (3, 3.0, 5.0)] + mock_tpl_svc = _make_tpl_svc_with_segments(segments) + mock_plan_svc = _make_plan_svc(replace_return_count=4) + + mock_asset_repo = MagicMock() + + def get_asset(aid): + return _make_rich_asset(aid, 60.0) + + mock_asset_repo.get.side_effect = get_asset + + body = ClipsFromAssetsRequest(asset_ids=["a1", "a2"]) + create_clips_from_assets_editor( + template_id="tmpl-1", + body=body, + plan_id="plan-1", + services=(mock_tpl_svc, mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) + + clips_data = _get_clips_data(mock_plan_svc) + asset_ids = [c["asset_id"] for c in clips_data] + assert asset_ids == ["a1", "a2", "a1", "a2"] + + def test_orders_start_from_zero(self): + """片段 order 从 0 开始递增。""" + from app.api.routes.templates_editor.clips import create_clips_from_assets_editor + from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + + segments = [(0, 3.0, 5.0), (1, 3.0, 5.0), (2, 3.0, 5.0)] + mock_tpl_svc = _make_tpl_svc_with_segments(segments) + mock_plan_svc = _make_plan_svc(replace_return_count=3) + + mock_asset_repo = MagicMock() + mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0) + + body = ClipsFromAssetsRequest(asset_ids=["a1"]) + create_clips_from_assets_editor( + template_id="tmpl-1", + body=body, + plan_id="plan-1", + services=(mock_tpl_svc, mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) + + clips_data = _get_clips_data(mock_plan_svc) + orders = [c["order"] for c in clips_data] + assert orders == [0, 1, 2] + + def test_no_segments_raises_400(self): + """模板没有 segment 配置时返回 400。""" + from app.api.routes.templates_editor.clips import create_clips_from_assets_editor + from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + from fastapi import HTTPException + + mock_tpl_svc = MagicMock() + mock_tpl_svc.list_clip_configs.return_value = [] + mock_plan_svc = _make_plan_svc() + + with patch("app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository") as MockRepo: + mock_repo = MagicMock() + mock_repo.list_segments.return_value = [] + MockRepo.return_value = mock_repo + + body = ClipsFromAssetsRequest(asset_ids=["a1"]) + with pytest.raises(HTTPException) as exc_info: + create_clips_from_assets_editor( + template_id="tmpl-1", + body=body, + plan_id="plan-1", + services=(mock_tpl_svc, mock_plan_svc), + asset_repo=MagicMock(), + db=MagicMock(), + current_user=_make_auth_user(), + ) + assert exc_info.value.status_code == 400 + + mock_plan_svc.replace_all_clips_transactional.assert_not_called() + + def test_duration_capped_by_asset_duration(self): + """素材时长不足时 clip duration 被缩短。""" + from app.api.routes.templates_editor.clips import create_clips_from_assets_editor + from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + + segments = [(0, 10.0, 20.0)] + mock_tpl_svc = _make_tpl_svc_with_segments(segments) + mock_plan_svc = _make_plan_svc(replace_return_count=1) + + mock_asset_repo = MagicMock() + mock_asset_repo.get.return_value = _make_rich_asset("a1", 5.0) + + body = ClipsFromAssetsRequest(asset_ids=["a1"]) + create_clips_from_assets_editor( + template_id="tmpl-1", + body=body, + plan_id="plan-1", + services=(mock_tpl_svc, mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) + + clips_data = _get_clips_data(mock_plan_svc) + assert clips_data[0]["duration"] <= 5.0 + + def test_zero_duration_asset_raises_400(self): + """素材时长为 0 时抛出 400。""" + from app.api.routes.templates_editor.clips import create_clips_from_assets_editor + from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + from fastapi import HTTPException + + segments = [(0, 3.0, 5.0)] + mock_tpl_svc = _make_tpl_svc_with_segments(segments) + mock_plan_svc = _make_plan_svc() + + mock_asset_repo = MagicMock() + mock_asset_repo.get.return_value = _make_rich_asset("bad", 0.0) + + body = ClipsFromAssetsRequest(asset_ids=["bad"]) + with pytest.raises(HTTPException) as exc_info: + create_clips_from_assets_editor( + template_id="tmpl-1", + body=body, + plan_id="plan-1", + services=(mock_tpl_svc, mock_plan_svc), + asset_repo=mock_asset_repo, + db=MagicMock(), + current_user=_make_auth_user(), + ) + + assert exc_info.value.status_code == 400 + mock_plan_svc.replace_all_clips_transactional.assert_not_called() + + def test_empty_asset_ids_raises_400(self): + """asset_ids 为空列表时返回 400（defense-in-depth，schema 层也有 min_length=1）。""" + from app.api.routes.templates_editor.clips import create_clips_from_assets_editor + from fastapi import HTTPException + + segments = [(0, 3.0, 5.0)] + mock_tpl_svc = _make_tpl_svc_with_segments(segments) + mock_plan_svc = _make_plan_svc() + + # 用 MagicMock 模拟 body，绕过 Pydantic schema 的 min_length 校验 + mock_body = MagicMock() + mock_body.asset_ids = [] + mock_body.required_clips_count = None + + with pytest.raises(HTTPException) as exc_info: + create_clips_from_assets_editor( + template_id="tmpl-1", + body=mock_body, + plan_id="plan-1", + services=(mock_tpl_svc, mock_plan_svc), + asset_repo=MagicMock(), + db=MagicMock(), + current_user=_make_auth_user(), + ) + assert exc_info.value.status_code == 400 + assert "素材" in exc_info.value.detail + mock_plan_svc.replace_all_clips_transactional.assert_not_called() @patch("app.api.routes.templates_editor.clips.get_mediakit_client") @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_uses_mediakit_recommendation(self, mock_storage, mock_client_fn): - """MediaKit 推荐时间被优先使用。""" + def test_mediakit_first_clip_uses_recommendation(self, mock_storage, mock_client_fn): + """MediaKit 推荐时间用于每个素材的第一个片段。""" from app.api.routes.templates_editor.clips import create_clips_from_assets_editor from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest @@ -311,186 +620,71 @@ class TestMediakitIntegrationInFromAssets: mock_client_fn.return_value = mock_client mock_storage_svc = MagicMock() - mock_storage_svc.get_download_url.return_value = "https://example.com/test.mp4" + mock_storage_svc.get_download_url.return_value = "https://example.com/v.mp4" mock_storage.return_value = mock_storage_svc - mock_plan_svc = _make_plan_svc() + segments = [(0, 3.0, 5.0), (1, 3.0, 5.0)] + mock_tpl_svc = _make_tpl_svc_with_segments(segments) + mock_plan_svc = _make_plan_svc(replace_return_count=2) + mock_asset_repo = MagicMock() + mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0) - # asset needs storage_key and mime_type for MediaKit, plus duration for clip creation - asset_for_mediakit = MagicMock() - asset_for_mediakit.storage_key = "videos/test.mp4" - asset_for_mediakit.mime_type = "video/mp4" - asset_for_mediakit.duration = 60.0 - - def asset_get_side_effect(aid): - return _make_mock_asset(aid, 60.0) if aid else None - - mock_asset_repo.get = MagicMock(side_effect=asset_get_side_effect) - - # We need to make the asset have storage_key and mime_type for the mediakit function - # The mock_asset from _make_mock_asset doesn't have these, so let's use a richer mock - rich_asset = MagicMock() - rich_asset.id = "a1" - rich_asset.duration = 60.0 - rich_asset.storage_key = "videos/test.mp4" - rich_asset.mime_type = "video/mp4" - mock_asset_repo.get.return_value = rich_asset - - body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1) + body = ClipsFromAssetsRequest(asset_ids=["a1"]) create_clips_from_assets_editor( template_id="tmpl-1", body=body, - plan_id="plan-mk1", - services=(MagicMock(), mock_plan_svc), + plan_id="plan-1", + services=(mock_tpl_svc, mock_plan_svc), asset_repo=mock_asset_repo, + db=MagicMock(), current_user=_make_auth_user(), ) - create_calls = mock_plan_svc.create_clip.call_args_list - assert len(create_calls) == 1 - start_time = create_calls[0].kwargs.get("start_time") or create_calls[0][1].get("start_time") - assert start_time == 15.0 + clips_data = _get_clips_data(mock_plan_svc) + # 第一个片段应使用推荐时间 15.0 + assert clips_data[0]["start_time"] == 15.0 + # 第二个片段（同一素材）不应使用推荐时间 + assert clips_data[1]["start_time"] != 15.0 - @patch("app.api.routes.templates_editor.clips.get_mediakit_client") - @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_fallback_to_random_on_conflict(self, mock_storage, mock_client_fn): - """推荐时间与已有片段冲突时，降级为随机选择。""" - from app.api.routes.templates_editor.clips import create_clips_from_assets_editor - from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest - mock_client = MagicMock() - mock_client.is_available = True - # 推荐 2.0s，但已有片段占用了 [0, 10] - mock_client.analyze_videos.return_value = [ - '[{"asset_id": "a1", "recommended_start_time": 2.0, "reason": "test"}]' - ] - mock_client_fn.return_value = mock_client +# ── _safe_segment_duration 单元测试 ───────────────────────────────────────── - mock_storage_svc = MagicMock() - mock_storage_svc.get_download_url.return_value = "https://example.com/test.mp4" - mock_storage.return_value = mock_storage_svc - # 已有片段占用 [0, 10] - existing = [_make_mock_clip("c1", order=0, duration=10.0, start_time=0.0, asset_id="a1")] - mock_plan_svc = _make_plan_svc(existing_clips=existing) +class TestSafeSegmentDuration: + """测试片段时长安全转换。""" - rich_asset = MagicMock() - rich_asset.id = "a1" - rich_asset.duration = 60.0 - rich_asset.storage_key = "videos/test.mp4" - rich_asset.mime_type = "video/mp4" + def test_normal_float(self): + from app.api.routes.templates_editor.clips import _safe_segment_duration - mock_asset_repo = MagicMock() - mock_asset_repo.get.return_value = rich_asset + assert _safe_segment_duration(3.5, 1.0) == 3.5 - body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1) - create_clips_from_assets_editor( - template_id="tmpl-1", - body=body, - plan_id="plan-mk2", - services=(MagicMock(), mock_plan_svc), - asset_repo=mock_asset_repo, - current_user=_make_auth_user(), - ) + def test_none_returns_default(self): + from app.api.routes.templates_editor.clips import _safe_segment_duration - create_calls = mock_plan_svc.create_clip.call_args_list - assert len(create_calls) == 1 - start_time = create_calls[0].kwargs.get("start_time") or create_calls[0][1].get("start_time") - # 推荐时间 2.0 与 [0, 10] 冲突，应降级为随机，不应等于 2.0 - # 随机起始应在 [10, 55] 范围内（避开 [0,10]，5s clip 在 60s 素材中） - assert start_time is not None - assert start_time != 2.0 + assert _safe_segment_duration(None, 5.0) == 5.0 - @patch("app.api.routes.templates_editor.clips.get_mediakit_client") - @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_fallback_when_mediakit_unavailable(self, mock_storage, mock_client_fn): - """MediaKit 不可用时降级为随机选择，功能正常。""" - from app.api.routes.templates_editor.clips import create_clips_from_assets_editor - from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + def test_string_number(self): + from app.api.routes.templates_editor.clips import _safe_segment_duration - mock_client = MagicMock() - mock_client.is_available = False - mock_client_fn.return_value = mock_client + assert _safe_segment_duration("4.2", 1.0) == 4.2 - mock_plan_svc = _make_plan_svc() - mock_asset_repo = MagicMock() - mock_asset_repo.get.return_value = _make_mock_asset("a1", 30.0) + def test_invalid_string_returns_default(self): + from app.api.routes.templates_editor.clips import _safe_segment_duration - body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1) - result = create_clips_from_assets_editor( - template_id="tmpl-1", - body=body, - plan_id="plan-mk3", - services=(MagicMock(), mock_plan_svc), - asset_repo=mock_asset_repo, - current_user=_make_auth_user(), - ) + assert _safe_segment_duration("abc", 5.0) == 5.0 - assert result.created_count == 1 - # analyze_videos 不应被调用 - mock_client.analyze_videos.assert_not_called() + def test_negative_returns_default(self): + from app.api.routes.templates_editor.clips import _safe_segment_duration - @patch("app.api.routes.templates_editor.clips.get_mediakit_client") - @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_fallback_on_unparseable_response(self, mock_storage, mock_client_fn): - """MediaKit 返回不可解析内容时降级为随机选择。""" - from app.api.routes.templates_editor.clips import create_clips_from_assets_editor - from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest + assert _safe_segment_duration(-1.0, 5.0) == 5.0 - mock_client = MagicMock() - mock_client.is_available = True - mock_client.analyze_videos.return_value = ["这段视频内容很精彩，有很多好看的画面"] - mock_client_fn.return_value = mock_client + def test_zero_returns_default(self): + from app.api.routes.templates_editor.clips import _safe_segment_duration - mock_storage_svc = MagicMock() - mock_storage_svc.get_download_url.return_value = "https://example.com/test.mp4" - mock_storage.return_value = mock_storage_svc + assert _safe_segment_duration(0, 5.0) == 5.0 - mock_plan_svc = _make_plan_svc() - rich_asset = MagicMock() - rich_asset.id = "a1" - rich_asset.duration = 30.0 - rich_asset.storage_key = "videos/test.mp4" - rich_asset.mime_type = "video/mp4" - mock_asset_repo = MagicMock() - mock_asset_repo.get.return_value = rich_asset + def test_integer_value(self): + from app.api.routes.templates_editor.clips import _safe_segment_duration - body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1) - result = create_clips_from_assets_editor( - template_id="tmpl-1", - body=body, - plan_id="plan-mk4", - services=(MagicMock(), mock_plan_svc), - asset_repo=mock_asset_repo, - current_user=_make_auth_user(), - ) - - # 功能正常，降级为随机 - assert result.created_count == 1 - - @patch("app.api.routes.templates_editor.clips.get_mediakit_client") - @patch("app.api.routes.templates_editor.clips.get_storage_service") - def test_fallback_on_mediakit_exception(self, mock_storage, mock_client_fn): - """MediaKit 抛异常时优雅降级，不影响片段创建。""" - from app.api.routes.templates_editor.clips import create_clips_from_assets_editor - from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest - - mock_client_fn.side_effect = RuntimeError("MediaKit connection failed") - - mock_plan_svc = _make_plan_svc() - mock_asset_repo = MagicMock() - mock_asset_repo.get.return_value = _make_mock_asset("a1", 30.0) - - body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=1) - result = create_clips_from_assets_editor( - template_id="tmpl-1", - body=body, - plan_id="plan-mk5", - services=(MagicMock(), mock_plan_svc), - asset_repo=mock_asset_repo, - current_user=_make_auth_user(), - ) - - # 功能正常，降级为随机 - assert result.created_count == 1 + assert _safe_segment_duration(10, 1.0) == 10.0