""" GCAS Search Engine – Fuzzy Entity Matcher ========================================== Responsibilities ---------------- 1. At index-time: build an entity vocabulary from the indexed data (college names, university names, districts, talukas, programs, etc.) 2. At query-time: a. Scan the query for tokens that partially / phonetically match known entity names. b. Compute a per-query confidence score from FAISS result scores. c. Return "did you mean?" suggestions when confidence is low. Matching strategy (layered) ---------------------------- Layer 1 – Exact match after normalisation (fastest) Layer 2 – RapidFuzz token_set_ratio ≥ FUZZY_THRESHOLD (handles word-order variation and partial matches) Layer 3 – Levenshtein edit-distance for short tokens (handles single-character ASR errors) Layer 4 – Soundex phonetic matching (handles phonetic ASR mistakes) Confidence levels ----------------- HIGH score > HIGH_THRESHOLD → return results, no "did you mean" MEDIUM score > LOW_THRESHOLD → return results + gentle suggestion LOW score ≤ LOW_THRESHOLD → "did you mean?" + results with caveat """ from __future__ import annotations import logging import re import unicodedata from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Set, Tuple logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Thresholds # --------------------------------------------------------------------------- FUZZY_THRESHOLD = 75 # RapidFuzz score 0-100 to accept a match HIGH_CONFIDENCE = 0.72 # FAISS cosine score (L2-normalised inner product) LOW_CONFIDENCE = 0.52 # below this → trigger "did you mean?" # --------------------------------------------------------------------------- # Entity vocabulary (populated at index-time via build_vocabulary()) # --------------------------------------------------------------------------- @dataclass class EntityVocabulary: colleges: List[str] = field(default_factory=list) universities: List[str] = field(default_factory=list) districts: List[str] = field(default_factory=list) talukas: List[str] = field(default_factory=list) programs: List[str] = field(default_factory=list) major_subjects:List[str] = field(default_factory=list) def all_entities(self) -> List[Tuple[str, str]]: """Return (entity_name, entity_type) for every known entity.""" result: List[Tuple[str, str]] = [] for name in self.colleges: result.append((name, "college")) for name in self.universities: result.append((name, "university")) for name in self.districts: result.append((name, "district")) for name in self.talukas: result.append((name, "taluka")) for name in self.programs: result.append((name, "program")) for name in self.major_subjects:result.append((name, "subject")) return result # Module-level vocabulary – populated by build_vocabulary() _vocab: EntityVocabulary = EntityVocabulary() def build_vocabulary(data_store: Dict[str, List[Dict[str, Any]]]) -> EntityVocabulary: """ Extract unique entity names from the indexed row data. Should be called after indexer.load_and_index() completes. """ global _vocab colleges: Set[str] = set() universities: Set[str] = set() districts: Set[str] = set() talukas: Set[str] = set() programs: Set[str] = set() major_subjects: Set[str] = set() COLUMN_MAP = { # column name → which set to add to "CollegeName": colleges, "College": colleges, "UniversityName": universities, "University": universities, "DistrictName": districts, "CollegeDistrict": districts, "TalukaName": talukas, "AdmissionName": programs, "Program": programs, "MajorSubject": major_subjects, "MajorBasketName": major_subjects, } for _table, rows in data_store.items(): for row in rows: for col, target_set in COLUMN_MAP.items(): val = row.get(col) if val and isinstance(val, str) and val.strip(): target_set.add(val.strip()) _vocab = EntityVocabulary( colleges=sorted(colleges), universities=sorted(universities), districts=sorted(districts), talukas=sorted(talukas), programs=sorted(programs), major_subjects=sorted(major_subjects), ) logger.info( "Vocabulary built: %d colleges, %d universities, %d districts, " "%d talukas, %d programs, %d subjects", len(_vocab.colleges), len(_vocab.universities), len(_vocab.districts), len(_vocab.talukas), len(_vocab.programs), len(_vocab.major_subjects), ) return _vocab def get_vocabulary() -> EntityVocabulary: return _vocab # --------------------------------------------------------------------------- # College keyword lookup (uses CollegeNameSearchKeyword1-4 columns) # --------------------------------------------------------------------------- # Module-level map built alongside the vocabulary. # Structure: normalised_keyword → list of (full_college_name, college_id) # List because multiple colleges can share a keyword (e.g. "ARB"). _keyword_map: Dict[str, List[Tuple[str, str]]] = {} def build_college_keyword_map(data_store: Dict[str, List[Dict[str, Any]]]) -> None: """ Build a lookup from every CollegeNameSearchKeyword* value to the corresponding full CollegeName. Called once at startup alongside build_vocabulary(). """ global _keyword_map kmap: Dict[str, List[Tuple[str, str]]] = {} for _table, rows in data_store.items(): for row in rows: college_name = (row.get("CollegeName") or "").strip() college_id = str(row.get("CollegeId") or "") if not college_name: continue for i in range(1, 5): kw = (row.get(f"CollegeNameSearchKeyword{i}") or "").strip() if not kw or kw.lower() in ("nan", "none", ""): continue key = _normalize_for_matching(kw) if key: kmap.setdefault(key, []) entry = (college_name, college_id) if entry not in kmap[key]: kmap[key].append(entry) # Filter out keywords whose every token is a stopword (e.g. "Girls College", # "Boys College"). These are too generic and cause false fuzzy matches # (e.g. "gls college" fuzzy-matches "girls college" at partial_ratio=95). filtered_kmap: Dict[str, List[Tuple[str, str]]] = {} for key, entries in kmap.items(): tokens = key.split() if all(t in _STOPWORDS or len(t) <= 2 for t in tokens): logger.debug("Skipping all-stopword keyword: %r", key) continue filtered_kmap[key] = entries _keyword_map = filtered_kmap logger.info( "College keyword map built: %d unique keys (%d skipped as all-stopword)", len(_keyword_map), len(kmap) - len(filtered_kmap), ) def resolve_college_in_query(query: str) -> Tuple[str, List[EntityMatch]]: """ Scan n-grams of `query` against the college keyword map. Returns (rewritten_query, list_of_EntityMatch). Noise guards: - Single-word spans: must be an all-uppercase abbreviation (≥ 3 alpha chars) OR a long-enough non-stopword (≥ 6 chars, not in _STOPWORDS). This prevents common words like "for", "college", "fees", "girls" from matching college keyword entries. - Multi-word spans: at least one token must pass the above guard. - Exact match is only used when the keyword maps to exactly ONE college. - Fuzzy match threshold: 95 (partial) / 90 (ratio) — intentionally tight. """ if not _keyword_map: return query, [] # Common words that appear in college keyword lists but are NOT abbreviations. # Includes major Gujarat cities/districts — they appear in thousands of # college keywords but are NOT useful for uniquely identifying a college. _STOPWORDS: Set[str] = { # English function / generic education words "for", "in", "of", "at", "by", "to", "the", "a", "an", "and", "or", "with", "college", "colleges", "university", "universities", "institute", "institutes", "arts", "commerce", "science", "law", "education", "girls", "boys", "women", "men", "fees", "hostel", "naac", "contact", "under", "grade", "center", "centre", "general", "department", "school", "faculty", # Hindi/Gujarati common words "mein", "ke", "ka", "ki", "hai", "kya", "liye", "sath", "aur", # Major Gujarat cities / districts (appear in ~100s of college keywords) "ahmedabad", "surat", "vadodara", "baroda", "rajkot", "gandhinagar", "bhavnagar", "jamnagar", "junagadh", "anand", "mehsana", "nadiad", "patan", "navsari", "valsad", "bharuch", "surendranagar", "amreli", "kutch", "kachchh", "porbandar", "morbi", "botad", "dwarka", "aravalli", "kheda", "mahisagar", "dahod", "panchmahal", "chhota udaipur", "narmada", "tapi", "dang", "sabar kantha", "sabarkantha", "banaskantha", "visnagar", "gondal", "veraval", "sidhpur", "unjha", "palanpur", "himmatnagar", } def _is_plausible_token(t: str) -> bool: """True if t is a meaningful abbreviation or a long non-stopword.""" t = t.strip() tl = t.lower() if tl in _STOPWORDS: return False # Short abbreviation: 3–5 alpha chars, not a stopword (e.g. "arb", "imn") if t.isalpha() and 3 <= len(t) <= 5: return True # Long word (≥ 6 chars) if len(t) >= 6: return True return False def _is_plausible_span(span_tokens: List[str]) -> bool: # Allow spans made up of initials: "m n", "a r b", "m n college" # → two or more single alpha chars (likely college initials like M.N., A.R.B.) single_initials = [t for t in span_tokens if len(t.strip()) == 1 and t.strip().isalpha()] if len(single_initials) >= 2: return True return any(_is_plausible_token(t) for t in span_tokens) tokens = query.split() matches: List[EntityMatch] = [] matched_token_indices: Set[int] = set() # Try n-grams from longest to shortest (up to 6 words for multi-word names) for n in range(min(6, len(tokens)), 0, -1): for i in range(len(tokens) - n + 1): # Skip if any token already claimed by a previous match if any(idx in matched_token_indices for idx in range(i, i + n)): continue span_tokens = tokens[i : i + n] if not _is_plausible_span(span_tokens): continue span = " ".join(span_tokens) norm_span = _normalize_for_matching(span) if not norm_span or len(norm_span) < 2: continue # --- Exact match (unambiguous only) --- if norm_span in _keyword_map: entries = _keyword_map[norm_span] if len(entries) == 1: # reject ambiguous keywords matches.append(EntityMatch( query_span=span, matched=entries[0][0], entity_type="college", score=100, method="keyword_exact", )) matched_token_indices.update(range(i, i + n)) continue # --- Fuzzy match (≥ 2 tokens, OR ≥ 3 chars if a plausible abbreviation) --- if n < 2 and len(norm_span) < 3: continue best_score = 0 best_name = "" try: from rapidfuzz import fuzz as _fuzz for key, entries in _keyword_map.items(): if len(norm_span) < len(key) - 4: score = int(_fuzz.partial_ratio(norm_span, key)) threshold = 95 else: score = int(_fuzz.ratio(norm_span, key)) threshold = 90 if score > best_score and score >= threshold: best_score = score best_name = entries[0][0] except ImportError: for key, entries in _keyword_map.items(): score = _levenshtein_similarity(norm_span, key) if score > best_score and score >= 90: best_score = score best_name = entries[0][0] if best_name: matches.append(EntityMatch( query_span=span, matched=best_name, entity_type="college", score=best_score, method="keyword_fuzzy", )) matched_token_indices.update(range(i, i + n)) if not matches: return query, [] # Rewrite: replace matched spans with canonical college name (longest first) rewritten = query for m in sorted(matches, key=lambda x: -len(x.query_span)): pattern = re.compile(re.escape(m.query_span), re.IGNORECASE) rewritten = pattern.sub(m.matched, rewritten, count=1) # Clean up duplicate consecutive words that can arise when the matched # span overlaps with a surrounding word (e.g. "Law College" → "Law College college") rewritten = re.sub(r'\b(\w+)\s+\1\b', r'\1', rewritten, flags=re.IGNORECASE) rewritten = re.sub(r'\s+', ' ', rewritten).strip() return rewritten, matches # --------------------------------------------------------------------------- # Soundex (pure-Python, no deps) # --------------------------------------------------------------------------- _SOUNDEX_TABLE = str.maketrans( "AEHIOUYWBFPVCGJKQSXZDTLMNR", "00000000111122222222334556" ) def soundex(word: str) -> str: """Classic Soundex for English-like strings.""" word = word.upper() # Keep first letter first = word[0] if word else "" # Encode coded = word.translate(_SOUNDEX_TABLE) # Remove zeros filtered = first + re.sub(r'0', '', coded[1:]) # Remove consecutive duplicates deduped = first for ch in filtered[1:]: if ch != deduped[-1]: deduped += ch # Pad / truncate to length 4 return (deduped + "000")[:4] # --------------------------------------------------------------------------- # Core fuzzy matching helpers # --------------------------------------------------------------------------- def _normalize_for_matching(s: str) -> str: """Lowercase, strip punctuation, collapse whitespace.""" s = s.lower() s = unicodedata.normalize("NFKD", s).encode("ascii", "ignore").decode() s = re.sub(r'[^\w\s]', ' ', s) s = re.sub(r'\s+', ' ', s).strip() return s def _fuzzy_score(query_token: str, candidate: str) -> int: """ Return a 0-100 RapidFuzz score between a query token and a candidate. Falls back to a simple Levenshtein-based score if rapidfuzz is missing. """ q = _normalize_for_matching(query_token) c = _normalize_for_matching(candidate) try: from rapidfuzz import fuzz return fuzz.token_set_ratio(q, c) except ImportError: # Fallback: character-level Levenshtein similarity return _levenshtein_similarity(q, c) def _levenshtein_similarity(a: str, b: str) -> int: """Returns 0-100 similarity via edit distance (no external deps).""" if not a or not b: return 0 m, n = len(a), len(b) dp = list(range(n + 1)) for i in range(1, m + 1): prev = dp[0] dp[0] = i for j in range(1, n + 1): temp = dp[j] if a[i-1] == b[j-1]: dp[j] = prev else: dp[j] = 1 + min(prev, dp[j], dp[j-1]) prev = temp dist = dp[n] max_len = max(m, n) return int((1 - dist / max_len) * 100) def _phonetic_match(query_token: str, candidate: str) -> bool: """True if Soundex codes match (useful for ASR phonetic errors).""" q_words = _normalize_for_matching(query_token).split() c_words = _normalize_for_matching(candidate).split() if not q_words or not c_words: return False # Compare first significant words return soundex(q_words[0]) == soundex(c_words[0]) # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- @dataclass class EntityMatch: query_span: str # the substring from the query that matched matched: str # canonical entity name from vocabulary entity_type: str # "college" | "university" | "district" | ... score: int # 0-100 fuzzy score method: str # "exact" | "fuzzy" | "phonetic" @dataclass class QueryAnalysis: corrected_query: str detected_entities: List[EntityMatch] confidence_level: str # "high" | "medium" | "low" did_you_mean: List[str] # human-readable suggestions top_faiss_score: float def analyze_entities(query: str) -> List[EntityMatch]: """ Scan the query for tokens that match known geographic entity names (districts and talukas only). Colleges/universities are found by FAISS semantic search — rewriting the query with a college name guessed from common words like "college" or "Gujarat" causes catastrophically wrong results because token_set_ratio gives 100 for any word that appears inside a long college name string. University abbreviations (GTU, VNSGU, …) and program abbreviations (BCA, B.TECH, …) are already expanded by the normalizer pipeline before this function is called, so those entity types are also excluded here. """ # Only match against geographic entities that are short, specific, # and unambiguous — districts and talukas. geo_entities: List[Tuple[str, str]] = ( [(name, "district") for name in _vocab.districts] + [(name, "taluka") for name in _vocab.talukas] ) if not geo_entities: return [] # Tokenise query into n-grams (1-word to 3-word phrases only) tokens = query.split() matches: List[EntityMatch] = [] matched_spans: Set[str] = set() for n in range(min(3, len(tokens)), 0, -1): for i in range(len(tokens) - n + 1): span = " ".join(tokens[i:i+n]) if span in matched_spans: continue best_score = 0 best_entity = "" best_type = "" best_method = "" for entity_name, entity_type in geo_entities: if len(span) < 3: continue # Layer 1: exact match if _normalize_for_matching(span) == _normalize_for_matching(entity_name): best_score, best_entity, best_type, best_method = ( 100, entity_name, entity_type, "exact" ) break # Layer 2: fuzzy token match (use ratio, not token_set_ratio, # so that a single word does NOT score 100 against a long entity) try: from rapidfuzz import fuzz as _fuzz q_norm = _normalize_for_matching(span) c_norm = _normalize_for_matching(entity_name) score = int(_fuzz.ratio(q_norm, c_norm)) except ImportError: score = _levenshtein_similarity( _normalize_for_matching(span), _normalize_for_matching(entity_name), ) if score > best_score and score >= FUZZY_THRESHOLD: best_score = score best_entity = entity_name best_type = entity_type best_method = "fuzzy" # Layer 3: phonetic match (single-word spans only) if n == 1 and score < FUZZY_THRESHOLD: if _phonetic_match(span, entity_name): phonetic_score = max(score, 70) if phonetic_score > best_score: best_score = phonetic_score best_entity = entity_name best_type = entity_type best_method = "phonetic" if best_entity: matches.append(EntityMatch( query_span=span, matched=best_entity, entity_type=best_type, score=best_score, method=best_method, )) matched_spans.add(span) # Sort by score desc matches.sort(key=lambda m: m.score, reverse=True) return matches def rewrite_query_with_entities(query: str, entities: List[EntityMatch]) -> str: """ Replace matched (possibly mis-spelled) spans with canonical entity names. Only replaces if the fuzzy score is high enough to be confident. """ if not entities: return query result = query # Replace from longest span to shortest to avoid partial overlaps for match in sorted(entities, key=lambda m: -len(m.query_span)): if match.score >= FUZZY_THRESHOLD and match.method != "exact": # Case-insensitive replacement pattern = re.compile(re.escape(match.query_span), re.IGNORECASE) result = pattern.sub(match.matched, result, count=1) return result def assess_confidence( top_faiss_score: float, entities: List[EntityMatch], ) -> Tuple[str, float]: """ Compute a final confidence level combining FAISS score and entity match quality. Returns (level, adjusted_score) level: "high" | "medium" | "low" """ entity_boost = 0.0 if entities: best_entity_score = max(m.score for m in entities) / 100.0 entity_boost = best_entity_score * 0.08 # max +0.08 boost adjusted = min(1.0, top_faiss_score + entity_boost) if adjusted >= HIGH_CONFIDENCE: level = "high" elif adjusted >= LOW_CONFIDENCE: level = "medium" else: level = "low" return level, round(adjusted, 4) def generate_did_you_mean( query: str, entities: List[EntityMatch], top_faiss_score: float, top_results: List[Dict[str, Any]], ) -> List[str]: """ Generate human-readable "Did you mean …?" suggestions. Sources for suggestions: 1. Entity matches with score in the 60–74 range (below main threshold but plausible alternatives) 2. Top FAISS results' college/university names if score is medium 3. Near-miss entities from vocabulary scan """ suggestions: List[str] = [] seen: Set[str] = set() # 1. Near-miss entity suggestions (60 ≤ score < FUZZY_THRESHOLD) try: from rapidfuzz import process as rf_process # For each unmatched token, find closest entity matched_spans = {m.query_span.lower() for m in entities} tokens = [t for t in query.split() if len(t) >= 4] for token in tokens[:5]: # limit to first 5 tokens if token.lower() in matched_spans: continue all_names = ( _vocab.colleges[:200] # sample to keep it fast + _vocab.universities + _vocab.districts + _vocab.programs[:100] ) results = rf_process.extract( token, all_names, limit=2, score_cutoff=60 ) for name, score, _ in results: if score < FUZZY_THRESHOLD and name not in seen: suggestions.append(f'Did you mean "{name}"?') seen.add(name) except ImportError: pass # 2. Top result names as suggestions when confidence is low if top_faiss_score < LOW_CONFIDENCE and top_results: for r in top_results[:2]: data = r.get("data", {}) college = data.get("CollegeName") or data.get("College") if college and college not in seen: suggestions.append(f'Did you mean "{college}"?') seen.add(college) return suggestions[:3] # cap at 3 suggestions