feat: 素材批量操作后端接口（软删除/批量打标签/批量分类/批量智能视图标记）

- 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同步更新

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