apps/api/app/api/routes/videos.py

@@ -5,13 +5,15 @@ from app.core.celery_app import celery_app
 from app.core.storage import OSSStorageService, get_storage_service
 from app.dependencies import get_generated_video_repository
 from app.schemas.video_center import (
+    BatchDeleteRequest,
     BatchDownloadRequest,
     BatchDownloadResponse,
+    BatchOperationResponse,
     ListVideosResponse,
     UpdateVideoReviewRequest,
     VideoItemResponse,
 )
-from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi import APIRouter, Depends, HTTPException, Query, Response
 
 from packages.application import (
     GetGeneratedVideoUseCase,
@@ -114,6 +116,61 @@ def update_video_review_status(
     return _to_video_response(item, storage)
 
 
+@router.delete("/videos/{video_id}", status_code=204, response_class=Response)
+def delete_video(
+    video_id: str,
+    repo=Depends(get_generated_video_repository),
+    storage: OSSStorageService = Depends(get_storage_service),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+):
+    """删除单个成片（硬删除）。"""
+    use_case = GetGeneratedVideoUseCase(repo)
+    item = use_case.execute(video_id)
+    if item is None:
+        raise HTTPException(status_code=404, detail="Video not found")
+
+    # 尝试删除 OSS 文件，失败不影响数据库删除
+    if item.file_url:
+        try:
+            key = storage._normalize_storage_key(item.file_url)
+            storage.delete_file(key)
+        except Exception:
+            logger.warning("删除 OSS 视频文件失败，跳过: video_id=%s", video_id)
+
+    repo.delete(video_id)
+    return Response(status_code=204)
+
+
+@router.post("/videos/batch-delete", response_model=BatchOperationResponse)
+def batch_delete_videos(
+    request: BatchDeleteRequest,
+    repo=Depends(get_generated_video_repository),
+    storage: OSSStorageService = Depends(get_storage_service),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+):
+    """批量删除成片。"""
+    videos = repo.get_by_ids(request.video_ids)
+    existing_ids = {v.id for v in videos}
+    failed_ids = [vid for vid in request.video_ids if vid not in existing_ids]
+    failed_details = {vid: "Video not found" for vid in failed_ids}
+
+    # 尝试删除 OSS 文件
+    for video in videos:
+        if video.file_url:
+            try:
+                key = storage._normalize_storage_key(video.file_url)
+                storage.delete_file(key)
+            except Exception:
+                logger.warning("批量删除 OSS 视频文件失败，跳过: video_id=%s", video.id)
+
+    success_count = repo.batch_delete(list(existing_ids))
+    return BatchOperationResponse(
+        success_count=success_count,
+        failed_ids=failed_ids,
+        failed_details=failed_details,
+    )
+
+
 @router.post("/videos/batch-download", response_model=BatchDownloadResponse)
 def batch_download_videos(
     request: BatchDownloadRequest,

apps/api/app/schemas/video_center.py

@@ -35,6 +35,19 @@ class UpdateVideoReviewRequest(BaseModel):
     review_status: VideoReviewStatus
 
 
+MAX_BATCH_DELETE_SIZE = 200
+
+
+class BatchDeleteRequest(BaseModel):
+    video_ids: list[str] = Field(..., min_length=1, max_length=MAX_BATCH_DELETE_SIZE, description="要删除的成片ID列表")
+
+
+class BatchOperationResponse(BaseModel):
+    success_count: int = Field(..., ge=0, description="成功删除数量")
+    failed_ids: list[str] = Field(default_factory=list, description="失败的ID列表")
+    failed_details: dict[str, str] = Field(default_factory=dict, description="失败详情")
+
+
 class BatchDownloadRequest(BaseModel):
     video_ids: list[str]
 

packages/adapters/sqlalchemy_impl/generated_video_repository.py

@@ -157,6 +157,27 @@ class SQLAlchemyGeneratedVideoRepository:
         models = self.session.query(GeneratedVideoModel).filter(GeneratedVideoModel.id.in_(video_ids)).all()
         return [self._to_domain(model) for model in models]
 
+    def delete(self, video_id: str) -> bool:
+        """删除单个成片记录（硬删除）。"""
+        model = self.session.query(GeneratedVideoModel).filter(GeneratedVideoModel.id == video_id).first()
+        if model is None:
+            return False
+        self.session.delete(model)
+        self.session.commit()
+        return True
+
+    def batch_delete(self, video_ids: list[str]) -> int:
+        """批量删除成片记录，返回成功删除数量。"""
+        if not video_ids:
+            return 0
+        result = (
+            self.session.query(GeneratedVideoModel)
+            .filter(GeneratedVideoModel.id.in_(video_ids))
+            .delete(synchronize_session=False)
+        )
+        self.session.commit()
+        return result
+
     @staticmethod
     def _to_domain(model: GeneratedVideoModel) -> GeneratedVideo:
         return GeneratedVideo(

packages/ports/generated_video_repository.py

@@ -31,3 +31,7 @@ class GeneratedVideoRepository(Protocol):
     def update_thumbnail(self, video_id: str, thumbnail_url: str) -> bool: ...
 
     def get_by_ids(self, video_ids: list[str]) -> list[GeneratedVideo]: ...
+
+    def delete(self, video_id: str) -> bool: ...
+
+    def batch_delete(self, video_ids: list[str]) -> int: ...

packages/shared/storage.py

@@ -8,7 +8,7 @@ import json
 import logging
 import os
 from typing import Optional
-from urllib.parse import urlparse
+from urllib.parse import unquote, urlparse
 
 try:
     import oss2
@@ -159,10 +159,15 @@ class SharedStorageService:
             return self.get_url(storage_key)
 
     def _normalize_storage_key(self, storage_key_or_url: str) -> str:
-        """Extract storage key from URL."""
+        """Extract storage key from URL.
+
+        从完整 URL 提取 OSS 存储键，并做 URL 解码 — 否则 URL 编码的字符
+        （如空格=%20、中文=%XX）会导致 sign_url 计算的签名与 OSS 服务端
+        不匹配（SignatureDoesNotMatch）。原始 key 传入时直接返回。
+        """
         if storage_key_or_url.startswith("http://") or storage_key_or_url.startswith("https://"):
             parsed = urlparse(storage_key_or_url)
-            return parsed.path.lstrip("/")
+            return unquote(parsed.path.lstrip("/"))
         return storage_key_or_url.lstrip("/")
 
     def create_direct_upload_post(