You've already forked DataMate
feat: Enhance file tag update functionality with automatic format conversion (#84)
- Updated `update_file_tags` to support both simplified and full tag formats. - Introduced `TagFormatConverter` to handle conversion from simplified external tags to internal storage format. - Added logic to fetch and utilize the appropriate annotation template for conversion. - Improved error handling for missing templates and unknown controls during tag updates. - Created example script demonstrating the usage of the new tag format conversion feature. - Added unit tests for `TagFormatConverter` to ensure correct functionality and edge case handling.
This commit is contained in:
@@ -165,14 +165,20 @@ class Service:
|
||||
async def update_file_tags_partial(
|
||||
self,
|
||||
file_id: str,
|
||||
new_tags: List[Dict[str, Any]]
|
||||
new_tags: List[Dict[str, Any]],
|
||||
template_id: Optional[str] = None
|
||||
) -> tuple[bool, Optional[str], Optional[datetime]]:
|
||||
"""
|
||||
部分更新文件标签
|
||||
部分更新文件标签,支持自动格式转换
|
||||
|
||||
如果提供了 template_id,会自动将简化格式的标签转换为完整格式。
|
||||
简化格式: {"from_name": "x", "to_name": "y", "values": [...]}
|
||||
完整格式: {"id": "...", "from_name": "x", "to_name": "y", "type": "...", "value": {"type": [...]}}
|
||||
|
||||
Args:
|
||||
file_id: 文件ID
|
||||
new_tags: 新的标签列表(部分更新)
|
||||
new_tags: 新的标签列表(部分更新),可以是简化格式或完整格式
|
||||
template_id: 可选的模板ID,用于格式转换
|
||||
|
||||
Returns:
|
||||
(成功标志, 错误信息, 更新时间)
|
||||
@@ -190,6 +196,38 @@ class Service:
|
||||
logger.error(f"File not found: {file_id}")
|
||||
return False, f"File not found: {file_id}", None
|
||||
|
||||
# 如果提供了 template_id,尝试进行格式转换
|
||||
processed_tags = new_tags
|
||||
if template_id:
|
||||
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:
|
||||
# 使用 converter 转换标签格式
|
||||
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")
|
||||
|
||||
# 获取现有标签
|
||||
existing_tags: List[Dict[str, Any]] = file_record.tags or [] # type: ignore
|
||||
|
||||
@@ -197,7 +235,7 @@ class Service:
|
||||
tag_id_map = {tag.get('id'): idx for idx, tag in enumerate(existing_tags) if tag.get('id')}
|
||||
|
||||
# 更新或追加标签
|
||||
for new_tag in new_tags:
|
||||
for new_tag in processed_tags:
|
||||
tag_id = new_tag.get('id')
|
||||
if tag_id and tag_id in tag_id_map:
|
||||
# 更新现有标签
|
||||
|
||||
Reference in New Issue
Block a user