refactor: Replace complex OSS manager with simplified version - remove compassflow dependency

src/oss/oss_file_manager.py

@@ -1,517 +1,254 @@
-# %%
-import datetime as dt
-import os
-import re
-from logging import Logger
-from multiprocessing import Pool
-from multiprocessing.pool import ThreadPool
-from pathlib import Path
-from typing import List, Union
+#!/usr/bin/env python3
+"""
+简化的OSS文件管理器 - 专为SAGE-Bench HuggingFace Space设计
+移除了对 compassflow 的依赖，只保留必需的OSS功能
+"""
 
+import os
 import oss2
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import List, Dict, Optional
 from loguru import logger
-from oss2.credentials import EnvironmentVariableCredentialsProvider
-from tqdm import tqdm
-
-from compassflow.constants import DATADIR
-from compassflow.oss.oss import OssBucket
-from compassflow.utils import starstarmap
 
 
 class OSSFileManager:
+    """简化的OSS文件管理器"""
+    
     def __init__(
         self,
         oss_access_key_id: str = None,
         oss_access_key_secret: str = None,
-        region: str = "http://oss-cn-shanghai.aliyuncs.com",
-        bucket_name: str = "opencompass",
-        oss_block_name: str = None,
-        logger: Logger = logger,
-    ) -> None:
-        """OSS File Manager
-
+        oss_region: str = None,
+        oss_bucket_name: str = None
+    ):
+        """
+        初始化OSS文件管理器
+        
         Args:
-            oss_access_key_id (str, optional): _description_. Defaults to None.
-            oss_access_key_secret (str, optional): _description_. Defaults to None.
-            region (_type_, optional): _description_. Defaults to 'http://oss-cn-shanghai.aliyuncs.com'.
-            bucket_name (str, optional): _description_. Defaults to 'opencompass'.
-            oss_block_name: oss_block_name which is defined in the prefect
-            logger (Logger, optional): _description_. Defaults to logger.
+            oss_access_key_id: OSS访问密钥ID
+            oss_access_key_secret: OSS访问密钥Secret
+            oss_region: OSS区域端点
+            oss_bucket_name: OSS存储桶名称
         """
-        self.logger = logger
-        if oss_block_name is not None:
-            oss_bucket = OssBucket.load(oss_block_name)
-            self._bucket = oss_bucket._get_bucket()
-            return
-
-        # 阿里云账号AccessKey拥有所有API的访问权限，风险很高。强烈建议您创建并使用RAM账号进行API访问或日常运维，请登录RAM控制台创建RAM账号。
-        if oss_access_key_id is not None and oss_access_key_secret is not None:
-            os.environ["OSS_ACCESS_KEY_ID"] = oss_access_key_id
-            os.environ["OSS_ACCESS_KEY_SECRET"] = oss_access_key_secret
-
-        if (
-            os.getenv("OSS_ACCESS_KEY_ID") is None
-            or os.getenv("OSS_ACCESS_KEY_SECRET") is None
-        ):
-            raise ValueError("Access Key ID and Access Key Secret cannot be empty.")
-
-        auth = oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())
-
-        # Endpoint以杭州为例，其它Region请按实际情况填写。
-        # 填写Bucket名称，例如examplebucket。
-        bucket = oss2.Bucket(
-            auth=auth,
-            endpoint=region,
-            bucket_name=bucket_name,
-        )
-
-        self._bucket = bucket
-
-    def list_latest_files_by_date(
-        self,
-        object_dir: str = "",
-        delimiter: str = "/",
-        start_date: Union[str, dt.date, dt.datetime] = None,
-        end_date: Union[str, dt.date, dt.datetime] = None,
-        max_num_files: int = 10,
-        date_pattern: str = r"([0-9]{4}-[0-9]{2}-[0-9]{2})",
-        file_date_format: str = "%Y-%m-%d",
-        suffix: str = "",
-    ) -> List[Union[str, Path]]:
-        """List the latest files by date in an OSS bucket directory
-
+        # 从环境变量获取配置
+        self.access_key_id = oss_access_key_id or os.getenv('OSS_ACCESS_KEY_ID')
+        self.access_key_secret = oss_access_key_secret or os.getenv('OSS_ACCESS_KEY_SECRET')
+        self.region = oss_region or os.getenv('OSS_REGION', 'http://oss-cn-shanghai.aliyuncs.com')
+        self.bucket_name = oss_bucket_name or os.getenv('OSS_BUCKET_NAME', 'opencompass')
+        
+        if not self.access_key_id or not self.access_key_secret:
+            raise ValueError("OSS访问密钥未设置。请设置 OSS_ACCESS_KEY_ID 和 OSS_ACCESS_KEY_SECRET 环境变量。")
+        
+        # 初始化OSS客户端
+        auth = oss2.Auth(self.access_key_id, self.access_key_secret)
+        self.bucket = oss2.Bucket(auth, self.region, self.bucket_name)
+        
+        logger.info(f"OSS初始化成功: {self.bucket_name} @ {self.region}")
+    
+    def list_files(
+        self, 
+        oss_dir: str = "", 
+        after_date: datetime = None,
+        file_extension: str = None
+    ) -> List[Dict]:
+        """
+        列出OSS目录中的文件
+        
         Args:
-            object_dir (str, optional): _description_. Defaults to ''.
-            delimiter (str, optional): _description_. Defaults to '/'.
-            start_date (Union[str, dt.date], optional): _description_. Defaults to None.
-            end_date (Union[str, dt.date], optional): _description_. Defaults to None.
-            max_num_files (int, optional): _description_. Defaults to 10.
-            date_pattern (str, optional): _description_. Defaults to r'^([0-9]{4}-[0-9]{2}-[0-9]{2})'.
-            suffix (str, optional): _description_. Defaults to ''.
-
+            oss_dir: OSS目录路径
+            after_date: 只返回此日期之后的文件
+            file_extension: 文件扩展名过滤 (如 ".json")
+            
         Returns:
-            List[Union[str, Path]]: _description_
+            文件信息列表
         """
