diff --git a/apps/api/app/api/routes/templates_editor/clips.py b/apps/api/app/api/routes/templates_editor/clips.py index ee130f805..b7106b3eb 100755 --- a/apps/api/app/api/routes/templates_editor/clips.py +++ b/apps/api/app/api/routes/templates_editor/clips.py @@ -724,7 +724,12 @@ def create_clips_from_assets_editor( else: transition_compensation = 0.0 - for i, (_seg_order, dur_min, dur_max) in enumerate(segments): + # 打乱 segments 的处理顺序（分配素材的顺序随机化），但最终 clips_data 按原始 order 排序 + shuffled_indices = list(range(len(segments))) + random.shuffle(shuffled_indices) + + for idx in shuffled_indices: + _seg_order, dur_min, dur_max = segments[idx] # 在 segment 的 duration_min ~ duration_max 之间随机取值（保留一位小数） raw_duration = random.uniform(dur_min, dur_max) # 加上转场补偿，确保最终输出时长 = 模板设定总时长 @@ -805,7 +810,7 @@ def create_clips_from_assets_editor( clips_data.append( { - "order": i, + "order": _seg_order, "asset_id": asset_id, "start_time": start_time, "duration": clip_duration, @@ -813,6 +818,9 @@ def create_clips_from_assets_editor( } ) + # 按原始 segment order 排序，确保 clips_data 的 order 字段有序（0,1,2,3...） + clips_data.sort(key=lambda c: c["order"]) + # 4. 事务性替换：清空旧片段 → 创建新片段 → 标记ready（单事务，失败自动回滚） created_count = plan_svc.replace_all_clips_transactional(plan_id, clips_data) @@ -860,11 +868,58 @@ def create_clips_from_assets_editor( ) +def _build_scene_segments( + scene_changes: list[float], + asset_duration: float, +) -> list[tuple[float, float]]: + """根据场景切换点构建镜头段列表. + + Args: + scene_changes: 场景切换点时间戳列表（已排序，首位为 0.0） + asset_duration: 素材总时长 + + Returns: + 镜头段列表 [(start, end), ...] + """ + segments: list[tuple[float, float]] = [] + for i, ts in enumerate(scene_changes): + end = scene_changes[i + 1] if i + 1 < len(scene_changes) else asset_duration + # 只保留有效长度的镜头段（至少 0.5 秒） + if end - ts >= 0.5: + segments.append((ts, end)) + return segments + + +def _pick_start_in_scene_segment( + seg_start: float, + seg_end: float, + clip_duration: float, +) -> float | None: + """在镜头段内随机选取一个起始时间点. + + 确保 start + clip_duration <= seg_end。 + 若镜头段长度不足以容纳片段，返回 None。 + """ + available = seg_end - seg_start - clip_duration + if available < 0: + return None + max_start = seg_start + available + return random.uniform(seg_start, max_start) + + def _update_mediakit_recommendations_async( # pragma: no cover plan_id: str, asset_ids: list[str], ) -> None: - """后台任务：调用 MediaKit 智能选片并更新片段的起始时间. + """后台任务：使用 SceneChange 智能选帧并更新片段的起始时间. + + 优先使用 SceneChange 策略检测视频镜头切换点，将每个素材按镜头段拆分， + 各片段优先从不同镜头段中选取起始时间，实现「不同片段展示不同场景」的效果。 + + 降级策略： + 1. SceneChange 优先 → detect_scene_changes 内部已含 TimeInterval 降级 + 2. 若 detect_scene_changes 仍返回 None → 回退到旧的 analyze_videos 方式 + 3. 所有方式都失败 → 保持现有随机 start_time，不影响视频生成 此函数在后台异步执行，不影响接口响应时间。 失败时静默处理，不影响已创建的片段。 @@ -886,12 +941,6 @@ def _update_mediakit_recommendations_async( # pragma: no cover asset_repo = SQLAlchemyAssetRepository(db) plan_svc = EditPlanService(db) - # 调用 MediaKit 获取推荐时间 - recommendations = _get_mediakit_recommendations(asset_ids, asset_repo) - if not recommendations: - logger.info("后台任务: MediaKit 无推荐结果，跳过更新") - return - # 查询该 plan 的所有片段（分批获取，避免硬编码 limit 截断） batch_size = 500 all_clips = [] @@ -914,15 +963,16 @@ def _update_mediakit_recommendations_async( # pragma: no cover unique_asset_ids = list({getattr(c, "asset_id", "") or "" for c in clips} - {""}) assets_map: dict[str, object] = {a.id: a for a in asset_repo.find_by_ids(unique_asset_ids)} - # 按 asset_id 预分组片段时间段（消除 O(N^2) 嵌套循环） - clips_by_asset: dict[str, list[tuple[str, float, float]]] = defaultdict(list) + # 按 asset_id 预分组片段对象（按 order 排序，保证按模板顺序分配镜头段） + clips_by_asset: dict[str, list] = defaultdict(list) for clip in clips: aid = getattr(clip, "asset_id", "") or "" - if aid and clip.start_time is not None: - clips_by_asset[aid].append((clip.id, clip.start_time, clip.start_time + clip.duration)) + if aid: + clips_by_asset[aid].append(clip) + for aid in clips_by_asset: + clips_by_asset[aid].sort(key=lambda c: c.order) - # 读取素材全部历史已用区间（跨任务/跨 plan 持久化记录）： - # MediaKit 挪点必须与随机选片一样避让历史区间，否则会把片段挪回已用过的画面 + # 读取素材全部历史已用区间（跨任务/跨 plan 持久化记录） historical_segments = get_used_segments(db, unique_asset_ids) # 已更新的片段ID（用于排除已移动的旧时间段） @@ -931,16 +981,22 @@ def _update_mediakit_recommendations_async( # pragma: no cover updated_segments: dict[str, list[tuple[float, float]]] = {} updated_count = 0 - # 遍历片段，按 asset_id 匹配推荐时间 - for clip in clips: - asset_id = getattr(clip, "asset_id", "") or "" - if not asset_id or asset_id not in recommendations: + # 尝试获取存储服务（用于生成视频 URL） + try: + storage = get_storage_service() + except Exception: + logger.warning("后台任务: 获取存储服务失败，跳过 SceneChange 更新") + return + + # 获取 MediaKit 客户端 + client = get_mediakit_client() + + # 对每个素材，检测场景切换点并分配镜头段 + for asset_id in unique_asset_ids: + asset_clips = clips_by_asset.get(asset_id, []) + if not asset_clips: continue - recommended_start = recommendations[asset_id] - clip_duration = clip.duration - - # 从预加载字典获取素材（O(1) 查找） asset = assets_map.get(asset_id) if not asset: continue @@ -948,89 +1004,143 @@ def _update_mediakit_recommendations_async( # pragma: no cover if asset_total <= 0: continue - # 推荐时间 + 片段时长不能超过素材总时长 - if recommended_start + clip_duration > asset_total: - logger.info( - "后台任务: 推荐时间越界，跳过: asset_id=%s recommended=%.2f duration=%.1f total=%.1f", - asset_id, - recommended_start, - clip_duration, - asset_total, - ) - continue - - # 构建排除当前片段及已更新片段后的占用列表（O(M)，M=同素材片段数） - other_segments: list[tuple[float, float]] = [ - (cs, ce) - for cid, cs, ce in clips_by_asset.get(asset_id, []) - if cid != clip.id and cid not in updated_clip_ids - ] - other_segments.extend(updated_segments.get(asset_id, [])) - - # 并入该素材全部历史已用区间（含其他 plan/其他任务），set 去重： - # 本 plan 片段创建时已写入历史记录 - # 并入该素材全部历史已用区间（含其他 plan/其他任务）。 - # set 去重前先归一化精度（round 3 位），避免浮点尾差导致逻辑相同的 - # 区间（如 1.0 与 1.0000000001）被误判为不同区间 - def _norm(segs): - return {(round(float(a), 3), round(float(b), 3)) for a, b in segs} - - other_segments = list(_norm(other_segments) | _norm(historical_segments.get(asset_id, []))) - - # 检查推荐时间是否与同 plan 片段或历史已用区间冲突（含 0.3s 边缘间隙）： - # 冲突时放弃该推荐、保留原随机起点（不硬挪到已用过的画面） - if _recommended_time_conflicts(recommended_start, clip_duration, other_segments): - logger.info( - "后台任务: 推荐时间与同片/历史区间冲突，保留原起点: asset_id=%s recommended=%.2f", - asset_id, - recommended_start, - ) - continue - - # 逐个更新并捕获异常（单点失败不影响其他片段） - try: - old_start = clip.start_time - old_end = old_start + clip_duration - # MediaKit 移动片段起点 + 同步素材 metadata 区间记录放在同一事务： - # 删旧区间记录（按 plan_id + 旧 start 匹配，兼容无 plan_id 的旧数据）、 - # 写新区间，最后统一 commit；任一步失败整体 rollback， - # 保证 clip.start_time 与 metadata.used_time_ranges 不出现不一致。 - plan_svc.update_clip(clip.id, start_time=recommended_start) + # 获取素材视频 URL + video_url: str | None = None + storage_key = getattr(asset, "storage_key", None) or "" + mime = getattr(asset, "mime_type", "") or "" + if storage_key and mime.startswith("video/"): try: - if remove_used_segment(db, asset_id, old_start, old_end, plan_id=plan_id): - record_used_segments( - db, - asset_id, - recommended_start, - recommended_start + clip_duration, - plan_id, - ) - except Exception as me: - logger.warning( - "后台任务: 同步素材区间记录失败，回滚本次片段更新: clip_id=%s error=%s", - clip.id, - me, + video_url = storage.get_download_url(storage_key) + except Exception as e: + logger.warning("后台任务: 获取素材URL失败: asset_id=%s error=%s", asset_id, e) + + # 构建该素材的占用区间列表（排除已更新片段） + def _get_other_segments(asset_id_inner, clip_id_inner): + segs: list[tuple[float, float]] = [] + for c in clips_by_asset.get(asset_id_inner, []): + cid = c.id + if cid != clip_id_inner and cid not in updated_clip_ids: + segs.append((c.start_time, c.start_time + c.duration)) + segs.extend(updated_segments.get(asset_id_inner, [])) + # 并入历史已用区间 + def _norm(segs_in): + return {(round(float(a), 3), round(float(b), 3)) for a, b in segs_in} + return list(_norm(segs) | _norm(historical_segments.get(asset_id_inner, []))) + + # 优先使用 SceneChange 策略 + scene_segments: list[tuple[float, float]] = [] + if client.is_available and video_url: + scene_changes = client.detect_scene_changes(video_url) + if scene_changes is not None: + scene_segments = _build_scene_segments(scene_changes, asset_total) + logger.info( + "后台任务: 素材场景检测完成: asset_id=%s scenes=%d", + asset_id, len(scene_segments), ) - db.rollback() - continue - db.commit() - updated_count += 1 - updated_clip_ids.add(clip.id) - except Exception as ue: - logger.warning("后台任务: 单个片段更新失败: clip_id=%s error=%s", clip.id, ue) - try: - db.rollback() - except Exception: - pass + + # SceneChange 未获得有效结果 → 尝试 analyze_videos 作为 fallback + if not scene_segments and video_url: + fallback_recs = _get_mediakit_recommendations([asset_id], asset_repo) + if fallback_recs and asset_id in fallback_recs: + # analyze_videos 只返回单个推荐点，转为单镜头段 + rec_start = fallback_recs[asset_id] + scene_segments = [(rec_start, asset_total)] + logger.info( + "后台任务: 使用 analyze_videos fallback: asset_id=%s start=%.2f", + asset_id, rec_start, + ) + + if not scene_segments: + # 所有方式都失败 → 保持现有随机 start_time + logger.info( + "后台任务: SceneChange 与 analyze_videos 均无结果，保持随机起点: asset_id=%s", + asset_id, + ) continue - updated_segments.setdefault(asset_id, []).append((recommended_start, recommended_start + clip_duration)) - logger.info( - "后台任务: 更新片段起始时间: clip_id=%s asset_id=%s start_time=%.2f", - clip.id, - asset_id, - recommended_start, - ) + # 为每个片段分配不同的镜头段 + scene_segments_pool = list(scene_segments) # 可消费的镜头段池 + for clip in asset_clips: + clip_duration = clip.duration + recommended_start: float | None = None + + # 从镜头段池中依次尝试，选一个不冲突的 + for seg_idx, (seg_start, seg_end) in enumerate(scene_segments_pool): + candidate_start = _pick_start_in_scene_segment(seg_start, seg_end, clip_duration) + if candidate_start is None: + continue # 镜头段太短，跳过 + + # 检查越界 + if candidate_start + clip_duration > asset_total: + continue + + # 检查与已用区间冲突 + other_segs = _get_other_segments(asset_id, clip.id) + if _recommended_time_conflicts(candidate_start, clip_duration, other_segs): + continue + + recommended_start = candidate_start + # 消费该镜头段（从池中移除，下一个片段用不同镜头段） + scene_segments_pool.pop(seg_idx) + break + + if recommended_start is None: + # 镜头段用完或都冲突 → 尝试 _calc_random_start_time 兜底 + used_segs_for_calc: dict[str, list[tuple[float, float]]] = { + asset_id: _get_other_segments(asset_id, clip.id) + } + fallback_start = _calc_random_start_time( + asset_id, + clip_duration, + {asset_id: asset_total}, + used_segs_for_calc, + ) + if fallback_start is None: + continue # 完全无法分配，保持原起点 + recommended_start = fallback_start + + # 更新片段起始时间 + try: + old_start = clip.start_time + old_end = old_start + clip_duration + + plan_svc.update_clip(clip.id, start_time=recommended_start) + try: + if remove_used_segment(db, asset_id, old_start, old_end, plan_id=plan_id): + record_used_segments( + db, + asset_id, + recommended_start, + recommended_start + clip_duration, + plan_id, + ) + except Exception as me: + logger.warning( + "后台任务: 同步素材区间记录失败，回滚本次片段更新: clip_id=%s error=%s", + clip.id, + me, + ) + db.rollback() + continue + db.commit() + updated_count += 1 + updated_clip_ids.add(clip.id) + updated_segments.setdefault(asset_id, []).append( + (recommended_start, recommended_start + clip_duration) + ) + logger.info( + "后台任务: 更新片段起始时间(场景选帧): clip_id=%s asset_id=%s start_time=%.2f", + clip.id, + asset_id, + recommended_start, + ) + except Exception as ue: + logger.warning("后台任务: 单个片段更新失败: clip_id=%s error=%s", clip.id, ue) + try: + db.rollback() + except Exception: + pass + continue logger.info("后台任务完成: plan_id=%s 成功更新 %d 个片段", plan_id, updated_count) diff --git a/packages/shared/mediakit_client.py b/packages/shared/mediakit_client.py index 94557cbe5..e18228ec9 100755 --- a/packages/shared/mediakit_client.py +++ b/packages/shared/mediakit_client.py @@ -119,6 +119,72 @@ class MediaKitClient: return None + def detect_scene_changes( + self, + video_url: str, + max_frames: int = 20, + poll_interval: float = 2.0, + max_poll_attempts: int = 30, + ) -> Optional[List[float]]: + """检测视频场景切换点，返回时间戳列表. + + 降级策略： + 1. 先尝试 SceneChange 策略 + 2. SceneChange 失败（OOM等）→ 退回 TimeInterval（5秒间隔） + 3. MediaKit 不可用 → 返回 None + + Returns: + 场景切换点时间戳列表，如 [0.0, 3.2, 7.8, 12.5] + 失败返回 None + """ + if not self.is_available: + logger.warning("MediaKit 未配置，跳过场景检测") + return None + + # 策略1：尝试 SceneChange + frames = self.extract_frames( + video_url=video_url, + strategy="SceneChange", + max_frames=max_frames, + poll_interval=poll_interval, + max_poll_attempts=max_poll_attempts, + ) + + # 策略2：SceneChange 失败 → 退回 TimeInterval（5秒间隔） + if frames is None: + logger.info("SceneChange 策略失败，降级为 TimeInterval（5秒间隔）") + # 估算帧数：假设视频最长60秒，每5秒一帧 + ti_max_frames = max(max_frames, 12) + frames = self.extract_frames( + video_url=video_url, + strategy="TimeInterval", + max_frames=ti_max_frames, + poll_interval=poll_interval, + max_poll_attempts=max_poll_attempts, + ) + + if frames is None: + return None + + # 从帧列表中提取 timestamp，排序 + timestamps = sorted( + {float(f.get("timestamp", 0.0)) for f in frames if "timestamp" in f} + ) + + if not timestamps: + return None + + # 始终在列表开头加 0.0（素材起始点） + if timestamps[0] != 0.0: + timestamps.insert(0, 0.0) + + logger.info( + "场景检测完成: video_url=%s scene_changes=%s", + video_url[:80], + timestamps, + ) + return timestamps + def _submit_extract_task( self, video_url: str, diff --git a/tests/unit/test_editor_clips_random_start.py b/tests/unit/test_editor_clips_random_start.py index d9d902c92..fb101e1cb 100644 --- a/tests/unit/test_editor_clips_random_start.py +++ b/tests/unit/test_editor_clips_random_start.py @@ -418,8 +418,13 @@ class TestEditorClipsDurationAndStartTime: 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 + # clips_data 按 order 排序，但分配顺序因 shuffle 而随机， + # 因此只验证两个 start_time 值都存在 + start_times = {c["start_time"] for c in clips_data} + assert start_times == {12.5, 18.0} + # 验证 order 仍然有序 + orders = [c["order"] for c in clips_data] + assert orders == sorted(orders) @patch("app.api.routes.templates_editor.clips.get_storage_service") def test_asset_durations_deduped(self, mock_storage): diff --git a/tests/unit/test_scene_change_shuffle.py b/tests/unit/test_scene_change_shuffle.py new file mode 100644 index 000000000..291789289 --- /dev/null +++ b/tests/unit/test_scene_change_shuffle.py @@ -0,0 +1,262 @@ +"""Tests for scene-change smart frame selection + random shuffle of segment processing.""" + +from __future__ import annotations + +import random +from unittest.mock import MagicMock, patch + +import pytest + +from apps.api.app.api.routes.templates_editor.clips import ( + _build_scene_segments, + _pick_start_in_scene_segment, +) +from packages.shared.mediakit_client import MediaKitClient + + +# ── Part 1: Random shuffle tests ────────────────────────────────────────── + + +class TestRandomShuffle: + """验证 segments 处理顺序随机打乱逻辑.""" + + def test_same_segments_produce_different_asset_orders(self): + """同一批 segments 多次处理，asset 分配顺序有变化. + + 模拟打乱后的处理顺序，验证多次运行中 asset_id 分配顺序 + 存在差异（概率性验证，运行 50 次应该至少出现 2 种排列）。 + """ + segments = [(0, 3.0, 5.0), (1, 4.0, 6.0), (2, 3.0, 5.0), (3, 4.0, 6.0)] + asset_ids = ["A", "B", "C", "D"] + + observed_orders: list[tuple] = set() + + for _ in range(50): + shuffled_indices = list(range(len(segments))) + random.shuffle(shuffled_indices) + order_tuple = tuple(shuffled_indices) + observed_orders.add(order_tuple) + + # 50 次打乱，4! = 24 种排列，应出现多种不同排列 + assert len(observed_orders) > 1, "打乱应该产生多种不同顺序" + + def test_clips_data_order_always_sorted(self): + """clips_data 按 order 排序后始终有序. + + 模拟打乱处理后 clips_data 按 order 排序，验证最终 order 为 [0,1,2,3]。 + """ + segments = [(0, 3.0, 5.0), (1, 4.0, 6.0), (2, 3.0, 5.0), (3, 4.0, 6.0)] + + for _ in range(20): + shuffled_indices = list(range(len(segments))) + random.shuffle(shuffled_indices) + + # 模拟构建 clips_data（用 _seg_order 作为 order） + clips_data = [] + for idx in shuffled_indices: + seg_order, _, _ = segments[idx] + clips_data.append({"order": seg_order, "asset_id": f"asset_{idx}"}) + + # 按 order 排序 + clips_data.sort(key=lambda c: c["order"]) + + # 验证 order 始终有序 + orders = [c["order"] for c in clips_data] + assert orders == [0, 1, 2, 3], f"排序后 order 应为 [0,1,2,3]，实际为 {orders}" + + +# ── Part 2: detect_scene_changes tests ──────────────────────────────────── + + +class TestDetectSceneChanges: + """验证 MediaKitClient.detect_scene_changes 方法.""" + + def _make_client(self) -> MediaKitClient: + """创建一个可用的 MediaKitClient（mock 配置）.""" + with patch( + "packages.shared.mediakit_client.get_shared_settings" + ) as mock_settings: + mock_settings.return_value.mediakit_api_key = "test-key" + mock_settings.return_value.mediakit_base_url = "http://test" + mock_settings.return_value.mediakit_timeout = 30 + client = MediaKitClient() + return client + + def test_scene_change_success(self): + """SceneChange 策略成功返回时间戳列表.""" + client = self._make_client() + + mock_frames = [ + {"image_url": "url1", "timestamp": 0.0}, + {"image_url": "url2", "timestamp": 3.2}, + {"image_url": "url3", "timestamp": 7.8}, + {"image_url": "url4", "timestamp": 12.5}, + ] + + with patch.object(client, "extract_frames", return_value=mock_frames): + result = client.detect_scene_changes("https://example.com/video.mp4") + + assert result is not None + assert result[0] == 0.0 # 始终以 0.0 开头 + assert 3.2 in result + assert 7.8 in result + assert 12.5 in result + assert result == sorted(result) # 应已排序 + + def test_scene_change_fallback_to_time_interval(self): + """SceneChange 失败降级到 TimeInterval 策略.""" + client = self._make_client() + + # 第一次调用（SceneChange）返回 None，第二次（TimeInterval）返回结果 + fallback_frames = [ + {"image_url": "url1", "timestamp": 0.0}, + {"image_url": "url2", "timestamp": 5.0}, + {"image_url": "url3", "timestamp": 10.0}, + ] + + call_count = 0 + + def side_effect(*args, **kwargs): + nonlocal call_count + call_count += 1 + if call_count == 1: + # 第一次 SceneChange 失败 + return None + else: + # 第二次 TimeInterval 成功 + assert kwargs.get("strategy") == "TimeInterval" + return fallback_frames + + with patch.object(client, "extract_frames", side_effect=side_effect): + result = client.detect_scene_changes("https://example.com/video.mp4") + + assert result is not None + assert result[0] == 0.0 + assert 5.0 in result + assert 10.0 in result + + def test_mediakit_not_available_returns_none(self): + """MediaKit 不可用时返回 None.""" + with patch( + "packages.shared.mediakit_client.get_shared_settings" + ) as mock_settings: + mock_settings.return_value.mediakit_api_key = "" # 未配置 + mock_settings.return_value.mediakit_base_url = "http://test" + mock_settings.return_value.mediakit_timeout = 30 + client = MediaKitClient() + + result = client.detect_scene_changes("https://example.com/video.mp4") + assert result is None + + def test_both_strategies_fail_returns_none(self): + """SceneChange 和 TimeInterval 都失败时返回 None.""" + client = self._make_client() + + with patch.object(client, "extract_frames", return_value=None): + result = client.detect_scene_changes("https://example.com/video.mp4") + + assert result is None + + def test_prepends_zero_if_not_present(self): + """若帧列表中不包含 0.0，自动在开头添加.""" + client = self._make_client() + + # 帧列表中没有 timestamp=0.0 + mock_frames = [ + {"image_url": "url1", "timestamp": 2.0}, + {"image_url": "url2", "timestamp": 5.5}, + ] + + with patch.object(client, "extract_frames", return_value=mock_frames): + result = client.detect_scene_changes("https://example.com/video.mp4") + + assert result is not None + assert result[0] == 0.0 + assert 2.0 in result + assert 5.5 in result + + +# ── Part 2.2: Scene segment building and assignment ─────────────────────── + + +class TestSceneSegments: + """验证镜头段构建和分配逻辑.""" + + def test_build_scene_segments(self): + """从场景切换点正确构建镜头段.""" + scene_changes = [0.0, 3.2, 7.8, 12.5] + asset_duration = 15.0 + + segments = _build_scene_segments(scene_changes, asset_duration) + + assert len(segments) == 4 + assert segments[0] == (0.0, 3.2) + assert segments[1] == (3.2, 7.8) + assert segments[2] == (7.8, 12.5) + assert segments[3] == (12.5, 15.0) + + def test_build_scene_segments_filters_short(self): + """过滤掉过短的镜头段（< 0.5秒）.""" + scene_changes = [0.0, 0.1, 5.0, 5.3, 10.0] + asset_duration = 12.0 + + segments = _build_scene_segments(scene_changes, asset_duration) + + # (0.0, 0.1) 长度 0.1 < 0.5 → 过滤 + # (0.1, 5.0) → 保留 + # (5.0, 5.3) 长度 0.3 < 0.5 → 过滤 + # (5.3, 10.0) → 保留 + # (10.0, 12.0) → 保留 + assert len(segments) == 3 + assert segments[0] == (0.1, 5.0) + assert segments[1] == (5.3, 10.0) + assert segments[2] == (10.0, 12.0) + + def test_pick_start_in_segment(self): + """在镜头段内随机选取起始时间.""" + seg_start = 3.0 + seg_end = 8.0 + clip_duration = 2.0 + + starts = set() + for _ in range(100): + start = _pick_start_in_scene_segment(seg_start, seg_end, clip_duration) + assert start is not None + assert seg_start <= start <= seg_end - clip_duration + starts.add(round(start, 2)) + + # 应该有多个不同的起始时间 + assert len(starts) > 1 + + def test_pick_start_segment_too_short(self): + """镜头段太短无法容纳片段时返回 None.""" + result = _pick_start_in_scene_segment(0.0, 1.0, 2.0) + assert result is None + + def test_different_clips_from_different_scenes(self): + """不同片段应来自不同的镜头段（模拟分配逻辑）.""" + scene_changes = [0.0, 5.0, 10.0, 15.0] + asset_duration = 18.0 + clip_duration = 3.0 + + segments = _build_scene_segments(scene_changes, asset_duration) + assert len(segments) == 4 # (0,5), (5,10), (10,15), (15,18) + + # 模拟 3 个片段从不同镜头段取点 + scene_pool = list(segments) + assigned_starts = [] + + for _ in range(3): + if not scene_pool: + break + seg_start, seg_end = scene_pool.pop(0) + start = _pick_start_in_scene_segment(seg_start, seg_end, clip_duration) + assert start is not None + assigned_starts.append(start) + + # 3 个片段分别从 3 个不同镜头段中选取 + assert len(assigned_starts) == 3 + # 第一个来自 [0, 2]，第二个来自 [5, 7]，第三个来自 [10, 12] + assert 0.0 <= assigned_starts[0] <= 2.0 + assert 5.0 <= assigned_starts[1] <= 7.0 + assert 10.0 <= assigned_starts[2] <= 12.0