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,
+    }

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]:
         """获取计划及其所有片段

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