diff --git a/apps/api/app/services/plan_generator_service.py b/apps/api/app/services/plan_generator_service.py index 07d1ec2bf..28dc0a94d 100755 --- a/apps/api/app/services/plan_generator_service.py +++ b/apps/api/app/services/plan_generator_service.py @@ -49,9 +49,10 @@ class PlanGeneratorService: 基于模板 + 素材,自动生成 EditPlan 及 EditPlanClip 列表。 """ - def __init__(self, db: Session) -> None: + def __init__(self, db: Session, asset_repo=None) -> None: self._plan_repo = SQLAlchemyEditPlanRepository(db) self._clip_repo = SQLAlchemyEditPlanClipRepository(db) + self._asset_repo = asset_repo # ── 公开接口 ───────────────────────────────────────────────────────────── @@ -64,6 +65,7 @@ class PlanGeneratorService: project_id: str = "", created_by_user_id: str = "", name: str = "", + random_preview: bool = False, ) -> dict[str, Any]: """基于模板+素材生成剪辑计划 @@ -74,6 +76,7 @@ class PlanGeneratorService: project_id: 所属项目 ID created_by_user_id: 创建者用户 ID name: 计划名称(为空则自动取模板名) + random_preview: 是否启用随机预览模式(随机选素材+随机截取片段) Returns: dict: {"plan": EditPlan, "clips": List[EditPlanClip]} @@ -115,7 +118,17 @@ class PlanGeneratorService: # 4. 按 editing_mode 分配素材 if asset_ids: - self._distribute_assets(clips, asset_ids, editing_mode) + # 如果是随机预览模式,获取素材时长信息 + asset_durations = None + if random_preview and self._asset_repo: + asset_durations = self._fetch_asset_durations(asset_ids) + self._distribute_assets( + clips, + asset_ids, + editing_mode, + random_selection=random_preview, + asset_durations=asset_durations, + ) # 5. 持久化所有 clips 并计算总时长 created_clips: List[EditPlanClip] = [] @@ -199,9 +212,34 @@ class PlanGeneratorService: clips: List[EditPlanClip], asset_ids: List[str], editing_mode: str, + *, + random_selection: bool = False, + asset_durations: dict[str, float] | None = None, ) -> None: """按 editing_mode 将素材分配到 clips(就地修改,未持久化). 委托给 plan_generator_utils.distribute_assets 纯函数。 """ - distribute_assets(clips, asset_ids, editing_mode) + distribute_assets( + clips, + asset_ids, + editing_mode, + random_selection=random_selection, + asset_durations=asset_durations, + ) + + def _fetch_asset_durations(self, asset_ids: List[str]) -> dict[str, float]: + """从数据库获取素材时长信息. + + Args: + asset_ids: 素材 ID 列表 + + Returns: + dict: 素材 ID -> 时长(秒)映射 + """ + durations: dict[str, float] = {} + for asset_id in asset_ids: + asset = self._asset_repo.get(asset_id) + if asset and hasattr(asset, "duration"): + durations[asset_id] = float(asset.duration or 0.0) + return durations diff --git a/packages/domain/edit_plan_clip.py b/packages/domain/edit_plan_clip.py index b5da75911..a913b988a 100755 --- a/packages/domain/edit_plan_clip.py +++ b/packages/domain/edit_plan_clip.py @@ -127,11 +127,18 @@ class EditPlanClip: config=config or {}, ) - def assign_asset(self, asset_id: str) -> None: - """分配素材""" + def assign_asset(self, asset_id: str, *, start_time: float | None = None) -> None: + """分配素材 + + Args: + asset_id: 素材 ID + start_time: 可选,素材播放起始时间(秒)。如果提供且在有效范围内,则设置;否则保持默认 0.0 + """ if not asset_id.strip(): raise ValueError("asset_id 不能为空") self.asset_id = asset_id.strip() + if start_time is not None and start_time >= 0: + self.start_time = start_time self.updated_at = datetime.now(timezone.utc) def mark_ready(self) -> None: diff --git a/packages/domain/plan_generator_utils.py b/packages/domain/plan_generator_utils.py index 155a38516..cbaa0f4c1 100755 --- a/packages/domain/plan_generator_utils.py +++ b/packages/domain/plan_generator_utils.py @@ -11,6 +11,7 @@ from __future__ import annotations +import random from typing import List from packages.domain.edit_plan_clip import EditPlanClip @@ -30,6 +31,9 @@ def distribute_assets( clips: List[EditPlanClip], asset_ids: List[str], editing_mode: str, + *, + random_selection: bool = False, + asset_durations: dict[str, float] | None = None, ) -> None: """按 editing_mode 将素材分配到 clips(就地修改). @@ -43,66 +47,85 @@ def distribute_assets( clips: 剪辑片段列表(就地修改 asset_id) asset_ids: 素材 ID 列表 editing_mode: 剪辑模式字符串 + random_selection: 是否随机选择素材(用于预览生成) + asset_durations: 素材 ID -> 时长(秒)映射,用于设置随机 start_time """ if not asset_ids or not clips: return + # 如果需要随机选择,先打乱素材顺序 + if random_selection: + asset_ids = list(asset_ids) # 复制避免修改原列表 + random.shuffle(asset_ids) + if editing_mode == EditingMode.ONE_TAKE.value: - _distribute_one_take(clips, asset_ids) + _distribute_one_take(clips, asset_ids, asset_durations) elif editing_mode == EditingMode.PIP.value: - _distribute_pip(clips, asset_ids) + _distribute_pip(clips, asset_ids, asset_durations) elif editing_mode == EditingMode.VOICE_OVER.value: - _distribute_voice_over(clips, asset_ids) + _distribute_voice_over(clips, asset_ids, asset_durations) elif editing_mode == EditingMode.VOICE_PIP.value: - _distribute_voice_pip(clips, asset_ids) + _distribute_voice_pip(clips, asset_ids, asset_durations) else: # 未知模式,退化为 one_take - _distribute_one_take(clips, asset_ids) + _distribute_one_take(clips, asset_ids, asset_durations) def _distribute_one_take( clips: List[EditPlanClip], asset_ids: List[str], + asset_durations: dict[str, float] | None = None, ) -> None: """ONE_TAKE: 素材按顺序依次分配给 main 类型 clips.""" main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value] for i, clip in enumerate(main_clips): if i < len(asset_ids): - clip.assign_asset(asset_ids[i]) + asset_id = asset_ids[i] + start_time = _calc_random_start_time(asset_id, clip.duration, asset_durations) + clip.assign_asset(asset_id, start_time=start_time) def _distribute_pip( clips: List[EditPlanClip], asset_ids: List[str], + asset_durations: dict[str, float] | None = None, ) -> None: """PIP: 第1个素材→main(全屏背景),其余→overlay clips.""" # 第1个素材 → main clip main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value] if main_clips and asset_ids: - main_clips[0].assign_asset(asset_ids[0]) + asset_id = asset_ids[0] + start_time = _calc_random_start_time(asset_id, main_clips[0].duration, asset_durations) + main_clips[0].assign_asset(asset_id, start_time=start_time) # 其余素材 → overlay clips overlay_clips = [c for c in clips if c.clip_type == "overlay"] remaining = asset_ids[1:] for i, clip in enumerate(overlay_clips): if i < len(remaining): - clip.assign_asset(remaining[i]) + asset_id = remaining[i] + start_time = _calc_random_start_time(asset_id, clip.duration, asset_durations) + clip.assign_asset(asset_id, start_time=start_time) def _distribute_voice_over( clips: List[EditPlanClip], asset_ids: List[str], + asset_durations: dict[str, float] | None = None, ) -> None: """VOICE_OVER: 素材→main clips (B-roll).""" main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value] for i, clip in enumerate(main_clips): if i < len(asset_ids): - clip.assign_asset(asset_ids[i]) + asset_id = asset_ids[i] + start_time = _calc_random_start_time(asset_id, clip.duration, asset_durations) + clip.assign_asset(asset_id, start_time=start_time) def _distribute_voice_pip( clips: List[EditPlanClip], asset_ids: List[str], + asset_durations: dict[str, float] | None = None, ) -> None: """VOICE_PIP: 第1个→background, 第2个→corner_voice, 其余→b_roll.""" bg_clips = [c for c in clips if c.clip_type == "background"] @@ -113,19 +136,61 @@ def _distribute_voice_pip( # 第1个 → background if idx < len(asset_ids) and bg_clips: - bg_clips[0].assign_asset(asset_ids[idx]) + asset_id = asset_ids[idx] + start_time = _calc_random_start_time(asset_id, bg_clips[0].duration, asset_durations) + bg_clips[0].assign_asset(asset_id, start_time=start_time) idx += 1 # 第2个 → corner_voice if idx < len(asset_ids) and voice_clips: - voice_clips[0].assign_asset(asset_ids[idx]) + asset_id = asset_ids[idx] + start_time = _calc_random_start_time(asset_id, voice_clips[0].duration, asset_durations) + voice_clips[0].assign_asset(asset_id, start_time=start_time) idx += 1 # 剩余 → b_roll clips remaining = asset_ids[idx:] for i, clip in enumerate(broll_clips): if i < len(remaining): - clip.assign_asset(remaining[i]) + asset_id = remaining[i] + start_time = _calc_random_start_time(asset_id, clip.duration, asset_durations) + clip.assign_asset(asset_id, start_time=start_time) + + +# ── 随机 start_time 计算 ──────────────────────────────────────────────────── + + +def _calc_random_start_time( + asset_id: str, + clip_duration: float, + asset_durations: dict[str, float] | None, +) -> float | None: + """计算随机 start_time. + + 在素材总时长范围内随机取点,确保 clip_duration 不超出素材边界。 + 如果 asset_durations 为 None 或素材不在其中,返回 None(使用默认 0.0)。 + + Args: + asset_id: 素材 ID + clip_duration: 片段时长(秒) + asset_durations: 素材 ID -> 时长映射 + + Returns: + 随机 start_time 或 None + """ + if asset_durations is None: + return None + + total_duration = asset_durations.get(asset_id) + if total_duration is None or total_duration <= 0: + return None + + # 最大起始点 = 素材总时长 - 片段时长 + max_start = max(0.0, total_duration - clip_duration) + if max_start <= 0: + return 0.0 + + return random.uniform(0.0, max_start) # ── clip_type 映射 ──────────────────────────────────────────────────────── diff --git a/tests/unit/test_1256_random_asset_selection.py b/tests/unit/test_1256_random_asset_selection.py new file mode 100644 index 000000000..76514b7c7 --- /dev/null +++ b/tests/unit/test_1256_random_asset_selection.py @@ -0,0 +1,200 @@ +"""Tests for #1256: random asset selection and random start_time in preview generation.""" + +from unittest.mock import MagicMock + +import pytest + +from packages.domain.edit_plan_clip import EditPlanClip +from packages.domain.plan_generator_utils import ( + _calc_random_start_time, + distribute_assets, +) + + +class TestAssignAssetWithStartTime: + """Test assign_asset() with optional start_time parameter.""" + + def test_assign_asset_without_start_time(self): + """Backward compatible: assign_asset without start_time keeps default 0.0""" + clip = EditPlanClip.create(plan_id="p1", clip_type="main", order=1, duration=5.0) + clip.assign_asset("asset1") + assert clip.asset_id == "asset1" + assert clip.start_time == 0.0 + + def test_assign_asset_with_valid_start_time(self): + """assign_asset with valid start_time sets it correctly""" + clip = EditPlanClip.create(plan_id="p1", clip_type="main", order=1, duration=5.0) + clip.assign_asset("asset1", start_time=3.5) + assert clip.asset_id == "asset1" + assert clip.start_time == 3.5 + + def test_assign_asset_with_zero_start_time(self): + """assign_asset with start_time=0.0 sets it to 0.0""" + clip = EditPlanClip.create(plan_id="p1", clip_type="main", order=1, duration=5.0, start_time=5.0) + clip.assign_asset("asset1", start_time=0.0) + assert clip.start_time == 0.0 + + def test_assign_asset_with_none_start_time(self): + """assign_asset with start_time=None keeps existing start_time""" + clip = EditPlanClip.create(plan_id="p1", clip_type="main", order=1, duration=5.0, start_time=2.0) + clip.assign_asset("asset1", start_time=None) + assert clip.start_time == 2.0 # unchanged + + def test_assign_asset_with_negative_start_time(self): + """assign_asset with negative start_time is ignored""" + clip = EditPlanClip.create(plan_id="p1", clip_type="main", order=1, duration=5.0, start_time=2.0) + clip.assign_asset("asset1", start_time=-1.0) + assert clip.start_time == 2.0 # unchanged, negative ignored + + +class TestCalcRandomStartTime: + """Test _calc_random_start_time helper function.""" + + def test_returns_none_when_no_durations(self): + """Returns None when asset_durations is None""" + result = _calc_random_start_time("asset1", 5.0, None) + assert result is None + + def test_returns_none_when_asset_not_in_durations(self): + """Returns None when asset_id not in durations dict""" + result = _calc_random_start_time("asset1", 5.0, {"other": 30.0}) + assert result is None + + def test_returns_zero_when_duration_too_short(self): + """Returns 0.0 when asset duration <= clip duration""" + result = _calc_random_start_time("asset1", 10.0, {"asset1": 5.0}) + assert result == 0.0 + + def test_returns_valid_random_start_time(self): + """Returns start_time within valid range""" + import random + + random.seed(42) + result = _calc_random_start_time("asset1", 5.0, {"asset1": 30.0}) + # max_start = 30.0 - 5.0 = 25.0 + assert 0.0 <= result <= 25.0 + + def test_returns_zero_when_duration_zero(self): + """Returns None when asset duration is 0""" + result = _calc_random_start_time("asset1", 5.0, {"asset1": 0.0}) + assert result is None + + +class TestDistributeAssetsRandom: + """Test distribute_assets() with random_selection parameter.""" + + def test_one_take_without_random(self): + """ONE_TAKE without random: assets assigned in order""" + clips = [EditPlanClip.create(plan_id="p1", clip_type="main", order=i, duration=5.0) for i in range(3)] + assets = ["a1", "a2", "a3", "a4"] + distribute_assets(clips, assets, "one_take", random_selection=False) + + assert clips[0].asset_id == "a1" + assert clips[1].asset_id == "a2" + assert clips[2].asset_id == "a3" + + def test_one_take_with_random(self): + """ONE_TAKE with random: assets assigned in random order""" + clips = [EditPlanClip.create(plan_id="p1", clip_type="main", order=i, duration=5.0) for i in range(3)] + assets = ["a1", "a2", "a3", "a4"] + + # Run multiple times to verify randomness + results = set() + for _ in range(10): + distribute_assets(clips, assets, "one_take", random_selection=True) + results.add(tuple(c.asset_id for c in clips)) + + # Should have multiple different orderings + assert len(results) > 1, "Random selection should produce different orderings" + + def test_with_asset_durations_sets_start_time(self): + """distribute_assets with asset_durations sets random start_time""" + clips = [EditPlanClip.create(plan_id="p1", clip_type="main", order=0, duration=5.0)] + durations = {"a1": 30.0} + distribute_assets(clips, ["a1"], "one_take", asset_durations=durations) + + assert clips[0].asset_id == "a1" + # start_time should be set (0 <= start_time <= 25.0) + assert 0.0 <= clips[0].start_time <= 25.0 + + def test_pip_mode_with_random(self): + """PIP mode with random selection works correctly""" + clips = [ + EditPlanClip.create(plan_id="p1", clip_type="main", order=0, duration=5.0), + EditPlanClip.create(plan_id="p1", clip_type="overlay", order=1, duration=5.0), + EditPlanClip.create(plan_id="p1", clip_type="overlay", order=2, duration=5.0), + ] + assets = ["a1", "a2", "a3"] + distribute_assets(clips, assets, "pip", random_selection=True) + + # All clips should have assets assigned + assert clips[0].asset_id + assert clips[1].asset_id + assert clips[2].asset_id + + def test_voice_pip_mode_with_random(self): + """VOICE_PIP mode with random selection works correctly""" + clips = [ + EditPlanClip.create(plan_id="p1", clip_type="background", order=0, duration=5.0), + EditPlanClip.create(plan_id="p1", clip_type="corner_voice", order=1, duration=5.0), + EditPlanClip.create(plan_id="p1", clip_type="b_roll", order=2, duration=5.0), + ] + assets = ["a1", "a2", "a3"] + durations = {"a1": 30.0, "a2": 25.0, "a3": 20.0} + distribute_assets(clips, assets, "voice_pip", random_selection=True, asset_durations=durations) + + # All clips should have assets and start_times + for clip in clips: + assert clip.asset_id + assert 0.0 <= clip.start_time <= 25.0 # max_start = duration - clip_duration + + def test_does_not_modify_original_list(self): + """random_selection should not modify the original asset_ids list""" + clips = [EditPlanClip.create(plan_id="p1", clip_type="main", order=0, duration=5.0)] + assets = ["a1", "a2", "a3"] + original = list(assets) + distribute_assets(clips, assets, "one_take", random_selection=True) + + assert assets == original, "Original asset_ids list should not be modified" + + +class TestDistributeAssetsBackwardCompatible: + """Ensure backward compatibility - existing calls without new params still work.""" + + def test_distribute_assets_default_params(self): + """distribute_assets works with just required params""" + clips = [ + EditPlanClip.create(plan_id="p1", clip_type="main", order=0, duration=5.0), + EditPlanClip.create(plan_id="p1", clip_type="main", order=1, duration=5.0), + ] + distribute_assets(clips, ["a1", "a2"], "one_take") + + assert clips[0].asset_id == "a1" + assert clips[1].asset_id == "a2" + assert clips[0].start_time == 0.0 # default + + def test_all_modes_work_without_new_params(self): + """All editing modes work without random_selection/asset_durations""" + # one_take / voice_over: use main clips + for mode in ["one_take", "voice_over"]: + clips = [ + EditPlanClip.create(plan_id="p1", clip_type="main", order=0, duration=5.0), + ] + distribute_assets(clips, ["a1"], mode) + assert clips[0].asset_id == "a1" + + # pip: main + overlay + clips = [ + EditPlanClip.create(plan_id="p1", clip_type="main", order=0, duration=5.0), + EditPlanClip.create(plan_id="p1", clip_type="overlay", order=1, duration=5.0), + ] + distribute_assets(clips, ["a1"], "pip") + assert clips[0].asset_id == "a1" # first asset goes to main + + # voice_pip: background + corner_voice + b_roll + clips = [ + EditPlanClip.create(plan_id="p1", clip_type="background", order=0, duration=5.0), + EditPlanClip.create(plan_id="p1", clip_type="corner_voice", order=1, duration=5.0), + ] + distribute_assets(clips, ["a1"], "voice_pip") + assert clips[0].asset_id == "a1" # first asset goes to background