DetectifAI_db/setup_minio.py

"""
S3-compatible Storage Setup and Test Script for DetectifAI (Backblaze B2)
"""
from minio import Minio
from dotenv import load_dotenv
import os
import logging

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Load environment variables
load_dotenv()

def setup_minio():
    """Setup S3-compatible storage (Backblaze B2)"""
    try:
        endpoint = os.getenv('MINIO_ENDPOINT', 's3.eu-central-003.backblazeb2.com')
        access_key = os.getenv('MINIO_ACCESS_KEY', '00367479ffb7e4e0000000001')
        secret_key = os.getenv('MINIO_SECRET_KEY', 'K003opTvf92ijRj5dM7H1dgrlwcGTdA')
        secure = os.getenv('MINIO_SECURE', 'true').lower() == 'true'
        region = os.getenv('MINIO_REGION', 'eu-central-003')

        # S3 client setup
        client = Minio(
            endpoint,
            access_key=access_key,
            secret_key=secret_key,
            secure=secure,
            region=region or None
        )

        # Define required buckets
        buckets = [
            "detectifai-videos",       # Original and compressed videos
            "detectifai-keyframes",    # Extracted keyframes
            "detectifai-reports"       # Generated reports (HTML/PDF)
        ]

        # Verify buckets exist (don't create — buckets managed in B2 dashboard)
        for bucket in buckets:
            found = client.bucket_exists(bucket)
            if found:
                logger.info(f"✅ Bucket exists: {bucket}")
            else:
                logger.warning(f"⚠️ Bucket NOT found: {bucket} — create it in Backblaze B2 dashboard")

        # Test upload to each bucket
        test_data = b"DetectifAI Test Data"
        for bucket in buckets:
            try:
                test_object = f"test_{bucket}.txt"
                client.put_object(
                    bucket,
                    test_object,
                    bytes(test_data),
                    len(test_data)
                )
                logger.info(f"✅ Test upload successful to {bucket}")

                # Clean up test file
                client.remove_object(bucket, test_object)

            except Exception as bucket_error:
                logger.error(f"❌ Failed to upload test file to {bucket}: {str(bucket_error)}")

        # List objects in each bucket
        logger.info("\nCurrent bucket contents:")
        for bucket in buckets:
            logger.info(f"\nBucket: {bucket}")
            try:
                objects = client.list_objects(bucket, recursive=True)
                for obj in objects:
                    logger.info(f"- {obj.object_name} (size: {obj.size} bytes)")
            except Exception as list_error:
                logger.error(f"❌ Failed to list objects in {bucket}: {str(list_error)}")

        return True, "MinIO setup completed successfully"

    except Exception as e:
        error_message = f"MinIO setup failed: {str(e)}"
        logger.error(f"❌ {error_message}")
        return False, error_message

if __name__ == "__main__":
    success, message = setup_minio()
    if success:
        logger.info("✅ MinIO setup completed successfully!")
    else:
        logger.error(f"❌ MinIO setup failed: {message}")