import boto3 from typing import Optional from botocore.config import Config from botocore.exceptions import ClientError from app.config import settings from loguru import logger # Initialize boto3 S3 client for Supabase Storage (S3-compatible) s3_client = boto3.client( "s3", endpoint_url=settings.CLOUDFLARE_R2_ENDPOINT_URL, aws_access_key_id=settings.CLOUDFLARE_R2_ACCESS_KEY, aws_secret_access_key=settings.CLOUDFLARE_R2_SECRET_KEY, config=Config(signature_version="s3v4"), region_name=settings.CLOUDFLARE_R2_REGION_NAME ) def upload_file(file_content: bytes, object_key: str, content_type: str = "application/octet-stream") -> bool: try: s3_client.put_object( Bucket=settings.CLOUDFLARE_R2_BUCKET, Key=object_key, Body=file_content, ContentType=content_type ) logger.info(f"Successfully uploaded {object_key} to {settings.CLOUDFLARE_R2_BUCKET}") return True except ClientError as e: logger.error(f"Failed to upload {object_key} to S3: {e}") return False def get_file(object_key: str) -> Optional[bytes]: try: response = s3_client.get_object( Bucket=settings.CLOUDFLARE_R2_BUCKET, Key=object_key ) return response["Body"].read() except ClientError as e: logger.error(f"Failed to download {object_key} from S3: {e}") return None def generate_download_url(object_key: str, expires_in: int = 3600) -> str: try: url = s3_client.generate_presigned_url( "get_object", Params={ "Bucket": settings.CLOUDFLARE_R2_BUCKET, "Key": object_key }, ExpiresIn=expires_in ) return url except ClientError as e: logger.error(f"Failed to generate presigned URL for {object_key}: {e}") # Fallback to direct construction if presigned fails return f"{settings.CLOUDFLARE_R2_ENDPOINT_URL}/{settings.CLOUDFLARE_R2_BUCKET}/{object_key}"