新增格式转换 1.创建meta的参数skip_meta 2.将meta.json改成meta.log

Author: 727854320

README_zh.md

@@ -10,6 +10,7 @@ pip install -r docker/requirements.txt
 ### 启动api-server
 ```cmd
 uvicorn data_server.main:app --reload
+uvicorn data_server.main:app --host 0.0.0.0 --port 8001
 ```
 
 ### 启动celery

data_celery/formatify/tasks.py

@@ -222,7 +222,7 @@ def get_progress_from_mongodb_logs(task_uid: str) -> Optional[dict]:
             collection = get_formatity_collection(client, task_uid)
 
             # Find logs containing progress information (sorted by time descending)
-            # Match format: Updated and uploaded meta.json (total: X, success: Y, failure: Z)
+            # Match format: Updated and uploaded meta.log (total: X, success: Y, failure: Z)
             # Or: All files processed. Total: X, Success: Y, Failure: Z
             progress_patterns = [
                 r'\(total:\s*(\d+),\s*success:\s*(\d+),\s*failure:\s*(\d+)\)',  # (total: X, success: Y, failure: Z)

data_server/api/endpoints/formatify.py

@@ -185,6 +185,31 @@ def add_mineru_backend_column():
                 logger.info("Column 'mineru_backend' added successfully to data_format_tasks table")
 
 
+def add_skip_meta_column():
+    """Add skip_meta column to data_format_tasks table"""
+    try:
+        logger.info("Checking if skip_meta column exists in data_format_tasks table...")
+        with get_sync_session() as session:
+            with session.begin():
+                result = session.execute(text("""
+                    SELECT column_name 
+                    FROM information_schema.columns 
+                    WHERE table_name = 'data_format_tasks' AND column_name = 'skip_meta';
+                """))
+
+                if not result.fetchone():
+                    logger.info("skip_meta column does not exist, adding it...")
+                    session.execute(text("ALTER TABLE data_format_tasks ADD COLUMN skip_meta BOOLEAN DEFAULT FALSE;"))
+                    logger.info("Column 'skip_meta' added successfully to data_format_tasks table")
+                else:
+                    logger.info("Column 'skip_meta' already exists in data_format_tasks table")
+    except Exception as e:
+        logger.error(f"Error adding skip_meta column: {e}")
+        import traceback
+        logger.error(traceback.format_exc())
+        logger.warning("Continuing despite error...")
+
+
 _initialized = False
 from data_server.database.bean.work import Worker
 from data_server.job.JobModels import Job
@@ -215,9 +240,28 @@ def create_tables():
 
     _initialized = True
 
-    add_first_op_column()
-    add_mineru_api_url_column()
-    add_mineru_backend_column()
+    logger.info("Starting database column migrations...")
+    try:
+        add_first_op_column()
+    except Exception as e:
+        logger.error(f"Error in add_first_op_column: {e}")
+    
+    try:
+        add_mineru_api_url_column()
+    except Exception as e:
+        logger.error(f"Error in add_mineru_api_url_column: {e}")
+    
+    try:
+        add_mineru_backend_column()
+    except Exception as e:
+        logger.error(f"Error in add_mineru_backend_column: {e}")
+    
+    try:
+        add_skip_meta_column()
+    except Exception as e:
+        logger.error(f"Error in add_skip_meta_column: {e}")
+    
+    logger.info("Database column migrations completed")
 def is_table_initialized(table_name: str) -> bool:
     """
     Check if a specific table contains any data.
@@ -236,6 +280,12 @@ def initialize_database():
     Automatically executes one-time deletion on first startup (tracked in deletion_status table).
     This should be called once when the application starts.
     """
+    # Ensure database schema migrations are applied (check on every startup)
+    try:
+        add_skip_meta_column()
+    except Exception as e:
+        logger.warning(f"Could not check/add skip_meta column: {e}")
+    
     tables_to_initialize = [
         'operator_info',
         'operator_config',

data_server/database/session.py

@@ -4,6 +4,7 @@
 import uuid, os
 from typing import List, Tuple, Optional
 from data_server.formatify.FormatifyTask import run_format_task, stop_celery_task
+from loguru import logger
 
 
 def greate_task_uid():
@@ -16,6 +17,10 @@ def create_formatify_task(db_session: Session, dataFormatTask: DataFormatTaskReq
 
     # create db model
     task_uid = greate_task_uid()
+    
+    # Prepare skip_meta value (use provided value or default to False)
+    skip_meta_value = dataFormatTask.skip_meta if dataFormatTask.skip_meta is not None else False
+    
     data_format_task_db = DataFormatTask(name=dataFormatTask.name,
                                          des=dataFormatTask.des,
                                          from_csg_hub_dataset_name=dataFormatTask.from_csg_hub_dataset_name,
@@ -30,9 +35,12 @@ def create_formatify_task(db_session: Session, dataFormatTask: DataFormatTaskReq
                                          to_data_type=dataFormatTask.to_data_type,
                                          mineru_api_url=dataFormatTask.mineru_api_url,
                                          mineru_backend=dataFormatTask.mineru_backend,
+                                         skip_meta=skip_meta_value,
                                          task_uid=task_uid,
                                          task_status=DataFormatTaskStatusEnum.WAITING.value,
                                          owner_id=user_id)
+    
+    logger.info(f"Created task with skip_meta={skip_meta_value} for task {task_uid}")
 
     db_session.add(data_format_task_db)
     db_session.commit()
@@ -76,7 +84,7 @@ def update_formatify_task(db_session: Session, formatify_id: int, dataFormatTask
         'name', 'des', 'from_csg_hub_dataset_name', 'from_csg_hub_dataset_id',
         'from_csg_hub_dataset_branch', 'from_data_type', 'to_csg_hub_dataset_name',
         'to_csg_hub_dataset_id', 'to_csg_hub_dataset_default_branch', 'to_data_type',
-        'mineru_api_url', 'mineru_backend'
+        'mineru_api_url', 'mineru_backend'  # 'skip_meta' temporarily removed
     }
     for field in updatable_fields:
         value = getattr(dataFormatTaskRequest, field, None)

data_server/formatify/FormatifyManager.py

@@ -55,6 +55,7 @@ class DataFormatTask(Base):
     owner_id = Column(Integer, comment="所属用户")
     mineru_api_url = Column(String(500), comment="MinerU API 地址")
     mineru_backend = Column(String(100), comment="MinerU 后端类型")
+    skip_meta = Column(Boolean, default=False, comment="如果为 True，则生成并上传 meta.log 文件；如果为 False，则不生成 meta.log 文件")
     start_run_at = Column(DateTime, comment='运行开始时间')
     end_run_at = Column(DateTime, comment='运行结束时间')
     created_at = Column(DateTime, default=datetime.datetime.now, comment='任务创建时间')
@@ -81,6 +82,7 @@ def to_dict(self):
             "owner_id": self.owner_id,
             "mineru_api_url": self.mineru_api_url,
             "mineru_backend": self.mineru_backend,
+            "skip_meta": getattr(self, 'skip_meta', False),  # Use getattr to handle missing column
             "start_run_at": self.start_run_at,
             "end_run_at": self.end_run_at,
             "created_at": self.created_at.strftime("%Y-%m-%d %H:%M:%S") if self.created_at else None,

data_server/formatify/FormatifyModels.py

@@ -17,3 +17,4 @@ class DataFormatTaskRequest(BaseModel):
     to_data_type: Optional[int] = None
     mineru_api_url: Optional[str] = None
     mineru_backend: Optional[str] = None
+    skip_meta: Optional[bool] = False  # 如果为 True，则生成并上传 meta.log 文件；如果为 False，则不生成 meta.log 文件