diff --git a/apps/api/app/api/routes/edit_plans.py b/apps/api/app/api/routes/edit_plans.py index df1c22e66..a0459910c 100644 --- a/apps/api/app/api/routes/edit_plans.py +++ b/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 背景音乐 ─────────────────────────────────────────────────────────── diff --git a/apps/api/app/services/edit_plan_service.py b/apps/api/app/services/edit_plan_service.py old mode 100755 new mode 100644 index 30c95e54e..7696029ad --- a/apps/api/app/services/edit_plan_service.py +++ b/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]: """获取计划及其所有片段 diff --git a/tests/unit/test_edit_plan_service.py b/tests/unit/test_edit_plan_service.py index c67cabb77..df4fc6307 100755 --- a/tests/unit/test_edit_plan_service.py +++ b/tests/unit/test_edit_plan_service.py @@ -849,6 +849,234 @@ class TestClipSplit: assert right.asset_id == "asset-001" +class TestSubtitleManagement: + """字幕管理测试""" + + def test_add_subtitle_basic(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) + + subtitle = svc.add_subtitle(clip.id, start=1.0, end=3.0, text="大家好") + + assert subtitle["text"] == "大家好" + assert subtitle["start"] == 1.0 + assert subtitle["end"] == 3.0 + assert "id" in subtitle + assert len(subtitle["id"]) > 0 + + def test_add_subtitle_with_style(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) + + style = {"font_size": 24, "color": "#ffffff", "position": "bottom"} + subtitle = svc.add_subtitle(clip.id, start=0.0, end=2.0, text="测试", style=style) + + assert subtitle["style"]["font_size"] == 24 + assert subtitle["style"]["color"] == "#ffffff" + + def test_list_subtitles_sorted_by_time(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) + + svc.add_subtitle(clip.id, start=5.0, end=6.0, text="第二") + svc.add_subtitle(clip.id, start=1.0, end=2.0, text="第一") + svc.add_subtitle(clip.id, start=8.0, end=9.0, text="第三") + + subtitles = svc.list_subtitles(clip.id) + assert len(subtitles) == 3 + assert subtitles[0]["text"] == "第一" + assert subtitles[1]["text"] == "第二" + assert subtitles[2]["text"] == "第三" + + def test_add_subtitle_invalid_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.add_subtitle(clip.id, start=-1.0, end=2.0, text="test") + + # 结束时间 <= 开始时间 + with pytest.raises(ValueError, match="时间非法"): + svc.add_subtitle(clip.id, start=5.0, end=3.0, text="test") + + # 超过片段时长 + with pytest.raises(ValueError, match="不能超过片段时长"): + svc.add_subtitle(clip.id, start=8.0, end=15.0, text="test") + + def test_add_subtitle_empty_text_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.add_subtitle(clip.id, start=1.0, end=2.0, text=" ") + + def test_get_subtitle(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) + + sub = svc.add_subtitle(clip.id, start=1.0, end=2.0, text="测试") + found = svc.get_subtitle(clip.id, sub["id"]) + + assert found is not None + assert found["text"] == "测试" + + # 不存在的返回 None + assert svc.get_subtitle(clip.id, "nonexistent") is None + + def test_update_subtitle_text(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) + + sub = svc.add_subtitle(clip.id, start=1.0, end=2.0, text="原文") + updated = svc.update_subtitle(clip.id, sub["id"], text="修改后") + + assert updated["text"] == "修改后" + assert updated["start"] == 1.0 # 时间不变 + + def test_update_subtitle_time(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) + + sub = svc.add_subtitle(clip.id, start=1.0, end=2.0, text="测试") + updated = svc.update_subtitle(clip.id, sub["id"], start=3.0, end=5.0) + + assert updated["start"] == 3.0 + assert updated["end"] == 5.0 + + def test_update_subtitle_not_found_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.update_subtitle(clip.id, "fake-id", text="test") + + def test_delete_subtitle(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) + + sub = svc.add_subtitle(clip.id, start=1.0, end=2.0, text="要删的") + assert svc.count_clips(p.id) == 1 # 片段还在 + + deleted = svc.delete_subtitle(clip.id, sub["id"]) + assert deleted is True + + subtitles = svc.list_subtitles(clip.id) + assert len(subtitles) == 0 + + def test_delete_subtitle_not_found(self): + """删除不存在的字幕返回 False""" + 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) + + deleted = svc.delete_subtitle(clip.id, "nonexistent") + assert deleted is False + + def test_batch_update_subtitles(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=20.0) + + # 先加一条 + svc.add_subtitle(clip.id, start=1.0, end=2.0, text="旧字幕") + + # 全量替换为 3 条 + new_subs = [ + {"start": 0.0, "end": 3.0, "text": "第一条"}, + {"start": 4.0, "end": 7.0, "text": "第二条"}, + {"start": 8.0, "end": 12.0, "text": "第三条"}, + ] + result = svc.batch_update_subtitles(clip.id, new_subs) + + assert len(result) == 3 + assert result[0]["text"] == "第一条" + # 都有 id + assert all("id" in s for s in result) + # 旧字幕没了 + subtitles = svc.list_subtitles(clip.id) + assert len(subtitles) == 3 + + def test_batch_update_preserves_existing_ids(self): + """批量更新时已有 id 的字幕保留原 id""" + 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) + + sub = svc.add_subtitle(clip.id, start=1.0, end=2.0, text="原字幕") + original_id = sub["id"] + + # 带 id 批量更新，修改文本 + updated_list = svc.batch_update_subtitles( + clip.id, + [{"id": original_id, "start": 1.0, "end": 3.0, "text": "修改了"}], + ) + + assert len(updated_list) == 1 + assert updated_list[0]["id"] == original_id + assert updated_list[0]["text"] == "修改了" + + def test_batch_update_skips_empty_text(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) + + subs = [ + {"start": 0.0, "end": 1.0, "text": "有效"}, + {"start": 2.0, "end": 3.0, "text": " "}, # 空白，跳过 + {"start": 4.0, "end": 5.0, "text": "也有效"}, + ] + result = svc.batch_update_subtitles(clip.id, subs) + + assert len(result) == 2 + + def test_empty_clip_returns_empty_list(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) + + subtitles = svc.list_subtitles(clip.id) + assert subtitles == [] + + class TestClipMerge: """片段合并测试"""