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

Merged
xiaoxia merged 4 commits from feat/subtitle-management into develop 2026-07-17 10:50:45 +08:00
3 changed files with 698 additions and 0 deletions
+239
View File
@@ -520,6 +520,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 背景音乐 ───────────────────────────────────────────────────────────
+231
View File
@@ -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]:
"""获取计划及其所有片段
+228
View File
@@ -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:
"""片段合并测试"""