feat: add PUT /{file_id}/replace endpoint for clearing file tags

runtime/datamate-python/app/module/annotation/interface/task.py

@@ -321,3 +321,81 @@ async def update_file_tags(
         message="标签更新成功",
         data=response_data
     )
+
+
+@router.put(
+    "/{file_id}/replace",
+    response_model=StandardResponse[UpdateFileTagsResponse],
+)
+async def replace_file_tags(
+    request: UpdateFileTagsRequest,
+    file_id: str = Path(..., description="文件ID"),
+    db: AsyncSession = Depends(get_db)
+):
+    """
+    Replace File Tags (Full Replacement with Auto Format Conversion)
+
+    完全替换文件的标签列表。与部分更新不同，此方法会：
+    - 空列表会清空所有标签
+    - 新标签列表完全替换原有标签（不合并）
+
+    支持两种标签格式：
+    1. 简化格式:
+       [{"from_name": "label", "to_name": "image", "values": ["cat", "dog"]}]
+
+    2. 完整格式:
+       [{"id": "...", "from_name": "label", "to_name": "image", "type": "choices",
+         "value": {"choices": ["cat", "dog"]}}]
+
+    系统会自动根据数据集关联的模板将简化格式转换为完整格式。
+    """
+    service = DatasetManagementService(db)
+
+    result = await db.execute(
+        select(DatasetFiles).where(DatasetFiles.id == file_id)
+    )
+    file_record = result.scalar_one_or_none()
+
+    if not file_record:
+        raise HTTPException(status_code=404, detail=f"File not found: {file_id}")
+
+    dataset_id = str(file_record.dataset_id)
+
+    mapping_service = DatasetMappingService(db)
+    template_id = await mapping_service.get_template_id_by_dataset_id(dataset_id)
+
+    if template_id:
+        logger.info(f"Found template {template_id} for dataset {dataset_id}, will auto-convert tag format")
+    else:
+        logger.warning(f"No template found for dataset {dataset_id}, tags must be in full format")
+
+    success, error_msg, updated_at = await service.replace_file_tags(
+        file_id=file_id,
+        new_tags=request.tags,
+        template_id=template_id
+    )
+
+    if not success:
+        if "not found" in (error_msg or "").lower():
+            raise HTTPException(status_code=404, detail=error_msg)
+        raise HTTPException(status_code=500, detail=error_msg or "替换标签失败")
+
+    result = await db.execute(
+        select(DatasetFiles).where(DatasetFiles.id == file_id)
+    )
+    file_record = result.scalar_one_or_none()
+
+    if not file_record:
+        raise HTTPException(status_code=404, detail=f"File not found: {file_id}")
+
+    response_data = UpdateFileTagsResponse(
+        fileId=file_id,
+        tags=file_record.tags or [],
+        tagsUpdatedAt=updated_at or datetime.now()
+    )
+
+    return StandardResponse(
+        code="0",
+        message="标签替换成功",
+        data=response_data
+    )

runtime/datamate-python/app/module/dataset/service/service.py

@@ -470,6 +470,84 @@ def _index_tag(idx: int, tag: Dict[str, Any]) -> None:
             await self.db.rollback()
             return False, str(e), None
 
+    async def replace_file_tags(
+        self,
+        file_id: str,
+        new_tags: List[Dict[str, Any]],
+        template_id: Optional[str] = None
+    ) -> tuple[bool, Optional[str], Optional[datetime]]:
+        """
+        完全替换文件标签（非部分更新）
+
+        与部分更新不同，此方法会完全替换标签列表：
+        - 空列表会清空所有标签
+        - 新标签列表会完全替换原有标签（不合并）
+
+        如果提供了 template_id，会自动将简化格式的标签转换为完整格式。
+
+        Args:
+            file_id: 文件ID
+            new_tags: 新的标签列表（完全替换），可以是简化格式或完整格式，空列表会清空所有标签
+            template_id: 可选的模板ID，用于格式转换
+
+        Returns:
+            (成功标志, 错误信息, 更新时间)
+        """
+        try:
+            logger.info(f"Replacing tags for file: {file_id}, new_tags count: {len(new_tags)}")
+
+            result = await self.db.execute(
+                select(DatasetFiles).where(DatasetFiles.id == file_id)
+            )
+            file_record = result.scalar_one_or_none()
+
+            if not file_record:
+                logger.error(f"File not found: {file_id}")
+                return False, f"File not found: {file_id}", None
+
+            processed_tags = new_tags
+            if template_id and new_tags:
+                logger.debug(f"Converting tags using template: {template_id}")
+
+                try:
+                    from app.db.models import AnnotationTemplate
+                    template_result = await self.db.execute(
+                        select(AnnotationTemplate).where(
+                            AnnotationTemplate.id == template_id,
+                            AnnotationTemplate.deleted_at.is_(None)
+                        )
+                    )
+                    template = template_result.scalar_one_or_none()
+
+                    if not template:
+                        logger.warning(f"Template {template_id} not found, skipping conversion")
+                    else:
+                        from app.module.annotation.utils import create_converter_from_template_config
+
+                        converter = create_converter_from_template_config(template.configuration)  # type: ignore
+                        processed_tags = converter.convert_if_needed(new_tags)
+
+                        logger.info(f"Converted {len(new_tags)} tags to full format")
+
+                except Exception as e:
+                    logger.error(f"Failed to convert tags using template: {e}")
+                    logger.warning("Continuing with original tag format")
+
+            update_time = datetime.utcnow()
+            file_record.tags = processed_tags  # type: ignore
+            file_record.tags_updated_at = update_time  # type: ignore
+
+            await self.db.commit()
+            await self.db.refresh(file_record)
+
+            logger.info(f"Successfully replaced tags for file: {file_id}, new tags count: {len(processed_tags)}")
+            return True, None, update_time
+
+        except Exception as e:
+            logger.error(f"Failed to replace tags for file {file_id}: {e}")
+            await self.db.rollback()
+            return False, str(e), None
+
     @staticmethod
     async def _get_or_create_dataset_directory(dataset: Dataset) -> str:
         """Get or create dataset directory"""