feat: 字幕管理 - 增删改查 + 批量导入（基于片段config） (#394)

feat: 字幕管理 - 增删改查 + 批量导入（基于片段config） (#394)

apps/api/app/api/routes/edit_plans.py

@@ -522,6 +522,245 @@ def copy_plan(
     return _to_response(new_plan)
 
 
+# ── 字幕管理 ────────────────────────────────────────────────────────────────
+
+
+class SubtitleCreateRequest(BaseModel):
+    """添加字幕请求体"""
+
+    start: float = Field(..., ge=0, description="开始时间（秒）")
+    end: float = Field(..., gt=0, description="结束时间（秒）")
+    text: str = Field(..., min_length=1, max_length=500, description="字幕文本")
+    style: Optional[dict[str, Any]] = Field(default=None, description="字幕样式")
+
+
+class SubtitleUpdateRequest(BaseModel):
+    """更新字幕请求体"""
+
+    start: Optional[float] = Field(default=None, ge=0, description="开始时间（秒）")
+    end: Optional[float] = Field(default=None, gt=0, description="结束时间（秒）")
+    text: Optional[str] = Field(default=None, min_length=1, max_length=500, description="字幕文本")
+    style: Optional[dict[str, Any]] = Field(default=None, description="字幕样式")
+
+
+class SubtitleBatchUpdateRequest(BaseModel):
+    """批量更新字幕请求体"""
+
+    subtitles: list[dict[str, Any]] = Field(
+        default_factory=list,
+        description="字幕列表（全量替换），每条包含 start/end/text，可选 id/style",
+    )
+
+
+@router.get(
+    "/clips/{clip_id}/subtitles",
+    response_model=list[dict[str, Any]],
+    summary="获取片段的所有字幕",
+)
+def list_subtitles(
+    clip_id: str,
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
+    project_repo=Depends(get_project_repository),
+) -> list[dict[str, Any]]:
+    """获取指定片段的所有字幕，按时间排序。"""
+    service = EditPlanService(db)
+
+    clip = service.get_clip(clip_id)
+    if clip is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"片段不存在: {clip_id}",
+        )
+    plan = service.get_plan(clip.plan_id)
+    if plan and plan.project_id:
+        check_project_access(project_repo, current_user, plan.project_id)
+
+    return service.list_subtitles(clip_id)
+
+
+@router.post(
+    "/clips/{clip_id}/subtitles",
+    response_model=dict[str, Any],
+    summary="添加一条字幕",
+    status_code=status.HTTP_201_CREATED,
+)
+def add_subtitle(
+    clip_id: str,
+    body: SubtitleCreateRequest,
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
+    project_repo=Depends(get_project_repository),
+) -> dict[str, Any]:
+    """给片段添加一条字幕。"""
+    service = EditPlanService(db)
+
+    clip = service.get_clip(clip_id)
+    if clip is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"片段不存在: {clip_id}",
+        )
+    plan = service.get_plan(clip.plan_id)
+    if plan and plan.project_id:
+        check_project_access(project_repo, current_user, plan.project_id)
+
+    try:
+        subtitle = service.add_subtitle(
+            clip_id,
+            start=body.start,
+            end=body.end,
+            text=body.text,
+            style=body.style,
+        )
+    except ValueError as e:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=str(e),
+        ) from e
+
+    logger.info("添加字幕: clip_id=%s by user=%s", clip_id, current_user.user.id)
+    return subtitle
+
+
+@router.put(
+    "/clips/{clip_id}/subtitles/{subtitle_id}",
+    response_model=dict[str, Any],
+    summary="更新一条字幕",
+)
+def update_subtitle(
+    clip_id: str,
+    subtitle_id: str,
+    body: SubtitleUpdateRequest,
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
+    project_repo=Depends(get_project_repository),
+) -> dict[str, Any]:
+    """更新一条字幕的时间、文本或样式。"""
+    service = EditPlanService(db)
+
+    clip = service.get_clip(clip_id)
+    if clip is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"片段不存在: {clip_id}",
+        )
+    plan = service.get_plan(clip.plan_id)
+    if plan and plan.project_id:
+        check_project_access(project_repo, current_user, plan.project_id)
+
+    try:
+        subtitle = service.update_subtitle(
+            clip_id,
+            subtitle_id,
+            start=body.start,
+            end=body.end,
+            text=body.text,
+            style=body.style,
+        )
+    except ValueError as e:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=str(e),
+        ) from e
+
+    logger.info(
+        "更新字幕: clip_id=%s subtitle_id=%s by user=%s",
+        clip_id,
+        subtitle_id,
+        current_user.user.id,
+    )
+    return subtitle
+
+
+@router.delete(
+    "/clips/{clip_id}/subtitles/{subtitle_id}",
+    summary="删除一条字幕",
+    status_code=status.HTTP_204_NO_CONTENT,
+    response_model=None,
+    response_class=Response,
+)
+def delete_subtitle(
+    clip_id: str,
+    subtitle_id: str,
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
+    project_repo=Depends(get_project_repository),
+) -> None:
+    """删除一条字幕。"""
+    service = EditPlanService(db)
+
+    clip = service.get_clip(clip_id)
+    if clip is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"片段不存在: {clip_id}",
+        )
+    plan = service.get_plan(clip.plan_id)
+    if plan and plan.project_id:
+        check_project_access(project_repo, current_user, plan.project_id)
+
+    deleted = service.delete_subtitle(clip_id, subtitle_id)
+    if not deleted:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"字幕不存在: {subtitle_id}",
+        )
+
+    logger.info(
+        "删除字幕: clip_id=%s subtitle_id=%s by user=%s",
+        clip_id,
+        subtitle_id,
+        current_user.user.id,
+    )
+
+
+@router.put(
+    "/clips/{clip_id}/subtitles",
+    response_model=list[dict[str, Any]],
+    summary="批量更新字幕（全量替换）",
+)
+def batch_update_subtitles(
+    clip_id: str,
+    body: SubtitleBatchUpdateRequest,
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
+    project_repo=Depends(get_project_repository),
+) -> list[dict[str, Any]]:
+    """批量更新片段的所有字幕（全量替换）。
+
+    用于批量编辑、SRT导入、ASR结果导入等场景。
+    每条字幕包含 start/end/text，已有 id 则保留，否则生成新 id。
+    """
+    service = EditPlanService(db)
+
+    clip = service.get_clip(clip_id)
+    if clip is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"片段不存在: {clip_id}",
+        )
+    plan = service.get_plan(clip.plan_id)
+    if plan and plan.project_id:
+        check_project_access(project_repo, current_user, plan.project_id)
+
+    try:
+        subtitles = service.batch_update_subtitles(clip_id, body.subtitles)
+    except ValueError as e:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=str(e),
+        ) from e
+
+    logger.info(
+        "批量更新字幕: clip_id=%s count=%d by user=%s",
+        clip_id,
+        len(subtitles),
+        current_user.user.id,
+    )
+    return subtitles
+
+
 # ── BGM 背景音乐 ───────────────────────────────────────────────────────────
 
 

