diff --git a/apps/api/app/api/routes/edit_plans_clips.py b/apps/api/app/api/routes/edit_plans_clips.py index 81342a132..ae2b422aa 100755 --- a/apps/api/app/api/routes/edit_plans_clips.py +++ b/apps/api/app/api/routes/edit_plans_clips.py @@ -276,3 +276,140 @@ def delete_clip( logger.info("删除剪辑片段: plan_id=%s clip_id=%s by user=%s", plan_id, clip_id, current_user.user.id) return None + + +# ── 片段分割与合并 ────────────────────────────────────────────────────────── + + +class SplitClipRequest(BaseModel): + """分割片段请求体""" + + split_time: float = Field(..., gt=0, description="分割点(秒,相对于片段起始)") + + +class MergeClipsRequest(BaseModel): + """合并片段请求体""" + + clip_ids: list[str] = Field(..., min_length=2, description="要合并的片段 ID 列表") + + +@router.post( + "/{clip_id}/split", + response_model=dict[str, Any], + summary="分割片段", + status_code=status.HTTP_200_OK, +) +def split_clip( + plan_id: str, + clip_id: str, + body: SplitClipRequest, + current_user: AuthenticatedUser = Depends(get_current_user), + db: Session = Depends(get_db_session), + project_repository: Any = Depends(get_project_repository), +) -> dict[str, Any]: + """将一个片段从指定时间点分割为两个片段。 + + 分割后原片段变为左半部分,新增右半部分片段,后续片段顺序自动后移。 + 若片段有关联素材,会自动设置 trim_start/trim_end 标记裁剪范围。 + """ + _check_plan_access(plan_id, current_user.user.id, project_repository, db) + + svc = _get_svc(db) + clip = svc.get_clip(clip_id) + if clip is None or clip.plan_id != plan_id: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"片段不存在: {clip_id}", + ) + + try: + result = svc.split_clip(clip_id, body.split_time) + except ValueError as e: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=str(e), + ) from e + + left = result["left_clip"] + right = result["right_clip"] + logger.info("分割片段: plan_id=%s clip_id=%s by user=%s", plan_id, clip_id, current_user.user.id) + + return { + "left_clip": { + "id": left.id, + "plan_id": left.plan_id, + "clip_type": left.clip_type, + "order": left.order, + "duration": left.duration, + "start_time": left.start_time, + }, + "right_clip": { + "id": right.id, + "plan_id": right.plan_id, + "clip_type": right.clip_type, + "order": right.order, + "duration": right.duration, + "start_time": right.start_time, + }, + } + + +@router.post( + "/merge", + response_model=dict[str, Any], + summary="合并多个连续片段", + status_code=status.HTTP_200_OK, +) +def merge_clips( + plan_id: str, + body: MergeClipsRequest, + current_user: AuthenticatedUser = Depends(get_current_user), + db: Session = Depends(get_db_session), + project_repository: Any = Depends(get_project_repository), +) -> dict[str, Any]: + """将多个连续的同类型片段合并为一个片段。 + + 合并要求: + - 至少 2 个片段 + - 属于同一剪辑计划 + - order 连续 + - 类型相同 + + 合并后保留第一个片段,其余删除,后续片段顺序自动前移。 + """ + _check_plan_access(plan_id, current_user.user.id, project_repository, db) + + svc = _get_svc(db) + + # 校验所有片段都属于该 plan + for cid in body.clip_ids: + clip = svc.get_clip(cid) + if clip is None or clip.plan_id != plan_id: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"片段不存在: {cid}", + ) + + try: + merged = svc.merge_clips(body.clip_ids) + except ValueError as e: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=str(e), + ) from e + + logger.info( + "合并片段: plan_id=%s clip_count=%d by user=%s", + plan_id, + len(body.clip_ids), + current_user.user.id, + ) + + return { + "id": merged.id, + "plan_id": merged.plan_id, + "clip_type": merged.clip_type, + "order": merged.order, + "duration": merged.duration, + "text_content": merged.text_content, + } diff --git a/apps/api/app/services/edit_plan_service.py b/apps/api/app/services/edit_plan_service.py index 9e9df4eca..30c95e54e 100755 --- a/apps/api/app/services/edit_plan_service.py +++ b/apps/api/app/services/edit_plan_service.py @@ -504,6 +504,175 @@ class EditPlanService: return created # ── 渲染生成流程 ──────────────────────────────────────────────────────── + # ── 片段分割与合并 ────────────────────────────────────────────────────── + + def split_clip(self, clip_id: str, split_time: float) -> Dict[str, Any]: + """将一个片段从指定位置分割为两个片段 + + Args: + clip_id: 要分割的片段 ID + split_time: 分割点(相对于片段起始的秒数),必须在 (0, duration) 范围内 + + Returns: + dict: {"left_clip": EditPlanClip, "right_clip": EditPlanClip} + + Raises: + ValueError: 片段不存在、分割时间越界 + """ + clip = self.get_clip_or_raise(clip_id) + plan_id = clip.plan_id + + if split_time <= 0 or split_time >= clip.duration: + raise ValueError(f"分割时间必须在 (0, {clip.duration:.3f}) 范围内,当前: {split_time}") + + self._auto_resume_editing(plan_id) + + original_duration = clip.duration + left_duration = round(split_time, 3) + right_duration = round(original_duration - split_time, 3) + original_order = clip.order + + # 更新左半部分(原片段) + clip.duration = left_duration + left_clip = self._clip_repo.update(clip) + + # 后面片段的 order 全部 +1(给右半部分腾位置) + all_clips = self._clip_repo.list_by_plan(plan_id) + for c in all_clips: + if c.order > original_order and c.id != clip_id: + c.order += 1 + self._clip_repo.update(c) + + # 创建右半部分新片段(继承原片段的大部分属性) + right_config = dict(clip.config) if clip.config else {} + # 素材裁剪信息 + if clip.asset_id: + # 右半部分从 split_time 开始播放 + right_config["trim_start"] = left_duration + # 左半部分在 split_time 处结束 + left_config = dict(left_clip.config) if left_clip.config else {} + left_config["trim_end"] = right_duration + left_clip.config = left_config + left_clip = self._clip_repo.update(left_clip) + + right_clip = EditPlanClip.create( + plan_id=plan_id, + clip_type=clip.clip_type, + order=original_order + 1, + template_clip_config_id=clip.template_clip_config_id, + asset_id=clip.asset_id, + text_content=clip.text_content, + start_time=clip.start_time + left_duration, + duration=right_duration, + transition_effect=clip.transition_effect, + transition_duration=clip.transition_duration, + playback_speed=clip.playback_speed, + config=right_config, + ) + created_right = self._clip_repo.create(right_clip) + + logger.info( + "分割片段: clip_id=%s plan_id=%s split_time=%.3fs left_dur=%.3fs right_dur=%.3fs", + clip_id, + plan_id, + split_time, + left_duration, + right_duration, + ) + + return { + "left_clip": left_clip, + "right_clip": created_right, + } + + def merge_clips(self, clip_ids: List[str]) -> EditPlanClip: + """合并多个连续片段为一个片段 + + Args: + clip_ids: 要合并的片段 ID 列表(至少2个),必须属于同一个计划且 order 连续 + + Returns: + EditPlanClip: 合并后的新片段 + + Raises: + ValueError: 数量不足、不属于同一计划、不连续、类型不一致 + """ + if len(clip_ids) < 2: + raise ValueError("至少需要 2 个片段才能合并") + + # 读取所有片段 + clips = [] + for cid in clip_ids: + clip = self.get_clip_or_raise(cid) + clips.append(clip) + + # 校验:同一计划 + plan_id = clips[0].plan_id + for c in clips[1:]: + if c.plan_id != plan_id: + raise ValueError("只能合并同一计划下的片段") + + # 按 order 排序 + clips.sort(key=lambda c: c.order) + + # 校验:order 连续 + for i in range(1, len(clips)): + if clips[i].order != clips[i - 1].order + 1: + raise ValueError(f"片段不连续:order {clips[i-1].order} → {clips[i].order}") + + # 校验:类型一致 + clip_type = clips[0].clip_type + for c in clips[1:]: + if c.clip_type != clip_type: + raise ValueError("只能合并相同类型的片段") + + self._auto_resume_editing(plan_id) + + # 计算合并后的属性 + first_clip = clips[0] + total_duration = round(sum(c.duration for c in clips), 3) + first_order = first_clip.order + + # 合并文案(用换行连接) + merged_text = "\n".join(c.text_content for c in clips if c.text_content.strip()) + + # 合并 config(后面的覆盖前面的) + merged_config: Dict[str, Any] = {} + for c in clips: + if c.config: + merged_config.update(c.config) + # 清理 trim 相关字段(合并后就是完整片段了) + merged_config.pop("trim_start", None) + merged_config.pop("trim_end", None) + + # 更新第一个片段(保留它作为合并结果) + first_clip.duration = total_duration + first_clip.text_content = merged_text + first_clip.config = merged_config + # 转场保留第一个的(合并后的入点转场) + # playback_speed 取第一个的 + merged_clip = self._clip_repo.update(first_clip) + + # 删除其余片段 + for c in clips[1:]: + self._clip_repo.delete(c.id) + + # 后面的片段 order 前移 (len - 1) 位 + shift = len(clips) - 1 + all_clips = self._clip_repo.list_by_plan(plan_id) + for c in all_clips: + if c.order > first_order and c.id != merged_clip.id: + c.order -= shift + self._clip_repo.update(c) + + logger.info( + "合并片段: plan_id=%s count=%d total_duration=%.3fs", + plan_id, + len(clips), + total_duration, + ) + + return merged_clip def get_plan_with_clips(self, plan_id: str) -> Dict[str, Any]: """获取计划及其所有片段 diff --git a/tests/unit/test_edit_plan_service.py b/tests/unit/test_edit_plan_service.py index aae0e2b5e..c67cabb77 100755 --- a/tests/unit/test_edit_plan_service.py +++ b/tests/unit/test_edit_plan_service.py @@ -727,3 +727,255 @@ class TestResumeEditingAndRegenerate: p = EditPlan.create("tpl-001", "测试") with pytest.raises(ValueError): p.resume_editing() + + +# =========================================================================== +# 片段分割与合并测试 +# =========================================================================== + + +class TestClipSplit: + """片段分割测试""" + + def test_split_basic(self): + """基础分割:10秒片段在第3秒处分割""" + svc = _make_service() + p = svc.create_plan("tpl-001", "测试计划") + svc.transition_status(p.id, EditPlanStatus.EDITING) + clip = svc.create_clip(p.id, "main", 0, duration=10.0, text_content="测试文案") + + result = svc.split_clip(clip.id, 3.0) + + assert result["left_clip"].duration == 3.0 + assert result["left_clip"].order == 0 + assert result["right_clip"].duration == 7.0 + assert result["right_clip"].order == 1 + assert result["right_clip"].clip_type == "main" + assert result["right_clip"].text_content == "测试文案" + # 总片段数 = 2 + assert svc.count_clips(p.id) == 2 + + def test_split_preserves_clip_properties(self): + """分割后属性继承正确""" + svc = _make_service() + p = svc.create_plan("tpl-001", "测试计划") + svc.transition_status(p.id, EditPlanStatus.EDITING) + clip = svc.create_clip( + p.id, + "intro", + 0, + duration=10.0, + transition_effect="fade", + playback_speed=1.5, + config={"filter": "vivid"}, + ) + + result = svc.split_clip(clip.id, 5.0) + + right = result["right_clip"] + assert right.clip_type == "intro" + assert right.transition_effect == "fade" + assert right.playback_speed == 1.5 + assert right.config.get("filter") == "vivid" + + def test_split_shifts_following_clips(self): + """分割后,后面的片段 order 自动 +1""" + svc = _make_service() + p = svc.create_plan("tpl-001", "测试计划") + svc.transition_status(p.id, EditPlanStatus.EDITING) + clip0 = svc.create_clip(p.id, "main", 0, duration=5.0) + clip1 = svc.create_clip(p.id, "main", 1, duration=5.0) + clip2 = svc.create_clip(p.id, "main", 2, duration=5.0) + + svc.split_clip(clip1.id, 2.0) + + # clip0: order 0 + # clip1(left): order 1 + # new right: order 2 + # clip2: order 3 + clips = svc.list_clips(p.id) + order_map = {c.id: c.order for c in clips} + assert order_map[clip0.id] == 0 + assert order_map[clip1.id] == 1 + assert order_map[clip2.id] == 3 + assert len(clips) == 4 + + def test_split_at_boundary_raises(self): + """分割点为0或等于时长时,报错""" + svc = _make_service() + p = svc.create_plan("tpl-001", "测试计划") + svc.transition_status(p.id, EditPlanStatus.EDITING) + clip = svc.create_clip(p.id, "main", 0, duration=10.0) + + with pytest.raises(ValueError, match="分割时间"): + svc.split_clip(clip.id, 0.0) + + with pytest.raises(ValueError, match="分割时间"): + svc.split_clip(clip.id, 10.0) + + def test_split_negative_time_raises(self): + """负数分割点报错""" + svc = _make_service() + p = svc.create_plan("tpl-001", "测试计划") + svc.transition_status(p.id, EditPlanStatus.EDITING) + clip = svc.create_clip(p.id, "main", 0, duration=10.0) + + with pytest.raises(ValueError, match="分割时间"): + svc.split_clip(clip.id, -1.0) + + def test_split_nonexistent_clip_raises(self): + """不存在的片段报错""" + svc = _make_service() + + with pytest.raises(ValueError, match="片段不存在"): + svc.split_clip("nonexistent", 5.0) + + def test_split_with_asset_adds_trim_info(self): + """有素材的片段分割后,添加trim_start/trim_end""" + svc = _make_service() + p = svc.create_plan("tpl-001", "测试计划") + svc.transition_status(p.id, EditPlanStatus.EDITING) + clip = svc.create_clip(p.id, "main", 0, duration=10.0, asset_id="asset-001") + + result = svc.split_clip(clip.id, 3.0) + + left = result["left_clip"] + right = result["right_clip"] + # 左半部分有 trim_end + assert left.config.get("trim_end") == 7.0 + # 右半部分有 trim_start + assert right.config.get("trim_start") == 3.0 + # 右半部分也关联同一个素材 + assert right.asset_id == "asset-001" + + +class TestClipMerge: + """片段合并测试""" + + def test_merge_two_clips(self): + """基础合并:两个5秒片段合并为10秒""" + svc = _make_service() + p = svc.create_plan("tpl-001", "测试计划") + svc.transition_status(p.id, EditPlanStatus.EDITING) + clip1 = svc.create_clip(p.id, "main", 0, duration=5.0, text_content="第一段") + clip2 = svc.create_clip(p.id, "main", 1, duration=5.0, text_content="第二段") + + merged = svc.merge_clips([clip1.id, clip2.id]) + + assert merged.duration == 10.0 + assert merged.order == 0 + assert merged.clip_type == "main" + assert "第一段" in merged.text_content + assert "第二段" in merged.text_content + # 总片段数 = 1 + assert svc.count_clips(p.id) == 1 + + def test_merge_shifts_following_clips(self): + """合并后,后面的片段 order 前移""" + svc = _make_service() + p = svc.create_plan("tpl-001", "测试计划") + svc.transition_status(p.id, EditPlanStatus.EDITING) + clip0 = svc.create_clip(p.id, "main", 0, duration=5.0) + clip1 = svc.create_clip(p.id, "main", 1, duration=5.0) + clip2 = svc.create_clip(p.id, "main", 2, duration=5.0) + clip3 = svc.create_clip(p.id, "main", 3, duration=5.0) + + svc.merge_clips([clip1.id, clip2.id]) + + clips = svc.list_clips(p.id) + order_map = {c.id: c.order for c in clips} + assert order_map[clip0.id] == 0 + assert order_map[clip3.id] == 2 # 原来order=3,前移1位=2 + assert len(clips) == 3 + + def test_merge_three_clips(self): + """合并3个片段""" + svc = _make_service() + p = svc.create_plan("tpl-001", "测试计划") + svc.transition_status(p.id, EditPlanStatus.EDITING) + clips = [] + for i in range(3): + c = svc.create_clip(p.id, "main", i, duration=3.0) + clips.append(c) + + merged = svc.merge_clips([c.id for c in clips]) + + assert merged.duration == 9.0 + assert svc.count_clips(p.id) == 1 + + def test_merge_different_types_raises(self): + """不同类型片段不能合并""" + svc = _make_service() + p = svc.create_plan("tpl-001", "测试计划") + svc.transition_status(p.id, EditPlanStatus.EDITING) + clip1 = svc.create_clip(p.id, "intro", 0, duration=5.0) + clip2 = svc.create_clip(p.id, "main", 1, duration=5.0) + + with pytest.raises(ValueError, match="相同类型"): + svc.merge_clips([clip1.id, clip2.id]) + + def test_merge_non_contiguous_raises(self): + """不连续的片段不能合并""" + svc = _make_service() + p = svc.create_plan("tpl-001", "测试计划") + svc.transition_status(p.id, EditPlanStatus.EDITING) + clip0 = svc.create_clip(p.id, "main", 0, duration=5.0) + svc.create_clip(p.id, "main", 1, duration=5.0) + clip2 = svc.create_clip(p.id, "main", 2, duration=5.0) + + with pytest.raises(ValueError, match="不连续"): + svc.merge_clips([clip0.id, clip2.id]) + + def test_merge_single_clip_raises(self): + """单个片段不能合并""" + svc = _make_service() + p = svc.create_plan("tpl-001", "测试计划") + svc.transition_status(p.id, EditPlanStatus.EDITING) + clip = svc.create_clip(p.id, "main", 0, duration=5.0) + + with pytest.raises(ValueError, match="至少需要 2 个"): + svc.merge_clips([clip.id]) + + def test_merge_different_plans_raises(self): + """不同计划的片段不能合并""" + svc = _make_service() + p1 = svc.create_plan("tpl-001", "计划1") + p2 = svc.create_plan("tpl-001", "计划2") + svc.transition_status(p1.id, EditPlanStatus.EDITING) + svc.transition_status(p2.id, EditPlanStatus.EDITING) + clip1 = svc.create_clip(p1.id, "main", 0, duration=5.0) + clip2 = svc.create_clip(p2.id, "main", 0, duration=5.0) + + with pytest.raises(ValueError, match="同一计划"): + svc.merge_clips([clip1.id, clip2.id]) + + def test_merge_clears_trim_fields(self): + """合并后清理trim字段""" + svc = _make_service() + p = svc.create_plan("tpl-001", "测试计划") + svc.transition_status(p.id, EditPlanStatus.EDITING) + clip1 = svc.create_clip(p.id, "main", 0, duration=5.0, config={"trim_end": 2.0, "filter": "vivid"}) + clip2 = svc.create_clip(p.id, "main", 1, duration=5.0, config={"trim_start": 1.0}) + + merged = svc.merge_clips([clip1.id, clip2.id]) + + assert "trim_start" not in merged.config + assert "trim_end" not in merged.config + # 非 trim 字段保留(后面的覆盖前面的) + assert merged.config.get("filter") == "vivid" + + def test_split_then_merge_recovers(self): + """分割后再合并,时长基本恢复(浮点精度内)""" + svc = _make_service() + p = svc.create_plan("tpl-001", "测试计划") + svc.transition_status(p.id, EditPlanStatus.EDITING) + original = svc.create_clip(p.id, "main", 0, duration=10.0) + + result = svc.split_clip(original.id, 3.5) + left = result["left_clip"] + right = result["right_clip"] + + merged = svc.merge_clips([left.id, right.id]) + + assert abs(merged.duration - 10.0) < 0.001 + assert svc.count_clips(p.id) == 1