feat: Integrate ImageKit.io as a fallback for media uploads in social media publishers with new utilities, configuration, and tests.

requirements.txt

@@ -30,3 +30,4 @@ fal-client
 xai-sdk
 json5
 xdk
+imagekitio==4.0.0

social_media_publishers/instagram/publisher.py

@@ -170,18 +170,24 @@ class InstagramPublisher(SocialPublisher):
                 
                 video_url = upload_result.get("public_url")
                 if not video_url:
-                    return {"error": "Failed to get public URL for video"}
-                    
-                    return {"error": "Failed to get public URL for video"}
+                    raise ValueError("Failed to get public URL from GCS upload result")
                     
                 logger.info(f"🔗 Public URL: {video_url}")
                 
-            except ImportError as e:
-                logger.error(f"❌ Import Error for GCS: {e}")
-                return {"error": f"gcs_utils not available: {e}"}
             except Exception as e:
-                logger.error(f"❌ Failed to upload video: {e}", exc_info=True)
-                return {"error": f"Failed to upload video: {e}"}
+                logger.warning(f"⚠️ GCS Upload failed, attempting ImageKit as last resort: {e}")
+                try:
+                    from src.imagekit_utils import upload_file_to_imagekit
+                    ik_result = upload_file_to_imagekit(content_source)
+                    if ik_result and ik_result.get("url"):
+                        video_url = ik_result["url"]
+                        metadata["_imagekit_file_id"] = ik_result["file_id"]
+                        logger.info(f"🔗 ImageKit Public URL: {video_url}")
+                    else:
+                        return {"error": f"Both GCS and ImageKit uploads failed: {e}"}
+                except Exception as ik_error:
+                    logger.error(f"❌ ImageKit upload also failed: {ik_error}")
+                    return {"error": f"Failed to upload video to any provider: {e} | {ik_error}"}
         else:
              return {"error": f"Invalid content source: {content_source}"}
         
@@ -198,7 +204,17 @@ class InstagramPublisher(SocialPublisher):
         caption = "\n\n".join(parts)
         share_to_feed = metadata.get('share_to_feed', True)
         
