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