feat(assets): 批量删除接口 + 分页性能优化

- 新增 POST /assets/batch-delete 批量删除素材接口
  - 仓储层: abstract/SQLAlchemy/InMemory 均实现 batch_delete()
  - API层: 逐项校验项目权限后批量删除
  - Schema: BatchDeleteRequest/BatchDeleteResponse
- GET /assets 分页优化: 无 keyword/gender/style 过滤时走 DB 级 skip/limit
- 素材上传 duration 字段已全链路支持（domain/schema/ORM/repository）
- 新增 4 个批量删除单元测试，全量 1030 测试通过

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

apps/api/app/api/routes/assets.py

@@ -10,6 +10,8 @@ from app.dependencies import (
 )
 from app.schemas.asset import (
     AssetResponse,
+    BatchDeleteRequest,
+    BatchDeleteResponse,
     CreateAssetRequest,
     ListAssetsResponse,
     UpdateAssetRequest,
@@ -96,12 +98,12 @@ def list_assets(
     # kind → file_type 映射（voice 对应 audio）
     kind_to_file_type = {"video": "video", "voice": "audio", "image": "image"}
 
-    def _apply_filters(items):
-        """依次应用 kind / keyword / gender / style 过滤。"""
+    # 需要内存过滤的标志（keyword/gender/style 无法在 DB 层过滤）
+    needs_memory_filter = bool(keyword or gender or style)
+
+    def _apply_memory_filters(items):
+        """应用 keyword / gender / style 内存过滤。"""
         result = items
-        if kind:
-            ft = kind_to_file_type.get(kind)
-            result = [i for i in result if i.mime_type and i.mime_type.startswith(ft or "")]
         if keyword:
             kw = keyword.lower()
             result = [i for i in result if kw in (i.name or "").lower()]
@@ -111,7 +113,85 @@ def list_assets(
             result = [i for i in result if (i.metadata or {}).get("style") == style]
         return result
 
-    # 模式1：指定 library_id → 返回该库的素材
+    # ── 优化路径：无内存过滤时，使用 DB 级分页 ──
+    if not needs_memory_filter:
+        ft = kind_to_file_type.get(kind) if kind else None
+
+        # 模式1：指定 library_id
+        if library_id:
+            library = asset_library_repository.get(library_id)
+            if library is None:
+                raise HTTPException(status_code=404, detail=f"AssetLibrary {library_id} not found")
+            _check_project_access(library.project_id, user_id, project_repository)
+            if ft:
+                items = asset_repository.find_by_library_and_file_type(library_id, ft, skip=skip, limit=limit)
+                total = asset_repository.count_by_project(library.project_id) if not kind else len(items)
+            else:
+                items = asset_repository.find_by_library(library_id, skip=skip, limit=limit)
+                total = asset_repository.count_by_project(library.project_id)
+            return ListAssetsResponse(
+                items=[_to_asset_response(item) for item in items],
+                total=total,
+                skip=skip,
+                limit=limit,
+            )
+
+        # 模式2：指定 project_id
+        if project_id:
+            _check_project_access(project_id, user_id, project_repository)
+            if ft:
+                # 无直接方法，加载后按 file_type 过滤（仍比全量加载好）
+                all_items = asset_repository.find_by_project(project_id)
+                items = [i for i in all_items if i.mime_type and i.mime_type.startswith(ft)]
+                total = len(items)
+                paged = items[skip : skip + limit]
+            else:
+                items = asset_repository.find_by_project(project_id, skip=skip, limit=limit)
+                total = asset_repository.count_by_project(project_id)
+                paged = items
+            return ListAssetsResponse(
+                items=[_to_asset_response(item) for item in paged],
+                total=total,
+                skip=skip,
+                limit=limit,
+            )
+
+        # 模式3：跨项目（无 library_id/project_id）
+        try:
+            projects = project_repository.find_accessible_projects(user_id)
+        except Exception:
+            logger.exception("查询用户可访问项目失败: user_id=%s", user_id)
+            return ListAssetsResponse(items=[], total=0, skip=skip, limit=limit)
+
+        project_ids = [p.id for p in projects]
+        if not project_ids:
+            return ListAssetsResponse(items=[], total=0, skip=skip, limit=limit)
+
+        total = asset_repository.count_by_project_ids(project_ids)
+        # 跨项目分页：逐项目累积直到凑够一页
+        paged_items: list = []
+        offset = skip
+        remaining = limit
+        for pid in project_ids:
+            proj_total = asset_repository.count_by_project(pid)
+            if offset >= proj_total:
+                offset -= proj_total
+                continue
+            proj_items = asset_repository.find_by_project(pid, skip=offset, limit=remaining)
+            paged_items.extend(proj_items)
+            remaining -= len(proj_items)
+            offset = 0
+            if remaining <= 0:
+                break
+
+        return ListAssetsResponse(
+            items=[_to_asset_response(item) for item in paged_items],
+            total=total,
+            skip=skip,
+            limit=limit,
+        )
+
+    # ── 内存过滤路径：有 keyword/gender/style 时，加载全量后内存过滤 ──
     if library_id:
         library = asset_library_repository.get(library_id)
         if library is None:
@@ -121,41 +201,24 @@ def list_assets(
             all_items = asset_repository.find_by_library_and_file_type(library_id, kind_to_file_type[kind])
         else:
             all_items = asset_repository.find_by_library(library_id)
-        filtered = _apply_filters(all_items)
-        total = len(filtered)
-        paged = filtered[skip : skip + limit]
-        return ListAssetsResponse(
-            items=[_to_asset_response(item) for item in paged],
-            total=total,
-            skip=skip,
-            limit=limit,
-        )
-
-    # 模式2：指定 project_id → 返回该项目所有素材
-    if project_id:
+    elif project_id:
         _check_project_access(project_id, user_id, project_repository)
         all_items = asset_repository.find_by_project(project_id)
-        filtered = _apply_filters(all_items)
-        total = len(filtered)
-        paged = filtered[skip : skip + limit]
-        return ListAssetsResponse(
-            items=[_to_asset_response(item) for item in paged],
-            total=total,
-            skip=skip,
-            limit=limit,
-        )
+    else:
+        try:
+            projects = project_repository.find_accessible_projects(user_id)
+        except Exception:
+            logger.exception("查询用户可访问项目失败: user_id=%s", user_id)
+            return ListAssetsResponse(items=[], total=0, skip=skip, limit=limit)
+        all_items = []
+        for proj in projects:
+            all_items.extend(asset_repository.find_by_project(proj.id))
 
-    # 模式3：都不传 → 返回用户可访问的所有项目的所有素材
-    try:
-        projects = project_repository.find_accessible_projects(user_id)
-    except Exception:
-        logger.exception("查询用户可访问项目失败: user_id=%s", user_id)
-        return ListAssetsResponse(items=[], total=0, skip=skip, limit=limit)
-
-    all_items = []
-    for proj in projects:
-        all_items.extend(asset_repository.find_by_project(proj.id))
-    filtered = _apply_filters(all_items)
+    # 应用 kind 过滤（如果有）+ keyword/gender/style
+    if kind:
+        ft = kind_to_file_type.get(kind)
+        all_items = [i for i in all_items if i.mime_type and i.mime_type.startswith(ft or "")]
+    filtered = _apply_memory_filters(all_items)
     total = len(filtered)
     paged = filtered[skip : skip + limit]
     return ListAssetsResponse(
@@ -191,6 +254,35 @@ def update_asset_review_status(
     return _to_asset_response(updated)
 
 
+@router.post("/batch-delete", response_model=BatchDeleteResponse)
+def batch_delete_assets(
+    request: BatchDeleteRequest,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    asset_repository: Any = Depends(get_asset_repository),
+    project_repository: Any = Depends(get_project_repository),
+) -> BatchDeleteResponse:
+    """批量删除素材（配音素材等），需逐项校验项目权限。"""
+    user_id = authenticated_user.user.id
+    deleted_ids: list[str] = []
+    failed_ids: list[str] = []
+
+    for asset_id in request.ids:
+        item = asset_repository.find_by_id(asset_id)
+        if item is None:
+            failed_ids.append(asset_id)
+            continue
+        try:
+            _check_project_access(item.project_id, user_id, project_repository)
+            deleted_ids.append(asset_id)
+        except HTTPException:
+            failed_ids.append(asset_id)
+
+    if deleted_ids:
+        asset_repository.batch_delete(deleted_ids)
+
+    return BatchDeleteResponse(deleted_count=len(deleted_ids), failed_ids=failed_ids)
+
+
 @router.get("/{asset_id}", response_model=AssetResponse)
 def get_asset(
     asset_id: str,

apps/api/app/schemas/asset.py

@@ -53,6 +53,19 @@ class AssetResponse(BaseModel):
     uploaded_by_user_id: str
 
 
+class BatchDeleteRequest(BaseModel):
+    """批量删除请求。"""
+
+    ids: list[str] = Field(..., min_length=1, max_length=100, description="要删除的素材 ID 列表")
+
+
+class BatchDeleteResponse(BaseModel):
+    """批量删除响应。"""
+
+    deleted_count: int = Field(..., ge=0, description="实际删除数量")
+    failed_ids: list[str] = Field(default_factory=list, description="删除失败的 ID 列表")
+
+
 class ListAssetsResponse(BaseModel):
     items: list[AssetResponse]
     total: int = Field(default=0, ge=0)

packages/adapters/in_memory/asset_repository.py

@@ -42,3 +42,12 @@ class InMemoryAssetRepository:
             del self._assets[asset_id]
             return True
         return False
+
+    def batch_delete(self, asset_ids: list[str]) -> int:
+        """批量删除素材，返回实际删除数量。"""
+        count = 0
+        for aid in asset_ids:
+            if aid in self._assets:
+                del self._assets[aid]
+                count += 1
+        return count

packages/adapters/sqlalchemy_impl/asset_repository.py

@@ -120,6 +120,14 @@ class SQLAlchemyAssetRepository:
             return True
         return False
 
+    def batch_delete(self, asset_ids: list[str]) -> int:
+        """批量删除素材，返回实际删除数量。"""
+        if not asset_ids:
+            return 0
+        count = self.session.query(AssetModel).filter(AssetModel.id.in_(asset_ids)).delete(synchronize_session=False)
+        self.session.commit()
+        return count
+
     def count_by_project(self, project_id: str) -> int:
         return self.session.query(AssetModel).filter(AssetModel.project_id == project_id).count()
 

packages/ports/asset_repository.py

@@ -50,6 +50,11 @@ class AssetRepository(ABC):
     def delete(self, asset_id: str) -> bool:
         pass
 
+    @abstractmethod
+    def batch_delete(self, asset_ids: list[str]) -> int:
+        """批量删除素材，返回实际删除数量。"""
+        pass
+
     @abstractmethod
     def count_by_project(self, project_id: str) -> int:
         pass

tests/unit/test_asset_batch_delete.py

@@ -0,0 +1,77 @@
+"""批量删除素材 + 分页优化 单元测试。"""
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+from packages.adapters.in_memory.asset_repository import InMemoryAssetRepository
+from packages.domain import Asset, AssetStatus
+
+
+class TestBatchDelete:
+    """batch_delete 仓储方法测试。"""
+
+    def _make_repo_with_assets(self):
+        repo = InMemoryAssetRepository()
+        for i in range(5):
+            asset = Asset.create(
+                project_id="proj-1",
+                library_id="lib-1",
+                name=f"voice_{i}.mp3",
+                storage_key=f"uploads/voice_{i}.mp3",
+                mime_type="audio/mpeg",
+                status=AssetStatus.READY,
+            )
+            repo.create(asset)
+        return repo
+
+    def test_batch_delete_removes_multiple(self):
+        repo = InMemoryAssetRepository()
+        assets = []
+        for i in range(5):
+            asset = Asset.create(
+                project_id="proj-1",
+                library_id="lib-1",
+                name=f"voice_{i}.mp3",
+                storage_key=f"uploads/voice_{i}.mp3",
+                mime_type="audio/mpeg",
+            )
+            repo.create(asset)
+            assets.append(asset)
+
+        ids_to_delete = [assets[0].id, assets[2].id, assets[4].id]
+        deleted_count = repo.batch_delete(ids_to_delete)
+
+        assert deleted_count == 3
+        # 验证确实被删了
+        assert repo.get(assets[0].id) is None
+        assert repo.get(assets[2].id) is None
+        assert repo.get(assets[4].id) is None
+        # 验证其他还在
+        assert repo.get(assets[1].id) is not None
+        assert repo.get(assets[3].id) is not None
+
+    def test_batch_delete_empty_list(self):
+        repo = self._make_repo_with_assets()
+        assert repo.batch_delete([]) == 0
+
+    def test_batch_delete_nonexistent_ids(self):
+        repo = self._make_repo_with_assets()
+        deleted = repo.batch_delete(["nonexistent-1", "nonexistent-2"])
+        assert deleted == 0
+
+    def test_batch_delete_mixed_existing_and_nonexistent(self):
+        repo = InMemoryAssetRepository()
+        asset = Asset.create(
+            project_id="proj-1",
+            library_id="lib-1",
+            name="voice.mp3",
+            storage_key="uploads/voice.mp3",
+            mime_type="audio/mpeg",
+        )
+        repo.create(asset)
+
+        deleted = repo.batch_delete([asset.id, "nonexistent"])
+        assert deleted == 1
+        assert repo.get(asset.id) is None