修改数据流向显示错误问题

Author: zhanglongbin

data_celery/datasource/mongo/tasks.py

@@ -11,7 +11,6 @@
 from data_server.datasource.services.datasource import get_datasource_connector
 from data_celery.mongo_tools.tools import insert_datasource_run_task_log_info, insert_datasource_run_task_log_error
 from data_engine.exporter.load import load_exporter
-from pathlib import Path
 import pandas as pd
 from loguru import logger
 
@@ -256,46 +255,21 @@ def upload_to_csg_hub_server(csg_hub_dataset_id: str,
             branch=csg_hub_dataset_default_branch,
             user_name=user_name,
             user_token=user_token,
-            work_dir=datasource_csg_hub_server_dir
+            work_dir=datasource_csg_hub_server_dir,
+            path_is_dir=True
         )
-        upload_path: Path = Path(datasource_temp_json_dir)
-        # Check whether the uploaded directory exists and is not empty
-        if not os.path.exists(upload_path):
-            insert_datasource_run_task_log_error(collection_task.task_uid,
-                                                 f"the task[{collection_task.task_uid}] upload csg hub-server fail: upload path {upload_path} does not exist")
-            return False
-
-        # List all files in the upload directory for debugging
-        file_list = []
-        for root, dirs, files in os.walk(upload_path):
-            for file in files:
-                file_list.append(os.path.join(root, file))
-        insert_datasource_run_task_log_info(collection_task.task_uid,
-                                            f"Files to upload: {len(file_list)} files found in {upload_path}")
-        if len(file_list) == 0:
-            insert_datasource_run_task_log_error(collection_task.task_uid,
-                                                 f"the task[{collection_task.task_uid}] upload csg hub-server fail: upload path {upload_path} is empty")
-            return False
-
-        output_branch_name = exporter.export_from_files(upload_path)
-
-        if output_branch_name:
-            collection_task.csg_hub_branch = output_branch_name
+        exporter.export_large_folder()
+        if csg_hub_dataset_default_branch:
+            collection_task.csg_hub_branch = csg_hub_dataset_default_branch
             db_session.commit()
             insert_datasource_run_task_log_info(collection_task.task_uid,
                                                 f"the task[{collection_task.task_uid}] upload csg hub-server success...")
         else:
             insert_datasource_run_task_log_error(collection_task.task_uid,
-                                                 f"the task[{collection_task.task_uid}] upload csg hub-server fail: export_from_files returned None")
+                                                 f"the task[{collection_task.task_uid}] upload csg hub-server fail...")
     except Exception as e:
         logger.error(e)
-        error_msg = str(e)
-        # Check if this is a "nothing to commit" error
-        if "nothing to commit" in error_msg.lower() or "working tree clean" in error_msg.lower():
-            insert_datasource_run_task_log_error(collection_task.task_uid,
-                                                 f"the task[{collection_task.task_uid}] upload csg hub-server fail: No files to commit. This may happen if: 1) Files are already committed in the branch, 2) Files are ignored by .gitignore, 3) File paths are incorrect. Error: {error_msg}")
-        else:
-            insert_datasource_run_task_log_error(collection_task.task_uid,
-                                                 f"Task UID {collection_task.task_uid} Error occurred while uploading to CSG Hub server: {error_msg}")
+        insert_datasource_run_task_log_error(collection_task.task_uid,
+                                             f"Task UID {collection_task.task_uid} Error occurred while uploading to CSG Hub server: {e}")
         return False
     return True

data_engine/core/executor.py

@@ -74,7 +74,7 @@ def __init__(
 
         # Check if this is the specific output_only tool by tool name
         tool_name = getattr(self.cfg, 'tool_name', '')
-        is_specific_output_only = (tool_name == 'template_executor_06_common_internal')
+        is_specific_output_only = (tool_name == 'smoltalk_chinese_common_internal')
 
         # normal_logic
         if not is_specific_output_only:
@@ -127,7 +127,8 @@ def __init__(
             branch = self.cfg.branch,
             user_name=self.user_name,
             user_token=self.user_token,
-            work_dir=self.work_dir
+            work_dir=self.work_dir,
+            auto_version=True  # Pipeline jobs use auto versioning
         )
 
         # setup tracer

data_engine/core/ray_executor.py

@@ -101,7 +101,8 @@ def __init__(self, cfg=None):
             branch=self.cfg.branch,
             user_name=self.user_name,
             user_token=self.user_token,
-            work_dir=self.work_dir
+            work_dir=self.work_dir,
+            auto_version=True  # Pipeline jobs use auto versioning
         )
 
         # setup tracer

