Spaces:
Running
Running
| import logging | |
| from urllib.parse import urlsplit, urlunsplit | |
| import cloudinary | |
| import cloudinary.uploader | |
| from app.core.config import settings | |
| logger = logging.getLogger(__name__) | |
| AUTO_IMAGE_TRANSFORMATION = "f_auto,q_auto" | |
| AUTO_IMAGE_TRANSFORMATION_PARAMS = {"fetch_format": "auto", "quality": "auto"} | |
| CLOUDINARY_IMAGE_UPLOAD_SEGMENT = "/image/upload/" | |
| def init_cloudinary(): | |
| """Initialize Cloudinary with credentials from settings.""" | |
| if settings.CLOUDINARY_CLOUD_NAME: | |
| cloudinary.config( | |
| cloud_name=settings.CLOUDINARY_CLOUD_NAME, | |
| api_key=settings.CLOUDINARY_API_KEY, | |
| api_secret=settings.CLOUDINARY_API_SECRET, | |
| secure=True, | |
| ) | |
| def optimize_delivery_image_url(url: str | None) -> str | None: | |
| """Add automatic format and quality delivery transforms to Cloudinary image URLs.""" | |
| if not url: | |
| return url | |
| parsed_url = urlsplit(url) | |
| if "cloudinary.com" not in parsed_url.netloc: | |
| return url | |
| path = parsed_url.path | |
| if CLOUDINARY_IMAGE_UPLOAD_SEGMENT not in path: | |
| return url | |
| if "f_auto" in path and "q_auto" in path: | |
| return url | |
| optimized_path = path.replace( | |
| CLOUDINARY_IMAGE_UPLOAD_SEGMENT, | |
| f"{CLOUDINARY_IMAGE_UPLOAD_SEGMENT}{AUTO_IMAGE_TRANSFORMATION}/", | |
| 1, | |
| ) | |
| return urlunsplit(parsed_url._replace(path=optimized_path)) | |
| def optimize_delivery_image_urls(image_urls: dict | None) -> dict | None: | |
| """Add automatic delivery transforms to a mapping of Cloudinary image URLs.""" | |
| if not image_urls: | |
| return image_urls | |
| return { | |
| key: optimize_delivery_image_url(value) if isinstance(value, str) else value | |
| for key, value in image_urls.items() | |
| } | |
| def upload_image(file_bytes: bytes, folder: str = "gate-prep/questions") -> dict: | |
| """ | |
| Upload image to Cloudinary. | |
| Returns dict with url and public_id. | |
| Falls back to local storage if Cloudinary not configured. | |
| """ | |
| if not settings.CLOUDINARY_CLOUD_NAME: | |
| return {"url": None, "public_id": None, "error": "Cloudinary not configured"} | |
| try: | |
| result = cloudinary.uploader.upload( | |
| file_bytes, | |
| folder=folder, | |
| resource_type="image", | |
| transformation=[ | |
| dict(AUTO_IMAGE_TRANSFORMATION_PARAMS), | |
| {"width": 1200, "crop": "limit"}, | |
| ], | |
| ) | |
| return { | |
| "url": optimize_delivery_image_url(result["secure_url"]), | |
| "public_id": result["public_id"], | |
| } | |
| except Exception as e: | |
| logger.error(f"Cloudinary upload error: {e}") | |
| return {"url": None, "public_id": None, "error": str(e)} | |
| def delete_image(public_id: str) -> bool: | |
| """Delete image from Cloudinary.""" | |
| if not settings.CLOUDINARY_CLOUD_NAME or not public_id: | |
| return False | |
| try: | |
| cloudinary.uploader.destroy(public_id) | |
| return True | |
| except Exception as e: | |
| logger.error(f"Cloudinary delete error: {e}") | |
| return False | |
| # Initialize on import | |
| init_cloudinary() | |