chore: 修复black/isort格式

- AssetStatus 新增 DELETED 枚举值
- batch_delete 从硬删除改为软删除（标记 status=deleted）
- Repository 新增 batch_update_metadata / batch_add_tags / batch_replace_tags
- 新增4个API接口：
  - POST /assets/batch-delete（软删除，字段从ids改为asset_ids）
  - POST /assets/batch-tag（add/replace两种模式）
  - POST /assets/batch-classify（设置metadata.category）
  - POST /assets/batch-mark（设置metadata.smart_view）
- 单操作上限200个，返回成功/失败明细
- 素材库删除改为硬删除（循环调用单条delete）
- 20个新增单测 + 更新旧单测，共88个素材相关测试通过
- 集成测试MockRepository同步更新

apps/api/app/api/routes/asset_libraries.py

@@ -163,11 +163,10 @@ def delete_asset_library(
     # 权限校验：检查用户是否有项目访问权限
     check_project_access(library.project_id, authenticated_user.user.id, project_repository)
 
-    # 删除库内所有素材（无 FK 级联，需手动清理）
+    # 删除库内所有素材（硬删除，素材库已删除，无需保留软删除状态）
     assets_in_library = asset_repository.find_by_library(library_id)
-    if assets_in_library:
-        asset_ids_to_delete = [a.id for a in assets_in_library]
-        asset_repository.batch_delete(asset_ids_to_delete)
+    for asset in assets_in_library:
+        asset_repository.delete(asset.id)
 
     # 删除素材库本身
     asset_library_repository.delete(library_id)

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

@@ -12,8 +12,11 @@ from app.dependencies import (
 )
 from app.schemas.asset import (
     AssetResponse,
+    BatchClassifyRequest,
     BatchDeleteRequest,
-    BatchDeleteResponse,
+    BatchMarkRequest,
+    BatchOperationResponse,
+    BatchTagRequest,
     CreateAssetRequest,
     ListAssetsResponse,
     UpdateAssetRequest,
@@ -260,33 +263,157 @@ def update_asset_review_status(
     return _to_asset_response(updated)
 
 
-@router.post("/batch-delete", response_model=BatchDeleteResponse)
+@router.post("/batch-delete", response_model=BatchOperationResponse)
 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:
-    """批量删除素材（配音素材等），需逐项校验项目权限。"""
+) -> BatchOperationResponse:
+    """批量删除素材（软删除，标记 status=deleted），需逐项校验项目权限。"""
     user_id = authenticated_user.user.id
-    deleted_ids: list[str] = []
-    failed_ids: list[str] = []
+    success_ids: list[str] = []
+    failed_details: dict[str, str] = {}
 
-    for asset_id in request.ids:
+    for asset_id in request.asset_ids:
         item = asset_repository.find_by_id(asset_id)
         if item is None:
-            failed_ids.append(asset_id)
+            failed_details[asset_id] = "not_found"
             continue
         try:
             check_project_access(item.project_id, user_id, project_repository)
-            deleted_ids.append(asset_id)
+            success_ids.append(asset_id)
         except HTTPException:
-            failed_ids.append(asset_id)
+            failed_details[asset_id] = "access_denied"
 
-    if deleted_ids:
-        asset_repository.batch_delete(deleted_ids)
+    if success_ids:
+        asset_repository.batch_delete(success_ids)
 
-    return BatchDeleteResponse(deleted_count=len(deleted_ids), failed_ids=failed_ids)
+    return BatchOperationResponse(
+        success_count=len(success_ids),
+        failed_ids=list(failed_details.keys()),
+        failed_details=failed_details,
+    )
+
+
+@router.post("/batch-tag", response_model=BatchOperationResponse)
+def batch_tag_assets(
+    request: BatchTagRequest,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    asset_repository: Any = Depends(get_asset_repository),
+    project_repository: Any = Depends(get_project_repository),
+    tag_repository: Any = Depends(get_tag_repository),
+) -> BatchOperationResponse:
+    """批量打标签（添加或替换模式），需逐项校验项目权限和标签权限。"""
+    user_id = authenticated_user.user.id
+    success_ids: list[str] = []
+    failed_details: dict[str, str] = {}
+
+    # 校验标签存在且属于当前用户
+    for tag_id in request.tag_ids:
+        tag = tag_repository.get(tag_id)
+        if tag is None:
+            return BatchOperationResponse(
+                success_count=0,
+                failed_ids=list(request.asset_ids),
+                failed_details={aid: f"tag_not_found:{tag_id}" for aid in request.asset_ids},
+            )
+        if tag.user_id != user_id:
+            return BatchOperationResponse(
+                success_count=0,
+                failed_ids=list(request.asset_ids),
+                failed_details={aid: f"tag_access_denied:{tag_id}" for aid in request.asset_ids},
+            )
+
+    # 校验素材权限
+    for asset_id in request.asset_ids:
+        item = asset_repository.find_by_id(asset_id)
+        if item is None:
+            failed_details[asset_id] = "not_found"
+            continue
+        try:
+            check_project_access(item.project_id, user_id, project_repository)
+            success_ids.append(asset_id)
+        except HTTPException:
+            failed_details[asset_id] = "access_denied"
+
+    if success_ids:
+        if request.mode == "replace":
+            asset_repository.batch_replace_tags(success_ids, request.tag_ids)
+        else:
+            asset_repository.batch_add_tags(success_ids, request.tag_ids)
+
+    return BatchOperationResponse(
+        success_count=len(success_ids),
+        failed_ids=list(failed_details.keys()),
+        failed_details=failed_details,
+    )
+
+
+@router.post("/batch-classify", response_model=BatchOperationResponse)
+def batch_classify_assets(
+    request: BatchClassifyRequest,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    asset_repository: Any = Depends(get_asset_repository),
+    project_repository: Any = Depends(get_project_repository),
+) -> BatchOperationResponse:
+    """批量修改素材内容分类（person/scenic/product等），存在metadata.category中。"""
+    user_id = authenticated_user.user.id
+    success_ids: list[str] = []
+    failed_details: dict[str, str] = {}
+
+    for asset_id in request.asset_ids:
+        item = asset_repository.find_by_id(asset_id)
+        if item is None:
+            failed_details[asset_id] = "not_found"
+            continue
+        try:
+            check_project_access(item.project_id, user_id, project_repository)
+            success_ids.append(asset_id)
+        except HTTPException:
+            failed_details[asset_id] = "access_denied"
+
+    if success_ids:
+        asset_repository.batch_update_metadata(success_ids, {"category": request.category})
+
+    return BatchOperationResponse(
+        success_count=len(success_ids),
+        failed_ids=list(failed_details.keys()),
+        failed_details=failed_details,
+    )
+
+
+@router.post("/batch-mark", response_model=BatchOperationResponse)
+def batch_mark_assets(
+    request: BatchMarkRequest,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    asset_repository: Any = Depends(get_asset_repository),
+    project_repository: Any = Depends(get_project_repository),
+) -> BatchOperationResponse:
+    """批量设置智能视图标记（recommended/caution/high_risk），存在metadata.smart_view中。"""
+    user_id = authenticated_user.user.id
+    success_ids: list[str] = []
+    failed_details: dict[str, str] = {}
+
+    for asset_id in request.asset_ids:
+        item = asset_repository.find_by_id(asset_id)
+        if item is None:
+            failed_details[asset_id] = "not_found"
+            continue
+        try:
+            check_project_access(item.project_id, user_id, project_repository)
+            success_ids.append(asset_id)
+        except HTTPException:
+            failed_details[asset_id] = "access_denied"
+
+    if success_ids:
+        asset_repository.batch_update_metadata(success_ids, {"smart_view": request.smart_view})
+
+    return BatchOperationResponse(
+        success_count=len(success_ids),
+        failed_ids=list(failed_details.keys()),
+        failed_details=failed_details,
+    )
 
 
 @router.get("/{asset_id}", response_model=AssetResponse)

apps/api/app/schemas/asset.py

@@ -54,17 +54,45 @@ class AssetResponse(BaseModel):
     tag_ids: list[str] = Field(default_factory=list)
 
 
+MAX_BATCH_SIZE = 200
+
+
 class BatchDeleteRequest(BaseModel):
-    """批量删除请求。"""
+    """批量删除请求（软删除）。"""
 
-    ids: list[str] = Field(..., min_length=1, max_length=100, description="要删除的素材 ID 列表")
+    asset_ids: list[str] = Field(..., min_length=1, max_length=MAX_BATCH_SIZE, description="要删除的素材 ID 列表")
 
 
-class BatchDeleteResponse(BaseModel):
-    """批量删除响应。"""
+class BatchOperationResponse(BaseModel):
+    """批量操作通用响应。"""
 
-    deleted_count: int = Field(..., ge=0, description="实际删除数量")
-    failed_ids: list[str] = Field(default_factory=list, description="删除失败的 ID 列表")
+    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="失败详情 {asset_id: reason}")
+
+
+class BatchTagRequest(BaseModel):
+    """批量打标签请求。"""
+
+    asset_ids: list[str] = Field(..., min_length=1, max_length=MAX_BATCH_SIZE, description="素材 ID 列表")
+    tag_ids: list[str] = Field(..., min_length=1, max_length=50, description="标签 ID 列表")
+    mode: str = Field(default="add", pattern="^(add|replace)$", description="add=添加合并，replace=全量替换")
+
+
+class BatchClassifyRequest(BaseModel):
+    """批量修改分类请求。"""
+
+    asset_ids: list[str] = Field(..., min_length=1, max_length=MAX_BATCH_SIZE, description="素材 ID 列表")
+    category: str = Field(..., min_length=1, max_length=50, description="内容分类，如 person/scenic/product")
+
+
+class BatchMarkRequest(BaseModel):
+    """批量设置智能视图标记请求。"""
+
+    asset_ids: list[str] = Field(..., min_length=1, max_length=MAX_BATCH_SIZE, description="素材 ID 列表")
+    smart_view: str = Field(
+        ..., pattern="^(recommended|caution|high_risk)$", description="智能视图标记：recommended/caution/high_risk"
+    )
 
 
 class ListAssetsResponse(BaseModel):

packages/adapters/in_memory/asset_repository.py

@@ -44,11 +44,61 @@ class InMemoryAssetRepository:
         return False
 
     def batch_delete(self, asset_ids: list[str]) -> int:
-        """批量删除素材，返回实际删除数量。"""
+        """批量删除素材（软删除，标记 status=deleted），返回实际影响数量。"""
+        from datetime import datetime, timezone
+
+        from packages.domain import AssetStatus
+
         count = 0
         for aid in asset_ids:
-            if aid in self._assets:
-                del self._assets[aid]
+            asset = self._assets.get(aid)
+            if asset and asset.status != AssetStatus.DELETED:
+                asset.status = AssetStatus.DELETED
+                asset.updated_at = datetime.now(timezone.utc)
+                count += 1
+        return count
+
+    def batch_update_metadata(self, asset_ids: list[str], metadata_patch: dict[str, object]) -> int:
+        """批量更新素材 metadata（合并 patch），返回实际影响数量。"""
+        from datetime import datetime, timezone
+
+        count = 0
+        for aid in asset_ids:
+            asset = self._assets.get(aid)
+            if asset:
+                asset.metadata = {**asset.metadata, **metadata_patch}
+                asset.updated_at = datetime.now(timezone.utc)
+                count += 1
+        return count
+
+    def batch_add_tags(self, asset_ids: list[str], tag_ids: list[str]) -> int:
+        """批量给素材添加标签（合并去重），返回实际影响数量。"""
+        from datetime import datetime, timezone
+
+        count = 0
+        for aid in asset_ids:
+            asset = self._assets.get(aid)
+            if asset:
+                changed = False
+                for tid in tag_ids:
+                    if tid not in asset.tag_ids:
+                        asset.tag_ids.append(tid)
+                        changed = True
+                if changed:
+                    asset.updated_at = datetime.now(timezone.utc)
+                    count += 1
+        return count
+
+    def batch_replace_tags(self, asset_ids: list[str], tag_ids: list[str]) -> int:
+        """批量替换素材标签（全量覆盖），返回实际影响数量。"""
+        from datetime import datetime, timezone
+
+        count = 0
+        for aid in asset_ids:
+            asset = self._assets.get(aid)
+            if asset:
+                asset.tag_ids = list(tag_ids)
+                asset.updated_at = datetime.now(timezone.utc)
                 count += 1
         return count
 

packages/adapters/sqlalchemy_impl/asset_repository.py

@@ -127,10 +127,90 @@ class SQLAlchemyAssetRepository:
         return False
 
     def batch_delete(self, asset_ids: list[str]) -> int:
-        """批量删除素材，返回实际删除数量。"""
+        """批量删除素材（软删除，标记 status=deleted），返回实际影响数量。"""
         if not asset_ids:
             return 0
-        count = self.session.query(AssetModel).filter(AssetModel.id.in_(asset_ids)).delete(synchronize_session=False)
+        from datetime import datetime, timezone
+
+        now = datetime.now(timezone.utc)
+        count = (
+            self.session.query(AssetModel)
+            .filter(AssetModel.id.in_(asset_ids), AssetModel.status != "deleted")
+            .update({AssetModel.status: "deleted", AssetModel.updated_at: now}, synchronize_session=False)
+        )
+        self.session.commit()
+        return count
+
+    def batch_update_metadata(self, asset_ids: list[str], metadata_patch: dict[str, object]) -> int:
+        """批量更新素材 metadata（合并 patch），返回实际影响数量。"""
+        if not asset_ids:
+            return 0
+        from datetime import datetime, timezone
+
+        now = datetime.now(timezone.utc)
+        # 逐条读取 + 合并 + 更新，保证 JSON 合并正确
+        models = self.session.query(AssetModel).filter(AssetModel.id.in_(asset_ids)).all()
+        count = 0
+        for model in models:
+            existing = {}
+            if model.classification_result:
+                try:
+                    existing = json.loads(model.classification_result)
+                except Exception:
+                    existing = {}
+            merged = {**existing, **metadata_patch}
+            model.classification_result = json.dumps(merged, ensure_ascii=False)
+            model.updated_at = now
+            count += 1
+        self.session.commit()
+        return count
+
+    def batch_add_tags(self, asset_ids: list[str], tag_ids: list[str]) -> int:
+        """批量给素材添加标签（合并去重），返回实际影响数量。"""
+        if not asset_ids or not tag_ids:
+            return 0
+        from datetime import datetime, timezone
+
+        now = datetime.now(timezone.utc)
+        clean_tag_ids = list(set(tag_ids))
+        count = 0
+        for aid in asset_ids:
+            # 查询现有标签
+            existing = {
+                row.tag_id
+                for row in self.session.query(AssetTagModel.tag_id).filter(AssetTagModel.asset_id == aid).all()
+            }
+            new_tags = [t for t in clean_tag_ids if t not in existing]
+            if new_tags:
+                for tid in new_tags:
+                    self.session.add(AssetTagModel(asset_id=aid, tag_id=tid))
+                # 更新 updated_at
+                self.session.query(AssetModel).filter(AssetModel.id == aid).update(
+                    {AssetModel.updated_at: now}, synchronize_session=False
+                )
+                count += 1
+        self.session.commit()
+        return count
+
+    def batch_replace_tags(self, asset_ids: list[str], tag_ids: list[str]) -> int:
+        """批量替换素材标签（全量覆盖），返回实际影响数量。"""
+        if not asset_ids:
+            return 0
+        from datetime import datetime, timezone
+
+        now = datetime.now(timezone.utc)
+        clean_tag_ids = list(set(tag_ids))
+        count = 0
+        for aid in asset_ids:
+            # 先删再加
+            self.session.query(AssetTagModel).filter(AssetTagModel.asset_id == aid).delete(synchronize_session=False)
+            for tid in clean_tag_ids:
+                self.session.add(AssetTagModel(asset_id=aid, tag_id=tid))
+            # 更新 updated_at
+            self.session.query(AssetModel).filter(AssetModel.id == aid).update(
+                {AssetModel.updated_at: now}, synchronize_session=False
+            )
+            count += 1
         self.session.commit()
         return count
 

packages/domain/entities.py

@@ -133,6 +133,7 @@ class AssetStatus(StrEnum):
     READY = "ready"
     PROCESSING = "processing"
     ERROR = "error"
+    DELETED = "deleted"
 
     @classmethod
     def _missing_(cls, value: object) -> "AssetStatus":

packages/ports/asset_repository.py

@@ -52,7 +52,22 @@ class AssetRepository(ABC):
 
     @abstractmethod
     def batch_delete(self, asset_ids: list[str]) -> int:
-        """批量删除素材，返回实际删除数量。"""
+        """批量删除素材（软删除，标记 status=deleted），返回实际影响数量。"""
+        pass
+
+    @abstractmethod
+    def batch_update_metadata(self, asset_ids: list[str], metadata_patch: dict[str, object]) -> int:
+        """批量更新素材 metadata（合并 patch），返回实际影响数量。"""
+        pass
+
+    @abstractmethod
+    def batch_add_tags(self, asset_ids: list[str], tag_ids: list[str]) -> int:
+        """批量给素材添加标签（合并去重），返回实际影响数量。"""
+        pass
+
+    @abstractmethod
+    def batch_replace_tags(self, asset_ids: list[str], tag_ids: list[str]) -> int:
+        """批量替换素材标签（全量覆盖），返回实际影响数量。"""
         pass
 
     @abstractmethod

tests/integration/test_assets_api.py

@@ -129,10 +129,58 @@ class StubAssetRepository:
         return False
 
     def batch_delete(self, asset_ids: list[str]) -> int:
+        """软删除：标记 status=deleted。"""
+        from datetime import datetime, timezone
+
+        from packages.domain import AssetStatus
+
         count = 0
         for aid in asset_ids:
-            if aid in self._assets:
-                del self._assets[aid]
+            asset = self._assets.get(aid)
+            if asset and asset.status != AssetStatus.DELETED:
+                asset.status = AssetStatus.DELETED
+                asset.updated_at = datetime.now(timezone.utc)
+                count += 1
+        return count
+
+    def batch_update_metadata(self, asset_ids: list[str], metadata_patch: dict) -> int:
+        from datetime import datetime, timezone
+
+        count = 0
+        for aid in asset_ids:
+            asset = self._assets.get(aid)
+            if asset:
+                asset.metadata = {**asset.metadata, **metadata_patch}
+                asset.updated_at = datetime.now(timezone.utc)
+                count += 1
+        return count
+
+    def batch_add_tags(self, asset_ids: list[str], tag_ids: list[str]) -> int:
+        from datetime import datetime, timezone
+
+        count = 0
+        for aid in asset_ids:
+            asset = self._assets.get(aid)
+            if asset:
+                changed = False
+                for tid in tag_ids:
+                    if tid not in asset.tag_ids:
+                        asset.tag_ids.append(tid)
+                        changed = True
+                if changed:
+                    asset.updated_at = datetime.now(timezone.utc)
+                    count += 1
+        return count
+
+    def batch_replace_tags(self, asset_ids: list[str], tag_ids: list[str]) -> int:
+        from datetime import datetime, timezone
+
+        count = 0
+        for aid in asset_ids:
+            asset = self._assets.get(aid)
+            if asset:
+                asset.tag_ids = list(tag_ids)
+                asset.updated_at = datetime.now(timezone.utc)
                 count += 1
         return count
 
@@ -613,17 +661,22 @@ class TestBatchDeleteAssets:
         return ids
 
     def test_batch_delete_success(self, client):
-        """批量删除成功。"""
+        """批量删除成功（软删除）。"""
         ids = self._create_assets(client, 3)
 
         resp = client.post(
             "/api/v1/assets/batch-delete",
-            json={"ids": ids[:2]},
+            json={"asset_ids": ids[:2]},
         )
         assert resp.status_code == 200
         data = resp.json()
-        assert data["deleted_count"] == 2
+        assert data["success_count"] == 2
         assert len(data["failed_ids"]) == 0
+        # 软删除：记录仍在，status 变为 deleted
+        for aid in ids[:2]:
+            r = client.get(f"/api/v1/assets/{aid}")
+            assert r.status_code == 200
+            assert r.json()["status"] == "deleted"
 
     def test_batch_delete_with_nonexistent_ids(self, client):
         """批量删除包含不存在的 ID，失败的计入 failed_ids。"""
@@ -632,18 +685,20 @@ class TestBatchDeleteAssets:
 
         resp = client.post(
             "/api/v1/assets/batch-delete",
-            json={"ids": ids},
+            json={"asset_ids": ids},
         )
         assert resp.status_code == 200
         data = resp.json()
-        assert data["deleted_count"] == 2
+        assert data["success_count"] == 2
         assert "nonexistent-id" in data["failed_ids"]
+        assert "nonexistent-id" in data["failed_details"]
+        assert data["failed_details"]["nonexistent-id"] == "not_found"
 
     def test_batch_delete_empty_list_returns_422(self, client):
         """空列表返回 422。"""
         resp = client.post(
             "/api/v1/assets/batch-delete",
-            json={"ids": []},
+            json={"asset_ids": []},
         )
         assert resp.status_code == 422
 

tests/unit/test_asset_batch_delete.py

@@ -1,4 +1,7 @@
-"""批量删除素材 + 分页优化 单元测试。"""
+"""批量删除素材 + 分页优化 单元测试。
+
+注意：batch_delete 现在是软删除（标记 status=deleted），不是硬删除。
+"""
 
 import sys
 from pathlib import Path
@@ -10,7 +13,7 @@ from packages.domain import Asset, AssetStatus
 
 
 class TestBatchDelete:
-    """batch_delete 仓储方法测试。"""
+    """batch_delete 仓储方法测试（软删除）。"""
 
     def _make_repo_with_assets(self):
         repo = InMemoryAssetRepository()
@@ -26,7 +29,8 @@ class TestBatchDelete:
             repo.create(asset)
         return repo
 
-    def test_batch_delete_removes_multiple(self):
+    def test_batch_delete_marks_deleted_status(self):
+        """软删除：status 变为 deleted，记录仍然存在。"""
         repo = InMemoryAssetRepository()
         assets = []
         for i in range(5):
@@ -36,6 +40,7 @@ class TestBatchDelete:
                 name=f"voice_{i}.mp3",
                 storage_key=f"uploads/voice_{i}.mp3",
                 mime_type="audio/mpeg",
+                status=AssetStatus.READY,
             )
             repo.create(asset)
             assets.append(asset)