-        if start_date is not None:
-            if isinstance(start_date, str):
-                start_date = dt.datetime.strptime(start_date.replace("-", ""), "%Y%m%d")
-
-            if isinstance(start_date, dt.date):
-                start_date = dt.datetime(
-                    start_date.year, start_date.month, start_date.day, 0, 0, 0
-                )
-
-        if end_date is not None:
-            if isinstance(end_date, str):
-                end_date = dt.datetime.strptime(end_date.replace("-", ""), "%Y%m%d")
-
-            if isinstance(end_date, dt.date):
-                end_date = dt.datetime(
-                    end_date.year, end_date.month, end_date.day, 0, 0, 0
-                )
-
-        object_iter = oss2.ObjectIterator(
-            bucket=self._bucket, prefix=object_dir, delimiter=delimiter
-        )
-
-        root_dir = Path(object_dir)
-        filenames = []
-        for filename in object_iter:
-            filename = filename.key.replace(object_dir, "")
-            # print(filename)
-
-            if filename.endswith(suffix):
-                # Match date pattern in filename
-                date_search = re.search(date_pattern, filename)
-                if date_search:
-                    file_date = dt.datetime.strptime(
-                        date_search.group(1), file_date_format
-                    )
-                else:
-                    self.logger.warning(
-                        f"date pattern doesn't match, skipping file {filename}"
-                    )
+        try:
+            files = []
+            
+            # 确保目录路径以 / 结尾
+            if oss_dir and not oss_dir.endswith('/'):
+                oss_dir += '/'
+            
+            # 列出对象
+            for obj in oss2.ObjectIterator(self.bucket, prefix=oss_dir):
+                # 跳过目录本身
+                if obj.key.endswith('/'):
                     continue
-
-                # Check if file date within specified range
-                if start_date is not None:
-                    if start_date > file_date:
-                        continue
-
-                if end_date is not None:
-                    if end_date < file_date:
-                        continue
-
-                filepath = root_dir / filename
-                # name_tstamp_tuple = (filepath, os.path.getmtime(filepath))
-                # filenames.append(name_tstamp_tuple)
-                filenames.append(str(filepath))
-        # sort by tstamp
-        # filenames = sorted(filenames, key=lambda x: x[1])
-        filenames = sorted(filenames)
-        # filenames = [x[0] for x in filenames]
-
-        max_num_files = max_num_files or len(filenames)
-        filenames = filenames[-max_num_files:]
-
-        self.logger.info(f"{filenames=}")
-
-        return filenames
-
-    def download_object_to_file(
-        self,
-        oss_file_path: str | Path,
-        local_file_path: str | Path,
-        replace: bool = False,
-        make_dir: bool = False,
-        print_logs: bool = True,
-    ):
-        """Download a single OSS object to local file.
-
+                
+                # 文件扩展名过滤
+                if file_extension and not obj.key.endswith(file_extension):
+                    continue
+                
+                # 日期过滤
+                if after_date and obj.last_modified < after_date:
+                    continue
+                
+                file_info = {
+                    'key': obj.key,
+                    'name': os.path.basename(obj.key),
+                    'size': obj.size,
+                    'last_modified': obj.last_modified,
+                    'etag': obj.etag
+                }
+                files.append(file_info)
+            
+            logger.info(f"找到 {len(files)} 个文件在 {oss_dir}")
+            return files
+            
+        except Exception as e:
+            logger.error(f"列出文件失败: {e}")
+            raise
+    
+    def download_file(self, oss_file_path: str, local_file_path: str) -> bool:
+        """
+        从OSS下载文件到本地
+        
         Args:
-            oss_file_path (str): _description_
-            local_file_path (str): _description_
-            replace (bool, optional): _description_. Defaults to False.
-            make_dir (bool, optional): Whether to create intermediate dirs if they don't exist. Defaults to False.
-            print_logs: bool, optional): Whether to print logs. Defaults to True.
+            oss_file_path: OSS文件路径
+            local_file_path: 本地文件路径
+            
+        Returns:
+            下载是否成功
         """
-        if isinstance(local_file_path, str):
-            local_file_path = Path(local_file_path)
-
-        if not replace:
-            if local_file_path.exists():
-                if print_logs:
-                    err_msg = f"{local_file_path} already exists, skipping file..."
-                    self.logger.info(err_msg)
-
-                return
-
-        if print_logs:
-            if local_file_path.exists():
-                err_msg = f"{local_file_path} already exists, replacing file..."
-                self.logger.info(err_msg)
-
-        if make_dir:
-            os.makedirs(local_file_path.parent, exist_ok=True)
-
-        self._bucket.get_object_to_file(
-            key=str(oss_file_path),
-            filename=local_file_path,
-        )
-
-    def download_objects_to_files(
-        self,
-        file_download_mapping: list[tuple[str | Path, str | Path]],
-        oss_base_dir: str = None,
-        local_base_dir: str = None,
-        replace: bool = True,
-        num_threads: int = 1,
-        **kwargs,
-    ) -> None:
-        """Download objects from OSS to local storage.
-
+        try:
+            # 确保本地目录存在
+            local_dir = os.path.dirname(local_file_path)
+            if local_dir:
+                os.makedirs(local_dir, exist_ok=True)
+            
+            # 下载文件
+            self.bucket.get_object_to_file(oss_file_path, local_file_path)
+            
+            logger.info(f"下载成功: {oss_file_path} -> {local_file_path}")
+            return True
+            
+        except Exception as e:
+            logger.error(f"下载文件失败: {oss_file_path} -> {local_file_path}, 错误: {e}")
+            return False
+    
+    def upload_file_to_object(
+        self, 
+        local_file_path: str, 
+        oss_file_path: str,
+        replace: bool = False
+    ) -> bool:
+        """
+        上传本地文件到OSS
+        
         Args:
-            file_download_mapping (list[tuple[str | Path, str | Path]]): A list of file path pairs that maps the OSS file path (to download) to the local file path (download location).
-            oss_base_dir (str): OSS directory path to be prepended to all OSS file paths.
-            local_base_dir (str, optional): Base directory path to be prepended to all local file paths.
-            replace (bool, optional): _description_. Defaults to True.
-            num_threads (int, optional): _description_. Defaults to 1.
-            **kwargs: Additional keyword arguments passed to `list_latest_files_by_date`
+            local_file_path: 本地文件路径
+            oss_file_path: OSS文件路径
+            replace: 是否替换已存在的文件
+            
+        Returns:
+            上传是否成功
         """
