recommender.py

import os
import joblib
import pandas as pd
import numpy as np
import tensorflow as tf
from tensorflow.keras.models import load_model
from functools import lru_cache

# Custom FMLayer definition for Factorization Machine component in DeepFM
class FMLayer(tf.keras.layers.Layer):
    def call(self, inputs):
        square_of_sum = tf.square(tf.reduce_sum(inputs, axis=1))
        sum_of_square = tf.reduce_sum(tf.square(inputs), axis=1)
        return 0.5 * (square_of_sum - sum_of_square)

class RecommenderManager:
    def __init__(self, models_dir="Models", data_path="Processed/train_full.parquet"):
        self.models_dir = models_dir
        self.data_path = data_path
        
        # Placeholders
        self.model = None
        self.user_encoder = None
        self.movie_encoder = None
        self.scaler = None
        self.watch_history_df = None
        self.movie_embeddings = None
        self.loaded = False
        
    def load_resources(self):
        """
        Loads all encoders, scaler, watch history data, and TensorFlow model.
        """
        if self.loaded:
            return
            
        print("Loading recommender encoders and scaler...")
        try:
            self.user_encoder = joblib.load(os.path.join(self.models_dir, "user_encoder_full.pkl"))
            self.movie_encoder = joblib.load(os.path.join(self.models_dir, "movie_encoder_full.pkl"))
            self.scaler = joblib.load(os.path.join(self.models_dir, "scaler_full.pkl"))
        except Exception as e:
            print(f"Error loading encoders/scaler: {e}")
            
        print("Loading training watch history...")
        try:
            if os.path.exists(self.data_path):
                self.watch_history_df = pd.read_parquet(self.data_path, columns=["user_id", "movie_id"])
            else:
                print(f"Watch history not found at {self.data_path}, initializing empty")
                self.watch_history_df = pd.DataFrame(columns=["user_id", "movie_id"])
        except Exception as e:
            print(f"Error loading watch history: {e}")
            self.watch_history_df = pd.DataFrame(columns=["user_id", "movie_id"])
            
        print("Loading Keras DeepFM model (deepfm_full.keras)...")
        try:
            model_path = os.path.join(self.models_dir, "deepfm_full.keras")
            if not os.path.exists(model_path):
                model_path = os.path.join(self.models_dir, "deepfm_model.keras")
                
            self.model = load_model(model_path, custom_objects={"FMLayer": FMLayer})
            print("Model loaded successfully")
            
            # Extract movie embeddings
            try:
                self.movie_embeddings = self.model.get_layer("embedding_1").get_weights()[0]
                print(f"Extracted movie embeddings with shape {self.movie_embeddings.shape}")
            except Exception as emb_err:
                print(f"Could not extract embeddings from model layer: {emb_err}")
        except Exception as e:
            print(f"Error loading model: {e}")
            
        self.loaded = True

    def _mmr_rerank(self, candidates_df, top_n, lambda_param=0.6, seed=None, relevance_col="pred_rating"):
        """
        Maximal Marginal Relevance re-ranking to balance relevance and diversity.
        Uses movie embeddings for similarity when available, falls back to year-based.
        """
        if len(candidates_df) <= top_n:
            return candidates_df.copy()

        pool = candidates_df.copy().reset_index(drop=True)
        n = len(pool)

        if self.movie_embeddings is not None and "movie_idx" in pool.columns:
            idx_min, idx_max = int(pool["movie_idx"].min()), int(pool["movie_idx"].max())
            if idx_min >= 0 and idx_max < len(self.movie_embeddings):
                emb = self.movie_embeddings[pool["movie_idx"].values.astype(int)]
                norms = np.linalg.norm(emb, axis=1, keepdims=True)
                norms[norms == 0] = 1e-9
                sim_matrix = np.dot(emb, emb.T) / (norms * norms.T + 1e-9)
                sim_matrix = np.clip(sim_matrix, 0.0, 1.0)
            else:
                sim_matrix = self._fallback_sim_matrix(pool)
        else:
            sim_matrix = self._fallback_sim_matrix(pool)

        relevance = pool[relevance_col].values
        if seed is not None:
            rng = np.random.RandomState(seed)
            noise = rng.normal(0, 0.25 * (relevance.max() - relevance.min() + 1e-6), size=n)
            relevance = relevance + noise

        selected_mask = np.zeros(n, dtype=bool)
        remaining = np.ones(n, dtype=bool)

        for _ in range(min(top_n, n)):
            mmr_scores = np.full(n, -np.inf)
            for i in np.where(remaining)[0]:
                rel = relevance[i]
                if selected_mask.any():
                    div = float(sim_matrix[i, selected_mask].max())
                else:
                    div = 0.0
                mmr_scores[i] = lambda_param * rel - (1.0 - lambda_param) * div

            best = int(np.argmax(mmr_scores))
            selected_mask[best] = True
            remaining[best] = False

        return pool[selected_mask].copy().reset_index(drop=True)

    def _fallback_sim_matrix(self, pool):
        """Build similarity matrix from year when embeddings unavailable."""
        years = pool["year"].values.astype(float)
        year_diff = np.abs(years[:, None] - years[None, :])
        sim = 1.0 - (year_diff / 50.0)
        return np.clip(sim, 0.0, 1.0)

    @lru_cache(maxsize=128)
    def recommend_for_user(self, user_id, top_n=10, movie_metadata_file="movie_metadata.parquet"):
        """
        Generates predictions and recommendations for a given User ID.
        Uses caching to avoid redundant model runs.
        """
        if not self.loaded:
            self.load_resources()
            
        movie_df = pd.read_parquet(movie_metadata_file)
        
        is_cold_start = True
        if self.user_encoder is not None and user_id in self.user_encoder.classes_:
            is_cold_start = False
            
        if is_cold_start:
            movie_df["popularity_score"] = movie_df["avg_rating"] * np.log1p(movie_df["rating_count"])
            pool_size = min(300, len(movie_df))
            pool = movie_df.sort_values(by="popularity_score", ascending=False).head(pool_size).copy()
            pool["pred_rating"] = pool["popularity_score"] / (pool["popularity_score"].max() + 1e-9)
            if self.movie_encoder is not None:
                valid = pool["movie_id"].isin(self.movie_encoder.classes_)
                pool = pool[valid].copy()
                if not pool.empty:
                    pool["movie_idx"] = self.movie_encoder.transform(pool["movie_id"])
            diverse = self._mmr_rerank(pool, top_n, lambda_param=0.25, seed=int(user_id))
            diverse["rank"] = range(1, len(diverse) + 1)
            diverse["match_score"] = 0.85
            diverse["pred_rating"] = diverse["avg_rating"]
            diverse["reason"] = diverse.apply(
                lambda x: "Popular choice recommended for new profiles" if x["rating_count"] > 5000 
                else "Highly rated classic choice", axis=1
            )
            return diverse, True
            
        user_idx = self.user_encoder.transform([user_id])[0]
        
        watched_movies = set()
        if self.watch_history_df is not None and not self.watch_history_df.empty:
            watched_movies = set(self.watch_history_df.loc[self.watch_history_df["user_id"] == user_id, "movie_id"])
            
        candidates = movie_df[~movie_df["movie_id"].isin(watched_movies)].copy()
        if candidates.empty:
            candidates = movie_df.copy()
            
        if self.movie_encoder is not None:
            valid_movies_mask = candidates["movie_id"].isin(self.movie_encoder.classes_)
            candidates = candidates[valid_movies_mask].copy()
            if candidates.empty:
                candidates = movie_df.copy()
                
            candidates["movie_idx"] = self.movie_encoder.transform(candidates["movie_id"])
            
        candidates["user_idx"] = user_idx

        # Drop rows with NaN features (e.g., missing year)
        candidates = candidates.dropna(subset=["year", "avg_rating", "rating_count"]).copy()
        if candidates.empty:
            candidates = movie_df.copy()
            candidates["movie_idx"] = self.movie_encoder.transform(candidates["movie_id"])
            candidates["user_idx"] = user_idx
        
        scale_df = pd.DataFrame({
            "year_x": candidates["year"],
            "movie_avg_rating": candidates["avg_rating"],
            "movie_rating_count": candidates["rating_count"]
        })
        
        scaled_features = self.scaler.transform(scale_df)
        
        if self.model is not None:
            user_input = np.array(candidates["user_idx"]).reshape(-1, 1)
            movie_input = np.array(candidates["movie_idx"]).reshape(-1, 1)
            
            predictions = self.model.predict(
                [user_input, movie_input, scaled_features],
                batch_size=4096,
                verbose=0
            ).flatten()
            
            candidates["pred_rating"] = predictions
        else:
            candidates["pred_rating"] = candidates["avg_rating"]
            
        # Drop any NaN predictions
        candidates = candidates.dropna(subset=["pred_rating"]).copy()
        if candidates.empty:
            candidates = movie_df.copy()
            candidates["pred_rating"] = candidates["avg_rating"] / 5.0

        # Compute user-specific surprise: pred_rating vs expected (avg_rating/5)
        # This is user-specific because pred_rating varies per user
        candidates["global_expect"] = candidates["avg_rating"] / 5.0
        candidates["user_surprise"] = candidates["pred_rating"] - candidates["global_expect"]

        # Quality gate: consider movies above ~25th percentile by avg_rating
        min_quality = candidates["avg_rating"].quantile(0.25)
        quality_pool = candidates[candidates["avg_rating"] >= min_quality].copy()
        if len(quality_pool) < top_n * 20:
            quality_pool = candidates.copy()

        # Use raw user_surprise as primary ranking signal (breaks 96% prediction correlation)
        pool = quality_pool.nlargest(600, "user_surprise").copy()
        diverse = self._mmr_rerank(pool, top_n, lambda_param=0.15, seed=int(user_id), relevance_col="pred_rating")
        diverse["rank"] = range(1, len(diverse) + 1)
        
        if self.model is not None:
            raw_probs = diverse["pred_rating"].copy()
            diverse["match_score"] = np.clip(raw_probs, 0.0, 1.0)
            diverse["pred_rating"] = 1.0 + 4.0 * raw_probs
        else:
            diverse["match_score"] = np.clip((diverse["pred_rating"] - 1.0) / 4.0, 0.0, 1.0)
        
        def get_explanation(row):
            score = row["match_score"]
            avg_r = row["avg_rating"]
            count = row["rating_count"]
            if score >= 0.85:
                return "Highly matched with your implicit interest fingerprint."
            elif avg_r >= 4.3:
                return "Highly rated globally matching your cinematic quality standard."
            elif count >= 3000:
                return "Popular blockbuster choice aligning with your content depth."
            return "Recommended based on similar user preferences."
            
        diverse["reason"] = diverse.apply(get_explanation, axis=1)
        
        return diverse, False

    def get_recommendation_analytics(self, recommendations_df):
        """
        Calculates confidence, diversity, and personalization metrics.
        """
        if recommendations_df.empty:
            return {
                "avg_confidence": 0.0,
                "diversity_score": 0.0,
                "personalization_score": 0.0
            }
            
        avg_confidence = float(recommendations_df["match_score"].mean() * 100)
        
        years = recommendations_df["year"].dropna()
        if len(years) > 1:
            year_std = float(years.std())
            diversity = min(100.0, (year_std / 20.0) * 100.0)
        else:
            diversity = 50.0
            
        if "pred_rating" in recommendations_df.columns:
            rating_diffs = np.abs(recommendations_df["pred_rating"] - recommendations_df["avg_rating"])
            personalization = min(100.0, float(rating_diffs.mean() * 60.0 + 40.0))
        else:
            personalization = 50.0
            
        return {
            "avg_confidence": avg_confidence,
            "diversity_score": diversity,
            "personalization_score": personalization
        }

    @lru_cache(maxsize=128)
    def get_similar_movies(self, movie_id, top_n=10, movie_metadata_file="movie_metadata.parquet"):
        """
        Retrieves top N similar movies using cosine similarity on DeepFM embeddings.
        """
        if not self.loaded:
            self.load_resources()
            
        movie_df = pd.read_parquet(movie_metadata_file)
        
        is_known = False
        if (self.movie_encoder is not None 
                and self.movie_embeddings is not None 
                and movie_id in self.movie_encoder.classes_):
            is_known = True
            
        if not is_known:
            print(f"Fallback Similarity lookup for movie ID: {movie_id}")
            match_row = movie_df[movie_df["movie_id"] == movie_id]
            if not match_row.empty:
                ref_year = match_row.iloc[0]["year"]
                decade_start = (ref_year // 10) * 10
                candidates = movie_df[
                    (movie_df["year"] >= decade_start) & 
                    (movie_df["year"] < decade_start + 10) & 
                    (movie_df["movie_id"] != movie_id)
                ].copy()
                candidates["similarity_score"] = 0.70 + 0.15 * (candidates["avg_rating"] / 5.0)
                recs = candidates.sort_values(by="avg_rating", ascending=False).head(top_n)
                return recs
            else:
                candidates = movie_df.copy()
                candidates["similarity_score"] = 0.60
                return candidates.sort_values(by="rating_count", ascending=False).head(top_n)
                
        movie_idx = self.movie_encoder.transform([movie_id])[0]
        movie_vector = self.movie_embeddings[movie_idx]
        
        dot_products = np.dot(self.movie_embeddings, movie_vector)
        norms = np.linalg.norm(self.movie_embeddings, axis=1)
        ref_norm = np.linalg.norm(movie_vector)
        
        if ref_norm == 0:
            similarities = np.zeros(len(self.movie_embeddings))
        else:
            similarities = dot_products / (norms * ref_norm + 1e-9)
            
        sim_df = pd.DataFrame({
            "movie_idx": range(len(self.movie_embeddings)),
            "similarity_score": similarities
        })
        
        sim_df["movie_id"] = self.movie_encoder.inverse_transform(sim_df["movie_idx"])
        sim_df = sim_df[sim_df["movie_id"] != movie_id]
        
        merged = pd.merge(sim_df, movie_df, on="movie_id")
        top_recs = merged.sort_values(by="similarity_score", ascending=False).head(top_n).copy()
        top_recs["similarity_score"] = np.clip(top_recs["similarity_score"], 0.0, 1.0)
        
        return top_recs

    @lru_cache(maxsize=64)
    def get_movie_detailed_profile(self, movie_id, ratings_file="Processed/ratings_filtered.parquet"):
        """
        Loads all raw ratings for a movie using parquet filters,
        calculates percentiles, rating distributions, timelines, and generates semantic summaries.
        """
        # Load catalog for percentiles
        movie_df = pd.read_parquet("movie_metadata.parquet")
        match_row = movie_df[movie_df["movie_id"] == movie_id]
        
        if match_row.empty:
            return None
            
        ref_movie = match_row.iloc[0]
        title = str(ref_movie["title"])
        year = int(ref_movie["year"])
        avg_rating = float(ref_movie["avg_rating"])
        rating_count = int(ref_movie["rating_count"])
        
        # Calculate global rating percentile
        percentile = float((movie_df["avg_rating"] < avg_rating).mean() * 100)
        
        # Query raw ratings for detailed metrics
        ratings_df = pd.DataFrame(columns=["rating", "date"])
        rating_std = 1.0
        
        try:
            if os.path.exists(ratings_file):
                # Using parquet filter for efficient query
                ratings_df = pd.read_parquet(ratings_file, columns=["rating", "date"], filters=[("movie_id", "==", movie_id)])
                if not ratings_df.empty:
                    rating_std = float(ratings_df["rating"].std())
        except Exception as e:
            print(f"Error querying ratings timeline for movie {movie_id}: {e}")
            
        # Generate semantic AI-style movie summary
        decade_str = f"released in {year}" if year > 0 else "from our catalog"
        if 1990 <= year < 2000:
            decade_str = "released during the iconic 1990s era of filmmaking"
        elif 1980 <= year < 1990:
            decade_str = "released during the nostalgic 1980s cinematic wave"
        elif 2000 <= year < 2010:
            decade_str = "released in the early 2000s transition period of cinema"
            
        if avg_rating >= 4.5:
            quality = "widely regarded as a cinematic masterpiece of exceptional caliber"
        elif avg_rating >= 4.0:
            quality = "recognized as a highly polished, critically acclaimed production"
        elif avg_rating >= 3.5:
            quality = "offering a solid, engaging narrative with general audience appeal"
        else:
            quality = "a lighter, casual watch with mixed viewer receptions"
            
        if rating_std > 1.2:
            consensus = "a polarizing work that continues to spark passionate debates among film critics and enthusiasts alike"
        elif rating_std < 0.8:
            consensus = "a universally praised entry with a highly cohesive consensus and solid backing across all user demographics"
        else:
            consensus = "a stable favorite that maintains consistent appeal and strong, standard ratings"
            
        if rating_count >= 10000:
            popularity = "As a massive blockbuster sensation, it has amassed an enormous viewer base and high cultural footprint."
        elif rating_count >= 2000:
            popularity = "With a healthy, popular following, it remains a common recommendation in modern streaming circles."
        else:
            popularity = "Operating as a hidden gem, it appeals greatly to niche enthusiasts looking for unique narrative directions."
            
        summary_text = f"<strong>\"{title}\"</strong> is {quality}, {decade_str}. It is valued as {consensus}. {popularity}"
        
        # Process popularity timeline: group/sort by date to calculate cumulative votes
        timeline_data = pd.DataFrame(columns=["date", "cumulative_votes"])
        if not ratings_df.empty:
            ratings_df["date"] = pd.to_datetime(ratings_df["date"])
            sorted_ratings = ratings_df.sort_values(by="date").copy()
            sorted_ratings["cumulative_votes"] = range(1, len(sorted_ratings) + 1)
            
            # Sub-sample to keep Plotly line charts light
            if len(sorted_ratings) > 500:
                indices = np.linspace(0, len(sorted_ratings) - 1, 500, dtype=int)
                timeline_data = sorted_ratings.iloc[indices][["date", "cumulative_votes"]].copy()
            else:
                timeline_data = sorted_ratings[["date", "cumulative_votes"]].copy()
                
        # Process rating counts for histogram
        dist_counts = {str(i): 0 for i in range(1, 6)}
        if not ratings_df.empty:
            freqs = ratings_df["rating"].value_counts().to_dict()
            for star, val in freqs.items():
                dist_counts[str(int(star))] = int(val)
                
        return {
            "title": title,
            "year": year,
            "avg_rating": avg_rating,
            "rating_count": rating_count,
            "percentile": percentile,
            "summary": summary_text,
            "timeline": timeline_data,
            "distribution": dist_counts,
            "std": rating_std
        }