"""File upload service using Supabase Storage.""" from __future__ import annotations import uuid from io import BytesIO from typing import Optional from PIL import Image from app.core.config import get_settings from app.core.exceptions import BadRequestException from app.core.supabase import get_supabase_admin ALLOWED_CONTENT_TYPES = { "image/jpeg", "image/png", "image/webp", "image/gif", } MAX_DIMENSION = 2048 class UploadService: def __init__(self): self.db = get_supabase_admin() self.settings = get_settings() self.bucket = self.settings.SUPABASE_STORAGE_BUCKET def upload_image( self, file_bytes: bytes, content_type: str, folder: str = "images", max_size_mb: Optional[int] = None, ) -> str: max_bytes = (max_size_mb or self.settings.MAX_UPLOAD_SIZE_MB) * 1024 * 1024 if content_type not in ALLOWED_CONTENT_TYPES: raise BadRequestException( f"Invalid file type. Allowed: {', '.join(ALLOWED_CONTENT_TYPES)}" ) if len(file_bytes) > max_bytes: raise BadRequestException( f"File too large. Max: {max_size_mb or self.settings.MAX_UPLOAD_SIZE_MB}MB" ) # Validate and optionally resize try: img = Image.open(BytesIO(file_bytes)) img.verify() img = Image.open(BytesIO(file_bytes)) # Re-open after verify # Resize if too large if max(img.size) > MAX_DIMENSION: img.thumbnail((MAX_DIMENSION, MAX_DIMENSION), Image.LANCZOS) buffer = BytesIO() fmt = "JPEG" if content_type == "image/jpeg" else "PNG" img.save(buffer, format=fmt, quality=85) file_bytes = buffer.getvalue() except Exception: raise BadRequestException("Invalid image file") ext = content_type.split("/")[-1] if ext == "jpeg": ext = "jpg" filename = f"{folder}/{uuid.uuid4().hex}.{ext}" self.db.storage.from_(self.bucket).upload( path=filename, file=file_bytes, file_options={"content-type": content_type}, ) # Return public URL public_url = self.db.storage.from_(self.bucket).get_public_url(filename) return public_url def delete_image(self, file_url: str) -> bool: """Extract path from URL and delete from storage.""" try: bucket_prefix = f"/storage/v1/object/public/{self.bucket}/" if bucket_prefix in file_url: path = file_url.split(bucket_prefix)[-1] else: # Try extracting from full URL path = file_url.split(f"{self.bucket}/")[-1] self.db.storage.from_(self.bucket).remove([path]) return True except Exception: return False