diff --git a/apps/api/app/api/routes/asset_libraries.py b/apps/api/app/api/routes/asset_libraries.py old mode 100644 new mode 100755 index 6b7c297b6..7b75e54c2 --- a/apps/api/app/api/routes/asset_libraries.py +++ b/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) diff --git a/apps/api/app/api/routes/assets.py b/apps/api/app/api/routes/assets.py old mode 100644 new mode 100755 index 0861a2cfb..491440148 --- a/apps/api/app/api/routes/assets.py +++ b/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) diff --git a/apps/api/app/schemas/asset.py b/apps/api/app/schemas/asset.py old mode 100644 new mode 100755 index 1c686772f..afd5f74be --- a/apps/api/app/schemas/asset.py +++ b/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): diff --git a/packages/adapters/in_memory/asset_repository.py b/packages/adapters/in_memory/asset_repository.py index 02f323736..c3fc13e5c 100755 --- a/packages/adapters/in_memory/asset_repository.py +++ b/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 diff --git a/packages/adapters/sqlalchemy_impl/asset_repository.py b/packages/adapters/sqlalchemy_impl/asset_repository.py old mode 100644 new mode 100755 index b340b4d75..801679ac3 --- a/packages/adapters/sqlalchemy_impl/asset_repository.py +++ b/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 diff --git a/packages/domain/entities.py b/packages/domain/entities.py index 9a76e46bd..0280e36e2 100755 --- a/packages/domain/entities.py +++ b/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": diff --git a/packages/ports/asset_repository.py b/packages/ports/asset_repository.py old mode 100644 new mode 100755 index b2dda54e9..e3b2cdbb8 --- a/packages/ports/asset_repository.py +++ b/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 diff --git a/tests/integration/test_assets_api.py b/tests/integration/test_assets_api.py old mode 100644 new mode 100755 index 6eb9b23fa..d15ef7df1 --- a/tests/integration/test_assets_api.py +++ b/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 diff --git a/tests/unit/test_asset_batch_delete.py b/tests/unit/test_asset_batch_delete.py old mode 100644 new mode 100755 index ef9a2f468..263b74186 --- a/tests/unit/test_asset_batch_delete.py +++ b/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 diff --git a/tests/unit/test_asset_batch_operations.py b/tests/unit/test_asset_batch_operations.py new file mode 100755 index 000000000..783450ce2 --- /dev/null +++ b/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