apps/api/app/services/edit_plan_service.py

@@ -674,6 +674,237 @@ class EditPlanService:
 
         return merged_clip
 
+    # ── 字幕管理 ──────────────────────────────────────────────────────────
+
+    def list_subtitles(self, clip_id: str) -> List[Dict[str, Any]]:
+        """获取片段的所有字幕
+
+        Returns:
+            List[dict]: 字幕列表，按 start 时间排序
+        """
+        clip = self.get_clip_or_raise(clip_id)
+        config = clip.config or {}
+        subtitles = config.get("subtitles", [])
+        # 按开始时间排序
+        subtitles.sort(key=lambda s: s.get("start", 0))
+        return subtitles
+
+    def get_subtitle(self, clip_id: str, subtitle_id: str) -> Optional[Dict[str, Any]]:
+        """获取单条字幕"""
+        subtitles = self.list_subtitles(clip_id)
+        for s in subtitles:
+            if s.get("id") == subtitle_id:
+                return s
+        return None
+
+    def add_subtitle(
+        self,
+        clip_id: str,
+        start: float,
+        end: float,
+        text: str,
+        *,
+        style: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """添加一条字幕
+
+        Args:
+            clip_id: 片段 ID
+            start: 开始时间（秒，相对于片段）
+            end: 结束时间（秒）
+            text: 字幕文本
+            style: 样式配置（字体、大小、颜色、位置等）
+
+        Returns:
+            dict: 新增的字幕条目
+
+        Raises:
+            ValueError: 时间非法或文本为空
+        """
+        clip = self.get_clip_or_raise(clip_id)
+
+        if start < 0 or end <= start:
+            raise ValueError(f"字幕时间非法: start={start}, end={end}")
+        if not text.strip():
+            raise ValueError("字幕文本不能为空")
+        if end > clip.duration + 0.001:
+            raise ValueError(f"字幕结束时间不能超过片段时长: end={end:.3f}, duration={clip.duration:.3f}")
+
+        self._auto_resume_editing(clip.plan_id)
+
+        from uuid import uuid4
+
+        config = dict(clip.config) if clip.config else {}
+        subtitles = list(config.get("subtitles", []))
+
+        subtitle = {
+            "id": uuid4().hex,
+            "start": round(start, 3),
+            "end": round(end, 3),
+            "text": text.strip(),
+            "style": style or {},
+        }
+        subtitles.append(subtitle)
+        subtitles.sort(key=lambda s: s.get("start", 0))
+
+        config["subtitles"] = subtitles
+        clip.config = config
+        updated = self._clip_repo.update(clip)
+
+        logger.info(
+            "添加字幕: clip_id=%s subtitle_id=%s start=%.3fs end=%.3fs",
+            clip_id,
+            subtitle["id"],
+            start,
+            end,
+        )
+
+        return subtitle
+
+    def update_subtitle(
+        self,
+        clip_id: str,
+        subtitle_id: str,
+        *,
+        start: Optional[float] = None,
+        end: Optional[float] = None,
+        text: Optional[str] = None,
+        style: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """更新一条字幕
+
+        Returns:
+            dict: 更新后的字幕条目
+
+        Raises:
+            ValueError: 字幕不存在或参数非法
+        """
+        clip = self.get_clip_or_raise(clip_id)
+        config = dict(clip.config) if clip.config else {}
+        subtitles = list(config.get("subtitles", []))
+
+        found = False
+        for i, s in enumerate(subtitles):
+            if s.get("id") == subtitle_id:
+                # 更新字段
+                updated_s = dict(s)
+                if start is not None:
+                    updated_s["start"] = round(start, 3)
+                if end is not None:
+                    updated_s["end"] = round(end, 3)
+                if text is not None:
+                    if not text.strip():
+                        raise ValueError("字幕文本不能为空")
+                    updated_s["text"] = text.strip()
+                if style is not None:
+                    updated_s["style"] = style
+
+                # 校验时间
+                if updated_s["start"] < 0 or updated_s["end"] <= updated_s["start"]:
+                    raise ValueError(f"字幕时间非法: start={updated_s['start']}, end={updated_s['end']}")
+                if updated_s["end"] > clip.duration + 0.001:
+                    raise ValueError("字幕结束时间不能超过片段时长")
+
+                subtitles[i] = updated_s
+                found = True
+                break
+
+        if not found:
+            raise ValueError(f"字幕不存在: {subtitle_id}")
+
+        self._auto_resume_editing(clip.plan_id)
+
+        subtitles.sort(key=lambda s: s.get("start", 0))
+        config["subtitles"] = subtitles
+        clip.config = config
+        self._clip_repo.update(clip)
+
+        logger.info("更新字幕: clip_id=%s subtitle_id=%s", clip_id, subtitle_id)
+
+        return subtitles[next(i for i, s in enumerate(subtitles) if s["id"] == subtitle_id)]
+
+    def delete_subtitle(self, clip_id: str, subtitle_id: str) -> bool:
+        """删除一条字幕
+
+        Returns:
+            bool: 是否删除成功
+        """
+        clip = self.get_clip_or_raise(clip_id)
+        config = dict(clip.config) if clip.config else {}
+        subtitles = list(config.get("subtitles", []))
+
+        new_subtitles = [s for s in subtitles if s.get("id") != subtitle_id]
+        if len(new_subtitles) == len(subtitles):
+            return False
+
+        self._auto_resume_editing(clip.plan_id)
+
+        config["subtitles"] = new_subtitles
+        clip.config = config
+        self._clip_repo.update(clip)
+
+        logger.info("删除字幕: clip_id=%s subtitle_id=%s", clip_id, subtitle_id)
+        return True
+
+    def batch_update_subtitles(
+        self,
+        clip_id: str,
+        subtitles: List[Dict[str, Any]],
+    ) -> List[Dict[str, Any]]:
+        """批量更新字幕（全量替换，用于批量编辑或导入）
+
+        Args:
+            clip_id: 片段 ID
+            subtitles: 字幕列表，每条需包含 start/end/text，已有 id 则保留
+
+        Returns:
+            List[dict]: 更新后的字幕列表
+        """
+        clip = self.get_clip_or_raise(clip_id)
+
+        from uuid import uuid4
+
+        validated = []
+        for s in subtitles:
+            start = float(s.get("start", 0))
+            end = float(s.get("end", 0))
+            text = str(s.get("text", ""))
+
+            if start < 0 or end <= start:
+                raise ValueError(f"字幕时间非法: start={start}, end={end}")
+            if not text.strip():
+                continue  # 跳过空字幕
+            if end > clip.duration + 0.001:
+                raise ValueError(f"字幕结束时间不能超过片段时长: end={end}")
+
+            subtitle_id = s.get("id") or uuid4().hex
+            validated.append(
+                {
+                    "id": subtitle_id,
+                    "start": round(start, 3),
+                    "end": round(end, 3),
+                    "text": text.strip(),
+                    "style": s.get("style", {}),
+                }
+            )
+
+        validated.sort(key=lambda s: s["start"])
+
+        self._auto_resume_editing(clip.plan_id)
+
+        config = dict(clip.config) if clip.config else {}
+        config["subtitles"] = validated
+        clip.config = config
+        self._clip_repo.update(clip)
+
+        logger.info(
+            "批量更新字幕: clip_id=%s count=%d",
+            clip_id,
+            len(validated),
+        )
+
+        return validated
+
     def get_plan_with_clips(self, plan_id: str) -> Dict[str, Any]:
         """获取计划及其所有片段