breakpilot-pwa/backend/transcription_worker/storage.py

"""
BreakPilot MinIO Storage Helper

Provides file upload/download operations for MinIO object storage.
"""

import os
import io
import structlog
from typing import Optional, BinaryIO

log = structlog.get_logger(__name__)


class MinIOStorage:
    """
    MinIO storage client for recordings and transcriptions.

    Provides methods to upload, download, and manage files
    in MinIO object storage (S3-compatible).
    """

    def __init__(
        self,
        endpoint: str = "minio:9000",
        access_key: str = "breakpilot",
        secret_key: str = "breakpilot123",
        bucket: str = "breakpilot-recordings",
        secure: bool = False
    ):
        """
        Initialize MinIO client.

        Args:
            endpoint: MinIO server endpoint (host:port)
            access_key: Access key (username)
            secret_key: Secret key (password)
            bucket: Default bucket name
            secure: Use HTTPS
        """
        self.endpoint = endpoint
        self.access_key = access_key
        self.secret_key = secret_key
        self.bucket = bucket
        self.secure = secure
        self._client = None

    def _get_client(self):
        """Lazy initialize MinIO client."""
        if self._client is not None:
            return self._client

        try:
            from minio import Minio

            self._client = Minio(
                self.endpoint,
                access_key=self.access_key,
                secret_key=self.secret_key,
                secure=self.secure
            )

            log.info(
                "minio_client_initialized",
                endpoint=self.endpoint,
                bucket=self.bucket
            )

            return self._client

        except ImportError:
            log.error("minio_not_installed")
            raise ImportError(
                "minio is not installed. "
                "Install with: pip install minio"
            )

    def ensure_bucket(self) -> bool:
        """
        Ensure the bucket exists, create if needed.

        Returns:
            True if bucket exists or was created
        """
        client = self._get_client()

        if not client.bucket_exists(self.bucket):
            client.make_bucket(self.bucket)
            log.info("bucket_created", bucket=self.bucket)
            return True

        return True

    def download_file(
        self,
        object_name: str,
        local_path: str,
        bucket: Optional[str] = None
    ) -> str:
        """
        Download a file from MinIO.

        Args:
            object_name: Path in MinIO bucket
            local_path: Local destination path
            bucket: Optional bucket override

        Returns:
            Local file path
        """
        client = self._get_client()
        bucket = bucket or self.bucket

        log.info(
            "downloading_file",
            bucket=bucket,
            object_name=object_name,
            local_path=local_path
        )

        # Ensure directory exists
        os.makedirs(os.path.dirname(local_path), exist_ok=True)

        # Download
        client.fget_object(bucket, object_name, local_path)

        log.info(
            "file_downloaded",
            object_name=object_name,
            local_path=local_path,
            size=os.path.getsize(local_path)
        )

        return local_path

    def upload_file(
        self,
        local_path: str,
        object_name: str,
        content_type: Optional[str] = None,
        bucket: Optional[str] = None
    ) -> str:
        """
        Upload a file to MinIO.

        Args:
            local_path: Local file path
            object_name: Destination path in MinIO
            content_type: MIME type
            bucket: Optional bucket override

        Returns:
            Object name in MinIO
        """
        client = self._get_client()
        bucket = bucket or self.bucket

        # Ensure bucket exists
        self.ensure_bucket()

        log.info(
            "uploading_file",
            local_path=local_path,
            bucket=bucket,
            object_name=object_name
        )

        # Upload
        result = client.fput_object(
            bucket,
            object_name,
            local_path,
            content_type=content_type
        )

        log.info(
            "file_uploaded",
            object_name=object_name,
            etag=result.etag
        )

        return object_name

    def upload_content(
        self,
        content: str,
        object_name: str,
        content_type: str = "text/plain",
        bucket: Optional[str] = None
    ) -> str:
        """
        Upload string content directly to MinIO.

        Args:
            content: String content to upload
            object_name: Destination path in MinIO
            content_type: MIME type
            bucket: Optional bucket override

        Returns:
            Object name in MinIO
        """
        client = self._get_client()
        bucket = bucket or self.bucket

        # Ensure bucket exists
        self.ensure_bucket()

        # Convert to bytes
        data = content.encode("utf-8")
        data_stream = io.BytesIO(data)

        log.info(
            "uploading_content",
            bucket=bucket,
            object_name=object_name,
            size=len(data)
        )

        # Upload
        result = client.put_object(
            bucket,
            object_name,
            data_stream,
            length=len(data),
            content_type=content_type
        )

        log.info(
            "content_uploaded",
            object_name=object_name,
            etag=result.etag
        )

        return object_name

    def get_content(
        self,
        object_name: str,
        bucket: Optional[str] = None
    ) -> str:
        """
        Get string content from MinIO.

        Args:
            object_name: Path in MinIO bucket
            bucket: Optional bucket override

        Returns:
            File content as string
        """
        client = self._get_client()
        bucket = bucket or self.bucket

        response = client.get_object(bucket, object_name)
        content = response.read().decode("utf-8")
        response.close()
        response.release_conn()

        return content

    def delete_file(
        self,
        object_name: str,
        bucket: Optional[str] = None
    ) -> bool:
        """
        Delete a file from MinIO.

        Args:
            object_name: Path in MinIO bucket
            bucket: Optional bucket override

        Returns:
            True if deleted
        """
        client = self._get_client()
        bucket = bucket or self.bucket

        client.remove_object(bucket, object_name)

        log.info("file_deleted", object_name=object_name)
        return True

    def file_exists(
        self,
        object_name: str,
        bucket: Optional[str] = None
    ) -> bool:
        """
        Check if a file exists in MinIO.

        Args:
            object_name: Path in MinIO bucket
            bucket: Optional bucket override

        Returns:
            True if file exists
        """
        client = self._get_client()
        bucket = bucket or self.bucket

        try:
            client.stat_object(bucket, object_name)
            return True
        except Exception:
            return False

    def get_presigned_url(
        self,
        object_name: str,
        expires_hours: int = 24,
        bucket: Optional[str] = None
    ) -> str:
        """
        Get a presigned URL for temporary file access.

        Args:
            object_name: Path in MinIO bucket
            expires_hours: URL validity in hours
            bucket: Optional bucket override

        Returns:
            Presigned URL
        """
        from datetime import timedelta

        client = self._get_client()
        bucket = bucket or self.bucket

        url = client.presigned_get_object(
            bucket,
            object_name,
            expires=timedelta(hours=expires_hours)
        )

        return url

    def list_files(
        self,
        prefix: str = "",
        bucket: Optional[str] = None
    ) -> list:
        """
        List files with a given prefix.

        Args:
            prefix: Path prefix to filter
            bucket: Optional bucket override

        Returns:
            List of object names
        """
        client = self._get_client()
        bucket = bucket or self.bucket

        objects = client.list_objects(bucket, prefix=prefix, recursive=True)

        return [obj.object_name for obj in objects]