mirror of
http://88.130.71.182:3000/BlitTech/deals24togo_be.git
synced 2026-06-12 23:33:21 +00:00
95 lines
2.9 KiB
Python
95 lines
2.9 KiB
Python
"""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
|