utils.py

import re
import pandas as pd
import numpy as np
from rapidfuzz import fuzz

_STOP_WORDS = frozenset({'the', 'a', 'an', 'of', 'in', 'on', 'at', 'to', 'for', 'and', 'or', 'is', 'it', 'as', 'be', 'by', 'with', 'its', 'but', 'not', 'so', 'no', 'up', 'if', 'me', 'my', 'we', 'he', 'she', 'do', 'has', 'had', 'was', 'are', 'were', 'been'})

_ACRONYMS = {
    "lotr": "lord of the rings",
    "hp": "harry potter",
    "sw": "star wars",
    "st": "star trek",
    "tdk": "the dark knight",
    "tdkr": "the dark knight rises",
    "bb": "batman begins",
    "potc": "pirates of the caribbean",
    "twd": "the walking dead",
    "got": "game of thrones",
}

_FUZZY_THRESHOLD = 86
_TOKEN_SET_THRESHOLD = 62

def _tokenize(text):
    return [w for w in re.findall(r"[A-Za-z0-9]+", str(text).lower()) if len(w) >= 2]

def _significant_words(words):
    return [w for w in words if w not in _STOP_WORDS]

def _expand_acronyms(query):
    q = query.lower().strip()
    expanded = _ACRONYMS.get(q)
    if expanded:
        return [query, expanded]
    return [query]

def _score_title(t_lower, query_variants):
    best = 0
    for qv in query_variants:
        s = _score_single(t_lower, qv)
        if s > best:
            best = s
    return best

def _score_single(t_lower, query):
    q = query.lower().strip()
    qtokens = _tokenize(q)
    sig = _significant_words(qtokens)

    # Tier 1: Exact title match = 100
    if t_lower == q:
        return 100

    # Tier 2: Phrase match (consecutive whole words) = 90
    if qtokens:
        pr = r'\b' + r'\s+'.join(re.escape(w) for w in qtokens) + r'\b'
        if re.search(pr, t_lower):
            return 90

    # Tier 3: Starts-with match = 80
    if t_lower.startswith(q):
        return 80

    # No significant tokens — nothing more to match
    if not sig:
        return 0

    n_sig = len(sig)
    words_present = sum(1 for w in sig if re.search(r'\b' + re.escape(w) + r'\b', t_lower))

    # Tier 4: Multi-token overlap (ALL significant words present) = 70
    if words_present == n_sig:
        return 70

    # Tier 5: RapidFuzz fuzzy match = 60
    if len(q) >= 5:
        if len(qtokens) >= 2:
            # Multi-word: require BOTH token_set (word-level overlap) AND partial (character-level) above threshold
            if fuzz.token_set_ratio(q, t_lower) >= _TOKEN_SET_THRESHOLD and fuzz.partial_ratio(q, t_lower) >= _FUZZY_THRESHOLD:
                return 60
        else:
            if fuzz.partial_ratio(q, t_lower) >= _FUZZY_THRESHOLD:
                return 60

    # Tier 6: Single token match = 40  (filtered out — never returned)
    if words_present >= 1:
        return 40

    return 0

def classify_match(query, top_score, result_count):
    """Classify the search result into EXACT_MATCH, FRANCHISE_MATCH, PARTIAL_MATCH, or NO_MATCH."""
    if not query or not query.strip():
        return None
    if result_count == 0:
        return "NO_MATCH"
    if top_score >= 100:
        return "EXACT_MATCH"
    if top_score >= 80:
        return "FRANCHISE_MATCH"
    return "PARTIAL_MATCH"

def fuzzy_search_movies(df, query):
    """
    Production-grade ranked search pipeline.
    Returns (results_df, info_dict) where info_dict contains:
      - match_type: str (EXACT_MATCH, FRANCHISE_MATCH, PARTIAL_MATCH, NO_MATCH, or None for no query)
      - top_score: int (highest score among results)
      - threshold_applied: bool (whether score < 60 were dropped)
    """
    if not query or not query.strip():
        return df.copy(), {"match_type": None, "top_score": 0, "threshold_applied": False}

    query_variants = _expand_acronyms(query)
    titles_lower = df["title"].str.lower()

    scores = titles_lower.apply(lambda t: _score_title(t, query_variants))
    result = df.copy()
    result["search_score"] = scores.values

    # Apply threshold: never return weak single-token matches (score < 60)
    result = result[result["search_score"] >= 60].copy()

    if result.empty:
        return result.drop(columns=["search_score"]), {
            "match_type": "NO_MATCH", "top_score": 0, "threshold_applied": True
        }

    result = result.sort_values(by=["search_score", "avg_rating"], ascending=[False, False])
    top_score = result["search_score"].iloc[0]
    match_type = classify_match(query, top_score, len(result))
    result = result.drop(columns=["search_score"])

    return result, {
        "match_type": match_type,
        "top_score": int(top_score),
        "threshold_applied": True
    }

def filter_and_sort_movies(df, query, year_range, min_rating, min_rating_count, sort_by):
    """
    Performs full filtering and sorting operations on the movie catalog.
    Returns (filtered_df, search_info) where search_info is a dict with match_type and top_score.
    If no query, search_info is None.
    """
    filtered_df = df.copy()
    search_info = None
    
    # Apply Search Query
    if query and query.strip():
        filtered_df, search_info = fuzzy_search_movies(filtered_df, query)
        
    # Apply Release Year Filter
    if year_range and len(year_range) == 2:
        year_min, year_max = year_range
        filtered_df = filtered_df[
            (filtered_df["year"] >= year_min) & (filtered_df["year"] <= year_max)
        ]
        
    # Apply Minimum Average Rating Filter
    filtered_df = filtered_df[filtered_df["avg_rating"] >= min_rating]
    
    # Apply Minimum Rating Count Filter
    filtered_df = filtered_df[filtered_df["rating_count"] >= min_rating_count]
    
    # Apply Sorting
    if sort_by == "Highest Rated":
        filtered_df = filtered_df.sort_values(by=["avg_rating", "rating_count"], ascending=[False, False])
    elif sort_by == "Most Popular":
        filtered_df = filtered_df.sort_values(by="rating_count", ascending=False)
    elif sort_by == "Newest":
        filtered_df = filtered_df.sort_values(by="year", ascending=False)
    elif sort_by == "Oldest":
        has_year = filtered_df[filtered_df["year"] > 0]
        no_year = filtered_df[filtered_df["year"] == 0]
        sorted_has_year = has_year.sort_values(by="year", ascending=True)
        filtered_df = pd.concat([sorted_has_year, no_year])
    elif sort_by == "Alphabetical":
        filtered_df = filtered_df.sort_values(by="title", ascending=True)
        
    return filtered_df, search_info

def paginate_dataframe(df, page_idx, page_size=20):
    """
    Extracts a chunk of the dataframe for pagination.
    Returns: (sliced_dataframe, total_pages)
    """
    total_records = len(df)
    total_pages = max(1, int(np.ceil(total_records / page_size)))
    
    # Clamp page index
    page_idx = max(0, min(page_idx, total_pages - 1))
    
    start_row = page_idx * page_size
    end_row = start_row + page_size
    
    sliced_df = df.iloc[start_row:end_row]
    return sliced_df, total_pages