File size: 3,573 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 92 93 94 95 96 97 98 99 100 101 102 103 104 | """
Reset MinIO buckets and test storage paths for DetectifAI.
This script ensures that all required MinIO buckets and storage paths
are properly configured for video processing.
"""
from minio import Minio
from minio.error import S3Error
import os
from datetime import datetime
from dotenv import load_dotenv
import logging
# Load environment variables
load_dotenv()
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# MinIO configuration
MINIO_CONFIG = {
"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")
}
# Bucket configuration with descriptions
BUCKETS = {
"detectifai-videos": {
"description": "Main bucket for video storage",
"prefixes": {
"original": "Original uploaded videos",
"compressed": "Compressed video versions"
}
},
"detectifai-keyframes": {
"description": "Storage for extracted video frames",
"prefixes": {
"keyframes": "Extracted keyframes and annotated frames"
}
}
}
def reset_minio_storage():
"""Reset and verify MinIO storage configuration"""
client = Minio(**MINIO_CONFIG)
print("Checking MinIO connection and buckets...")
for bucket_name, config in BUCKETS.items():
try:
# Check if bucket exists
found = client.bucket_exists(bucket_name)
if not found:
print(f"Creating bucket: {bucket_name}")
client.make_bucket(bucket_name)
# Test each prefix path
for prefix in config["prefixes"]:
test_object = f"{prefix}/test.txt"
test_data = f"Test data for {bucket_name}/{prefix}"
print(f"\nTesting path: {bucket_name}/{test_object}")
# Upload test object
test_bytes = bytes(test_data, 'utf-8')
from io import BytesIO
test_stream = BytesIO(test_bytes)
client.put_object(
bucket_name,
test_object,
test_stream,
len(test_bytes)
)
# Verify upload
try:
client.stat_object(bucket_name, test_object)
print(f"✅ Test file uploaded successfully")
# Clean up test file
client.remove_object(bucket_name, test_object)
print(f"✅ Test file removed")
except:
print(f"❌ Could not verify test file")
print(f"\nListing objects in {bucket_name}:")
objects = client.list_objects(bucket_name, recursive=True)
for obj in objects:
print(f"- {obj.object_name} (size: {obj.size} bytes)")
except S3Error as e:
print(f"❌ Error with bucket {bucket_name}: {e}")
continue
if __name__ == "__main__":
reset_minio_storage() |