Spaces:
Sleeping
Sleeping
| import os | |
| import cloudinary | |
| import cloudinary.uploader | |
| from pathlib import Path | |
| import sys | |
| sys.path.append(str(Path(__file__).resolve().parent.parent.parent)) | |
| from backend.core.logger import logger | |
| from backend.core.config import config | |
| # Lazy initialize Cloudinary config | |
| _is_configured = False | |
| def _configure_cloudinary(): | |
| global _is_configured | |
| if not _is_configured: | |
| if not config.CLOUDINARY_CLOUD_NAME: | |
| logger.warning("Cloudinary config missing, skipping Cloudinary uploads.") | |
| return False | |
| cloudinary.config( | |
| cloud_name=config.CLOUDINARY_CLOUD_NAME, | |
| api_key=config.CLOUDINARY_API_KEY, | |
| api_secret=config.CLOUDINARY_API_SECRET, | |
| secure=True | |
| ) | |
| _is_configured = True | |
| return True | |
| def upload_file(file_path: str, folder_path: str, resource_type: str = "auto") -> str: | |
| """ | |
| Uploads a local file to Cloudinary. | |
| Returns the secure URL on success, or an empty string on failure/if missing. | |
| """ | |
| if not _configure_cloudinary(): | |
| return "" | |
| try: | |
| if not os.path.exists(file_path): | |
| logger.error(f"File not found for Cloudinary upload: {file_path}") | |
| return "" | |
| file_name = Path(file_path).stem | |
| folder_path = folder_path.strip("/") | |
| logger.info(f"Uploading to Cloudinary: {folder_path}/{file_name}") | |
| response = cloudinary.uploader.upload( | |
| file_path, | |
| folder=folder_path, | |
| public_id=file_name, | |
| resource_type=resource_type, | |
| use_filename=True, | |
| unique_filename=False, | |
| overwrite=True | |
| ) | |
| url = response.get('secure_url', "") | |
| if url: | |
| logger.success(f"Upload successful: {url}") | |
| return url | |
| except Exception as e: | |
| logger.error(f"Cloudinary upload failed: {str(e)}") | |
| return "" | |