File size: 2,830 Bytes
2d331ea 6b45746 3dfc77e 2d331ea f20025d 2d331ea faf693c 2d331ea 9fd2fea 2d331ea |
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 |
"""
ImageKit.io utilities for temporary file hosting.
Used as a fallback when GCS is unavailable.
"""
from typing import Optional, Dict, Any
import os
from imagekitio import ImageKit
from src.config import get_config_value
from src.logger_config import logger
def get_imagekit_client() -> Optional[ImageKit]:
"""Initialize and return ImageKit client using config values."""
# Debug (you will see they exist now)
print("IK public:", bool(os.environ.get("IMAGEKIT_PUBLIC_KEY")))
print("IK private:", bool(os.environ.get("IMAGEKIT_PRIVATE_KEY")))
print("IK url:", bool(os.environ.get("IMAGEKIT_URL_ENDPOINT")))
private_key = get_config_value("imagekit_private_key")
if not private_key:
logger.warning("⚠️ ImageKit private key missing. Cannot initialize client.")
return None
return ImageKit(
private_key=private_key
)
def upload_file_to_imagekit(local_path: str, filename: Optional[str] = None) -> Optional[Dict[str, Any]]:
"""
Upload a local file to ImageKit.
Returns a dict with 'url' and 'file_id' or None if failed.
"""
client = get_imagekit_client()
if not client:
return None
if not os.path.exists(local_path):
logger.error(f"❌ Local file not found for ImageKit upload: {local_path}")
return None
if not filename:
filename = os.path.basename(local_path)
try:
from imagekitio import file_from_path
logger.info(f"📤 Uploading {filename} to ImageKit...")
# Use v5 SDK upload method
upload_result = client.files.upload(
file=file_from_path(local_path),
file_name=filename,
folder="/temp_social_uploads",
use_unique_file_name=True
)
if upload_result and upload_result.url:
logger.info(f"✅ ImageKit upload successful: {upload_result.url}")
return {
"url": upload_result.url,
"file_id": upload_result.file_id,
"storage_type": "imagekit"
}
else:
logger.error(f"❌ ImageKit upload failed: No URL in response")
return None
except Exception as e:
logger.error(f"❌ ImageKit upload error: {e}")
return None
def delete_file_from_imagekit(file_id: str) -> bool:
"""Delete a file from ImageKit by file_id."""
client = get_imagekit_client()
if not client:
return False
try:
logger.info(f"🗑️ Deleting file from ImageKit: {file_id}")
# Use v5 SDK delete method
client.files.delete(file_id)
return True
except Exception as e:
logger.warning(f"⚠️ Failed to delete from ImageKit ({file_id}): {e}")
return False
|