import spacy from fastapi import FastAPI from pydantic import BaseModel from typing import Optional, Dict, Any # Load spaCy language model nlp = spacy.load("en_core_web_sm") # FastAPI instance app = FastAPI() # Define mappings for categories and filters KEYWORD_MAPPINGS = { "categories": { "salon": "Salon", "spa": "Spa", "gym": "Fitness", "fitness": "Fitness", "pet spa": "Pet Spa", "dental": "Dental", "nail art": "Nail Art", "tattoo": "Tattoo", }, "filters": { "top": {"top_rated": True}, "top-rated": {"top_rated": True}, "highly-rated": {"top_rated": True}, "best": {"top_rated": True}, "popular": {"popular": True}, "trending": {"trending": True}, "near me": {"radius": 500}, "around me": {"radius": 500}, "nearby": {"radius": 500}, }, } # Pydantic schema for the input query class QueryRequest(BaseModel): sentence: str latitude: Optional[float] = None longitude: Optional[float] = None def parse_sentence_to_query_ner(sentence: str, lat: Optional[float] = None, lng: Optional[float] = None) -> Dict[str, Any]: """ Parse a sentence using spaCy NER and build a search query. Args: sentence (str): Input sentence to parse. lat (Optional[float]): Latitude for geolocation. lng (Optional[float]): Longitude for geolocation. Returns: Dict[str, Any]: Parsed search query. """ # Initialize query with defaults query = { "location_id": None, "latitude": lat, "longitude": lng, "radius": None, "merchant_category": None, "business_name": None, "top_rated": False, "popular": False, "trending": False, } # Process the sentence using spaCy doc = nlp(sentence) # Extract categories from sentence using token matching for token in doc: for keyword, category in KEYWORD_MAPPINGS["categories"].items(): if keyword in token.text.lower(): query["merchant_category"] = category break # Extract filters using token and phrase matching for phrase, filter_dict in KEYWORD_MAPPINGS["filters"].items(): if phrase in sentence.lower(): query.update(filter_dict) # Extract potential business names using NER for ent in doc.ents: if ent.label_ in ["ORG", "PERSON", "GPE"]: # Relevant entity types query["business_name"] = ent.text break # Use NER to extract location context if "near me" in sentence.lower() or "around me" in sentence.lower() or "nearby" in sentence.lower(): if lat is not None and lng is not None: query["radius"] = 500 # Default radius for "near me" return query # FastAPI route @app.post("/parse-query/") async def parse_query(request: QueryRequest): """ API endpoint to parse a query and return a structured response. Args: request (QueryRequest): Input sentence and optional location data. Returns: Dict[str, Any]: Parsed search query. """ parsed_query = parse_sentence_to_query_ner( sentence=request.sentence, lat=request.latitude, lng=request.longitude ) return {"query": parsed_query}