data_engine/exporter/csghub_exporter.py

@@ -1,12 +1,12 @@
 import traceback
 from typing import List
+import os
+import re
 
 from pycsghub.cmd.repo_types import RepoType
 from pycsghub.upload_large_folder.main import upload_large_folder_internal
 
 from data_engine.exporter.base_exporter import Exporter
-import os
-import re
 from loguru import logger
 from pycsghub.repository import Repository
 from pycsghub.utils import (build_csg_headers,
@@ -38,6 +38,7 @@ def __init__(
             user_token: str = None,
             work_dir: str = None,
             path_is_dir: bool = False,
+            auto_version: bool = False,
     ):
         """
         Initialization method.
@@ -64,6 +65,7 @@ def __init__(
         self.user_name = user_name
         self.user_token = user_token
         self.work_dir = work_dir
+        self.auto_version = auto_version  # True for pipeline jobs, False for formatify tasks
         super().__init__(
             export_path=export_path,
             export_shard_size=export_shard_size,
@@ -193,26 +195,42 @@ def get_avai_branch(self, origin_branch: str) -> str:
         # valid_branches = ["main", "v3.1", "v1.11", "v1.5", "v2", "v1", "v1.2", "v1.11.2"]            
         valid_branches.sort()
         logger.info(f'repo {self.repo_id} all branches: {valid_branches}')
-        return self.find_next_version(origin_branch=origin_branch, valid_branches=valid_branches)
+        result = self.find_next_version(origin_branch=origin_branch, valid_branches=valid_branches)
+        return result
 
     def find_next_version(self, origin_branch: str, valid_branches: List):
+        """
+        根据任务类型选择版本生成逻辑：
+        - auto_version=False (文件转换任务): 直接返回用户指定的分支名
+        - auto_version=True (算子执行任务): 自动生成版本号 (v1, v2, ...)
+        """
+        if not self.auto_version:
+            # 文件转换任务：直接返回用户指定的分支名
+            if origin_branch in valid_branches:
+                return origin_branch
+            return origin_branch
+        
+        # 算子执行任务：自动生成版本号
         latestNum = 0
+        
         for b in valid_branches:
             if origin_branch == "main" and re.match(r"^v\d+", b):
+                # 处理 main 分支的情况，查找 v1, v2, v3 等
                 numStr = b.split(".")[0][1:]
                 if not numStr.isdigit():
                     continue
                 num = int(numStr)
                 latestNum = max(latestNum, num)
             elif b.startswith(origin_branch) and len(b) > len(origin_branch):
+                # 处理其他分支的情况，查找 origin_branch.1, origin_branch.2 等
                 numStr = b[len(origin_branch) + 1:]
                 if not numStr.isdigit():
                     continue
+                num = int(numStr)
+                latestNum = max(latestNum, num)
             else:
                 continue
-            num = int(numStr)
-            latestNum = max(latestNum, num)
-
+        
         if origin_branch == "main":
             if latestNum > 0:
                 return "v" + str(latestNum + 1)

data_engine/exporter/load.py

@@ -15,7 +15,8 @@ def load_exporter(
     user_name: str=None,
     user_token: str=None,
     work_dir: str=None,
-    path_is_dir: bool=False
+    path_is_dir: bool=False,
+    auto_version: bool=False
 ) -> Exporter:
     return ExporterCSGHUB(
         export_path=export_path,
@@ -31,6 +32,7 @@ def load_exporter(
         user_name=user_name,
         user_token=user_token,
         work_dir=work_dir,
-        path_is_dir=path_is_dir
+        path_is_dir=path_is_dir,
+        auto_version=auto_version
     )
 

data_server/datasource/DatasourceModels.py

@@ -1,6 +1,7 @@
 from sqlalchemy import create_engine, Column, Integer, String, Text, Boolean, ForeignKey, DateTime, JSON
 from sqlalchemy.orm import relationship
 import datetime
+import json
 from enum import Enum
 from data_server.database.bean.base import Base
 
@@ -47,6 +48,38 @@ class DataSource(Base):
     updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now, comment='更新时间')
 
     def to_json(self):
+        # 处理 extra_config，确保正确解析并返回
+        extra_config_raw = self.extra_config
+        extra_config_dict = None
+        
+        # 解析 extra_config
+        if isinstance(extra_config_raw, str):
+            try:
+                extra_config_dict = json.loads(extra_config_raw)
+            except (json.JSONDecodeError, TypeError):
+                extra_config_dict = None
+        elif isinstance(extra_config_raw, dict):
+            extra_config_dict = extra_config_raw.copy()
+        
+        # 处理分支字段：保持 csg_hub_dataset_default_branch 原样（从数据库读取，不做修改），同时添加 csg_hub_dataset_branch
+        if extra_config_dict is not None:
+            # 获取 csg_hub_dataset_default_branch 的值（如果存在），用于设置 csg_hub_dataset_branch
+            # 保持 csg_hub_dataset_default_branch 原样，不做任何修改
+            branch_value = extra_config_dict.get("csg_hub_dataset_default_branch")
+            if not branch_value or (isinstance(branch_value, str) and branch_value.strip() == ""):
+                branch_value = "main"
+            
+            # 添加 csg_hub_dataset_branch 字段（从 csg_hub_dataset_default_branch 获取值，如果不存在则使用 main）
+            extra_config_dict["csg_hub_dataset_branch"] = branch_value
+            
+            # csg_hub_dataset_default_branch 保持原样，不做修改（如果数据库中有就返回，如果没有就不添加）
+            
+            # 将更新后的字典转换回 JSON 字符串格式返回
+            extra_config = json.dumps(extra_config_dict, ensure_ascii=False, indent=4)
+        else:
+            # 如果 extra_config 为空，保持原样返回空配置
+            extra_config = extra_config_raw if extra_config_raw else "{}"
+        
         return {
             "id": self.id,
             "name": self.name,
@@ -57,7 +90,7 @@ def to_json(self):
             "username": self.username,
             "password": self.password,
             "database": self.database,
-            "extra_config": self.extra_config,
+            "extra_config": extra_config,
             "source_status": self.source_status,
             "owner_id": self.owner_id,
             "created_at": self.created_at.strftime("%Y-%m-%d %H:%M:%S") if self.created_at else None,

data_server/job/branch_tool.py

@@ -37,21 +37,37 @@ def get_avai_branch(self, origin_branch: str) -> str:
             raise
 
     def find_next_version(self, origin_branch: str, valid_branches: List) -> str:
+        """
+        算子执行任务的版本生成逻辑：自动生成 v1, v2 等版本号
+        """
         latestNum = 0
+        
         for b in valid_branches:
             if origin_branch == "main" and re.match(r"^v\d+", b):
+                # 处理 main 分支的情况，查找 v1, v2, v3 等
                 numStr = b.split(".")[0][1:]
-                if numStr.isdigit():
-                    num = int(numStr)
-                    latestNum = max(latestNum, num)
+                if not numStr.isdigit():
+                    continue
+                num = int(numStr)
+                latestNum = max(latestNum, num)
             elif b.startswith(origin_branch) and len(b) > len(origin_branch):
-                numStr = b[len(origin_branch)+1:]
-                if numStr.isdigit():
-                    num = int(numStr)
-                    latestNum = max(latestNum, num)
-
+                # 处理其他分支的情况，查找 origin_branch.1, origin_branch.2 等
+                numStr = b[len(origin_branch) + 1:]
+                if not numStr.isdigit():
+                    continue
+                num = int(numStr)
+                latestNum = max(latestNum, num)
+            else:
+                continue
+        
         if origin_branch == "main":
-            return f"v{latestNum + 1}" if latestNum > 0 else "v1"
+            if latestNum > 0:
+                return "v" + str(latestNum + 1)
+            else:
+                return "v1"
         else:
-            return f"{origin_branch}.{latestNum + 1}" if latestNum > 0 else f"{origin_branch}.1"
+            if latestNum > 0:
+                return origin_branch + "." + str(latestNum + 1)
+            else:
+                return origin_branch + ".1"