fix(annotation): fix file version check to compare with latest version by logical path

Problem:
check_file_version was comparing annotation version with the passed
file_id's version, but when files are updated, new file records are
created with higher versions and old ones are marked ARCHIVED.

Solution:
1. Query the latest ACTIVE file by logical_path
2. Compare annotation version with latest file version
3. Return latestFileId so frontend can switch to new version

Changes:
- check_file_version now queries latest version by logical_path
- Added latest_file_id to FileVersionCheckResponse schema
- Updated descriptions to clarify currentFileVersion is latest version

Database scenario:
- old file: id=6dae9f2f, version=1, status=ARCHIVED
- new file: id=3365b4e7, version=3, status=ACTIVE
- Both have same logical_path='rufus.ini'
- Now correctly detects version 3 > annotation version
This commit is contained in:
2026-02-06 15:11:54 +08:00
parent c6dccf5e29
commit 44a1f2193f
2 changed files with 34 additions and 8 deletions

View File

@@ -198,9 +198,9 @@ class UpsertAnnotationResponse(BaseModel):
class FileVersionCheckResponse(BaseModel):
"""文件版本检查响应"""
file_id: str = Field(..., alias="fileId", description="文件ID")
file_id: str = Field(..., alias="fileId", description="文件ID(标注关联的文件)")
current_file_version: int = Field(
..., alias="currentFileVersion", description="当前文件版本"
..., alias="currentFileVersion", description="当前最新文件版本"
)
annotation_file_version: Optional[int] = Field(
None, alias="annotationFileVersion", description="标注时的文件版本"
@@ -213,6 +213,9 @@ class FileVersionCheckResponse(BaseModel):
has_new_version: bool = Field(
..., alias="hasNewVersion", description="是否有新版本"
)
latest_file_id: Optional[str] = Field(
None, alias="latestFileId", description="最新版本文件的ID(如有新版本)"
)
model_config = ConfigDict(populate_by_name=True)

View File

@@ -1645,16 +1645,19 @@ class AnnotationEditorService:
"""
检查文件是否有新版本
通过比较同一逻辑路径下最新版本的文件与标注时的文件版本,
判断是否有新版本可用。
Args:
project_id: 标注项目ID
file_id: 文件ID
file_id: 文件ID(标注关联的文件ID)
Returns:
包含文件版本信息的字典
"""
project = await self._get_project_or_404(project_id)
# 获取文件信息
# 获取当前标注关联的文件信息
file_result = await self.db.execute(
select(DatasetFiles)
.join(LabelingProjectFile, LabelingProjectFile.file_id == DatasetFiles.id)
@@ -1670,7 +1673,22 @@ class AnnotationEditorService:
status_code=404, detail=f"文件不存在或不属于该项目: {file_id}"
)
# 获取标注信息
# 获取同一逻辑路径下的最新版本文件
# 用于判断是否有新版本
logical_path = file_record.logical_path
latest_file_result = await self.db.execute(
select(DatasetFiles)
.where(
DatasetFiles.dataset_id == project.dataset_id,
DatasetFiles.logical_path == logical_path,
DatasetFiles.status == "ACTIVE",
)
.order_by(DatasetFiles.version.desc())
.limit(1)
)
latest_file = latest_file_result.scalar_one_or_none()
# 获取标注信息(基于传入的 file_id)
annotation_result = await self.db.execute(
select(AnnotationResult).where(
AnnotationResult.project_id == project_id,
@@ -1679,24 +1697,29 @@ class AnnotationEditorService:
)
annotation = annotation_result.scalar_one_or_none()
current_file_version = file_record.version
# 最新文件版本(同一逻辑路径下版本最高的 ACTIVE 文件)
latest_file_version = latest_file.version if latest_file else file_record.version
# 标注时的文件版本
annotation_file_version = annotation.file_version if annotation else None
annotation_version_unknown = (
annotation is not None and annotation_file_version is None
)
# 判断是否有新版本:最新版本 > 标注时的版本
has_new_version = (
current_file_version > annotation_file_version
latest_file_version > annotation_file_version
if annotation_file_version is not None
else False
)
return {
"fileId": file_id,
"currentFileVersion": current_file_version,
"currentFileVersion": latest_file_version,
"annotationFileVersion": annotation_file_version,
"hasNewVersion": has_new_version,
"annotationVersionUnknown": annotation_version_unknown,
"latestFileId": latest_file.id if latest_file else file_id,
}
async def use_new_version(self, project_id: str, file_id: str) -> Dict[str, Any]: