Unified_Authentication_Platform/backend/app/core/minio.py

from minio import Minio
from minio.error import S3Error
from app.core.config import settings
import io
import uuid
from datetime import datetime
import logging

logger = logging.getLogger(__name__)

class MessageStorage:
    def __init__(self):
        try:
            # Handle endpoint protocol
            endpoint = settings.MINIO_ENDPOINT
            secure = settings.MINIO_SECURE
            
            if endpoint.startswith("http://"):
                endpoint = endpoint.replace("http://", "")
                secure = False
            elif endpoint.startswith("https://"):
                endpoint = endpoint.replace("https://", "")
                secure = True
                
            self.client = Minio(
                endpoint=endpoint,
                access_key=settings.MINIO_ACCESS_KEY,
                secret_key=settings.MINIO_SECRET_KEY,
                secure=secure
            )
            self.bucket_name = settings.MINIO_BUCKET_NAME
            self._ensure_bucket()
        except Exception as e:
            logger.error(f"MinIO init failed: {e}")
            self.client = None

    def _ensure_bucket(self):
        if not self.client: return
        try:
            if not self.client.bucket_exists(self.bucket_name):
                self.client.make_bucket(self.bucket_name)
                # 移除公开读策略，默认为私有
                # self.client.set_bucket_policy(self.bucket_name, json.dumps(policy))
        except S3Error as e:
            logger.error(f"MinIO bucket error: {e}")

    def get_presigned_url(self, object_name: str, expires=None):
        """生成预签名访问链接"""
        if not self.client: return None
        try:
            from datetime import timedelta
            if expires is None:
                expires = timedelta(hours=1)
                
            return self.client.get_presigned_url(
                "GET",
                self.bucket_name,
                object_name,
                expires=expires
            )
        except Exception as e:
            logger.error(f"Generate presigned url failed: {e}")
            return None

    def upload_message_file(self, file_data: bytes, filename: str, content_type: str, user_id: int) -> str:
        """
        上传消息附件
        路径格式: messages/{user_id}/{year}/{month}/{uuid}.ext
        返回: object_name (用于存储和生成签名URL)
        """
        if not self.client:
            raise Exception("Storage service unavailable")

        # 生成结构化路径
        ext = filename.split('.')[-1] if '.' in filename else 'bin'
        now = datetime.now()
        object_name = f"messages/{user_id}/{now.year}/{now.month:02d}/{uuid.uuid4()}.{ext}"
        
        try:
            self.client.put_object(
                bucket_name=self.bucket_name,
                object_name=object_name,
                data=io.BytesIO(file_data),
                length=len(file_data),
                content_type=content_type
            )
            
            return object_name
            
        except S3Error as e:
            logger.error(f"Upload failed: {e}")
            raise Exception("File upload failed")

# 单例实例
minio_storage = MessageStorage()