import uuid from datetime import datetime, timezone from pathlib import PurePosixPath import boto3 from botocore.config import Config as BotoConfig from app.config import settings def _get_client(): return boto3.client( "s3", endpoint_url=settings.minio.endpoint, aws_access_key_id=settings.minio.access_key, aws_secret_access_key=settings.minio.secret_key, config=BotoConfig(signature_version="s3v4"), ) def _generate_object_key(filename: str) -> str: now = datetime.now(timezone.utc) date_prefix = now.strftime("%Y/%m/%d") ext = PurePosixPath(filename).suffix or ".jpg" return f"{date_prefix}/{uuid.uuid4().hex}{ext}" def upload_image(file_bytes: bytes, original_filename: str) -> str: client = _get_client() object_key = _generate_object_key(original_filename) content_type = "image/jpeg" if original_filename.lower().endswith(".png"): content_type = "image/png" client.put_object( Bucket=settings.minio.bucket, Key=object_key, Body=file_bytes, ContentType=content_type, ) return object_key def generate_presigned_put_url(object_key: str) -> str: client = _get_client() url: str = client.generate_presigned_url( "put_object", Params={ "Bucket": settings.minio.bucket, "Key": object_key, }, ExpiresIn=300, ) # Replace internal endpoint with public endpoint if settings.minio.public_endpoint: from urllib.parse import urlparse, urlunparse internal = urlparse(settings.minio.endpoint) public = urlparse(settings.minio.public_endpoint) url = url.replace(f"{internal.scheme}://{internal.netloc}", f"{public.scheme}://{public.netloc}", 1) return url