Spaces:
Sleeping
Sleeping
File size: 1,982 Bytes
2cb327c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | 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 ""
|