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