-        if isinstance(oss_base_dir, str):
-            oss_base_dir = Path(oss_base_dir)
-
-        if isinstance(local_base_dir, str):
-            local_base_dir = Path(local_base_dir)
-
-        if not isinstance(file_download_mapping, list):
-            raise TypeError("file_download_mapping must be a list of 2 value tuples.")
-
-        for item in file_download_mapping:
-            if not isinstance(item, tuple):
-                raise TypeError(
-                    "Each item in the file_download_mapping list must be a 2 value tuple."
-                )
-
-            if len(item) != 2:
-                raise ValueError(
-                    "Each tuple in the file_download_mapping list must be length 2."
-                )
-
-        if num_threads == 1:
-            for local_file_path, oss_file_path in file_download_mapping:
-                if local_base_dir is not None:
-                    local_file_path = local_base_dir / local_file_path
-
-                if oss_base_dir is not None:
-                    oss_file_path = oss_base_dir / oss_file_path
-
-                self.download_object_to_file(
-                    oss_file_path=oss_file_path,
-                    local_file_path=local_file_path,
-                    replace=replace,
-                )
-
-            return
-
-        # Start multithreaded process if num_threads > 1
-        with ThreadPool(num_threads) as p:
-            pool_args_list = []
-            for local_file_path, oss_file_path in file_download_mapping:
-                if local_base_dir is not None:
-                    local_file_path = local_base_dir / local_file_path
-
-                if oss_base_dir is not None:
-                    oss_file_path = oss_base_dir / oss_file_path
-
-                args_dict = dict(
-                    oss_file_path=oss_file_path,
-                    local_file_path=local_file_path,
-                    replace=replace,
-                    print_logs=True,
-                )
-                pool_args_list.append(args_dict)
-
-            ret_all = list(
-                tqdm(
-                    starstarmap(
-                        pool=p,
-                        fn=self.download_object_to_file,
-                        kwargs_iter=pool_args_list,
-                    ),
-                    total=len(file_download_mapping),
-                )
-            )
-
-    def download_latest_objects_to_dir(
-        self,
-        oss_object_dir: str,
-        local_dir: str,
-        start_date: Union[str, dt.date, dt.datetime] = None,
-        end_date: Union[str, dt.date, dt.datetime] = None,
-        date_pattern: str = r"([0-9]{4}-[0-9]{2}-[0-9]{2})",
-        file_date_format: str = "%Y-%m-%d",
-        max_num_files: int = 5,
-        replace: bool = True,
-        make_dir: bool = True,
-        delimiter: str = "/",
-        num_threads: int = 1,
-        suffix: str = "",
-        **kwargs,
-    ) -> None:
-        """Download the latest objects from oss to local directory
-
+        try:
+            # 检查本地文件是否存在
+            if not os.path.exists(local_file_path):
+                logger.error(f"本地文件不存在: {local_file_path}")
+                return False
+            
+            # 检查OSS文件是否存在
+            if not replace and self.bucket.object_exists(oss_file_path):
+                logger.warning(f"OSS文件已存在: {oss_file_path}")
+                return False
+            
+            # 上传文件
+            self.bucket.put_object_from_file(oss_file_path, local_file_path)
+            
+            logger.info(f"上传成功: {local_file_path} -> {oss_file_path}")
+            return True
+            
+        except Exception as e:
+            logger.error(f"上传文件失败: {local_file_path} -> {oss_file_path}, 错误: {e}")
+            return False
+    
+    def file_exists(self, oss_file_path: str) -> bool:
+        """
+        检查OSS文件是否存在
+        
         Args:
-            oss_object_dir (str): _description_
-            local_dir (str): _description_
-            start_date (Union[str, dt.date, dt.datetime], optional): _description_. Defaults to None.
-            end_date (Union[str, dt.date, dt.datetime], optional): _description_. Defaults to None.
-            max_num_files (int, optional): _description_. Defaults to 5.
-            replace (bool, optional): _description_. Defaults to True.
-            delimiter (str, optional): _description_. Defaults to '/'.
-            num_threads (int, optional): _description_. Defaults to 1.
-            suffix (str, optional): _description_. Defaults to ''.
-            **kwargs: Additional keyword arguments passed to `list_latest_files_by_date`
+            oss_file_path: OSS文件路径
+            
+        Returns:
+            文件是否存在
         """
-        if isinstance(local_dir, str):
-            local_dir = Path(local_dir)
-
-        oss_file_list = self.list_latest_files_by_date(
-            object_dir=oss_object_dir,
-            delimiter=delimiter,
-            start_date=start_date,
-            end_date=end_date,
-            date_pattern=date_pattern,
-            file_date_format=file_date_format,
-            max_num_files=max_num_files,
-            suffix=suffix,
-            **kwargs,
-        )
-
-        if num_threads == 1:
-            for oss_file_path in oss_file_list:
-                file_name = Path(oss_file_path).name
-                local_file_path = local_dir / file_name
-
-                self.download_object_to_file(
-                    oss_file_path=oss_file_path,
-                    local_file_path=local_file_path,
-                    replace=replace,
-                    make_dir=make_dir,
-                )
-
-            return
-
-        # Start multithreaded process if num_threads > 1
-        with ThreadPool(num_threads) as p:
-            pool_args_list = []
-            for oss_file_path in oss_file_list:
-                file_name = Path(oss_file_path).name
-                local_file_path = local_dir / file_name
-
-                args_dict = dict(
-                    oss_file_path=str(oss_file_path),
-                    local_file_path=str(local_file_path),
-                    replace=replace,
-                    make_dir=make_dir,
-                    print_logs=True,
-                )
-                pool_args_list.append(args_dict)
-
-            ret_all = list(
-                tqdm(
-                    starstarmap(
-                        pool=p,
-                        fn=self.download_object_to_file,
-                        kwargs_iter=pool_args_list,
-                    ),
-                    total=len(oss_file_list),
-                )
-            )
-
-    def upload_file_to_object(
-        self,
-        local_file_path: str,
-        oss_file_path: str | Path,
-        replace: bool = False,
-        print_logs: bool = True,
-    ):
-        """Upload a single local file to OSS
-
+        try:
+            return self.bucket.object_exists(oss_file_path)
+        except Exception as e:
+            logger.error(f"检查文件存在性失败: {oss_file_path}, 错误: {e}")
+            return False
+    
+    def get_file_info(self, oss_file_path: str) -> Optional[Dict]:
+        """
+        获取OSS文件信息
+        
         Args:
-            oss_file_path (str): _description_
-            local_file_path (str): _description_
-            replace (bool, optional): _description_. Defaults to False.
+            oss_file_path: OSS文件路径
+            
+        Returns:
+            文件信息字典
         """
