minio_client.py

"""MinIO client module for S3-compatible object storage (MP3 files)."""

import os
import io
from typing import Optional, BinaryIO, Dict, Any
from datetime import timedelta
from minio import Minio
from minio.error import S3Error

# MinIO configuration from environment variables
MINIO_ENDPOINT = os.getenv("MINIO_ENDPOINT", "localhost:9000")
MINIO_ACCESS_KEY = os.getenv("MINIO_ACCESS_KEY", "minioadmin")
MINIO_SECRET_KEY = os.getenv("MINIO_SECRET_KEY", "minioadmin")
MINIO_BUCKET = os.getenv("MINIO_BUCKET", "music-memories")
MINIO_SECURE = os.getenv("MINIO_SECURE", "false").lower() == "true"

# Global MinIO client instance
_minio_client: Optional[Minio] = None


def get_minio_client() -> Optional[Minio]:
    """Get or create the MinIO client instance."""
    global _minio_client
    if _minio_client is None:
        try:
            _minio_client = Minio(
                MINIO_ENDPOINT,
                access_key=MINIO_ACCESS_KEY,
                secret_key=MINIO_SECRET_KEY,
                secure=MINIO_SECURE,
            )
        except Exception as e:
            print(f"⚠ MinIO client creation failed: {e}")
            return None
    return _minio_client


def init_minio() -> bool:
    """Initialize MinIO bucket."""
    client = get_minio_client()
    if client is None:
        print("⚠ MinIO not available")
        return False
    
    try:
        # Create bucket if it doesn't exist
        if not client.bucket_exists(MINIO_BUCKET):
            client.make_bucket(MINIO_BUCKET)
            print(f"✓ Created MinIO bucket: {MINIO_BUCKET}")
        else:
            print(f"✓ MinIO bucket exists: {MINIO_BUCKET}")
        return True
    except S3Error as e:
        print(f"⚠ MinIO bucket creation failed: {e}")
        return False


def upload_mp3_file(
    file_data: BinaryIO,
    file_size: int,
    song_id: int,
    filename: str = None
) -> Optional[Dict[str, Any]]:
    """Upload an MP3 file to MinIO."""
    client = get_minio_client()
    if client is None:
        return None
    
    # Generate object name
    object_name = f"songs/{song_id}/{filename or f'song_{song_id}.mp3'}"
    
    try:
        # Upload the file
        client.put_object(
            MINIO_BUCKET,
            object_name,
            file_data,
            length=file_size,
            content_type="audio/mpeg",
        )
        
        # Generate presigned URL for streaming (expires in 7 days)
        presigned_url = client.presigned_get_object(
            MINIO_BUCKET,
            object_name,
            expires=timedelta(days=7),
        )
        
        return {
            "object_name": object_name,
            "bucket": MINIO_BUCKET,
            "size": file_size,
            "presigned_url": presigned_url,
            "content_type": "audio/mpeg",
        }
    except S3Error as e:
        print(f"⚠ MinIO upload failed: {e}")
        return None


def download_mp3_file(song_id: int, filename: str = None, object_name: str = None) -> Optional[bytes]:
    """Download an MP3 file from MinIO."""
    client = get_minio_client()
    if client is None:
        return None

    # Use provided object_name, or construct it from song_id and filename
    if object_name is None:
        object_name = f"songs/{song_id}/{filename or f'song_{song_id}.mp3'}"

    try:
        response = client.get_object(MINIO_BUCKET, object_name)
        return response.read()
    except S3Error as e:
        print(f"⚠ MinIO download failed: {e}")
        return None


def stream_mp3_file(song_id: int, filename: str = None) -> Optional[Any]:
    """Get a streaming response for an MP3 file."""
    client = get_minio_client()
    if client is None:
        return None
    
    object_name = f"songs/{song_id}/{filename or f'song_{song_id}.mp3'}"
    
    try:
        # Generate presigned URL for direct streaming
        presigned_url = client.presigned_get_object(
            MINIO_BUCKET,
            object_name,
            expires=timedelta(hours=24),
        )
        return {"presigned_url": presigned_url, "object_name": object_name}
    except S3Error as e:
        print(f"⚠ MinIO stream URL generation failed: {e}")
        return None


def delete_mp3_file(song_id: int, filename: str = None) -> bool:
    """Delete an MP3 file from MinIO."""
    client = get_minio_client()
    if client is None:
        return False
    
    object_name = f"songs/{song_id}/{filename or f'song_{song_id}.mp3'}"
    
    try:
        client.remove_object(MINIO_BUCKET, object_name)
        return True
    except S3Error as e:
        print(f"⚠ MinIO delete failed: {e}")
        return False


def get_file_metadata(song_id: int, filename: str = None) -> Optional[Dict[str, Any]]:
    """Get metadata for an MP3 file."""
    client = get_minio_client()
    if client is None:
        return None
    
    object_name = f"songs/{song_id}/{filename or f'song_{song_id}.mp3'}"
    
    try:
        stat = client.stat_object(MINIO_BUCKET, object_name)
        return {
            "size": stat.size,
            "content_type": stat.content_type,
            "last_modified": stat.last_modified.isoformat() if stat.last_modified else None,
            "etag": stat.etag,
        }
    except S3Error as e:
        print(f"⚠ MinIO metadata fetch failed: {e}")
        return None


def list_all_mp3_files() -> list:
    """List all MP3 files in the bucket."""
    client = get_minio_client()
    if client is None:
        return []
    
    try:
        objects = client.list_objects(MINIO_BUCKET, prefix="songs/", recursive=True)
        return [
            {
                "object_name": obj.object_name,
                "size": obj.size,
                "last_modified": obj.last_modified.isoformat() if obj.last_modified else None,
            }
            for obj in objects
        ]
    except S3Error as e:
        print(f"⚠ MinIO list failed: {e}")
        return []


def check_minio_health() -> Dict[str, Any]:
    """Check MinIO health status."""
    client = get_minio_client()
    if client is None:
        return {"status": "disconnected", "error": "Client not initialized"}
    
    try:
        # Check if bucket exists (lightweight health check)
        exists = client.bucket_exists(MINIO_BUCKET)
        return {
            "status": "connected" if exists else "bucket_missing",
            "bucket": MINIO_BUCKET,
            "endpoint": MINIO_ENDPOINT,
        }
    except S3Error as e:
        return {"status": "error", "error": str(e)}