from __future__ import annotations import copy from collections import Counter import numpy as np import pandas as pd from scipy.sparse import csr_matrix from .config import RuntimeSettings from .mal import MalClient from .training import interaction_weight, normalize_status def _anime_lookup(dataframe: pd.DataFrame) -> dict[str, dict]: lookup_df = dataframe.copy() lookup_df["anime_id"] = lookup_df["anime_id"].astype(str) return lookup_df.set_index("anime_id").to_dict("index") # Number of "Because you watched ___" anchors to surface per recommendation. ANCHOR_COUNT = 2 def _compute_anchors( model, item_id_map: dict, user_entries: list[dict], *, top_k: int = ANCHOR_COUNT, ) -> dict[str, list[int]]: """For every item the model knows, return the user's top-k most similar watched anime by cosine similarity over LightFM item embeddings. Anchors are restricted to high-affinity user entries (score >= 7 if the user scores anime, otherwise any completed/watching entry) so that "because you watched X" only cites things the user actually liked. """ high_affinity = [ entry for entry in user_entries if normalize_status(entry.get("status", "")) in {"completed", "watching"} and int(entry.get("score") or 0) >= 7 ] if not high_affinity: high_affinity = [ entry for entry in user_entries if normalize_status(entry.get("status", "")) in {"completed", "watching"} ] if not high_affinity: return {} anchor_internal_ids: list[int] = [] anchor_anime_ids: list[int] = [] for entry in high_affinity: anime_id_str = str(entry["anime_id"]) internal_id = item_id_map.get(anime_id_str) if internal_id is None: continue anchor_internal_ids.append(int(internal_id)) anchor_anime_ids.append(int(entry["anime_id"])) if not anchor_internal_ids: return {} item_embeddings = np.asarray(model.item_embeddings, dtype=np.float32) norms = np.linalg.norm(item_embeddings, axis=1, keepdims=True) norms[norms == 0] = 1.0 item_emb_norm = item_embeddings / norms anchor_emb = item_emb_norm[anchor_internal_ids] sim = item_emb_norm @ anchor_emb.T # (num_items, num_anchors) # Hide each anchor from itself so an item never anchors to itself. for col, internal_id in enumerate(anchor_internal_ids): sim[internal_id, col] = -np.inf effective_k = min(top_k, sim.shape[1]) if effective_k == sim.shape[1]: order = np.argsort(-sim, axis=1) top_indices = order[:, :effective_k] else: partitioned = np.argpartition(-sim, effective_k, axis=1)[:, :effective_k] top_scores = np.take_along_axis(sim, partitioned, axis=1) order = np.argsort(-top_scores, axis=1) top_indices = np.take_along_axis(partitioned, order, axis=1) internal_to_original = {v: k for k, v in item_id_map.items()} anchors_map: dict[str, list[int]] = {} for internal_id in range(item_emb_norm.shape[0]): anime_id_str = internal_to_original.get(internal_id) if anime_id_str is None: continue picks = [ anchor_anime_ids[col] for col in top_indices[internal_id] if np.isfinite(sim[internal_id, col]) ] if picks: anchors_map[str(anime_id_str)] = picks return anchors_map def get_user_anime_status(username: str) -> dict[int, str]: client = MalClient.from_env() statuses: dict[int, str] = {} for entry in client.iter_user_anime_list(username, fields=["list_status"]): anime = entry.get("node", {}) list_status = entry.get("list_status", {}) anime_id = anime.get("id") status = list_status.get("status") if anime_id is not None and status: statuses[int(anime_id)] = str(status) return statuses def fetch_recs_from_filters( item_score_pairs_sorted, df, dataset, filters, page, page_size, anchors_by_anime_id: dict | None = None, ): _, _, item_id_map, _ = dataset.mapping() internal_to_original_anime_id = {v: k for k, v in item_id_map.items()} anime_data_lookup = _anime_lookup(df) anchors_by_anime_id = anchors_by_anime_id or {} include_plan_to_watch = filters.get("include_plan_to_watch", False) plan_to_watch_anime_ids = { str(anime_id) for anime_id in filters.get("plan_to_watch_anime_ids", []) } filtered_recommendations = [] for internal_id, score in item_score_pairs_sorted: original_anime_id = internal_to_original_anime_id.get(internal_id) if original_anime_id is None: continue if not include_plan_to_watch and str(original_anime_id) in plan_to_watch_anime_ids: continue anime_data = anime_data_lookup.get(str(original_anime_id)) if not anime_data: continue selected_genres = filters.get("genres", []) if selected_genres: anime_genres = { genre.strip() for genre in str(anime_data.get("genres", "")).split(",") if genre.strip() } if not all(genre in anime_genres for genre in selected_genres): continue selected_media_types = filters.get("media_types", []) if selected_media_types: media_type = str(anime_data.get("media_type", "")).lower() if media_type not in selected_media_types: continue min_users = filters.get("min_users", 0) max_users = filters.get("max_users", 4_200_000) num_list_users = int(anime_data.get("num_list_users", 0) or 0) if not (min_users <= num_list_users <= max_users): continue if filters.get("filter_sequels", False): if anime_data.get("relationship", "") != "s1": continue filtered_recommendations.append( { "anime_id": int(original_anime_id), "title": anime_data.get("title", "Unknown"), "score": float(score), "num_list_users": num_list_users, "mean": float(anime_data.get("mean", 0.0) or 0.0), "genres": anime_data.get("genres", ""), "synopsis": anime_data.get("synopsis", ""), "image_url": anime_data.get("image_url", ""), "media_type": anime_data.get("media_type", ""), "anchors": anchors_by_anime_id.get(str(original_anime_id), []), } ) total_filtered_count = len(filtered_recommendations) start_index = max(page - 1, 0) * page_size end_index = start_index + page_size return filtered_recommendations[start_index:end_index], total_filtered_count def _build_runtime_user_entries(username: str) -> list[dict]: client = MalClient.from_env() entries = [] fields = [ "list_status", "genres", "start_season", "media_type", "main_picture", ] for entry in client.iter_user_anime_list(username, fields=fields): anime = entry.get("node", {}) list_status = entry.get("list_status", {}) start_season = anime.get("start_season", {}) or {} main_picture = anime.get("main_picture", {}) or {} anime_id = anime.get("id") if anime_id is None: continue entries.append( { "username": username, "anime_id": int(anime_id), "title": anime.get("title", ""), "status": normalize_status(list_status.get("status", "")), "score": int(list_status.get("score", 0) or 0), "genres": anime.get("genres", []), "start_date": list_status.get("start_date"), "finish_date": list_status.get("finish_date"), "start_season_year": start_season.get("year"), "start_season_season": start_season.get("season"), "media_type": anime.get("media_type"), "image_url": main_picture.get("large") or main_picture.get("medium"), } ) return entries def _build_genre_preferences(user_entries: list[dict]) -> dict[str, float]: demographics = set(RuntimeSettings().demographic_genres) genre_counts: Counter[str] = Counter() total_anime = len(user_entries) if total_anime == 0: return {} for entry in user_entries: unique_genres = { genre_obj.get("name") for genre_obj in entry.get("genres", []) if genre_obj.get("name") and genre_obj.get("name") not in demographics } genre_counts.update(unique_genres) return { genre: count / total_anime for genre, count in genre_counts.items() } def _build_user_stats(user_entries: list[dict], genre_preferences: dict[str, float]) -> dict: score_distribution = {score: 0 for score in range(1, 11)} status_distribution = { "watching": 0, "completed": 0, "on_hold": 0, "dropped": 0, "plan_to_watch": 0, } genre_counts: Counter[str] = Counter() for entry in user_entries: for genre_obj in entry.get("genres", []): genre_name = genre_obj.get("name") if genre_name and genre_name not in RuntimeSettings().demographic_genres: genre_counts[genre_name] += 1 score = int(entry.get("score", 0) or 0) if score > 0: score_distribution[score] += 1 status = normalize_status(entry.get("status", "")) if status in status_distribution: status_distribution[status] += 1 total_anime = len(user_entries) scored_anime = sum(score_distribution.values()) completed_anime = status_distribution.get("completed", 0) average_score = 0.0 if scored_anime > 0: average_score = round( sum(score * count for score, count in score_distribution.items()) / scored_anime, 2, ) completion_rate = 0.0 if total_anime > 0: completion_rate = round((completed_anime / total_anime) * 100, 1) return { "genre_preferences": genre_preferences, "score_distribution": score_distribution, "status_distribution": status_distribution, "total_anime": total_anime, "scored_anime": scored_anime, "completed_anime": completed_anime, "average_score": average_score, "completion_rate": completion_rate, "genre_counts": dict(genre_counts.most_common(15)), } # How many rows to return in each compatibility section. FIGHTS_LIMIT = 10 LOVES_LIMIT = 10 FRIEND_RECS_LIMIT = 12 # Minimum |score_a - score_b| for a shared anime to count as a "fight". FIGHT_DIFF_THRESHOLD = 3 # Minimum min(score_a, score_b) for a shared anime to count as a mutual "love". LOVE_MIN_SCORE = 7 # Status + score bar for an anime in user B's list to qualify as something # they would actually recommend to user A. FRIEND_REC_MIN_SCORE = 8 def _pearson_correlation(a_scores: list[int], b_scores: list[int]) -> float | None: if len(a_scores) < 2: return None a_arr = np.asarray(a_scores, dtype=float) b_arr = np.asarray(b_scores, dtype=float) if a_arr.std() == 0 or b_arr.std() == 0: return None coef = float(np.corrcoef(a_arr, b_arr)[0, 1]) if not np.isfinite(coef): return None return coef def compute_compatibility( user_a_anime_details: list[dict], user_a_item_scores: list[list[float]], username_b: str, dataset, df: pd.DataFrame, ) -> dict: user_b_entries = _build_runtime_user_entries(username_b) if not user_b_entries: return {"error": "USER_B_NOT_FOUND"} a_by_id: dict[int, dict] = {} for entry in user_a_anime_details or []: if entry.get("id") is None: continue a_by_id[int(entry["id"])] = entry b_by_id: dict[int, dict] = {int(entry["anime_id"]): entry for entry in user_b_entries} shared_ids = set(a_by_id.keys()) & set(b_by_id.keys()) # Pearson is only defined over jointly *scored* anime (status alone isn't # an opinion). 0 means unrated on MAL. a_scores: list[int] = [] b_scores: list[int] = [] joint_scored: list[tuple[int, int, int]] = [] for anime_id in shared_ids: sa = int(a_by_id[anime_id].get("score") or 0) sb = int(b_by_id[anime_id].get("score") or 0) if sa > 0 and sb > 0: a_scores.append(sa) b_scores.append(sb) joint_scored.append((anime_id, sa, sb)) pearson = _pearson_correlation(a_scores, b_scores) similarity_percent = None if pearson is not None: similarity_percent = round((pearson + 1.0) / 2.0 * 100.0, 1) fights = sorted( [(aid, sa, sb) for aid, sa, sb in joint_scored if abs(sa - sb) >= FIGHT_DIFF_THRESHOLD], # Biggest disagreement first; among equal gaps, the one where someone # really loved it (max score) is the spicier fight. key=lambda t: (-abs(t[1] - t[2]), -max(t[1], t[2])), )[:FIGHTS_LIMIT] loves = sorted( [(aid, sa, sb) for aid, sa, sb in joint_scored if min(sa, sb) >= LOVE_MIN_SCORE], # Most aligned first, ties broken by who-loved-it-more (higher min wins, # so a mutual 10/10 ranks above a mutual 7/7). key=lambda t: (abs(t[1] - t[2]), -min(t[1], t[2])), )[:LOVES_LIMIT] # Friend-curated recs: B loved it (status active + score >= 8) and A hasn't # touched it (any status). Rank by A's predicted AI score. a_seen_ids = set(a_by_id.keys()) b_loved_ids = { int(entry["anime_id"]) for entry in user_b_entries if normalize_status(entry.get("status", "")) in {"completed", "watching"} and int(entry.get("score") or 0) >= FRIEND_REC_MIN_SCORE } candidate_ids = b_loved_ids - a_seen_ids _, _, item_id_map, _ = dataset.mapping() a_score_lookup: dict[int, float] = {} for pair in user_a_item_scores or []: if len(pair) < 2: continue a_score_lookup[int(pair[0])] = float(pair[1]) candidate_with_scores: list[tuple[int, float, int]] = [] for anime_id in candidate_ids: item_internal_id = item_id_map.get(str(anime_id)) if item_internal_id is None: continue ai_score = a_score_lookup.get(int(item_internal_id)) if ai_score is None: continue b_score = int(b_by_id[anime_id].get("score") or 0) candidate_with_scores.append((anime_id, ai_score, b_score)) candidate_with_scores.sort(key=lambda t: -t[1]) friend_recs_top = candidate_with_scores[:FRIEND_RECS_LIMIT] anime_data_lookup = _anime_lookup(df) def hydrate_pair(anime_id: int, sa: int, sb: int) -> dict: anime_data = anime_data_lookup.get(str(anime_id), {}) return { "anime_id": int(anime_id), "title": anime_data.get("title", "Unknown"), "image_url": anime_data.get("image_url", ""), "genres": anime_data.get("genres", ""), "score_a": int(sa), "score_b": int(sb), "diff": int(abs(sa - sb)), } def hydrate_friend_rec(anime_id: int, ai_score: float, b_score: int) -> dict: anime_data = anime_data_lookup.get(str(anime_id), {}) return { "anime_id": int(anime_id), "title": anime_data.get("title", "Unknown"), "image_url": anime_data.get("image_url", ""), "genres": anime_data.get("genres", ""), "media_type": anime_data.get("media_type", ""), "mean": float(anime_data.get("mean", 0.0) or 0.0), "num_list_users": int(anime_data.get("num_list_users", 0) or 0), "ai_score": float(ai_score), "friend_score": int(b_score), } return { "username_b": username_b, "similarity_pearson": pearson, "similarity_percent": similarity_percent, "shared_scored_count": len(joint_scored), "total_shared_count": len(shared_ids), "user_a_total": len(a_by_id), "user_b_total": len(b_by_id), "fights": [hydrate_pair(*t) for t in fights], "loves": [hydrate_pair(*t) for t in loves], "friend_recs": [hydrate_friend_rec(*t) for t in friend_recs_top], } def predict_scores(username, dataset, model, df): user_entries = _build_runtime_user_entries(username) if not user_entries: return [], [], {}, [] model_copy = copy.deepcopy(model) _, user_feature_map, item_id_map, _ = dataset.mapping() internal_to_original_anime_id = {v: k for k, v in item_id_map.items()} genre_preferences = _build_genre_preferences(user_entries) user_stats = _build_user_stats(user_entries, genre_preferences) user_anime_details = [ { "id": entry["anime_id"], "title": entry["title"], "score": entry["score"], "status": entry["status"], "genres": entry["genres"], "start_date": entry["start_date"], "finish_date": entry["finish_date"], "start_year": entry["start_season_year"], "start_season": entry["start_season_season"], "media_type": entry["media_type"], "image_url": entry["image_url"], } for entry in user_entries ] new_user_feature_indices = [] new_user_feature_data = [] for genre, weight in genre_preferences.items(): feature_internal_id = user_feature_map.get(genre) if feature_internal_id is None: continue new_user_feature_indices.append(feature_internal_id) new_user_feature_data.append(weight) num_total_user_features = dataset.user_features_shape()[1] new_user_features_sparse = csr_matrix( (new_user_feature_data, ([0] * len(new_user_feature_data), new_user_feature_indices)), shape=(1, num_total_user_features), ) rows = [] cols = [] values = [] for entry in user_entries: anime_id = str(entry["anime_id"]) item_internal_id = item_id_map.get(anime_id) if item_internal_id is None: continue rows.append(0) cols.append(item_internal_id) values.append(interaction_weight(entry["status"], entry["score"])) num_items_in_dataset = dataset.interactions_shape()[1] new_user_interactions = csr_matrix( (values, (rows, cols)), shape=(1, num_items_in_dataset), ) model_copy.fit_partial( interactions=new_user_interactions, user_features=new_user_features_sparse, item_features=None, epochs=10, ) all_item_internal_ids = np.arange(num_items_in_dataset) scores = model_copy.predict( user_ids=0, item_ids=all_item_internal_ids, user_features=new_user_features_sparse, item_features=None, ) if scores.size == 0: return [], [], {}, [] min_score = float(np.min(scores)) max_score = float(np.max(scores)) if max_score == min_score: normalized_scores = [0.5 for _ in scores] else: normalized_scores = [ float((score - min_score) / (max_score - min_score)) for score in scores ] excluded_item_internal_ids = { item_id_map[str(anime_id)] for anime_id in RuntimeSettings().bad_anime_ids if str(anime_id) in item_id_map } for entry in user_entries: anime_id = str(entry["anime_id"]) if anime_id in item_id_map and normalize_status(entry.get("status", "")) != "plan_to_watch": excluded_item_internal_ids.add(item_id_map[anime_id]) item_score_pairs_sorted = sorted( [ (int(item_id), float(score)) for item_id, score in zip(all_item_internal_ids, normalized_scores) if int(item_id) not in excluded_item_internal_ids ], key=lambda item: item[1], reverse=True, ) anchors_by_anime_id = _compute_anchors(model_copy, item_id_map, user_entries) anime_data_lookup = _anime_lookup(df) plan_to_watch_anime_ids = { int(entry["anime_id"]) for entry in user_entries if normalize_status(entry.get("status", "")) == "plan_to_watch" } recommendations = [] for item_internal_id, score in item_score_pairs_sorted: original_anime_id = internal_to_original_anime_id.get(item_internal_id) if int(original_anime_id) in plan_to_watch_anime_ids: continue anime_data = anime_data_lookup.get(str(original_anime_id), {}) recommendations.append( { "anime_id": int(original_anime_id), "title": anime_data.get("title", f"Unknown Anime (ID: {original_anime_id})"), "score": float(score), "num_list_users": int(anime_data.get("num_list_users", 0) or 0), "mean": float(anime_data.get("mean", 0.0) or 0.0), "genres": anime_data.get("genres", ""), "synopsis": anime_data.get("synopsis", ""), "image_url": anime_data.get("image_url", ""), "media_type": anime_data.get("media_type", ""), "anchors": anchors_by_anime_id.get(str(original_anime_id), []), } ) if len(recommendations) >= 20: break return ( recommendations, item_score_pairs_sorted, user_stats, user_anime_details, anchors_by_anime_id, )