-        if isinstance(local_file_path, Path):
-            local_file_path = str(local_file_path)
-
-        if isinstance(oss_file_path, Path):
-            oss_file_path = str(oss_file_path)
-
-        # Check if file already exists
-        is_file_exists = self._bucket.object_exists(
-            key=oss_file_path,
-        )
-
-        if is_file_exists:
-            if replace:
-                if print_logs:
-                    err_msg = f"{oss_file_path} already exists, replacing file..."
-                    self.logger.info(err_msg)
-
-                self._bucket.put_object_from_file(
-                    key=str(oss_file_path),
-                    filename=local_file_path,
-                )
-
-            else:
-                if print_logs:
-                    err_msg = f"{oss_file_path} already exists, skipping file..."
-                    self.logger.info(err_msg)
-
-                return
-
-        self._bucket.put_object_from_file(
-            key=oss_file_path,
-            filename=local_file_path,
-        )
-
-    def upload_files_to_objects(
-        self,
-        file_upload_mapping: list[tuple[str | Path, str | Path]],
-        local_base_dir: str = None,
-        oss_base_dir: str = None,
-        replace: bool = True,
-        num_threads: int = 1,
-        **kwargs,
-    ) -> None:
-        """Upload files from local storage to OSS.
-
+        try:
+            obj = self.bucket.get_object_meta(oss_file_path)
+            
+            return {
+                'key': oss_file_path,
+                'name': os.path.basename(oss_file_path),
+                'size': obj.content_length,
+                'last_modified': obj.last_modified,
+                'etag': obj.etag,
+                'content_type': obj.content_type
+            }
+            
+        except oss2.exceptions.NoSuchKey:
+            logger.warning(f"文件不存在: {oss_file_path}")
+            return None
+        except Exception as e:
+            logger.error(f"获取文件信息失败: {oss_file_path}, 错误: {e}")
+            return None
+    
+    def delete_file(self, oss_file_path: str) -> bool:
+        """
+        删除OSS文件
+        
         Args:
