fuzzy_matcher.py

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

    _keyword_map = kmap
    logger.info("College keyword map built: %d unique keyword keys", len(_keyword_map))


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