Added image upload service

cloudzy/agents/image_analyzer.py

@@ -46,6 +46,7 @@ result: {
   "caption": "a short description for the image"
 }
 """
+        
 
         # Send request
         completion = self.client.chat.completions.create(

cloudzy/routes/upload.py

@@ -12,6 +12,7 @@ from cloudzy.ai_utils import  ImageEmbeddingGenerator
 from cloudzy.search_engine import SearchEngine
 
 from cloudzy.agents.image_analyzer import ImageDescriber
+from cloudzy.utils.file_upload_service import ImgBBUploader
 
 
 import os
@@ -61,8 +62,8 @@ def validate_image_file(filename: str) -> bool:
     """Check if file has valid image extension"""
     return Path(filename).suffix.lower() in ALLOWED_EXTENSIONS
 
-# response_model=UploadResponse
-@router.post("/upload")
+
+@router.post("/upload", response_model=UploadResponse)
 async def upload_photo(
     file: UploadFile = File(...),
     session: Session = Depends(get_session),
@@ -99,29 +100,32 @@ async def upload_photo(
     filepath = f"uploads/{saved_filename}"
 
 
-    APP_DOMAIN = os.getenv("APP_DOMAIN")
-
 
-    image_url = f"{APP_DOMAIN}uploads/{saved_filename}"
 
+    try:
+        uploader = ImgBBUploader(expiration=600)
+        image_url = uploader.upload(filepath)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Image upload failed: {str(e)}")
+    
 
 
     try:
-
-        
-        
+          
         describer = ImageDescriber()
         # result = describer.describe_image("https://userx2000-cloudzy-ai-challenge.hf.space/uploads/img_1_20251024_064435_667.jpg")
         # result = describer.describe_image("https://userx2000-cloudzy-ai-challenge.hf.space/uploads/img_2_20251024_082115_102.jpeg")
         result = describer.describe_image(image_url)
        
         
-
     except Exception as e:
         raise HTTPException(status_code=500, detail=f"Error processing image: {str(e)}")
     
 
-    return result
+    APP_DOMAIN = os.getenv("APP_DOMAIN")
+
+    image_url = f"{APP_DOMAIN}uploads/{saved_filename}"
+
     
     # Generate AI analysis
     tags = result.get("tags", [])

cloudzy/search_engine.py

@@ -1,45 +1,41 @@
-"""FAISS-based semantic search engine"""
+"""FAISS-based semantic search engine using ID-mapped index"""
 import faiss
 import numpy as np
-from typing import List, Tuple, Optional
+from typing import List, Tuple
 import os
-import pickle
 
 
 class SearchEngine:
     """FAISS-based search engine for image embeddings"""
-    
+
     def __init__(self, dim: int = 1024, index_path: str = "faiss_index.bin"):
         self.dim = dim
         self.index_path = index_path
-        self.id_map: List[int] = []  # Map FAISS indices to photo IDs
-        
-        # Load existing index or create new one
+
+        # Load existing index or create a new one
         if os.path.exists(index_path):
             self.index = faiss.read_index(index_path)
         else:
-            self.index = faiss.IndexFlatL2(dim)
-    
+            base_index = faiss.IndexFlatL2(dim)
+            self.index = faiss.IndexIDMap(base_index)
+
     def add_embedding(self, photo_id: int, embedding: np.ndarray) -> None:
         """
         Add an embedding to the index.
-        
+
         Args:
             photo_id: Unique photo identifier
             embedding: 1D numpy array of shape (dim,)
         """
         # Ensure embedding is float32 and correct shape
         embedding = embedding.astype(np.float32).reshape(1, -1)
-        
-        # Add to FAISS index
-        self.index.add(embedding)
-        
-        # Track photo ID
-        self.id_map.append(photo_id)
-        
+
+        # Add embedding with its ID
+        self.index.add_with_ids(embedding, np.array([photo_id], dtype=np.int64))
+
         # Save index to disk
         self.save()
-    
+
     def search(self, query_embedding: np.ndarray, top_k: int = 5) -> List[Tuple[int, float]]:
         """
         Search for similar embeddings.
@@ -49,9 +45,8 @@ class SearchEngine:
             top_k: Number of results to return
 
         Returns:
