csg-server/app/utils/minio_util.py

from datetime import timedelta
from io import BytesIO
from pathlib import Path

from app.config.setting import settings
from app.core.exceptions import CustomException
from app.core.logger import logger


class MinioUtil:
    """MinIO对象存储工具类。"""

    _client = None

    @staticmethod
    def _check_enable() -> None:
        if not settings.MINIO_ENABLE:
            raise CustomException(msg="请先开启MinIO对象存储", data="请启用配置项 MINIO_ENABLE")

    @staticmethod
    def _region() -> str | None:
        return settings.MINIO_REGION or None

    @staticmethod
    def _load_minio_classes():
        try:
            from minio import Minio
            from minio.error import S3Error
        except ImportError as e:
            raise CustomException(msg="MinIO依赖未安装", data="请先执行 pip install minio 或更新项目依赖") from e
        return Minio, S3Error

    @classmethod
    def get_client(cls):
        cls._check_enable()
        if cls._client is None:
            Minio, _ = cls._load_minio_classes()
            cls._client = Minio(
                endpoint=settings.MINIO_ENDPOINT,
                access_key=settings.MINIO_ACCESS_KEY,
                secret_key=settings.MINIO_SECRET_KEY,
                secure=settings.MINIO_SECURE,
                region=cls._region(),
            )
        return cls._client

    @classmethod
    def ensure_bucket(cls, bucket_name: str | None = None) -> None:
        bucket = bucket_name or settings.MINIO_BUCKET_NAME
        client = cls.get_client()
        _, S3Error = cls._load_minio_classes()
        try:
            if not client.bucket_exists(bucket):
                client.make_bucket(bucket, location=cls._region())
        except S3Error as e:
            logger.error(f"MinIO存储桶初始化失败: {e}")
            raise CustomException(msg="MinIO存储桶初始化失败", data=str(e)) from e

    @classmethod
    def upload_file(
        cls,
        object_name: str,
        file_path: str | Path,
        bucket_name: str | None = None,
        content_type: str | None = None,
    ) -> str:
        bucket = bucket_name or settings.MINIO_BUCKET_NAME
        client = cls.get_client()
        _, S3Error = cls._load_minio_classes()
        try:
            cls.ensure_bucket(bucket)
            client.fput_object(
                bucket_name=bucket,
                object_name=object_name,
                file_path=str(file_path),
                content_type=content_type or "application/octet-stream",
            )
            return cls.get_public_url(object_name=object_name, bucket_name=bucket)
        except S3Error as e:
            logger.error(f"MinIO文件上传失败: {e}")
            raise CustomException(msg="MinIO文件上传失败", data=str(e)) from e

    @classmethod
    def upload_bytes(
        cls,
        object_name: str,
        data: bytes,
        bucket_name: str | None = None,
        content_type: str = "application/octet-stream",
    ) -> str:
        bucket = bucket_name or settings.MINIO_BUCKET_NAME
        client = cls.get_client()
        _, S3Error = cls._load_minio_classes()
        try:
            cls.ensure_bucket(bucket)
            client.put_object(
                bucket_name=bucket,
                object_name=object_name,
                data=BytesIO(data),
                length=len(data),
                content_type=content_type,
            )
            return cls.get_public_url(object_name=object_name, bucket_name=bucket)
        except S3Error as e:
            logger.error(f"MinIO字节流上传失败: {e}")
            raise CustomException(msg="MinIO字节流上传失败", data=str(e)) from e

    @classmethod
    def remove_object(cls, object_name: str, bucket_name: str | None = None) -> None:
        bucket = bucket_name or settings.MINIO_BUCKET_NAME
        client = cls.get_client()
        _, S3Error = cls._load_minio_classes()
        try:
            client.remove_object(bucket_name=bucket, object_name=object_name)
        except S3Error as e:
            logger.error(f"MinIO文件删除失败: {e}")
            raise CustomException(msg="MinIO文件删除失败", data=str(e)) from e

    @classmethod
    def presigned_get_url(
        cls,
        object_name: str,
        bucket_name: str | None = None,
        expires_seconds: int | None = None,
    ) -> str:
        bucket = bucket_name or settings.MINIO_BUCKET_NAME
        client = cls.get_client()
        _, S3Error = cls._load_minio_classes()
        expires = timedelta(seconds=expires_seconds or settings.MINIO_PRESIGNED_EXPIRE_SECONDS)
        try:
            return client.presigned_get_object(bucket_name=bucket, object_name=object_name, expires=expires)
        except S3Error as e:
            logger.error(f"MinIO预签名地址生成失败: {e}")
            raise CustomException(msg="MinIO预签名地址生成失败", data=str(e)) from e

    @staticmethod
    def get_public_url(object_name: str, bucket_name: str | None = None) -> str:
        bucket = bucket_name or settings.MINIO_BUCKET_NAME
        object_path = object_name.lstrip("/")
        if settings.MINIO_PUBLIC_URL:
            return f"{settings.MINIO_PUBLIC_URL.rstrip('/')}/{bucket}/{object_path}"
        scheme = "https" if settings.MINIO_SECURE else "http"
        return f"{scheme}://{settings.MINIO_ENDPOINT.rstrip('/')}/{bucket}/{object_path}"