Spaces:
Sleeping
Sleeping
File size: 4,657 Bytes
86e7db6 000d6aa 86e7db6 000d6aa 86e7db6 |
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 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 |
"""
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 |