""" Cloudinary client for file uploads """ import os import logging import cloudinary import cloudinary.uploader from typing import Union logger = logging.getLogger(__name__) class CloudinaryClient: """Handles file uploads to Cloudinary""" def __init__(self): # Configure Cloudinary cloudinary.config( cloud_name=os.environ.get("CLOUDINARY_CLOUD_NAME"), api_key=os.environ.get("CLOUDINARY_API_KEY"), api_secret=os.environ.get("CLOUDINARY_API_SECRET") ) # Verify configuration if not all([ os.environ.get("CLOUDINARY_CLOUD_NAME"), os.environ.get("CLOUDINARY_API_KEY"), os.environ.get("CLOUDINARY_API_SECRET") ]): logger.warning("⚠️ Cloudinary credentials not fully configured") else: logger.info("✅ Cloudinary client initialized") def upload_image_from_bytes(self, image_bytes: bytes, public_id: str) -> str: """Upload image from bytes to Cloudinary""" try: logger.info(f"☁️ Uploading image to Cloudinary: {public_id}") result = cloudinary.uploader.upload( image_bytes, public_id=f"text-to-3d/{public_id}", resource_type="image", unique_filename=True, overwrite=True, quality="auto" ) url = result["secure_url"] logger.info(f"✅ Image uploaded: {url}") return url except Exception as e: logger.error(f"❌ Error uploading image to Cloudinary: {str(e)}") raise e def upload_image_from_path(self, file_path: str, public_id: str) -> str: """Upload image from file path to Cloudinary""" try: logger.info(f"☁️ Uploading image file to Cloudinary: {public_id}") result = cloudinary.uploader.upload( file_path, public_id=f"text-to-3d/{public_id}", resource_type="image", unique_filename=True, overwrite=True, quality="auto" ) url = result["secure_url"] logger.info(f"✅ Image file uploaded: {url}") return url except Exception as e: logger.error(f"❌ Error uploading image file to Cloudinary: {str(e)}") raise e def upload_file(self, file_path: str, public_id: str) -> str: """Upload any file to Cloudinary""" try: logger.info(f"☁️ Uploading file to Cloudinary: {public_id}") # Check if file exists and is accessible if not os.path.exists(file_path): raise FileNotFoundError(f"File not found: {file_path}") file_size = os.path.getsize(file_path) logger.info(f"📁 File info - Path: {file_path}, Size: {file_size} bytes") # Verify file is readable with open(file_path, 'rb') as f: f.read(1) # Try to read first byte result = cloudinary.uploader.upload( file_path, public_id=f"text-to-3d/{public_id}", resource_type="raw", # For non-image files unique_filename=True, overwrite=True ) url = result["secure_url"] logger.info(f"✅ File uploaded: {url}") return url except Exception as e: logger.error(f"❌ Error uploading file to Cloudinary: {str(e)}") logger.error(f"📁 Failed file path: {file_path}") raise e def delete_file(self, public_id: str, resource_type: str = "image") -> bool: """Delete file from Cloudinary""" try: logger.info(f"🗑️ Deleting file from Cloudinary: {public_id}") result = cloudinary.uploader.destroy( f"text-to-3d/{public_id}", resource_type=resource_type ) success = result.get("result") == "ok" if success: logger.info(f"✅ File deleted: {public_id}") else: logger.warning(f"⚠️ File deletion may have failed: {public_id}") return success except Exception as e: logger.error(f"❌ Error deleting file from Cloudinary: {str(e)}") return False