make2/factory_sliceing/auto_sliceing_operate/utils/miniIo.py

import os
from minio import Minio
from minio.error import S3Error
from .logs import log

# MiniIO 客户端单例模式
class MiniIOClientSingleton:
    _instance = None
    _client = None
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(MiniIOClientSingleton, cls).__new__(cls)
        return cls._instance
    
    def get_client(self):
        if self._client is None:
            # MiniIO 连接配置
            # 可以从环境变量或配置文件中读取
            endpoint = os.getenv('MINIO_ENDPOINT', 'localhost:9000')
            access_key = os.getenv('MINIO_ACCESS_KEY', 'minioadmin')
            secret_key = os.getenv('MINIO_SECRET_KEY', 'minioadmin')
            secure = os.getenv('MINIO_SECURE', 'False').lower() == 'true'
            
            try:
                self._client = Minio(
                    endpoint,
                    access_key=access_key,
                    secret_key=secret_key,
                    secure=secure
                )
                log(f"MiniIO 客户端初始化成功: {endpoint}")
            except Exception as e:
                log(f"MiniIO 客户端初始化失败: {str(e)}")
                raise
        return self._client


def miniIOClient():
    """获取 MiniIO 客户端单例"""
    return MiniIOClientSingleton().get_client()


def upload_file(bucket_name, object_name, file_path):
    """
    上传文件到 MiniIO
    
    Args:
        bucket_name: 存储桶名称
        object_name: 对象名称（在 MiniIO 中的路径）
        file_path: 本地文件路径
    
    Returns:
        bool: 上传成功返回 True，失败返回 False
    """
    try:
        client = miniIOClient()
        
        # 检查存储桶是否存在，不存在则创建
        if not client.bucket_exists(bucket_name):
            client.make_bucket(bucket_name)
            log(f"创建存储桶: {bucket_name}")
        
        # 上传文件
        log(f"开始上传文件: {file_path} -> {bucket_name}/{object_name}")
        client.fput_object(bucket_name, object_name, file_path)
        log(f"文件上传成功: {bucket_name}/{object_name}")
        return True
    except S3Error as e:
        log(f"上传文件失败 (S3Error): {str(e)}")
        return False
    except Exception as e:
        log(f"上传文件失败: {str(e)}")
        return False


def download_file(bucket_name, object_name, file_path):
    """
    从 MiniIO 下载文件到本地
    
    Args:
        bucket_name: 存储桶名称
        object_name: 对象名称（在 MiniIO 中的路径）
        file_path: 本地保存路径
    
    Returns:
        bool: 下载成功返回 True，失败返回 False
    """
    try:
        client = miniIOClient()
        
        # 检查文件是否存在
        if not check_file_exists(bucket_name, object_name):
            log(f"文件不存在: {bucket_name}/{object_name}")
            return False
        
        # 确保本地目录存在
        local_dir = os.path.dirname(file_path)
        if local_dir and not os.path.exists(local_dir):
            os.makedirs(local_dir, exist_ok=True)
        
        # 下载文件
        log(f"开始下载文件: {bucket_name}/{object_name} -> {file_path}")
        client.fget_object(bucket_name, object_name, file_path)
        
        # 验证文件是否下载成功
        if os.path.exists(file_path):
            file_size = os.path.getsize(file_path)
            log(f"文件下载成功: {file_path} (大小: {file_size} 字节)")
            return True
        else:
            log("文件下载失败: 本地文件不存在")
            return False
    except S3Error as e:
        log(f"下载文件失败 (S3Error): {str(e)}")
        return False
    except Exception as e:
        log(f"下载文件失败: {str(e)}")
        return False


def delete_file(bucket_name, object_name):
    """
    从 MiniIO 删除文件
    
    Args:
        bucket_name: 存储桶名称
        object_name: 对象名称（在 MiniIO 中的路径）
    
    Returns:
        bool: 删除成功返回 True，失败返回 False
    """
    try:
        client = miniIOClient()
        
        # 检查文件是否存在
        if not check_file_exists(bucket_name, object_name):
            log(f"文件不存在，无需删除: {bucket_name}/{object_name}")
            return True
        
        # 删除文件
        log(f"开始删除文件: {bucket_name}/{object_name}")
        client.remove_object(bucket_name, object_name)
        log(f"文件删除成功: {bucket_name}/{object_name}")
        return True
    except S3Error as e:
        log(f"删除文件失败 (S3Error): {str(e)}")
        return False
    except Exception as e:
        log(f"删除文件失败: {str(e)}")
        return False


def check_file_exists(bucket_name, object_name):
    """
    检查文件是否存在于 MiniIO
    
    Args:
        bucket_name: 存储桶名称
        object_name: 对象名称（在 MiniIO 中的路径）
    
    Returns:
        bool: 文件存在返回 True，不存在返回 False
    """
    try:
        client = miniIOClient()
        
        # 检查存储桶是否存在
        if not client.bucket_exists(bucket_name):
            log(f"存储桶不存在: {bucket_name}")
            return False
        
        # 检查对象是否存在
        try:
            client.stat_object(bucket_name, object_name)
            return True
        except S3Error as e:
            if e.code == 'NoSuchKey':
                return False
            else:
                log(f"检查文件存在性时出错: {str(e)}")
                return False
    except S3Error as e:
        log(f"检查文件存在性失败 (S3Error): {str(e)}")
        return False
    except Exception as e:
        log(f"检查文件存在性失败: {str(e)}")
        return False