""" Scalable recommender core for BookRec on the FULL Goodbooks-10k dataset. We never build a dense item-item similarity matrix. Instead we use the standard production pattern: * Collaborative filtering : truncated SVD (scipy.sparse.linalg.svds) of the sparse user x item rating matrix -> low-dimensional item factors. "Books similar to X" = X's factor vector's nearest neighbours (ANN via sklearn NearestNeighbors / brute-force dot product). All CPU-feasible. * Popularity : rank the full catalogue by explicit-rating count and average rating (a strong baseline). * Content / search : the resilient encoders (TF-IDF default, or sentence-transformer) over book text (title + authors + genres) -> top-K by similarity. * Hybrid : combine the CF factor similarity and content similarity for a tuned 'more like this' list. All components degrade gracefully (empty catalogue, no factors, offline) so the app always renders something. """ from __future__ import annotations from typing import List, Optional, Sequence, Tuple import numpy as np import pandas as pd import scipy.sparse as sp from scipy.sparse.linalg import svds try: from . import loader from .encoders import build_content_encoder, build_item_matrix, cosine_similarity_vector except ImportError: # allow running as a script with src on sys.path import loader from encoders import build_content_encoder, build_item_matrix, cosine_similarity_vector # --------------------------------------------------------------------------- # Ratings matrix + factorization # --------------------------------------------------------------------------- def build_user_item_matrix(ratings: pd.DataFrame) -> Tuple[sp.csr_matrix, np.ndarray, np.ndarray]: """Return (user_item csr, user_ids, item_ids) for a ratings frame. Assumes the frame already holds the ratings we want to factor (e.g. explicit). Matrix is users x items, values = Book-Rating. """ ratings = ratings.copy() ratings["User-ID"] = ratings["User-ID"].astype(str) ratings["ISBN"] = ratings["ISBN"].astype(str) users, ui = np.unique(ratings["User-ID"], return_inverse=True) items, ii = np.unique(ratings["ISBN"], return_inverse=True) data = ratings["Book-Rating"].astype(np.float32).values mat = sp.csr_matrix((data, (ui, ii)), shape=(len(users), len(items))) return mat, users, items def factorize(mat: sp.csr_matrix, rank: int = 50, seed: int = 0) -> Tuple[np.ndarray, np.ndarray]: """Truncated SVD of a sparse user-item matrix. Returns (user_factors, item_factors), both (n, rank), where item_factors[i] is the latent vector of item i and can be used directly for nearest-neighbour "similar items". """ # Centre by subtracting the global mean of observed entries (helps SVD). nnz = mat.nnz if nnz == 0: return np.zeros((mat.shape[0], rank)), np.zeros((mat.shape[1], rank)) mu = mat.data.mean() m_c = mat.copy() m_c.data = m_c.data - mu rank = min(rank, min(mat.shape) - 1) if rank < 1: return np.zeros((mat.shape[0], 1)), np.zeros((mat.shape[1], 1)) try: u, s, vt = svds(m_c, k=rank, which="LM", random_state=seed) except Exception: # noqa: BLE001 -- fall back to uncentred u, s, vt = svds(mat, k=rank, which="LM", random_state=seed) # Reorder descending (svds returns ascending) order = np.argsort(s)[::-1] s, u, vt = s[order], u[:, order], vt[order, :] user_factors = u * s[np.newaxis, :] # (n_users, rank) item_factors = vt.T # (n_items, rank) return user_factors, item_factors # --------------------------------------------------------------------------- # Popularity # --------------------------------------------------------------------------- def popular_books(ratings: pd.DataFrame, books: pd.DataFrame, top_n: int = 50, min_ratings: int = 5) -> pd.DataFrame: """Rank the most-popular books from explicit ratings (count & avg).""" df = ratings[ratings["Book-Rating"] > 0].copy() df["ISBN"] = df["ISBN"].astype(str) stats = df.groupby("ISBN")["Book-Rating"].agg(["count", "mean"]).reset_index() stats.columns = ["ISBN", "num_ratings", "avg_rating"] stats = stats[stats["num_ratings"] >= min_ratings] stats = stats.sort_values(["avg_rating", "num_ratings"], ascending=False).head(top_n) m = books.copy() m["ISBN"] = m["ISBN"].astype(str) out = stats.merge(m.drop_duplicates("ISBN"), on="ISBN", how="left") out["cover"] = out.apply(lambda r: loader.pick_cover_url(r), axis=1) return out.reset_index(drop=True) # --------------------------------------------------------------------------- # ANN neighbours (sklearn, no extra deps) # --------------------------------------------------------------------------- class ApproxNN: """Tiny wrapper around scikit-learn NearestNeighbors. Keeps the factor/search index in memory and returns top-K indices. """ def __init__(self, X: np.ndarray, n_neighbors: int = 50, metric: str = "cosine"): from sklearn.neighbors import NearestNeighbors self.X = X self.metric = metric self.knn = NearestNeighbors(n_neighbors=min(n_neighbors, X.shape[0]), metric=metric, algorithm="auto", n_jobs=-1) self.knn.fit(X) def query(self, vec: np.ndarray, k: int = 10) -> Tuple[np.ndarray, np.ndarray]: """Return (distances, indices) of the k closest rows to `vec` (excl. self).""" k = min(k + 1, self.X.shape[0]) dists, idx = self.knn.kneighbors(np.asarray(vec).reshape(1, -1) if vec.ndim == 1 else vec, n_neighbors=k) return dists, idx # --------------------------------------------------------------------------- # Recommender # --------------------------------------------------------------------------- def fused_ranking(content_ranked, cf_ranked, alpha: float = 0.5, n: int = 10) -> list: """Blend two ordered lists by reciprocal rank, matching the reference notebook's hybrid fusion.""" from collections import defaultdict score: dict = defaultdict(float) for rank, it in enumerate(content_ranked): score[it] += alpha / (rank + 1.0) for rank, it in enumerate(cf_ranked): score[it] += (1.0 - alpha) / (rank + 1.0) ranked = sorted(score.items(), key=lambda kv: -kv[1]) return [it for it, _ in ranked[:n]] class BookRecommender: """Serves popularity, search, content-similarity and CF-similarity. The content encoder is resilient: defaults to TF-IDF, optionally uses a sentence-transformer, and never crashes if the model is unavailable. """ def __init__(self, books: pd.DataFrame, ratings: pd.DataFrame, users: pd.DataFrame | None = None, encoder_kind: Optional[str] = None, model_name: str = "sentence-transformers/all-MiniLM-L6-v2", svd_rank: int = 50, text_col: str = "Book-Title"): self.books = books.reset_index(drop=True) self.books["ISBN"] = self.books["ISBN"].astype(str) self.ratings = ratings self.users = users self.encoder_kind = encoder_kind or "tfidf" self.svd_rank = svd_rank # full-catalogue text: title + author (+ genres if present) parts = [self.books.get(col, pd.Series(index=self.books.index, dtype=str)) for col in ("Book-Title", "Book-Author", "Book-Genres")] self.texts = pd.concat(parts, axis=1).fillna("").agg(" ".join, axis=1) # --- content index (search + similar-by-text) ----------------------- print(f"[recommender] building content index ({self.encoder_kind}) ...") self.content_enc = build_content_encoder(encoder_kind, verbose=True) self.content_mat = self.content_enc.build_matrix(self.texts.tolist()) self.content_enc.matrix_ = self.content_mat # ensure cached # --- CF factorization ------------------------------------------------ print("[recommender] building user-item matrix ...") self.mat, self.user_ids, self.item_ids = build_user_item_matrix(ratings) self.item_pos = {isbn: i for i, isbn in enumerate(self.item_ids)} self.item_isbn_at = self.item_ids.copy() self.user_pos = {u: i for i, u in enumerate(self.user_ids)} print(f"[recommender] factorizing (rank={svd_rank}) ...") self.user_factors, self.item_factors = factorize(self.mat, rank=svd_rank) self.cf_nn = ApproxNN(self.item_factors) if self.item_factors.shape[0] else None # item -> row index in self.books self.isbn_to_row = {row.ISBN: i for i, row in self.books.iterrows()} # ---- helpers ----------------------------------------------------------- def row_for_title(self, title: str) -> Optional[int]: d = self.books[self.books["Book-Title"].str.strip().str.lower() == str(title).strip().lower()] return int(d.index[0]) if len(d) else None # ---- search ------------------------------------------------------------- def search(self, query: str, top_k: int = 10) -> pd.DataFrame: res = self.content_enc.search(query, top_k=top_k) rows = [] for idx, score in res: r = self.books.iloc[idx] rows.append({ "rank": len(rows) + 1, "title": r["Book-Title"], "author": r.get("Book-Author", ""), "isbn": r["ISBN"], "cover": loader.pick_cover_url(r), "score": float(score), }) return pd.DataFrame(rows) # ---- similar items (hybrid of content + CF) ------------------------------ def similar(self, title: str, top_k: int = 0, hybrid_alpha: float = 0.5, limit: int = 10, fusion: str = "score") -> pd.DataFrame: """Return similar books to a title, blending content & CF similarity. hybrid_alpha in [0,1]: 1 = pure content, 0 = pure CF. fusion is 'score' (weighted score sum) or 'reciprocal_rank' (reciprocal-rank fusion). If a factor path is unavailable, falls back to content. """ row = self.row_for_title(title) if row is None: return pd.DataFrame() isbn = self.books.iloc[row]["ISBN"] # content neighbours (text) - cosine similarity, higher is better q = self.content_enc.encode([self.texts[row]])[0] content_ids = self._nearest_content(q, limit * 5) content_scores = {int(i): float(s) for i, s in content_ids.items()} content_scores.pop(row, None) # exclude the seed book itself # CF neighbours (latent factor) -- only if the ISBN is in the rating set # score = 1/(1 + cosine distance), so higher is better in (0, 1] cf_scores: dict = {} if self.cf_nn is not None and isbn in self.item_pos: f = self.item_factors[self.item_pos[isbn]] dists, idx = self.cf_nn.query(f, k=limit * 5 + 1) for d, it in zip(dists[0], idx[0]): if it == self.item_pos[isbn]: continue iid = self.item_ids[it] if iid in self.isbn_to_row: cf_scores[self.isbn_to_row[iid]] = 1.0 / (1.0 + d) # Min-max normalise each source to [0, 1] so the alpha blend compares # content and CF on a fair scale. def _norm(src: dict) -> dict: if not src: return {} lo, hi = min(src.values()), max(src.values()) span = hi - lo if span <= 1e-9: return {k: 1.0 for k in src} return {k: (v - lo) / span for k, v in src.items()} content_scores = _norm(content_scores) cf_scores = _norm(cf_scores) # fuse if fusion == "reciprocal_rank": content_ranked = sorted(content_scores, key=content_scores.get, reverse=True) cf_ranked = sorted(cf_scores, key=cf_scores.get, reverse=True) picked = fused_ranking(content_ranked, cf_ranked, alpha=hybrid_alpha, n=limit) fused = {i: float(len(picked) - j) for j, i in enumerate(picked)} fused = sorted(fused.items(), key=lambda kv: -kv[1]) else: all_idx = set(content_scores) | set(cf_scores) fused = {} for i in all_idx: c = content_scores.get(i, 0.0) cf = cf_scores.get(i, 0.0) fused[i] = hybrid_alpha * c + (1 - hybrid_alpha) * cf fused = sorted(fused.items(), key=lambda kv: -kv[1])[:limit] out = [] for idx, score in fused: r = self.books.iloc[idx] out.append({"rank": len(out) + 1, "title": r["Book-Title"], "author": r.get("Book-Author", ""), "isbn": r["ISBN"], "cover": loader.pick_cover_url(r), "score": float(score)}) return pd.DataFrame(out) def _nearest_content(self, q: np.ndarray, k: int) -> dict: sims = cosine_similarity_vector(q, self.content_mat) order = np.argsort(-sims)[:k] return {int(i): float(sims[i]) for i in order} # ---- content-based user profile ---------------------------------------- def recommend_content_for_user(self, liked_isbns, k: int = 10, exclude_known: bool = True) -> pd.DataFrame: """Content-based recommendation for a user from their liked books. Averages the content vectors of the books a user liked into a user profile, then ranks the catalogue by text proximity to that profile. This is the content-based 'recommend for user' path (no other users needed), the natural first responder for a reader with no neighbours. """ liked = [str(x) for x in liked_isbns if str(x) in self.isbn_to_row] if not liked: return pd.DataFrame() rows = [self.isbn_to_row[x] for x in liked] profile = np.asarray(self.content_mat[rows].mean(axis=0)).ravel() sims = cosine_similarity_vector(profile, self.content_mat) know = set(rows) out = [] for i in np.argsort(-sims): if exclude_known and int(i) in know: continue if sims[i] <= 0: break r = self.books.iloc[i] out.append({"rank": len(out) + 1, "title": r["Book-Title"], "author": r.get("Book-Author", ""), "isbn": r["ISBN"], "cover": loader.pick_cover_url(r), "score": float(sims[i])}) if len(out) >= k: break return pd.DataFrame(out) # ---- cold start --------------------------------------------------------- def cold_start_for_new_user(self, k: int = 10, min_ratings: int = 20) -> pd.DataFrame: """A brand-new reader has no history. Start them with popular, well rated books (the bestseller shelf).""" return popular_books(self.ratings, self.books, top_n=k, min_ratings=min_ratings) def cold_start_for_new_item(self, title_text: str, k: int = 10) -> pd.DataFrame: """A brand-new book has no ratings. Surface it next to the closest known books by text, boosting the popular ones among them, so it does not vanish into the long tail.""" q = self.content_enc.encode([str(title_text)])[0] sims = cosine_similarity_vector(q, self.content_mat) counts = self.ratings["ISBN"].value_counts() order = np.argsort(-sims) scored = [] for i in order: if sims[i] <= 0: break isbn = str(self.books.iloc[i]["ISBN"]) scored.append((int(i), float(sims[i]), int(counts.get(isbn, 0)))) scored.sort(key=lambda t: (t[2] > 0, t[1]), reverse=True) out = [] for i, score, pop in scored[:k]: r = self.books.iloc[i] out.append({"rank": len(out) + 1, "title": r["Book-Title"], "author": r.get("Book-Author", ""), "isbn": r["ISBN"], "cover": loader.pick_cover_url(r), "score": float(score), "n_ratings": pop}) return pd.DataFrame(out)