Spaces:
Sleeping
Sleeping
File size: 3,328 Bytes
fd50325 2278049 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | """
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}") |