-            file_upload_mapping (list[tuple[str | Path, str | Path]]): A list of file path pairs that maps the local file path (to upload) to the OSS file path (upload location).
-            oss_base_dir (str): OSS directory path to be prepended to all OSS file paths.
-            local_base_dir (str, optional): Base directory path to be prepended to all local file paths.
-            replace (bool, optional): _description_. Defaults to True.
-            num_threads (int, optional): _description_. Defaults to 1.
-            **kwargs: Additional keyword arguments passed to `list_latest_files_by_date`
+            oss_file_path: OSS文件路径
+            
+        Returns:
+            删除是否成功
         """
-        if isinstance(oss_base_dir, str):
-            oss_base_dir = Path(oss_base_dir)
-
-        if isinstance(local_base_dir, str):
-            local_base_dir = Path(local_base_dir)
-
-        if not isinstance(file_upload_mapping, list):
-            raise TypeError("file_upload_mapping must be a list of 2 value tuples.")
-
-        for item in file_upload_mapping:
-            if not isinstance(item, tuple):
-                raise TypeError(
-                    "Each item in the file_upload_mapping list must be a 2 value tuple."
-                )
-
-            if len(item) != 2:
-                raise ValueError(
-                    "Each tuple in the file_upload_mapping list must be length 2."
-                )
-
-        if num_threads == 1:
-            for local_file_path, oss_file_path in file_upload_mapping:
-                if local_base_dir is not None:
-                    local_file_path = local_base_dir / local_file_path
-
-                if oss_base_dir is not None:
-                    oss_file_path = oss_base_dir / oss_file_path
+        try:
+            self.bucket.delete_object(oss_file_path)
+            logger.info(f"删除成功: {oss_file_path}")
+            return True
+            
+        except Exception as e:
+            logger.error(f"删除文件失败: {oss_file_path}, 错误: {e}")
+            return False
 
-                self.upload_file_to_object(
-                    oss_file_path=oss_file_path,
-                    local_file_path=local_file_path,
-                    replace=replace,
-                )
 
-            return
+# 兼容性别名 - 保持与原始���码的兼容性
+class SimpleOSSManager(OSSFileManager):
+    """兼容性别名"""
+    pass
 
-        # Start multithreaded process if num_threads > 1
-        with ThreadPool(num_threads) as p:
-            pool_args_list = []
-            for local_file_path, oss_file_path in file_upload_mapping:
-                if local_base_dir is not None:
-                    local_file_path = local_base_dir / local_file_path
 
-                if oss_base_dir is not None:
-                    oss_file_path = oss_base_dir / oss_file_path
-
-                args_dict = dict(
-                    oss_file_path=oss_file_path,
-                    local_file_path=local_file_path,
-                    replace=replace,
-                    print_logs=True,
-                )
-                pool_args_list.append(args_dict)
-
-            ret_all = list(
-                tqdm(
-                    starstarmap(
-                        pool=p,
-                        fn=self.upload_file_to_object,
-                        kwargs_iter=pool_args_list,
-                    ),
-                    total=len(file_upload_mapping),
-                )
-            )
-
-    @property
-    def bucket(self) -> oss2.Bucket:
-        return self._bucket
-
-
-# %%
 if __name__ == "__main__":
-    # %% Initialize
-    oss_file_manager = OSSFileManager(logger=logger)
-
-    # %% List the latest files by date (based on file suffix) in an OSS directory
-    oss_file_manager.list_latest_files_by_date(
-        "compass-arena/dev/data/conversations/", max_num_files=5
-    )
+    # 测试代码
+    try:
+        manager = OSSFileManager()
+        print("✅ OSS文件管理器初始化成功")
+        
+        # 测试列出文件
+        files = manager.list_files("atlas_eval/submissions/", file_extension=".json")
+        print(f"📁 找到 {len(files)} 个提交文件")
+        
+        for file_info in files[:3]:  # 只显示前3个
+            print(f"  - {file_info['name']} ({file_info['size']} bytes)")
+            
+    except Exception as e:
+        print(f"❌ 测试失败: {e}")

src/oss/oss_file_manager_old.py

@@ -0,0 +1,517 @@
+# %%
+import datetime as dt
+import os
+import re
+from logging import Logger
+from multiprocessing import Pool
+from multiprocessing.pool import ThreadPool
+from pathlib import Path
+from typing import List, Union
+
+import oss2
+from loguru import logger
+from oss2.credentials import EnvironmentVariableCredentialsProvider
+from tqdm import tqdm
+
+from compassflow.constants import DATADIR
+from compassflow.oss.oss import OssBucket
+from compassflow.utils import starstarmap
+
+
+class OSSFileManager:
+    def __init__(
+        self,
+        oss_access_key_id: str = None,
+        oss_access_key_secret: str = None,
+        region: str = "http://oss-cn-shanghai.aliyuncs.com",
+        bucket_name: str = "opencompass",
+        oss_block_name: str = None,
+        logger: Logger = logger,
+    ) -> None:
+        """OSS File Manager
+
+        Args:
+            oss_access_key_id (str, optional): _description_. Defaults to None.
+            oss_access_key_secret (str, optional): _description_. Defaults to None.
+            region (_type_, optional): _description_. Defaults to 'http://oss-cn-shanghai.aliyuncs.com'.
+            bucket_name (str, optional): _description_. Defaults to 'opencompass'.
+            oss_block_name: oss_block_name which is defined in the prefect
+            logger (Logger, optional): _description_. Defaults to logger.
+        """
+        self.logger = logger
+        if oss_block_name is not None:
+            oss_bucket = OssBucket.load(oss_block_name)
+            self._bucket = oss_bucket._get_bucket()
+            return
+
+        # 阿里云账号AccessKey拥有所有API的访问权限，风险很高。强烈建议您创建并使用RAM账号进行API访问或日常运维，请登录RAM控制台创建RAM账号。
+        if oss_access_key_id is not None and oss_access_key_secret is not None:
+            os.environ["OSS_ACCESS_KEY_ID"] = oss_access_key_id
+            os.environ["OSS_ACCESS_KEY_SECRET"] = oss_access_key_secret
+
+        if (
+            os.getenv("OSS_ACCESS_KEY_ID") is None
+            or os.getenv("OSS_ACCESS_KEY_SECRET") is None
+        ):
+            raise ValueError("Access Key ID and Access Key Secret cannot be empty.")
+
+        auth = oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())
+
+        # Endpoint以杭州为例，其它Region请按实际情况填写。
+        # 填写Bucket名称，例如examplebucket。
+        bucket = oss2.Bucket(
+            auth=auth,
+            endpoint=region,
+            bucket_name=bucket_name,
+        )
+
+        self._bucket = bucket
+
+    def list_latest_files_by_date(
+        self,
+        object_dir: str = "",
+        delimiter: str = "/",
+        start_date: Union[str, dt.date, dt.datetime] = None,
+        end_date: Union[str, dt.date, dt.datetime] = None,
+        max_num_files: int = 10,
+        date_pattern: str = r"([0-9]{4}-[0-9]{2}-[0-9]{2})",
+        file_date_format: str = "%Y-%m-%d",
+        suffix: str = "",
+    ) -> List[Union[str, Path]]:
+        """List the latest files by date in an OSS bucket directory
+
+        Args:
+            object_dir (str, optional): _description_. Defaults to ''.
+            delimiter (str, optional): _description_. Defaults to '/'.
+            start_date (Union[str, dt.date], optional): _description_. Defaults to None.
+            end_date (Union[str, dt.date], optional): _description_. Defaults to None.
+            max_num_files (int, optional): _description_. Defaults to 10.
+            date_pattern (str, optional): _description_. Defaults to r'^([0-9]{4}-[0-9]{2}-[0-9]{2})'.
+            suffix (str, optional): _description_. Defaults to ''.
+
+        Returns:
+            List[Union[str, Path]]: _description_
+        """
+        if start_date is not None:
+            if isinstance(start_date, str):
+                start_date = dt.datetime.strptime(start_date.replace("-", ""), "%Y%m%d")
+
+            if isinstance(start_date, dt.date):
+                start_date = dt.datetime(
+                    start_date.year, start_date.month, start_date.day, 0, 0, 0
+                )
+
+        if end_date is not None:
+            if isinstance(end_date, str):
+                end_date = dt.datetime.strptime(end_date.replace("-", ""), "%Y%m%d")
+
+            if isinstance(end_date, dt.date):
+                end_date = dt.datetime(
+                    end_date.year, end_date.month, end_date.day, 0, 0, 0
+                )
+
+        object_iter = oss2.ObjectIterator(
+            bucket=self._bucket, prefix=object_dir, delimiter=delimiter
+        )
+
+        root_dir = Path(object_dir)
+        filenames = []
+        for filename in object_iter:
+            filename = filename.key.replace(object_dir, "")
+            # print(filename)
+
+            if filename.endswith(suffix):
+                # Match date pattern in filename
+                date_search = re.search(date_pattern, filename)
+                if date_search:
+                    file_date = dt.datetime.strptime(
+                        date_search.group(1), file_date_format
+                    )
+                else:
+                    self.logger.warning(
+                        f"date pattern doesn't match, skipping file {filename}"
+                    )
+                    continue
+
+                # Check if file date within specified range
+                if start_date is not None:
+                    if start_date > file_date:
+                        continue
+
+                if end_date is not None:
+                    if end_date < file_date:
+                        continue
+
+                filepath = root_dir / filename
+                # name_tstamp_tuple = (filepath, os.path.getmtime(filepath))
+                # filenames.append(name_tstamp_tuple)
+                filenames.append(str(filepath))
+        # sort by tstamp
+        # filenames = sorted(filenames, key=lambda x: x[1])
+        filenames = sorted(filenames)
+        # filenames = [x[0] for x in filenames]
+
+        max_num_files = max_num_files or len(filenames)
+        filenames = filenames[-max_num_files:]
+
+        self.logger.info(f"{filenames=}")
+
+        return filenames
+
+    def download_object_to_file(
+        self,
+        oss_file_path: str | Path,
+        local_file_path: str | Path,
+        replace: bool = False,
+        make_dir: bool = False,
+        print_logs: bool = True,
+    ):
+        """Download a single OSS object to local file.
+
+        Args:
+            oss_file_path (str): _description_
+            local_file_path (str): _description_
+            replace (bool, optional): _description_. Defaults to False.
+            make_dir (bool, optional): Whether to create intermediate dirs if they don't exist. Defaults to False.
+            print_logs: bool, optional): Whether to print logs. Defaults to True.
+        """
+        if isinstance(local_file_path, str):
+            local_file_path = Path(local_file_path)
+
+        if not replace:
+            if local_file_path.exists():
+                if print_logs:
+                    err_msg = f"{local_file_path} already exists, skipping file..."
+                    self.logger.info(err_msg)
+
+                return
+
+        if print_logs:
+            if local_file_path.exists():
+                err_msg = f"{local_file_path} already exists, replacing file..."
+                self.logger.info(err_msg)
+
+        if make_dir:
+            os.makedirs(local_file_path.parent, exist_ok=True)
+
+        self._bucket.get_object_to_file(
+            key=str(oss_file_path),
+            filename=local_file_path,
+        )
+
+    def download_objects_to_files(
+        self,
+        file_download_mapping: list[tuple[str | Path, str | Path]],
+        oss_base_dir: str = None,
+        local_base_dir: str = None,
+        replace: bool = True,
+        num_threads: int = 1,
+        **kwargs,
+    ) -> None:
+        """Download objects from OSS to local storage.
+
+        Args:
+            file_download_mapping (list[tuple[str | Path, str | Path]]): A list of file path pairs that maps the OSS file path (to download) to the local file path (download location).
+            oss_base_dir (str): OSS directory path to be prepended to all OSS file paths.
+            local_base_dir (str, optional): Base directory path to be prepended to all local file paths.
+            replace (bool, optional): _description_. Defaults to True.
+            num_threads (int, optional): _description_. Defaults to 1.
+            **kwargs: Additional keyword arguments passed to `list_latest_files_by_date`
+        """
+        if isinstance(oss_base_dir, str):
+            oss_base_dir = Path(oss_base_dir)
+
+        if isinstance(local_base_dir, str):
+            local_base_dir = Path(local_base_dir)
+
+        if not isinstance(file_download_mapping, list):
+            raise TypeError("file_download_mapping must be a list of 2 value tuples.")
+
+        for item in file_download_mapping:
+            if not isinstance(item, tuple):
+                raise TypeError(
+                    "Each item in the file_download_mapping list must be a 2 value tuple."
+                )
+
+            if len(item) != 2:
+                raise ValueError(
+                    "Each tuple in the file_download_mapping list must be length 2."
+                )
+
+        if num_threads == 1:
+            for local_file_path, oss_file_path in file_download_mapping:
+                if local_base_dir is not None:
+                    local_file_path = local_base_dir / local_file_path
+
+                if oss_base_dir is not None:
+                    oss_file_path = oss_base_dir / oss_file_path
+
+                self.download_object_to_file(
+                    oss_file_path=oss_file_path,
+                    local_file_path=local_file_path,
+                    replace=replace,
+                )
+
+            return
+
+        # Start multithreaded process if num_threads > 1
+        with ThreadPool(num_threads) as p:
+            pool_args_list = []
+            for local_file_path, oss_file_path in file_download_mapping:
+                if local_base_dir is not None:
+                    local_file_path = local_base_dir / local_file_path
+
+                if oss_base_dir is not None:
+                    oss_file_path = oss_base_dir / oss_file_path
+
+                args_dict = dict(
+                    oss_file_path=oss_file_path,
+                    local_file_path=local_file_path,
+                    replace=replace,
+                    print_logs=True,
+                )
+                pool_args_list.append(args_dict)
+
+            ret_all = list(
+                tqdm(
+                    starstarmap(
+                        pool=p,
+                        fn=self.download_object_to_file,
+                        kwargs_iter=pool_args_list,
+                    ),
+                    total=len(file_download_mapping),
+                )
+            )
+
+    def download_latest_objects_to_dir(
+        self,
+        oss_object_dir: str,
+        local_dir: str,
+        start_date: Union[str, dt.date, dt.datetime] = None,
+        end_date: Union[str, dt.date, dt.datetime] = None,
+        date_pattern: str = r"([0-9]{4}-[0-9]{2}-[0-9]{2})",
+        file_date_format: str = "%Y-%m-%d",
+        max_num_files: int = 5,
+        replace: bool = True,
+        make_dir: bool = True,
+        delimiter: str = "/",
+        num_threads: int = 1,
+        suffix: str = "",
+        **kwargs,
+    ) -> None:
+        """Download the latest objects from oss to local directory
+
+        Args:
+            oss_object_dir (str): _description_
+            local_dir (str): _description_
+            start_date (Union[str, dt.date, dt.datetime], optional): _description_. Defaults to None.
+            end_date (Union[str, dt.date, dt.datetime], optional): _description_. Defaults to None.
+            max_num_files (int, optional): _description_. Defaults to 5.
+            replace (bool, optional): _description_. Defaults to True.
+            delimiter (str, optional): _description_. Defaults to '/'.
+            num_threads (int, optional): _description_. Defaults to 1.
+            suffix (str, optional): _description_. Defaults to ''.
+            **kwargs: Additional keyword arguments passed to `list_latest_files_by_date`
+        """
+        if isinstance(local_dir, str):
+            local_dir = Path(local_dir)
+
+        oss_file_list = self.list_latest_files_by_date(
+            object_dir=oss_object_dir,
+            delimiter=delimiter,
+            start_date=start_date,
+            end_date=end_date,
+            date_pattern=date_pattern,
+            file_date_format=file_date_format,
+            max_num_files=max_num_files,
+            suffix=suffix,
+            **kwargs,
+        )
+
+        if num_threads == 1:
+            for oss_file_path in oss_file_list:
+                file_name = Path(oss_file_path).name
+                local_file_path = local_dir / file_name
+
+                self.download_object_to_file(
+                    oss_file_path=oss_file_path,
+                    local_file_path=local_file_path,
+                    replace=replace,
+                    make_dir=make_dir,
+                )
+
+            return
+
+        # Start multithreaded process if num_threads > 1
+        with ThreadPool(num_threads) as p:
+            pool_args_list = []
+            for oss_file_path in oss_file_list:
+                file_name = Path(oss_file_path).name
+                local_file_path = local_dir / file_name
+
+                args_dict = dict(
+                    oss_file_path=str(oss_file_path),
+                    local_file_path=str(local_file_path),
+                    replace=replace,
+                    make_dir=make_dir,
+                    print_logs=True,
+                )
+                pool_args_list.append(args_dict)
+
+            ret_all = list(
+                tqdm(
+                    starstarmap(
+                        pool=p,
+                        fn=self.download_object_to_file,
+                        kwargs_iter=pool_args_list,
+                    ),
+                    total=len(oss_file_list),
+                )
+            )
+
+    def upload_file_to_object(
+        self,
+        local_file_path: str,
+        oss_file_path: str | Path,
+        replace: bool = False,
+        print_logs: bool = True,
+    ):
+        """Upload a single local file to OSS
+
+        Args:
+            oss_file_path (str): _description_
+            local_file_path (str): _description_
+            replace (bool, optional): _description_. Defaults to False.
+        """
+        if isinstance(local_file_path, Path):
+            local_file_path = str(local_file_path)
+
+        if isinstance(oss_file_path, Path):
+            oss_file_path = str(oss_file_path)
+
+        # Check if file already exists
+        is_file_exists = self._bucket.object_exists(
+            key=oss_file_path,
+        )
+
+        if is_file_exists:
+            if replace:
+                if print_logs:
+                    err_msg = f"{oss_file_path} already exists, replacing file..."
+                    self.logger.info(err_msg)
+
+                self._bucket.put_object_from_file(
+                    key=str(oss_file_path),
+                    filename=local_file_path,
+                )
+
+            else:
+                if print_logs:
+                    err_msg = f"{oss_file_path} already exists, skipping file..."
+                    self.logger.info(err_msg)
+
+                return
+
+        self._bucket.put_object_from_file(
+            key=oss_file_path,
+            filename=local_file_path,
+        )
+
+    def upload_files_to_objects(
+        self,
+        file_upload_mapping: list[tuple[str | Path, str | Path]],
+        local_base_dir: str = None,
+        oss_base_dir: str = None,
+        replace: bool = True,
+        num_threads: int = 1,
+        **kwargs,
+    ) -> None:
+        """Upload files from local storage to OSS.
+
+        Args:
+            file_upload_mapping (list[tuple[str | Path, str | Path]]): A list of file path pairs that maps the local file path (to upload) to the OSS file path (upload location).
+            oss_base_dir (str): OSS directory path to be prepended to all OSS file paths.
+            local_base_dir (str, optional): Base directory path to be prepended to all local file paths.
+            replace (bool, optional): _description_. Defaults to True.
+            num_threads (int, optional): _description_. Defaults to 1.
+            **kwargs: Additional keyword arguments passed to `list_latest_files_by_date`
+        """
+        if isinstance(oss_base_dir, str):
+            oss_base_dir = Path(oss_base_dir)
+
+        if isinstance(local_base_dir, str):
+            local_base_dir = Path(local_base_dir)
+
+        if not isinstance(file_upload_mapping, list):
+            raise TypeError("file_upload_mapping must be a list of 2 value tuples.")
+
+        for item in file_upload_mapping:
+            if not isinstance(item, tuple):
+                raise TypeError(
+                    "Each item in the file_upload_mapping list must be a 2 value tuple."
+                )
+
+            if len(item) != 2:
+                raise ValueError(
+                    "Each tuple in the file_upload_mapping list must be length 2."
+                )
+
+        if num_threads == 1:
+            for local_file_path, oss_file_path in file_upload_mapping:
+                if local_base_dir is not None:
+                    local_file_path = local_base_dir / local_file_path
+
+                if oss_base_dir is not None:
+                    oss_file_path = oss_base_dir / oss_file_path
+
+                self.upload_file_to_object(
+                    oss_file_path=oss_file_path,
+                    local_file_path=local_file_path,
+                    replace=replace,
+                )
+
+            return
+
+        # Start multithreaded process if num_threads > 1
+        with ThreadPool(num_threads) as p:
+            pool_args_list = []
+            for local_file_path, oss_file_path in file_upload_mapping:
+                if local_base_dir is not None:
+                    local_file_path = local_base_dir / local_file_path
+
+                if oss_base_dir is not None:
+                    oss_file_path = oss_base_dir / oss_file_path
+
+                args_dict = dict(
+                    oss_file_path=oss_file_path,
+                    local_file_path=local_file_path,
+                    replace=replace,
+                    print_logs=True,
+                )
+                pool_args_list.append(args_dict)
+
+            ret_all = list(
+                tqdm(
+                    starstarmap(
+                        pool=p,
+                        fn=self.upload_file_to_object,
+                        kwargs_iter=pool_args_list,
+                    ),
+                    total=len(file_upload_mapping),
+                )
+            )
+
+    @property
+    def bucket(self) -> oss2.Bucket:
+        return self._bucket
+
+
+# %%
+if __name__ == "__main__":
+    # %% Initialize
+    oss_file_manager = OSSFileManager(logger=logger)
+
+    # %% List the latest files by date (based on file suffix) in an OSS directory
+    oss_file_manager.list_latest_files_by_date(
+        "compass-arena/dev/data/conversations/", max_num_files=5
+    )

src/oss/oss_file_manager_simple.py

@@ -0,0 +1,254 @@
+#!/usr/bin/env python3
+"""
+简化的OSS文件管理器 - 专为SAGE-Bench HuggingFace Space设计
+移除了对 compassflow 的依赖，只保留必需的OSS功能
+"""
+
+import os
+import oss2
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import List, Dict, Optional
+from loguru import logger
+
+
+class OSSFileManager:
+    """简化的OSS文件管理器"""
+    
+    def __init__(
+        self,
+        oss_access_key_id: str = None,
+        oss_access_key_secret: str = None,
+        oss_region: str = None,
+        oss_bucket_name: str = None
+    ):
+        """
+        初始化OSS文件管理器
+        
+        Args:
+            oss_access_key_id: OSS访问密钥ID
+            oss_access_key_secret: OSS访问密钥Secret
+            oss_region: OSS区域端点
+            oss_bucket_name: OSS存储桶名称
+        """
+        # 从环境变量获取配置
+        self.access_key_id = oss_access_key_id or os.getenv('OSS_ACCESS_KEY_ID')
+        self.access_key_secret = oss_access_key_secret or os.getenv('OSS_ACCESS_KEY_SECRET')
+        self.region = oss_region or os.getenv('OSS_REGION', 'http://oss-cn-shanghai.aliyuncs.com')
+        self.bucket_name = oss_bucket_name or os.getenv('OSS_BUCKET_NAME', 'opencompass')
+        
+        if not self.access_key_id or not self.access_key_secret:
+            raise ValueError("OSS访问密钥未设置。请设置 OSS_ACCESS_KEY_ID 和 OSS_ACCESS_KEY_SECRET 环境变量。")
+        
+        # 初始化OSS客户端
+        auth = oss2.Auth(self.access_key_id, self.access_key_secret)
+        self.bucket = oss2.Bucket(auth, self.region, self.bucket_name)
+        
+        logger.info(f"OSS初始化成功: {self.bucket_name} @ {self.region}")
+    
+    def list_files(
+        self, 
+        oss_dir: str = "", 
+        after_date: datetime = None,
+        file_extension: str = None
+    ) -> List[Dict]:
+        """
+        列出OSS目录中的文件
+        
+        Args:
+            oss_dir: OSS目录路径
+            after_date: 只返回此日期之后的文件
+            file_extension: 文件扩展名过滤 (如 ".json")
+            
+        Returns:
+            文件信息列表
+        """
+        try:
+            files = []
+            
+            # 确保目录路径以 / 结尾
+            if oss_dir and not oss_dir.endswith('/'):
+                oss_dir += '/'
+            
+            # 列出对象
+            for obj in oss2.ObjectIterator(self.bucket, prefix=oss_dir):
+                # 跳过目录本身
+                if obj.key.endswith('/'):
+                    continue
+                
+                # 文件扩展名过滤
+                if file_extension and not obj.key.endswith(file_extension):
+                    continue
+                
+                # 日期过滤
+                if after_date and obj.last_modified < after_date:
+                    continue
+                
+                file_info = {
+                    'key': obj.key,
+                    'name': os.path.basename(obj.key),
+                    'size': obj.size,
+                    'last_modified': obj.last_modified,
+                    'etag': obj.etag
+                }
+                files.append(file_info)
+            
+            logger.info(f"找到 {len(files)} 个文件在 {oss_dir}")
+            return files
+            
+        except Exception as e:
+            logger.error(f"列出文件失败: {e}")
+            raise
+    
+    def download_file(self, oss_file_path: str, local_file_path: str) -> bool:
+        """
+        从OSS下载文件到本地
+        
+        Args:
+            oss_file_path: OSS文件路径
+            local_file_path: 本地文件路径
+            
+        Returns:
+            下载是否成功
+        """
+        try:
+            # 确保本地目录存在
+            local_dir = os.path.dirname(local_file_path)
+            if local_dir:
+                os.makedirs(local_dir, exist_ok=True)
+            
+            # 下载文件
+            self.bucket.get_object_to_file(oss_file_path, local_file_path)
+            
+            logger.info(f"下载成功: {oss_file_path} -> {local_file_path}")
+            return True
+            
+        except Exception as e:
+            logger.error(f"下载文件失败: {oss_file_path} -> {local_file_path}, 错误: {e}")
+            return False
+    
+    def upload_file_to_object(
+        self, 
+        local_file_path: str, 
+        oss_file_path: str,
+        replace: bool = False
+    ) -> bool:
+        """
+        上传本地文件到OSS
+        
+        Args:
+            local_file_path: 本地文件路径
+            oss_file_path: OSS文件路径
+            replace: 是否替换已存在的文件
+            
+        Returns:
+            上传是否成功
+        """
+        try:
+            # 检查本地文件是否存在
+            if not os.path.exists(local_file_path):
+                logger.error(f"本地文件不存在: {local_file_path}")
+                return False
+            
+            # 检查OSS文件是否存在
+            if not replace and self.bucket.object_exists(oss_file_path):
+                logger.warning(f"OSS文件已存在: {oss_file_path}")
+                return False
+            
+            # 上传文件
+            self.bucket.put_object_from_file(oss_file_path, local_file_path)
+            
+            logger.info(f"上传成功: {local_file_path} -> {oss_file_path}")
+            return True
+            
+        except Exception as e:
+            logger.error(f"上传文件失败: {local_file_path} -> {oss_file_path}, 错误: {e}")
+            return False
+    
+    def file_exists(self, oss_file_path: str) -> bool:
+        """
+        检查OSS文件是否存在
+        
+        Args:
+            oss_file_path: OSS文件路径
+            
+        Returns:
+            文件是否存在
+        """
+        try:
+            return self.bucket.object_exists(oss_file_path)
+        except Exception as e:
+            logger.error(f"检查文件存在性失败: {oss_file_path}, 错误: {e}")
+            return False
+    
+    def get_file_info(self, oss_file_path: str) -> Optional[Dict]:
+        """
+        获取OSS文件信息
+        
+        Args:
+            oss_file_path: OSS文件路径
+            
+        Returns:
+            文件信息字典
+        """
+        try:
+            obj = self.bucket.get_object_meta(oss_file_path)
+            
+            return {
+                'key': oss_file_path,
+                'name': os.path.basename(oss_file_path),
+                'size': obj.content_length,
+                'last_modified': obj.last_modified,
+                'etag': obj.etag,
+                'content_type': obj.content_type
+            }
+            
+        except oss2.exceptions.NoSuchKey:
+            logger.warning(f"文件不存在: {oss_file_path}")
+            return None
+        except Exception as e:
+            logger.error(f"获取文件信息失败: {oss_file_path}, 错误: {e}")
+            return None
+    
+    def delete_file(self, oss_file_path: str) -> bool:
+        """
+        删除OSS文件
+        
+        Args:
+            oss_file_path: OSS文件路径
+            
+        Returns:
+            删除是否成功
+        """
+        try:
+            self.bucket.delete_object(oss_file_path)
+            logger.info(f"删除成功: {oss_file_path}")
+            return True
+            
+        except Exception as e:
+            logger.error(f"删除文件失败: {oss_file_path}, 错误: {e}")
+            return False
+
+
+# 兼容性别名 - 保持与原始代码的兼容性
+class SimpleOSSManager(OSSFileManager):
+    """兼容性别名"""
+    pass
+
+
+if __name__ == "__main__":
+    # 测试代码
+    try:
+        manager = OSSFileManager()
+        print("✅ OSS文件管理器初始化成功")
+        
+        # 测试列出文件
+        files = manager.list_files("atlas_eval/submissions/", file_extension=".json")
+        print(f"📁 找到 {len(files)} 个提交文件")
+        
+        for file_info in files[:3]:  # 只显示前3个
+            print(f"  - {file_info['name']} ({file_info['size']} bytes)")
+            
+    except Exception as e:
+        print(f"❌ 测试失败: {e}")