@@ -44,13 +49,13 @@ class TestBatchDelete:
         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
+        # 软删除：记录仍在，status 变为 deleted
+        assert repo.get(assets[0].id).status == AssetStatus.DELETED
+        assert repo.get(assets[2].id).status == AssetStatus.DELETED
+        assert repo.get(assets[4].id).status == AssetStatus.DELETED
+        # 未删除的保持 ready
+        assert repo.get(assets[1].id).status == AssetStatus.READY
+        assert repo.get(assets[3].id).status == AssetStatus.READY
 
     def test_batch_delete_empty_list(self):
         repo = self._make_repo_with_assets()
@@ -69,9 +74,27 @@ class TestBatchDelete:
             name="voice.mp3",
             storage_key="uploads/voice.mp3",
             mime_type="audio/mpeg",
+            status=AssetStatus.READY,
         )
         repo.create(asset)
 
         deleted = repo.batch_delete([asset.id, "nonexistent"])
         assert deleted == 1
-        assert repo.get(asset.id) is None
+        assert repo.get(asset.id).status == AssetStatus.DELETED
+
+    def test_batch_delete_idempotent(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",
+            status=AssetStatus.READY,
+        )
+        repo.create(asset)
+
+        assert repo.batch_delete([asset.id]) == 1
+        assert repo.batch_delete([asset.id]) == 0
+        assert repo.get(asset.id).status == AssetStatus.DELETED

