|
|
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) |
