Spaces:
Sleeping
Sleeping
| """ | |
| Cloudinary Upload Utility | |
| ========================= | |
| Uploads files (audio, JSON) to Cloudinary, maintaining folder structure. | |
| Used by: | |
| - backend.text_to_speech.english_tts (upload MP3/WAV audio files) | |
| - backend.text_to_speech.hindi_tts (upload MP3 audio files) | |
| - backend.summarization.english_summary (upload summarized JSON) | |
| - backend.web_scraping.news_scrape (upload scraped JSON) | |
| Configuration (all required in .env): | |
| CLOUDINARY_CLOUD_NAME=your_cloud_name | |
| CLOUDINARY_API_KEY=your_api_key | |
| CLOUDINARY_API_SECRET=your_api_secret | |
| Usage: | |
| from backend.utils.cloudinary_utils import upload_to_cloudinary | |
| result = upload_to_cloudinary( | |
| file_path="/path/to/audio.mp3", | |
| folder_path="audios/english/categories/sports/24_mar_10_37_pm", | |
| resource_type="video" # Cloudinary uses "video" for audio files | |
| ) | |
| if result: | |
| print(result['secure_url']) | |
| """ | |
| import os | |
| import cloudinary | |
| import cloudinary.uploader | |
| from dotenv import load_dotenv | |
| from pathlib import Path | |
| load_dotenv() | |
| cloudinary.config( | |
| cloud_name=os.getenv('CLOUDINARY_CLOUD_NAME'), | |
| api_key=os.getenv('CLOUDINARY_API_KEY'), | |
| api_secret=os.getenv('CLOUDINARY_API_SECRET'), | |
| secure=True | |
| ) | |
| def upload_to_cloudinary(file_path: str, folder_path: str, resource_type: str = "auto"): | |
| """Upload a file to Cloudinary, preserving the specified folder layout. | |
| Args: | |
| file_path: Absolute path to the local file. | |
| folder_path: Cloudinary folder path (e.g. "audios/english/categories/sports"). | |
| resource_type: One of "auto", "image", "video", "raw". | |
| Use "video" for audio files, "raw" for JSON. | |
| Returns: | |
| dict with Cloudinary response (contains 'secure_url'), or None on failure. | |
| """ | |
| try: | |
| if not os.path.exists(file_path): | |
| print(f"File not found: {file_path}") | |
| return None | |
| file_name = Path(file_path).stem | |
| folder_path = folder_path.strip("/") | |
| print(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 | |
| ) | |
| print(f"Upload successful: {response.get('secure_url')}") | |
| return response | |
| except Exception as e: | |
| print(f"Cloudinary upload failed: {str(e)}") | |
| return None | |