feat: File and Annotation 2-way sync implementation (#63)

* feat: Refactor configuration and sync logic for improved dataset handling and logging

* feat: Enhance annotation synchronization and dataset file management

- Added new fields `tags_updated_at` to `DatasetFiles` model for tracking the last update time of tags.
- Implemented new asynchronous methods in the Label Studio client for fetching, creating, updating, and deleting task annotations.
- Introduced bidirectional synchronization for annotations between DataMate and Label Studio, allowing for flexible data management.
- Updated sync service to handle annotation conflicts based on timestamps, ensuring data integrity during synchronization.
- Enhanced dataset file response model to include tags and their update timestamps.
- Modified database initialization script to create a new column for `tags_updated_at` in the dataset files table.
- Updated requirements to ensure compatibility with the latest dependencies.

Author: JasonW404

runtime/datamate-python/.env.example

@@ -1,94 +1,19 @@
-# ====================================
-# Label Studio Adapter Configuration
-# ====================================
-
-# =========================
-# 应用程序配置
-# =========================
-APP_NAME="Label Studio Adapter"
-APP_VERSION="1.0.0"
-APP_DESCRIPTION="Adapter for integrating Data Management System with Label Studio"
-DEBUG=true
-
-# =========================
-# 服务器配置
-# =========================
+# Dev settings
 HOST=0.0.0.0
 PORT=18000
 
-# =========================
-# 日志配置
-# =========================
-LOG_LEVEL=INFO
-
-# =========================
-# Label Studio 服务配置
-# =========================
-# Label Studio 服务地址（根据部署方式调整）
-# Docker 环境：http://label-studio:8080
-# 本地开发：http://127.0.0.1:8000
-LABEL_STUDIO_BASE_URL=http://label-studio:8080
-
-# Label Studio 用户名和密码（用于自动创建用户）
-LABEL_STUDIO_USERNAME=[email protected]
-LABEL_STUDIO_PASSWORD=password
-
-# Label Studio API 认证 Token（Legacy Token，推荐使用）
-# 从 Label Studio UI 的 Account & Settings > Access Token 获取
-LABEL_STUDIO_USER_TOKEN=your-label-studio-token-here
-
-# Label Studio 本地文件存储基础路径（容器内路径，用于 Docker 部署时的权限检查）
-LABEL_STUDIO_LOCAL_BASE=/label-studio/local_files
-
-# Label Studio 本地文件服务路径前缀（任务数据中的文件路径前缀）
-LABEL_STUDIO_FILE_PATH_PREFIX=/data/local-files/?d=
-
-# Label Studio 容器中的本地存储路径（用于配置 Local Storage）
-LABEL_STUDIO_LOCAL_STORAGE_DATASET_BASE_PATH=/label-studio/local_files/dataset
-LABEL_STUDIO_LOCAL_STORAGE_UPLOAD_BASE_PATH=/label-studio/local_files/upload
-
-# Label Studio 任务列表分页大小
-LS_TASK_PAGE_SIZE=1000
-
-# =========================
-# Data Management 服务配置
-# =========================
-# DM 存储文件夹前缀（通常与 Label Studio 的 local-files 文件夹映射一致）
-DM_FILE_PATH_PREFIX=/
+DEBUG=true
+LOG_LEVEL=DEBUG
+LOG_FILE_DIR=./logs
 
-# =========================
-# Adapter 数据库配置 (MySQL)
-# =========================
-# 优先级1：如果配置了 MySQL，将优先使用 MySQL 数据库
-MYSQL_HOST=adapter-db
+# DataBase
+MYSQL_HOST=localhost
 MYSQL_PORT=3306
-MYSQL_USER=label_studio_user
-MYSQL_PASSWORD=user_password
-MYSQL_DATABASE=label_studio_adapter
-
-# =========================
-# CORS 配置
-# =========================
-# 允许的来源（生产环境建议配置具体域名）
-ALLOWED_ORIGINS=["*"]
-
-# 允许的 HTTP 方法
-ALLOWED_METHODS=["*"]
-
-# 允许的请求头
-ALLOWED_HEADERS=["*"]
-
-# =========================
-# Docker Compose 配置
-# =========================
-# Docker Compose 项目名称前缀
-COMPOSE_PROJECT_NAME=ls-adapter
+MYSQL_USER=root
+MYSQL_PASSWORD=password
+MYSQL_DATABASE=datamate
 
-# =========================
-# 同步配置（未来扩展）
-# =========================
-# 批量同步任务的批次大小
-SYNC_BATCH_SIZE=100
+# Label Studio settings
+LABEL_STUDIO_BASE_URL=http://localhost:8080
 
-# 同步失败时的最大重试次数
-MAX_RETRIES=3
+LABEL_STUDIO_USER_TOKEN="demo_dev_token"

runtime/datamate-python/.gitignore

@@ -3,4 +3,6 @@
 .dev.env
 
 # logs
-logs/
+logs/
+
+doc/

runtime/datamate-python/app/core/config.py

@@ -1,48 +1,41 @@
 from pydantic_settings import BaseSettings
 from pydantic import model_validator
-from typing import Optional, List
-import os
-from pathlib import Path
+from typing import Optional
 
 class Settings(BaseSettings):
     """应用程序配置"""
 
     class Config:
         env_file = ".env"
         case_sensitive = False
-        extra = 'ignore'  # 允许额外字段（如 Shell 脚本专用的环境变量）
+        extra = 'ignore'
 
-    # =========================
-    # Adapter 服务配置
-    # =========================
-    app_name: str = "Label Studio Adapter"
+    # Service
+    app_name: str = "DataMate Python Backend"
     app_version: str = "1.0.0"
     app_description: str = "Adapter for integrating Data Management System with Label Studio"
 
-    # 日志配置
-    log_level: str = "INFO"
-    debug: bool = True
-    log_file_dir: str = "/var/log/datamate"
-
-    # 服务器配置
     host: str = "0.0.0.0"
-    port: int = 8000
-
-    # CORS配置
+    port: int = 18000
+    
+    # CORS
     # allowed_origins: List[str] = ["*"]
     # allowed_methods: List[str] = ["*"]
     # allowed_headers: List[str] = ["*"]
 
-    # MySQL数据库配置 (优先级1)
+    # Log
+    log_level: str = "INFO"
+    debug: bool = True
+    log_file_dir: str = "/var/log/datamate"
+
+    # Database
     mysql_host: str = "datamate-database"
     mysql_port: int = 3306
     mysql_user: str = "root"
     mysql_password: str = "password"
     mysql_database: str = "datamate"
 
-    # 直接数据库URL配置（如果提供，将覆盖上述配置）
-    # 初始值为空字符串，在 model_validator 中会被设置为完整的 URL
-    database_url: str = ""
+    database_url: str = ""  # Will be overridden by build_database_url() if not provided
 
     @model_validator(mode='after')
     def build_database_url(self):
@@ -55,22 +48,18 @@ def build_database_url(self):
         return self
 
 
-    # =========================
-    # Label Studio 服务配置
-    # =========================
+    # Label Studio
     label_studio_base_url: str = "http://label-studio:8000"
-    label_studio_username: Optional[str] = "[email protected]"  # Label Studio 用户名（用于登录）
-    label_studio_password: Optional[str] = "demoadmin"  # Label Studio 密码（用于登录）
+    label_studio_username: Optional[str] = "[email protected]"
+    label_studio_password: Optional[str] = "demoadmin"
     label_studio_user_token: Optional[str] = "abc123abc123"  # Legacy Token
 
-    label_studio_local_storage_dataset_base_path: str = "/label-studio/local"  # Label Studio容器中的本地存储基础路径
-    label_studio_file_path_prefix: str = "/data/local-files/?d="  # Label Studio本地文件服务路径前缀
+    label_studio_local_document_root: str = "/label-studio/local"  # Label Studio local file storage path
+    label_studio_file_path_prefix: str = "/data/local-files/?d="  # Label Studio local file serving URL prefix
 
     ls_task_page_size: int = 1000
 
-    # =========================
-    # Data Management 服务配置
-    # =========================
+    # DataMate
     dm_file_path_prefix: str = "/dataset"  # DM存储文件夹前缀
 
 # 全局设置实例

runtime/datamate-python/app/db/models/dataset_management.py

@@ -64,6 +64,7 @@ class DatasetFiles(Base):
     file_size = Column(BigInteger, default=0, comment="文件大小(字节)")
     check_sum = Column(String(64), nullable=True, comment="文件校验和")
     tags = Column(JSON, nullable=True, comment="文件标签信息")
+    tags_updated_at = Column(TIMESTAMP, nullable=True, comment="标签最后更新时间")
     dataset_filemetadata = Column("metadata", JSON, nullable=True, comment="文件元数据")
     status = Column(String(50), default='ACTIVE', comment="文件状态：ACTIVE/DELETED/PROCESSING")
     upload_time = Column(TIMESTAMP, server_default=func.current_timestamp(), comment="上传时间")

runtime/datamate-python/app/main.py

@@ -45,7 +45,7 @@ async def lifespan(app: FastAPI):
     yield
 
     # @shutdown
-    logger.info("DataMate Python Backend shutting down ...")
+    logger.info("DataMate Python Backend shutting down ...\n\n")
 
 # 创建FastAPI应用
 app = FastAPI(
@@ -69,11 +69,7 @@ async def lifespan(app: FastAPI):
 app.include_router(router)
 
 # 输出注册的路由（每行一个）
-logger.debug("Registered routes:")
-for route in app.routes:
-    route_path = getattr(route, "path", None)
-    if route_path:
-        logger.debug(f"  {route_path}")
+logger.debug(f"Registered routes refer to http://localhost:{settings.port}/redoc")
 
 # 注册全局异常处理器
 app.add_exception_handler(StarletteHTTPException, starlette_http_exception_handler) # type: ignore
@@ -102,7 +98,7 @@ async def root():
         data={
             "message": f"{settings.app_name} is running",
             "version": settings.app_version,
-            "docs_url": "/docs",
+            "docs_url": "/redoc",
             "label_studio_url": settings.label_studio_base_url
         }
     )

runtime/datamate-python/app/module/annotation/client/labelstudio/client.py

@@ -380,6 +380,148 @@ async def delete_project(self, project_id: int) -> bool:
             logger.error(f"Error while deleting project {project_id}: {e}")
             return False
 
+    async def get_task_annotations(
+        self,
+        task_id: int
+    ) -> Optional[List[Dict[str, Any]]]:
+        """获取任务的标注结果
+        
+        Args:
+            task_id: 任务ID
+            
+        Returns:
+            标注结果列表，每个标注包含完整的annotation信息
+        """
+        try:
+            logger.debug(f"Fetching annotations for task: {task_id}")
+            
+            response = await self.client.get(f"/api/tasks/{task_id}/annotations")
+            response.raise_for_status()
+            
+            annotations = response.json()
+            logger.debug(f"Fetched {len(annotations)} annotations for task {task_id}")
+            
+            return annotations
+            
+        except httpx.HTTPStatusError as e:
+            logger.error(f"Get task annotations failed HTTP {e.response.status_code}: {e.response.text}")
+            return None
+        except Exception as e:
+            logger.error(f"Error while getting task annotations: {e}")
+            return None
+    
+    async def create_annotation(
+        self,
+        task_id: int,
+        result: List[Dict[str, Any]],
+        completed_by: Optional[int] = None
+    ) -> Optional[Dict[str, Any]]:
+        """为任务创建新的标注
+        
+        Args:
+            task_id: 任务ID
+            result: 标注结果列表
+            completed_by: 完成标注的用户ID（可选）
+            
+        Returns:
+            创建的标注信息，失败返回None
+        """
+        try:
+            logger.debug(f"Creating annotation for task: {task_id}")
+            
+            annotation_data = {
+                "result": result,
+                "task": task_id
+            }
+            
+            if completed_by:
+                annotation_data["completed_by"] = completed_by
+            
+            response = await self.client.post(
+                f"/api/tasks/{task_id}/annotations",
+                json=annotation_data
+            )
+            response.raise_for_status()
+            
+            annotation = response.json()
+            logger.debug(f"Created annotation {annotation.get('id')} for task {task_id}")
+            
+            return annotation
+            
+        except httpx.HTTPStatusError as e:
+            logger.error(f"Create annotation failed HTTP {e.response.status_code}: {e.response.text}")
+            return None
+        except Exception as e:
+            logger.error(f"Error while creating annotation: {e}")
+            return None
+    
+    async def update_annotation(
+        self,
+        annotation_id: int,
+        result: List[Dict[str, Any]]
+    ) -> Optional[Dict[str, Any]]:
+        """更新已存在的标注
+        
+        Args:
+            annotation_id: 标注ID
+            result: 新的标注结果列表
+            
+        Returns:
+            更新后的标注信息，失败返回None
+        """
+        try:
+            logger.debug(f"Updating annotation: {annotation_id}")
+            
+            annotation_data = {
+                "result": result
+            }
+            
+            response = await self.client.patch(
+                f"/api/annotations/{annotation_id}",
+                json=annotation_data
+            )
+            response.raise_for_status()
+            
+            annotation = response.json()
+            logger.debug(f"Updated annotation {annotation_id}")
+            
+            return annotation
+            
+        except httpx.HTTPStatusError as e:
+            logger.error(f"Update annotation failed HTTP {e.response.status_code}: {e.response.text}")
+            return None
+        except Exception as e:
+            logger.error(f"Error while updating annotation: {e}")
+            return None
+    
+    async def delete_annotation(
+        self,
+        annotation_id: int
+    ) -> bool:
+        """删除标注
+        
+        Args:
+            annotation_id: 标注ID
+            
+        Returns:
+            成功返回True，失败返回False
+        """
+        try:
+            logger.debug(f"Deleting annotation: {annotation_id}")
+            
+            response = await self.client.delete(f"/api/annotations/{annotation_id}")
+            response.raise_for_status()
+            
+            logger.debug(f"Deleted annotation {annotation_id}")
+            return True
+            
+        except httpx.HTTPStatusError as e:
+            logger.error(f"Delete annotation failed HTTP {e.response.status_code}: {e.response.text}")
+            return False
+        except Exception as e:
+            logger.error(f"Error while deleting annotation: {e}")
+            return False
+    
     async def create_local_storage(
         self,
         project_id: int,