-            List of (photo_id, distance) tuples with distance <= 0.4
+            List of (photo_id, distance) tuples with distance <= 0.5
         """
-
         self.load()
 
         if self.index.ntotal == 0:
@@ -61,35 +56,34 @@ class SearchEngine:
         query_embedding = query_embedding.astype(np.float32).reshape(1, -1)
 
         # Search in FAISS index
-        distances, indices = self.index.search(query_embedding, min(top_k, self.index.ntotal))
+        distances, ids = self.index.search(query_embedding, top_k)
 
-        # Map back to photo IDs and filter distances > 0.4
+        # Filter invalid and distant results
         results = [
-            (self.id_map[int(idx)], float(distance))
-            for distance, idx in zip(distances[0], indices[0])
-            if distance <= 0.5
+            (int(photo_id), float(distance))
+            for photo_id, distance in zip(ids[0], distances[0])
+            if photo_id != -1 and distance <= 0.5
         ]
 
         return results
 
     def save(self) -> None:
-        """Save index and id_map to disk"""
+        """Save FAISS index to disk"""
         faiss.write_index(self.index, self.index_path)
-        with open(self.index_path + ".ids", "wb") as f:
-            pickle.dump(self.id_map, f)
 
     def load(self) -> None:
-        """Load index and id_map from disk"""
+        """Load FAISS index from disk"""
         if os.path.exists(self.index_path):
             self.index = faiss.read_index(self.index_path)
-        if os.path.exists(self.index_path + ".ids"):
-            with open(self.index_path + ".ids", "rb") as f:
-                self.id_map = pickle.load(f)
-    
+        else:
+            # Recreate empty ID-mapped index if missing
+            base_index = faiss.IndexFlatL2(self.dim)
+            self.index = faiss.IndexIDMap(base_index)
+
     def get_stats(self) -> dict:
         """Get index statistics"""
         return {
             "total_embeddings": self.index.ntotal,
             "dimension": self.dim,
-            "id_map_size": len(self.id_map)
-        }
+            "index_type": type(self.index).__name__,
+        }

cloudzy/utils/file_upload_service.py

@@ -0,0 +1,77 @@
+import base64
+import requests
+from pathlib import Path
+from typing import Optional
+import os 
+
+
+from dotenv import load_dotenv
+load_dotenv()
+
+
+class ImgBBUploader:
+    """
+    Upload an image file to ImgBB and return only the direct public URL.
+    """
+
+    API_ENDPOINT = "https://api.imgbb.com/1/upload"
+
+    def __init__(self, expiration: Optional[int] = None, timeout: int = 30):
+        api_key = os.getenv("IMAGE_UPLOAD_API_KEY")
+        if not api_key:
+            raise ValueError("api_key is required")
+        self.api_key = api_key
+        self.expiration = expiration
+        self.timeout = timeout
+
+    def _encode_file_to_base64(self, image_path: Path) -> str:
+        if not image_path.exists():
+            raise FileNotFoundError(f"Image not found: {image_path}")
+        with image_path.open("rb") as f:
+            data = f.read()
+        return base64.b64encode(data).decode("ascii")
+
+    def upload(self, image_path: str) -> str:
+        """
+        Upload the image and return the direct URL of the uploaded file.
+        """
+        image_path_obj = Path(image_path)
+        b64_image = self._encode_file_to_base64(image_path_obj)
+
+        params = {"key": self.api_key}
+        if self.expiration is not None:
+            params["expiration"] = str(self.expiration)
+
+        try:
+            resp = requests.post(
+                self.API_ENDPOINT,
+                params=params,
+                data={"image": b64_image},
+                timeout=self.timeout,
+            )
+            resp.raise_for_status()
+            data = resp.json()
+            print(data)
+            if data.get("success"):
+                return data["data"]["url"]
+            raise RuntimeError(f"Upload failed: {data}")
+        except requests.RequestException as e:
+            raise RuntimeError(f"ImgBB upload failed: {e}") from e
+
+
+if __name__ == "__main__":
+    uploader = ImgBBUploader(expiration=86400)
+    result = uploader.upload("/Users/komeilfathi/Documents/hf_deploy_test/cloudzy_ai_challenge/uploads/img_1_20251024_080401_794.jpg")
+    print(result)
+
+
+        # resp = uploader.upload_file("/Users/komeilfathi/Documents/hf_deploy_test/cloudzy_ai_challenge/uploads/img_1_20251024_080401_794.jpg")
+
+
+
+
+
+# Example usage:
+# uploader = FreeImageHostUploader(api_key="your_api_key_here")
+# image_url = uploader.upload_image("path_to_your_image.jpg")
+# print(f"Image uploaded successfully: {image_url}")