deploy

.env

@@ -0,0 +1 @@
+PINECONE_API_KEY=pcsk_5NzQp8_MrApuQxQBU5P3YXYqipyVM4hm7BdA7tzB9tYPJQJSWySrtgW3KJHkS5gMYvLJZk

Dockerfile

@@ -0,0 +1,20 @@
+# Use official Python 3.9 image
+FROM python:3.10-slim
+
+# Set working directory
+WORKDIR /app
+
+# Copy requirements file
+COPY requirements.txt .
+
+# Install dependencies
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy application code
+COPY app/ .
+
+# Expose port 8000 (Hugging Face Spaces uses 7860 by default, but we'll map it)
+EXPOSE 7860
+
+# Start FastAPI with uvicorn
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -0,0 +1,8 @@
+---
+title: Image Search Engine FastAPI
+emoji: 🐨
+colorFrom: purple
+colorTo: gray
+sdk: docker
+pinned: false
+---

main.py

@@ -0,0 +1,11 @@
+from fastapi import FastAPI, UploadFile, File
+from routers import search
+
+app = FastAPI(title="Image Search API")
+
+# Include search routes
+app.include_router(search.router)
+
+@app.get("/")
+def root():
+    return {"message": "Image Search API is running!"}

models/search_models.py

@@ -0,0 +1,10 @@
+from pydantic import BaseModel
+from typing import List, Dict
+
+class SearchResult(BaseModel):
+    id: str
+    score: float
+    metadata: Dict[str, str]
+
+class SearchResponse(BaseModel):
+    matches: List[SearchResult]

requirements.txt

@@ -0,0 +1,8 @@
+fastapi>=0.95.0
+uvicorn>=0.20.0
+sentence-transformers>=2.2.0
+pillow>=9.0.0
+torch>=2.0.0
+pinecone
+python-dotenv>=1.0.0
+numpy>=1.24.0

routers/search.py

@@ -0,0 +1,37 @@
+from fastapi import APIRouter, UploadFile, File, Query
+from fastapi.responses import JSONResponse
+from services.embedding import get_text_embedding, get_image_embedding
+from services.pinecone_service import search_similar_images
+from typing import List, Dict, Any
+
+router = APIRouter(prefix="/search", tags=["Search"])
+
+@router.get("/text")
+async def search_by_text(query: str = Query(..., description="Search query")) -> Dict[str, Any]:
+    try:
+        query_embedding = get_text_embedding(query)
+        if not query_embedding or not isinstance(query_embedding, list):
+            return JSONResponse(content={"error": "Failed to generate embedding"}, status_code=500)
+            
+        results = search_similar_images(query_embedding)
+        if not results:
+            return JSONResponse(content={"matches": []}, status_code=200)
+            
+        return JSONResponse(content={"matches": results}, status_code=200)
+    except Exception as e:
+        return JSONResponse(content={"error": str(e)}, status_code=500)
+
+@router.post("/image")
+async def search_by_image(file: UploadFile = File(...)) -> Dict[str, Any]:
+    try:
+        image_embedding = get_image_embedding(file)
+        if not image_embedding or not isinstance(image_embedding, list):
+            return JSONResponse(content={"error": "Failed to generate embedding"}, status_code=500)
+            
+        results = search_similar_images(image_embedding)
+        if not results:
+            return JSONResponse(content={"matches": []}, status_code=200)
+            
+        return JSONResponse(content={"matches": results}, status_code=200)
+    except Exception as e:
+        return JSONResponse(content={"error": str(e)}, status_code=500)

services/embedding.py

@@ -0,0 +1,24 @@
+from sentence_transformers import SentenceTransformer
+from PIL import Image
+from fastapi import UploadFile
+from typing import List, Optional
+import torch
+
+model = SentenceTransformer("clip-ViT-B-32")
+
+def get_text_embedding(text: str) -> Optional[List[float]]:
+    try:
+        embedding = model.encode(text, convert_to_tensor=True).cpu().numpy().tolist()
+        return embedding
+    except Exception as e:
+        print(f"Error generating text embedding: {e}")
+        return None
+
+def get_image_embedding(image_file: UploadFile) -> Optional[List[float]]:
+    try:
+        image = Image.open(image_file.file).convert("RGB").resize((224, 224))
+        embedding = model.encode(image, convert_to_tensor=True).cpu().numpy().tolist()
+        return embedding
+    except Exception as e:
+        print(f"Error generating image embedding: {e}")
+        return None

services/pinecone_service.py

@@ -0,0 +1,43 @@
+import os
+from pinecone import Pinecone
+from dotenv import load_dotenv
+from typing import List, Dict, Any
+
+# Load environment variables
+load_dotenv()
+
+PINECONE_API_KEY = os.getenv("PINECONE_API_KEY")
+if not PINECONE_API_KEY:
+    raise ValueError("PINECONE_API_KEY is missing! Check your .env file.")
+
+INDEX_NAME = "unsplash-index-session"
+NAMESPACE = "image-search-dataset"
+
+# Initialize Pinecone Client
+pc = Pinecone(api_key=PINECONE_API_KEY)
+index = pc.Index(INDEX_NAME)
+
+def search_similar_images(query_embedding: List[float], top_k: int = 10) -> List[Dict[str, Any]]:
+    """Search for similar images in Pinecone using the given embedding."""
+    try:
+        results = index.query(
+            vector=query_embedding,
+            top_k=top_k,
+            include_metadata=True,
+            namespace=NAMESPACE
+        )
+
+        cleaned_results = []
+        for match in results.get("matches", []):
+            metadata = match.get("metadata", {})
+            cleaned_results.append({
+                "id": match["id"],
+                "score": float(match["score"]),
+                "url": metadata.get("url", "")
+            })
+            
+        return cleaned_results
+        
+    except Exception as e:
+        print(f"❌ Error querying Pinecone: {e}")
+        return []