tests/unit/test_asset_batch_operations.py

@@ -0,0 +1,288 @@
+"""素材批量操作单元测试：软删除、批量打标签、批量分类、批量智能视图标记。"""
+
+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 TestBatchSoftDelete:
+    """batch_delete 软删除测试。"""
+
+    def _make_assets(self, repo: InMemoryAssetRepository, count: int = 5) -> list[Asset]:
+        assets = []
+        for i in range(count):
+            asset = Asset.create(
+                project_id="proj-1",
+                library_id="lib-1",
+                name=f"asset_{i}.mp4",
+                storage_key=f"uploads/asset_{i}.mp4",
+                mime_type="video/mp4",
+                status=AssetStatus.READY,
+            )
+            repo.create(asset)
+            assets.append(asset)
+        return assets
+
+    def test_batch_soft_delete_marks_status_deleted(self):
+        """软删除：status 变为 deleted，记录仍然存在。"""
+        repo = InMemoryAssetRepository()
+        assets = self._make_assets(repo, 3)
+
+        ids_to_delete = [assets[0].id, assets[2].id]
+        count = repo.batch_delete(ids_to_delete)
+
+        assert count == 2
+        # 记录仍在，只是 status 变了
+        assert repo.get(assets[0].id) is not None
+        assert repo.get(assets[0].id).status == AssetStatus.DELETED
+        assert repo.get(assets[2].id).status == AssetStatus.DELETED
+        # 未删除的保持原样
+        assert repo.get(assets[1].id).status == AssetStatus.READY
+
+    def test_batch_soft_delete_idempotent(self):
+        """重复删除已删除的素材，计数不增加。"""
+        repo = InMemoryAssetRepository()
+        assets = self._make_assets(repo, 2)
+
+        count1 = repo.batch_delete([assets[0].id])
+        count2 = repo.batch_delete([assets[0].id])
+
+        assert count1 == 1
+        assert count2 == 0
+        assert repo.get(assets[0].id).status == AssetStatus.DELETED
+
+    def test_batch_soft_delete_empty_list(self):
+        repo = InMemoryAssetRepository()
+        self._make_assets(repo, 3)
+        assert repo.batch_delete([]) == 0
+
+    def test_batch_soft_delete_nonexistent_ids(self):
+        repo = InMemoryAssetRepository()
+        self._make_assets(repo, 3)
+        assert repo.batch_delete(["nonexistent-1", "nonexistent-2"]) == 0
+
+
+class TestBatchUpdateMetadata:
+    """batch_update_metadata 批量更新 metadata 测试。"""
+
+    def test_batch_update_category(self):
+        """批量修改分类（metadata.category）。"""
+        repo = InMemoryAssetRepository()
+        assets = []
+        for i in range(3):
+            asset = Asset.create(
+                project_id="proj-1",
+                library_id="lib-1",
+                name=f"v{i}.mp4",
+                storage_key=f"v{i}.mp4",
+                mime_type="video/mp4",
+                metadata={"existing_key": "existing_value"},
+                status=AssetStatus.READY,
+            )
+            repo.create(asset)
+            assets.append(asset)
+
+        ids = [a.id for a in assets]
+        count = repo.batch_update_metadata(ids, {"category": "person"})
+
+        assert count == 3
+        for a in assets:
+            updated = repo.get(a.id)
+            assert updated.metadata["category"] == "person"
+            assert updated.metadata["existing_key"] == "existing_value"  # 合并而非覆盖
+
+    def test_batch_update_smart_view(self):
+        """批量设置智能视图标记。"""
+        repo = InMemoryAssetRepository()
+        assets = []
+        for i in range(4):
+            asset = Asset.create(
+                project_id="proj-1",
+                library_id="lib-1",
+                name=f"v{i}.mp4",
+                storage_key=f"v{i}.mp4",
+                mime_type="video/mp4",
+                status=AssetStatus.READY,
+            )
+            repo.create(asset)
+            assets.append(asset)
+
+        # 标记前2个为 recommended
+        count = repo.batch_update_metadata([assets[0].id, assets[1].id], {"smart_view": "recommended"})
+        assert count == 2
+        assert repo.get(assets[0].id).metadata["smart_view"] == "recommended"
+        assert repo.get(assets[1].id).metadata["smart_view"] == "recommended"
+        # 其余不变
+        assert "smart_view" not in repo.get(assets[2].id).metadata
+
+        # 再标记后2个为 high_risk
+        count2 = repo.batch_update_metadata([assets[2].id, assets[3].id], {"smart_view": "high_risk"})
+        assert count2 == 2
+        assert repo.get(assets[2].id).metadata["smart_view"] == "high_risk"
+        assert repo.get(assets[3].id).metadata["smart_view"] == "high_risk"
+
+    def test_batch_update_metadata_partial_existing(self):
+        """部分素材存在时，只更新存在的。"""
+        repo = InMemoryAssetRepository()
+        asset = Asset.create(
+            project_id="proj-1",
+            library_id="lib-1",
+            name="v.mp4",
+            storage_key="v.mp4",
+            mime_type="video/mp4",
+            status=AssetStatus.READY,
+        )
+        repo.create(asset)
+
+        count = repo.batch_update_metadata([asset.id, "nonexistent"], {"category": "scenic"})
+        assert count == 1
+        assert repo.get(asset.id).metadata["category"] == "scenic"
+
+    def test_batch_update_metadata_empty_list(self):
+        repo = InMemoryAssetRepository()
+        assert repo.batch_update_metadata([], {"category": "x"}) == 0
+
+
+class TestBatchAddTags:
+    """batch_add_tags 批量添加标签测试。"""
+
+    def test_batch_add_tags_merges_and_dedups(self):
+        """添加模式：合并去重，已有标签不重复添加。"""
+        repo = InMemoryAssetRepository()
+        assets = []
+        for i in range(3):
+            asset = Asset.create(
+                project_id="proj-1",
+                library_id="lib-1",
+                name=f"v{i}.mp4",
+                storage_key=f"v{i}.mp4",
+                mime_type="video/mp4",
+                status=AssetStatus.READY,
+            )
+            asset.add_tag("tag-existing")
+            repo.create(asset)
+            assets.append(asset)
+
+        ids = [a.id for a in assets]
+        count = repo.batch_add_tags(ids, ["tag-1", "tag-2", "tag-existing"])
+
+        assert count == 3  # 都有新增标签，所以都算变更
+        for a in assets:
+            updated = repo.get(a.id)
+            assert set(updated.tag_ids) == {"tag-existing", "tag-1", "tag-2"}
+
+    def test_batch_add_tags_no_change_when_all_exist(self):
+        """所有标签都已存在时，返回0。"""
+        repo = InMemoryAssetRepository()
+        asset = Asset.create(
+            project_id="proj-1",
+            library_id="lib-1",
+            name="v.mp4",
+            storage_key="v.mp4",
+            mime_type="video/mp4",
+            status=AssetStatus.READY,
+        )
+        asset.add_tag("tag-a")
+        asset.add_tag("tag-b")
+        repo.create(asset)
+
+        count = repo.batch_add_tags([asset.id], ["tag-a", "tag-b"])
+        assert count == 0
+
+    def test_batch_add_tags_empty_input(self):
+        repo = InMemoryAssetRepository()
+        assert repo.batch_add_tags([], ["tag-1"]) == 0
+        assert repo.batch_add_tags(["aid"], []) == 0
+
+
+class TestBatchReplaceTags:
+    """batch_replace_tags 批量替换标签测试。"""
+
+    def test_batch_replace_tags_full_override(self):
+        """替换模式：全量覆盖原有标签。"""
+        repo = InMemoryAssetRepository()
+        assets = []
+        for i in range(3):
+            asset = Asset.create(
+                project_id="proj-1",
+                library_id="lib-1",
+                name=f"v{i}.mp4",
+                storage_key=f"v{i}.mp4",
+                mime_type="video/mp4",
+                status=AssetStatus.READY,
+            )
+            asset.add_tag(f"old-{i}")
+            asset.add_tag("old-common")
+            repo.create(asset)
+            assets.append(asset)
+
+        ids = [a.id for a in assets]
+        count = repo.batch_replace_tags(ids, ["new-1", "new-2"])
+
+        assert count == 3
+        for a in assets:
+            updated = repo.get(a.id)
+            assert set(updated.tag_ids) == {"new-1", "new-2"}
+
+    def test_batch_replace_tags_empty_tags_clears_all(self):
+        """替换为空列表：清空所有标签。"""
+        repo = InMemoryAssetRepository()
+        asset = Asset.create(
+            project_id="proj-1",
+            library_id="lib-1",
+            name="v.mp4",
+            storage_key="v.mp4",
+            mime_type="video/mp4",
+            status=AssetStatus.READY,
+        )
+        asset.add_tag("tag-a")
+        asset.add_tag("tag-b")
+        repo.create(asset)
+
+        count = repo.batch_replace_tags([asset.id], [])
+        assert count == 1
+        assert repo.get(asset.id).tag_ids == []
+
+    def test_batch_replace_tags_empty_assets(self):
+        repo = InMemoryAssetRepository()
+        assert repo.batch_replace_tags([], ["tag-1"]) == 0
+
+
+class TestBatchOperationLimits:
+    """批量操作上限与边界测试。"""
+
+    def test_large_batch_operations(self):
+        """大量素材的批量操作（验证性能基本可用）。"""
+        repo = InMemoryAssetRepository()
+        assets = []
+        for i in range(50):
+            asset = Asset.create(
+                project_id="proj-1",
+                library_id="lib-1",
+                name=f"v{i}.mp4",
+                storage_key=f"v{i}.mp4",
+                mime_type="video/mp4",
+                status=AssetStatus.READY,
+            )
+            repo.create(asset)
+            assets.append(asset)
+
+        ids = [a.id for a in assets]
+
+        # 批量打标签
+        count = repo.batch_add_tags(ids, ["bulk-tag"])
+        assert count == 50
+
+        # 批量分类
+        count = repo.batch_update_metadata(ids, {"category": "scenic"})
+        assert count == 50
+
+        # 批量软删除
+        count = repo.batch_delete(ids)
+        assert count == 50
+        for a in assets:
+            assert repo.get(a.id).status == AssetStatus.DELETED