File size: 3,413 Bytes
75e22f2 | 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 105 106 107 108 109 | import aiofiles
import aiohttp
from pathlib import Path
from uuid import uuid4
from typing import Optional
from fastapi import UploadFile
from Backend.core.config import settings
from Backend.core.logging import get_logger
logger = get_logger(__name__)
def generate_filename(original_filename: str) -> str:
ext = Path(original_filename).suffix.lower()
if not ext:
ext = ".jpg"
return f"{uuid4().hex}{ext}"
def get_supabase_public_url(file_path: str) -> str:
return f"{settings.supabase_url}/storage/v1/object/public/{settings.supabase_bucket}/{file_path}"
async def upload_to_supabase(file_data: bytes, remote_path: str, content_type: str = "image/jpeg") -> str:
url = f"{settings.supabase_url}/storage/v1/object/{settings.supabase_bucket}/{remote_path}"
headers = {
"Authorization": f"Bearer {settings.supabase_key}",
"Content-Type": content_type,
"x-upsert": "true",
}
async with aiohttp.ClientSession() as session:
async with session.post(url, data=file_data, headers=headers) as response:
if response.status not in (200, 201):
error_text = await response.text()
logger.error(f"Supabase upload failed: {response.status} - {error_text}")
raise Exception(f"Failed to upload to Supabase: {error_text}")
logger.info(f"Uploaded to Supabase: {remote_path}")
return get_supabase_public_url(remote_path)
async def save_upload(file: UploadFile, subfolder: str = "") -> str:
filename = generate_filename(file.filename or "image.jpg")
if subfolder:
remote_path = f"{subfolder}/{filename}"
else:
remote_path = filename
content = await file.read()
await file.seek(0)
content_type = file.content_type or "image/jpeg"
public_url = await upload_to_supabase(content, remote_path, content_type)
return remote_path
async def save_bytes(data: bytes, filename: str, subfolder: str = "", content_type: str = "image/jpeg") -> str:
if subfolder:
remote_path = f"{subfolder}/{filename}"
else:
remote_path = filename
public_url = await upload_to_supabase(data, remote_path, content_type)
return remote_path
async def save_local_temp(data: bytes, filename: str) -> str:
temp_dir = settings.local_temp_dir
temp_dir.mkdir(parents=True, exist_ok=True)
file_path = temp_dir / filename
async with aiofiles.open(file_path, "wb") as f:
await f.write(data)
return str(file_path)
async def download_from_supabase(remote_path: str) -> bytes:
url = get_supabase_public_url(remote_path)
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
if response.status != 200:
raise Exception(f"Failed to download from Supabase: {response.status}")
return await response.read()
def get_upload_url(file_path: str) -> str:
if file_path.startswith("http"):
return file_path
return get_supabase_public_url(file_path)
def validate_file_extension(filename: str) -> bool:
ext = Path(filename).suffix.lower().lstrip(".")
return ext in settings.allowed_extensions
def validate_file_size(size: int) -> bool:
max_bytes = settings.max_upload_size_mb * 1024 * 1024
return size <= max_bytes
|