Fix the bug in #84: The search function for models in the MD2Json tool should display all searched models.

Author: zhanglongbin

data_celery/formatify/tasks.py

@@ -266,9 +266,9 @@ def format_task(task_id: int, user_name: str, user_token: str):
                     continue
 
                 # Execute conversion
-                # For PDF to MD conversion, need to pass mineru_api_url parameter
+                # For PDF to MD conversion, need to pass mineru_api_url and mineru_backend parameters
                 if format_task.from_data_type == DataFormatTypeEnum.PDF.value and format_task.to_data_type == DataFormatTypeEnum.Markdown.value:
-                    result = convert_func(file_path_full, format_task.task_uid, format_task.mineru_api_url)
+                    result = convert_func(file_path_full, format_task.task_uid, format_task.mineru_api_url, format_task.mineru_backend)
                 else:
                     result = convert_func(file_path_full, format_task.task_uid)
 
@@ -665,7 +665,7 @@ def convert_ppt_to_markdown(file_path: str, task_uid) -> Optional[Dict[str, str]
         return None  # Not a target file, return None
 
 
-def convert_pdf_to_markdown(file_path: str, task_uid, mineru_api_url: Optional[str] = None) -> Optional[Dict[str, str]]:
+def convert_pdf_to_markdown(file_path: str, task_uid, mineru_api_url: Optional[str] = None, mineru_backend: Optional[str] = None) -> Optional[Dict[str, str]]:
     if file_path.lower().endswith('.pdf'):
         insert_formatity_task_log_info(task_uid, f'Source file address：{file_path}')
         try:
@@ -679,10 +679,14 @@ def convert_pdf_to_markdown(file_path: str, task_uid, mineru_api_url: Optional[s
                 server_url = mineru_api_url
             else:
                 server_url = os.getenv("MINERU_API_URL", "http://111.4.242.20:30000")
-            backend = "vlm-http-client"
+            # MinerU backend: Priority: passed parameter > environment variable > default value
+            if mineru_backend:
+                backend = mineru_backend
+            else:
+                backend = os.getenv("MINERU_BACKEND", "http-client")
 
-            # Record used MinerU API address for debugging
-            insert_formatity_task_log_info(task_uid, f'Using MinerU API server: {server_url}')
+            # Record used MinerU API address and backend for debugging
+            insert_formatity_task_log_info(task_uid, f'Using MinerU API server: {server_url}, backend: {backend}')
 
             pdf_file_name = Path(file_path).stem
             temp_output_dir = Path(file_path).parent / f"_temp_pdf_convert_{pdf_file_name}"

data_engine/ops/base_op.py

@@ -501,6 +501,7 @@ class DataType(Enum):
     ClosedUnitInterval = 2
     from_2_to_20 = 3
     SEARCH_SELECT = "search-select"
+    SELECT_MODEL = "select-model"
 
 @dataclass
 class Sample:

data_engine/tools/preprocess/md_to_jsonl_preprocess.py

@@ -168,7 +168,7 @@ def export_from_files(self, upload_path: Path):
                 :param upload_path: path with files to upload
                 :return: branch name
                 """
-                # 🔍 诊断日志: 检查上传路径
+
                 logger.info(f'='*80)
                 logger.info(f'[EXPORT DEBUG] Starting export_from_files()')
                 logger.info(f'[EXPORT DEBUG] Upload path type: {type(upload_path)}, value: {upload_path}')
@@ -179,7 +179,7 @@ def export_from_files(self, upload_path: Path):
                     logger.error(f'[EXPORT DEBUG] Upload path does not exist!')
                     return None
 
-                # 🔍 诊断日志: 列出所有要上传的文件
+
                 try:
                     files_to_upload = os.listdir(upload_path)
                     logger.info(f'[EXPORT DEBUG] Files in upload directory: {len(files_to_upload)} files')
@@ -204,7 +204,7 @@ def export_from_files(self, upload_path: Path):
                     logger.info('[EXPORT DEBUG] No repo_id specified, skip upload.')
                     return 'N/A'
 
-                # 🔍 诊断日志: repo信息
+
                 logger.info(f'[EXPORT DEBUG] Target repo_id: {self.parent.tool_def.repo_id}')
                 logger.info(f'[EXPORT DEBUG] User token (first 10 chars): {self.parent.executed_params.user_token[:10] if self.parent.executed_params.user_token else "None"}...')
 
@@ -216,7 +216,7 @@ def export_from_files(self, upload_path: Path):
                     output_branch_name = self.parent._get_available_branch(branch)
                     logger.info(f'[EXPORT DEBUG] Output branch name: {output_branch_name}')
 
-                    # 🔍 诊断日志: 上传参数
+
                     endpoint = get_endpoint(endpoint=GetHubEndpoint())
                     logger.info(f'[EXPORT DEBUG] Upload parameters:')
                     logger.info(f'[EXPORT DEBUG]   - repo_id: {self.parent.tool_def.repo_id}')
@@ -701,7 +701,7 @@ def init_params(cls, userid: str = None, isadmin: bool = False):
                 "token": "token",
                 "sentence": "sentence"
             }, "token"),
-            Param("hf_tokenizer", DataType.SEARCH_SELECT, {
+            Param("hf_tokenizer", DataType.SELECT_MODEL, {
                 "EleutherAI/pythia-6.9b-deduped": "EleutherAI/pythia-6.9b-deduped",
                 "hfl/chinese-bert-wwm-ext": "hfl/chinese-bert-wwm-ext"
             }, "EleutherAI/pythia-6.9b-deduped"),

data_server/api/endpoints/formatify.py

@@ -21,24 +21,35 @@
 @router.get("/get_mineru_api_url", response_model=dict)
 async def get_mineru_api_url():
     """
-    获取当前配置的 MinerU API 地址
+    获取当前配置的 MinerU 引擎参数
     Returns:
-        Dict: 包含当前 MinerU API 地址的响应
+        Dict: 包含当前 MinerU 引擎参数的响应
             - mineru_api_url: MinerU API 服务器地址
-            - source: 配置来源 ("environment" | "default")
+            - mineru_backend: MinerU 后端类型
+            - sources: 配置来源字典
+                - mineru_api_url_source: mineru_api_url 的配置来源 ("environment" | "default")
+                - mineru_backend_source: mineru_backend 的配置来源 ("environment" | "default")
     """
     try:
-        # Get from environment variable, use default value if not set
+        # Get mineru_api_url from environment variable, use default value if not set
         mineru_api_url = os.getenv("MINERU_API_URL", "http://111.4.242.20:30000")
-        source = "environment" if os.getenv("MINERU_API_URL") else "default"
+        mineru_api_url_source = "environment" if os.getenv("MINERU_API_URL") else "default"
+        
+        # Get mineru_backend from environment variable, use default value if not set
+        mineru_backend = os.getenv("MINERU_BACKEND", "http-client")
+        mineru_backend_source = "environment" if os.getenv("MINERU_BACKEND") else "default"
 
         return response_success(data={
             "mineru_api_url": mineru_api_url,
-            "source": source
+            "mineru_backend": mineru_backend,
+            "sources": {
+                "mineru_api_url_source": mineru_api_url_source,
+                "mineru_backend_source": mineru_backend_source
+            }
         })
     except Exception as e:
         logger.error(f"Failed to get mineru_api_url: {str(e)}")
-        return response_fail(msg="获取 MinerU API 地址失败")
+        return response_fail(msg="获取 MinerU 引擎参数失败")
 
 
 @router.get("/formatify/get_format_type_list", response_model=dict)

data_server/api/endpoints/model_validator.py

@@ -7,6 +7,7 @@
 from loguru import logger
 from ...schemas.responses import response_success, response_fail
 from data_engine.utils.cache_utils import DATA_JUICER_MODELS_CACHE
+from typing import Optional
 
 router = APIRouter()
 
@@ -55,7 +56,7 @@ def check_model_for_md_to_jsonl(
 
         # Get the first matching model information
         model_info = data['data'][0]
-        model_name_from_api = model_info.get('name', model_name)  # Get name field from API response
+        model_name_from_api = model_info.get('path', model_name)  # Get path field from API response
         repository = model_info.get('repository', {})
         http_clone_url = repository.get('http_clone_url')
 
@@ -219,3 +220,87 @@ def check_model_for_md_to_jsonl(
     except Exception as e:
         logger.error(f'Error checking model: {str(e)}')
         return response_fail(msg=f'检查模型时发生错误: {str(e)}')
+
+
+@router.get("/list-models", summary="获取模型列表")
+def list_models(
+    page: int = Query(1, description="页码，从1开始"),
+    per_page: int = Query(16, description="每页数量，默认16条"),
+    search: Optional[str] = Query(None, description="搜索关键词"),
+    sort: str = Query("trending", description="排序方式")
+):
+    """
+    Fetch model list from OpenCSG Hub and return basic information.
+    
+    Returns:
+    - path: Model path
+    - updated_at: Last update time
+    - first_tag: show_name of the first tag with category == "task"
+    - downloads: Download count
+    - description: Model description
+    
+    :param page: Page number
+    :param per_page: Items per page, default 16
+    :param search: Search keyword (optional)
+    :param sort: Sort method, default trending
+    :return: Model list
+    """
+    try:
+        # Get CSGHUB_ENDPOINT from environment
+        csghub_endpoint = os.getenv('CSGHUB_ENDPOINT', 'https://hub.opencsg.com')
+        api_url = f'{csghub_endpoint}/api/v1/models'
+        
+        # Set request parameters
+        params = {
+            'page': page,
+            'per': per_page,
+            'search': search or '',
+            'sort': sort,
+            'source': ''
+        }
+        
+        logger.info(f'Fetching models from {api_url} with params: {params}')
+        
+        # Send request
+        response = requests.get(api_url, params=params, timeout=30)
+        response.raise_for_status()
+        data = response.json()
+        
+        # Extract required fields
+        models_data = []
+        if data.get('data'):
+            for model in data['data']:
+                # Find the first tag with category == "task" and get its show_name
+                first_tag_show_name = ''
+                if model.get('tags'):
+                    for tag in model['tags']:
+                        if tag.get('category') == 'task':
+                            first_tag_show_name = tag.get('show_name', '')
+                            break
+                
+                model_info = {
+                    'path': model.get('path', ''),
+                    'updated_at': model.get('updated_at', ''),
+                    'first_tag': first_tag_show_name,
+                    'downloads': model.get('downloads', 0),
+                    'description': model.get('description', '')
+                }
+                models_data.append(model_info)
+        
+        # Build result
+        result = {
+            'models': models_data,
+            'total': data.get('total', 0),
+            'page': page,
+            'per_page': per_page
+        }
+        
+        logger.info(f'Successfully fetched {len(models_data)} models')
+        return response_success(data=result, msg='获取模型列表成功')
+        
+    except requests.RequestException as e:
+        logger.error(f'Failed to request OpenCSG API: {str(e)}')
+        return response_fail(msg=f'无法连接到OpenCSG Hub API: {str(e)}')
+    except Exception as e:
+        logger.error(f'Error fetching models: {str(e)}')
+        return response_fail(msg=f'获取模型列表时发生错误: {str(e)}')

data_server/database/session.py

@@ -170,6 +170,21 @@ def add_mineru_api_url_column():
                 logger.info("Column 'mineru_api_url' added successfully to data_format_tasks table")
 
 
+def add_mineru_backend_column():
+    """Add mineru_backend column to 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 = 'mineru_backend';
+            """))
+
+            if not result.fetchone():
+                session.execute(text("ALTER TABLE data_format_tasks ADD COLUMN mineru_backend VARCHAR(100);"))
+                logger.info("Column 'mineru_backend' added successfully to data_format_tasks table")
+
+
 _initialized = False
 from data_server.database.bean.work import Worker
 from data_server.job.JobModels import Job
@@ -202,6 +217,7 @@ def create_tables():
 
     add_first_op_column()
     add_mineru_api_url_column()
+    add_mineru_backend_column()
 def is_table_initialized(table_name: str) -> bool:
     """
     Check if a specific table contains any data.

data_server/formatify/FormatifyManager.py

@@ -29,6 +29,7 @@ def create_formatify_task(db_session: Session, dataFormatTask: DataFormatTaskReq
                                          to_csg_hub_repo_id=dataFormatTask.to_csg_hub_repo_id,
                                          to_data_type=dataFormatTask.to_data_type,
                                          mineru_api_url=dataFormatTask.mineru_api_url,
+                                         mineru_backend=dataFormatTask.mineru_backend,
                                          task_uid=task_uid,
                                          task_status=DataFormatTaskStatusEnum.WAITING.value,
                                          owner_id=user_id)
@@ -75,7 +76,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_api_url', 'mineru_backend'
     }
     for field in updatable_fields:
         value = getattr(dataFormatTaskRequest, field, None)

data_server/formatify/FormatifyModels.py

@@ -54,6 +54,7 @@ class DataFormatTask(Base):
     task_status = Column(Integer, nullable=False, comment="任务状态 DataFormatTaskStatusEnum 枚举")
     owner_id = Column(Integer, comment="所属用户")
     mineru_api_url = Column(String(500), comment="MinerU API 地址")
+    mineru_backend = Column(String(100), comment="MinerU 后端类型")
     start_run_at = Column(DateTime, comment='运行开始时间')
     end_run_at = Column(DateTime, comment='运行结束时间')
     created_at = Column(DateTime, default=datetime.datetime.now, comment='任务创建时间')
@@ -79,6 +80,7 @@ def to_dict(self):
             "task_status": self.task_status,
             "owner_id": self.owner_id,
             "mineru_api_url": self.mineru_api_url,
+            "mineru_backend": self.mineru_backend,
             "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/schemas.py

@@ -16,3 +16,4 @@ class DataFormatTaskRequest(BaseModel):
     to_csg_hub_repo_id: Optional[str] = None
     to_data_type: Optional[int] = None
     mineru_api_url: Optional[str] = None
+    mineru_backend: Optional[str] = None

data_server/logic/models.py

@@ -81,7 +81,7 @@ class Sample(BaseModelExtended):
 class Param(BaseModelExtended):
     name: str
     type: Optional[Union[Literal["STRING"], Literal["INTEGER"], Literal["FLOAT"],
-                Literal["BOOLEAN"], Literal["DICTIONARY"], Literal["TUPLE"], Literal["LIST"], Literal["PositiveFloat"], Literal["ClosedUnitInterval"], Literal["from_2_to_20"], Literal["search-select"]]] = None
+                Literal["BOOLEAN"], Literal["DICTIONARY"], Literal["TUPLE"], Literal["LIST"], Literal["PositiveFloat"], Literal["ClosedUnitInterval"], Literal["from_2_to_20"], Literal["search-select"], Literal["select-model"]]] = None
     option_values: Optional[list[Option]] = None
     value: Optional[Any] = None
     tempVal: Optional[Any] = None