Spaces:
Sleeping
Sleeping
File size: 3,155 Bytes
0225ffc | 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 | import asyncio
import os
import shutil
import uuid
from fastapi import UploadFile
from config.app_config import app_config
from src.constants import PUBLIC_TEMP_DIR
import cloudinary
import cloudinary.uploader
# Configure Cloudinary
cloudinary.config(
cloud_name=app_config.cloudinary_cloud_name,
api_key=app_config.cloudinary_api_key,
api_secret=app_config.cloudinary_api_secret,
secure=True
)
def extract_public_id(url: str) -> str | None:
"""
Extracts the public_id from a Cloudinary URL.
E.g., https://res.cloudinary.com/cloud_name/image/upload/v1234567/avatars/abc123xyz.png -> avatars/abc123xyz
"""
if not url or "res.cloudinary.com" not in url:
return None
try:
parts = url.split("/upload/")
if len(parts) < 2:
return None
# Remove version prefix (e.g. v1619000000/) if present
subparts = parts[1].split("/", 1)
if len(subparts) < 2:
return None
path_with_ext = subparts[1]
public_id = os.path.splitext(path_with_ext)[0]
return public_id
except Exception:
return None
async def upload_avatar_to_cloudinary(file: UploadFile) -> str:
"""
Saves an uploaded file locally to the public temp directory,
uploads it to Cloudinary, deletes the local file, and returns the Cloudinary URL.
"""
# Ensure public/temp directory exists
os.makedirs(PUBLIC_TEMP_DIR, exist_ok=True)
# Create a unique file name to avoid collisions
file_extension = os.path.splitext(file.filename)[1] if file.filename else ".jpg"
unique_filename = f"{uuid.uuid4()}{file_extension}"
temp_file_path = os.path.join(PUBLIC_TEMP_DIR, unique_filename)
try:
# Save file locally in public/temp
with open(temp_file_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
# Upload to Cloudinary using threadpool to keep it async
upload_result = await asyncio.to_thread(
cloudinary.uploader.upload,
temp_file_path,
folder="avatars"
)
secure_url = upload_result.get("secure_url")
if not secure_url:
raise Exception("Cloudinary upload failed: secure_url was not returned.")
return secure_url
finally:
# Delete local file after upload
if os.path.exists(temp_file_path):
os.remove(temp_file_path)
async def delete_avatar_from_cloudinary(url: str) -> bool:
"""
Deletes an asset from Cloudinary using its secure URL.
"""
public_id = extract_public_id(url)
if not public_id:
print(f"[DEBUG] Could not extract public_id from url: {url}")
return False
try:
result = await asyncio.to_thread(
cloudinary.uploader.destroy,
public_id,
invalidate=True
)
print(f"[DEBUG] Cloudinary destroy result for {public_id}: {result}")
return result.get("result") == "ok"
except Exception as e:
print(f"[ERROR] Failed to delete image from Cloudinary: {e}")
return False
|