-        return self._upload_reel(video_url, caption, share_to_feed)
+        result = self._upload_reel(video_url, caption, share_to_feed)
+        
+        # Cleanup ImageKit if used
+        if metadata.get("_imagekit_file_id"):
+            try:
+                from src.imagekit_utils import delete_file_from_imagekit
+                delete_file_from_imagekit(metadata["_imagekit_file_id"])
+            except Exception as cleanup_error:
+                logger.warning(f"⚠️ Failed to cleanup ImageKit file: {cleanup_error}")
+                
+        return result
 
     def _upload_reel(self, video_url: str, caption: str, share_to_feed: bool = True) -> Dict[str, Any]:
         """

social_media_publishers/threads/publisher.py

@@ -167,7 +167,7 @@ class ThreadsPublisher(SocialPublisher):
                     
                     public_url = upload_result.get("public_url")
                     if not public_url:
-                        return {"error": "Failed to get public URL for media"}
+                        raise ValueError("Failed to get public URL from GCS upload result")
                         
                     if media_type == "VIDEO":
                         video_url = public_url
@@ -177,9 +177,22 @@ class ThreadsPublisher(SocialPublisher):
                     print(f"🔗 Public URL: {public_url}")
                     
                 except Exception as e:
-                    import traceback
-                    traceback.print_exc()
-                    return {"error": f"Failed to upload media: {e}"}
+                    print(f"⚠️ GCS Upload failed, attempting ImageKit as last resort: {e}")
+                    try:
+                        from src.imagekit_utils import upload_file_to_imagekit
+                        ik_result = upload_file_to_imagekit(content_source)
+                        if ik_result and ik_result.get("url"):
+                            if media_type == "VIDEO":
+                                video_url = ik_result["url"]
+                            else:
+                                image_url = ik_result["url"]
+                            metadata["_imagekit_file_id"] = ik_result["file_id"]
+                            print(f"🔗 ImageKit Public URL: {ik_result['url']}")
+                        else:
+                            return {"error": f"Both GCS and ImageKit uploads failed: {e}"}
+                    except Exception as ik_error:
+                        print(f"❌ ImageKit upload also failed: {ik_error}")
+                        return {"error": f"Failed to upload media to any provider: {e} | {ik_error}"}
         
         # Step 1: Create Container
         # Endpoint: POST /<user_id>/threads
@@ -241,6 +254,13 @@ class ThreadsPublisher(SocialPublisher):
         if 'id' in pub_result:
             media_id = pub_result['id']
             print(f"✅ Published! Thread ID: {media_id}")
+            # Cleanup ImageKit if used
+            if metadata.get("_imagekit_file_id"):
+                try:
+                    from src.imagekit_utils import delete_file_from_imagekit
+                    delete_file_from_imagekit(metadata["_imagekit_file_id"])
+                except Exception as cleanup_error:
+                    print(f"⚠️ Failed to cleanup ImageKit file: {cleanup_error}")
             # Construct URL (Threads URLs are usually https://www.threads.net/@user/post/ID)
             # But the ID returned is Graph ID. We might need permalink field.
             return {
@@ -250,6 +270,13 @@ class ThreadsPublisher(SocialPublisher):
                 'url': f"https://www.threads.net/post/{media_id}" # Approximation
             }
         else:
+            # Cleanup ImageKit even on failure if it was uploaded
+            if metadata.get("_imagekit_file_id"):
+                try:
+                    from src.imagekit_utils import delete_file_from_imagekit
+                    delete_file_from_imagekit(metadata["_imagekit_file_id"])
+                except Exception as cleanup_error:
+                    print(f"⚠️ Failed to cleanup ImageKit file: {cleanup_error}")
             error_msg = pub_result.get('error', {}).get('message', 'Publishing failed')
             print(f"❌ Publish error: {error_msg}")
             return {'error': error_msg, 'full_response': pub_result}

src/config.py

@@ -168,6 +168,12 @@ def load_configuration(force_reload: bool = False) -> Dict[str, Any]:
         # X (Twitter)
         "x_client_id": os.getenv("X_CLIENT_ID"),
         "x_client_secret": os.getenv("X_CLIENT_SECRET"),
+
+        # ImageKit
+        "imagekit_public_key": os.getenv("IMAGEKIT_PUBLIC_KEY"),
+        "imagekit_private_key": os.getenv("IMAGEKIT_PRIVATE_KEY"),
+        "imagekit_id": os.getenv("IMAGEKIT_ID"),
+        "imagekit_url_endpoint": os.getenv("IMAGEKIT_URL_ENDPOINT") or (f"https://ik.imagekit.io/{os.getenv('IMAGEKIT_ID')}/" if os.getenv("IMAGEKIT_ID") else None),
     }
 
     # On-screen CTA options

src/imagekit_utils.py

@@ -0,0 +1,79 @@
+"""
+ImageKit.io utilities for temporary file hosting.
+Used as a fallback when GCS is unavailable.
+"""
+import os
+from typing import Optional, Dict, Any
+from imagekitio import ImageKit
+from src.config import get_config_value
+from src.utils import logger
+
+def get_imagekit_client() -> Optional[ImageKit]:
+    """Initialize and return ImageKit client using config values."""
+    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

tests/test_imagekit.py

@@ -0,0 +1,41 @@
+"""
+Test script to verify ImageKit integration.
+Run this script after setting IMAGEKIT_PUBLIC_KEY, IMAGEKIT_PRIVATE_KEY, and IMAGEKIT_ID.
+"""
+import os
+import sys
+
+# Add project root to path
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+from src.imagekit_utils import upload_file_to_imagekit, delete_file_from_imagekit
+from src.utils import logger
+
+def test_imagekit_flow():
+    # Create a small dummy file
+    test_file = "test_imagekit_upload.txt"
+    with open(test_file, "w") as f:
+        f.write("This is a test file for ImageKit integration.")
+        
+    try:
+        # Test Upload
+        result = upload_file_to_imagekit(test_file)
+        if result and result.get("url"):
+            print(f"✅ Upload successful! URL: {result['url']}")
+            print(f"File ID: {result['file_id']}")
+            
+            # Test Delete
+            success = delete_file_from_imagekit(result['file_id'])
+            if success:
+                print("✅ Delete successful!")
+            else:
+                print("❌ Delete failed!")
+        else:
+            print("❌ Upload failed! Check your credentials and logs.")
+            
+    finally:
+        if os.path.exists(test_file):
+            os.remove(test_file)
+
+if __name__ == "__main__":
+    test_imagekit_flow()