from __future__ import annotations import asyncio import base64 import binascii import ipaddress import logging import re from typing import Optional, Tuple from urllib.parse import urlparse import cloudinary import cloudinary.uploader from fastapi import HTTPException, status from core.config import settings LOGGER = logging.getLogger(__name__) _DATA_URI_PATTERN = re.compile(r"^data:(image\/[a-zA-Z0-9.+-]+);base64,(.*)$", re.DOTALL) class CloudinaryService: """Upload helper that converts client image payloads into secure Cloudinary URLs.""" def __init__(self) -> None: self._enabled = settings.CLOUDINARY_UPLOAD_ENABLED and bool( settings.CLOUDINARY_CLOUD_NAME and settings.CLOUDINARY_API_KEY and settings.CLOUDINARY_API_SECRET ) if self._enabled: cloudinary.config( cloud_name=settings.CLOUDINARY_CLOUD_NAME, api_key=settings.CLOUDINARY_API_KEY, api_secret=settings.CLOUDINARY_API_SECRET, secure=True, ) @property def enabled(self) -> bool: return self._enabled @staticmethod def _is_url(value: str) -> bool: lowered = value.lower() return lowered.startswith("https://") or lowered.startswith("http://") @staticmethod def _is_local_file_reference(value: str) -> bool: lowered = value.lower() return ( lowered.startswith("file://") or lowered.startswith("content://") or lowered.startswith("ph://") or lowered.startswith("assets-library://") ) @staticmethod def _validate_external_url(value: str) -> None: parsed = urlparse(value) host = (parsed.hostname or "").strip().lower() if parsed.scheme not in {"http", "https"} or not host: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid image URL", ) if host in {"localhost", "127.0.0.1", "::1"}: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Local image URLs are not allowed", ) try: ip_value = ipaddress.ip_address(host) if ( ip_value.is_private or ip_value.is_loopback or ip_value.is_link_local or ip_value.is_multicast or ip_value.is_reserved or ip_value.is_unspecified ): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Private network image URLs are not allowed", ) except ValueError: # Hostname is not a direct IP literal; allow DNS hostnames. pass @staticmethod def _extract_data_uri(value: str) -> Tuple[Optional[str], str]: match = _DATA_URI_PATTERN.match(value.strip()) if not match: return None, value mime_type = match.group(1).lower() encoded_payload = match.group(2) return mime_type, encoded_payload @staticmethod def _normalize_mime(mime_type: Optional[str]) -> str: if not mime_type: return "image/jpeg" normalized = mime_type.strip().lower() if normalized == "image/jpg": return "image/jpeg" return normalized def _validate_and_decode(self, image_input: str) -> Tuple[bytes, str]: mime_type, encoded_payload = self._extract_data_uri(image_input) normalized_mime = self._normalize_mime(mime_type) if normalized_mime not in settings.cloudinary_allowed_mime_types: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Unsupported image format", ) # Mobile clients often send base64 with line-breaks or missing padding. compact_payload = re.sub(r"\s+", "", encoded_payload.strip()) if not compact_payload: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Image payload is empty", ) # Support URL-safe base64 variants before strict validation. compact_payload = compact_payload.replace("-", "+").replace("_", "/") remainder = len(compact_payload) % 4 if remainder: compact_payload = f"{compact_payload}{'=' * (4 - remainder)}" try: raw_bytes = base64.b64decode(compact_payload, validate=True) except (binascii.Error, ValueError) as exc: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid base64 image payload. Send a valid data URI, base64 string, or public image URL.", ) from exc if not raw_bytes: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Image payload is empty", ) if len(raw_bytes) > settings.CLOUDINARY_MAX_IMAGE_BYTES: raise HTTPException( status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail=f"Image exceeds max size of {settings.CLOUDINARY_MAX_IMAGE_BYTES} bytes", ) return raw_bytes, normalized_mime async def _upload_to_cloudinary(self, image_bytes: bytes, folder: str) -> str: def _do_upload() -> dict: return cloudinary.uploader.upload( image_bytes, folder=folder, resource_type="image", overwrite=False, unique_filename=True, use_filename=False, allowed_formats=list(settings.cloudinary_allowed_formats), ) try: response = await asyncio.to_thread(_do_upload) except Exception as exc: LOGGER.exception("Cloudinary upload failed") raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail="Image upload service unavailable", ) from exc secure_url = response.get("secure_url") if not isinstance(secure_url, str) or not secure_url: raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail="Image upload failed", ) return secure_url async def normalize_scan_image(self, image_input: str, user_id: str, plant_id: str) -> str: """Return a URL for vision inference, uploading base64/data-URI inputs when enabled.""" value = (image_input or "").strip() if not value: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Image is required") if self._is_local_file_reference(value): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Local file URIs are not supported. Send base64/data URI or a public image URL.", ) if self._is_url(value): self._validate_external_url(value) return value if not self.enabled: # Backward-compatible path when Cloudinary is not configured yet. return value image_bytes, _mime_type = self._validate_and_decode(value) folder = f"greenbuddy/scans/{user_id}/{plant_id}" return await self._upload_to_cloudinary(image_bytes=image_bytes, folder=folder) async def normalize_plant_image(self, image_input: str, user_id: str, plant_id: str) -> str: """Return a persisted plant image URL, uploading base64/data-URI inputs when enabled.""" value = (image_input or "").strip() if not value: return "" if self._is_local_file_reference(value): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Local file URIs are not supported. Send base64/data URI or a public image URL.", ) if self._is_url(value): self._validate_external_url(value) return value if not self.enabled: # Backward-compatible path when Cloudinary is not configured yet. return value image_bytes, _mime_type = self._validate_and_decode(value) folder = f"greenbuddy/plants/{user_id}/{plant_id}" return await self._upload_to_cloudinary(image_bytes=image_bytes, folder=folder) async def normalize_user_image(self, image_input: str, user_id: str) -> str: """Return a persisted user profile image URL, uploading base64/data-URI inputs when enabled.""" value = (image_input or "").strip() if not value: return "" if self._is_local_file_reference(value): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Local file URIs are not supported. Send base64/data URI or a public image URL.", ) if self._is_url(value): self._validate_external_url(value) return value if not self.enabled: # Backward-compatible path when Cloudinary is not configured yet. return value image_bytes, _mime_type = self._validate_and_decode(value) folder = f"greenbuddy/users/{user_id}/profile" return await self._upload_to_cloudinary(image_bytes=image_bytes, folder=folder)