minio-manager/app/services/minio_service.py

from minio import Minio
from minio.error import S3Error
from datetime import timedelta
from app.providers.minio_client_provider import minio_client_provider
from app.utils.logger_utils import logger

class MinioService:
    def __init__(self):
        """初始化服务，不再创建固定的客户端"""
        pass

    def _ensure_bucket_exists(self, client: Minio, bucket_name: str):
        """确保桶存在"""
        if not client.bucket_exists(bucket_name):
            client.make_bucket(bucket_name)

    def list_files(self, prefix: str):
        """列出文件"""
        # 清理前缀
        if prefix == "/":
            prefix = ""
        elif prefix.startswith("/"):
            prefix = prefix[1:]

        if prefix and not prefix.endswith("/"):
            prefix += "/"

        try:
            objects = minio_client_provider.list_objects(prefix=prefix, recursive=False)

            results = []
            for obj in objects:
                # 跳过与前缀完全相同的对象（通常是文件夹标记）
                if obj.object_name == prefix:
                    continue

                is_dir = obj.is_dir

                # 获取显示名称
                if is_dir:
                    # 对于目录，去掉末尾的斜杠
                    name = obj.object_name[len(prefix):].rstrip('/')
                else:
                    name = obj.object_name[len(prefix):]

                # 跳过空名称
                if not name:
                    continue

                results.append({
                    "name": name,
                    "path": obj.object_name,
                    "is_dir": is_dir,
                    "size": obj.size if not is_dir else 0,
                    "last_modified": str(obj.last_modified) if not is_dir else None,
                    "content_type": obj.content_type
                })

            # 排序
            results.sort(key=lambda x: (not x['is_dir'], x['name'].lower()))
            return results

        except S3Error as e:
            logger.error(f"列出文件错误: {e}", exc_info=True)
            raise e

    def upload_file(self, file_obj, file_path: str, content_type: str, file_size: int):
        """上传文件"""
        minio_client_provider.put_object(
            path=file_path,
            data=file_obj,
            length=file_size,
            content_type=content_type
        )

    def create_folder(self, folder_path: str):
        """创建文件夹"""
        if not folder_path.endswith("/"):
            folder_path += "/"
        # 创建一个空的 Object 以模拟文件夹
        minio_client_provider.put_empty_object(folder_path)

    def get_file_stream(self, file_path: str):
        """获取文件流"""
        return minio_client_provider.get_object(file_path)

    def delete_file(self, path: str, is_dir: bool):
        """删除文件或文件夹"""
        if is_dir:
            minio_client_provider.remove_objects_with_prefix(prefix=path)
        else:
            minio_client_provider.remove_object(path)

    def get_presigned_url(self, path: str):
        """获取预签名URL"""
        return minio_client_provider.get_presigned_url(path, expires=timedelta(hours=1))

    def get_storage_info(self):
        """获取存储桶使用情况"""
        return minio_client_provider.get_storage_info()


# 实例化单例
minio_service = MinioService()