"""Backend recommendations service. Candidate **relevance** still comes from weighted random walks on stored similarity edges (standard graph-based personalization). **Diversification** of the ranked pool uses **MMR** (Maximal Marginal Relevance) via :mod:`backend.recsys_mmr` over lightweight discovery-lane features, not broad catalog clusters. See Carbonell & Goldstein, SIGIR 1998; implementation references scikit-learn. """ from __future__ import annotations import random from collections import Counter, defaultdict from pathlib import Path from typing import Any import numpy as np from backend.constants import ( BLOCKED_RATINGS, EDGE_TYPE_WEIGHTS, POSITIVE_RATINGS, RATING_WEIGHTS, SUPPRESSED_RATINGS, VALID_RATINGS, ) from backend.discovery_lanes import discovery_lane_for_kink from backend.recsys_graph import ( personalized_pagerank_scores, ppr_transition_ready, ) from backend.recsys_mmr import mmr_reorder_items from backend.recsys_ots import cached_candidates_for_user, read_candidate_cache from backend.scenarios import play_excluded_from_surfacing def _recommendation_sort_key(item: dict[str, Any]) -> tuple[float, int, float]: """Sort by (score, has_definition, popularity) so kinks with descriptions surface above near-tie variants without one (e.g. canonical "ffm threesome" with a definition wins over the higher-popularity but undefined "ffm threesomes" sibling).""" kink = item.get("kink", {}) or {} has_def = 1 if (kink.get("definition") or kink.get("detail_summary") or kink.get("summary") or "").strip() else 0 return (float(item.get("score", 0.0) or 0.0), has_def, float(kink.get("popularity", 0.0) or 0.0)) # Stochastic recs: sample from a graph-aware pool instead of always taking global score order. # Higher temperature → more uniform; pool cap bounds work per request. _REC_SAMPLE_TEMPERATURE = 0.86 _REC_AGG_BLEND = 0.58 _REC_POOL_CAP = 280 _STARTER_POOL_MULT = 4 # Breadth before stochastic sampling: graph hubs over-concentrate; MMR repools the head. _REC_DIV_POOL_MIN = 40 # MMR λ: balance relevance vs diversity in cluster space (Carbonell & Goldstein 1998). _REC_MMR_LAMBDA = 0.72 # Scale PPR mass to roughly match prior edge-sum magnitudes after per-node max normalization. _PPR_SCORE_SCALE = 5.0 _STARTER_DISCOVERY_TARGET = 12 _BLENDED_DISCOVERY_TARGET = 40 _EDGE_FETCH_LIMIT = 80 _EDGE_FETCH_CHUNK = 700 def _discovery_phase_for_seed_count(seed_count: int) -> str: if int(seed_count) < _STARTER_DISCOVERY_TARGET: return "starter" if int(seed_count) < _BLENDED_DISCOVERY_TARGET: return "blended" return "mature" def _default_user_reason_label(discovery_source: str, reasons: list[str] | set[str]) -> str: source = str(discovery_source or "").strip().lower() if source == "starter": return "Popular place to start" if source == "partner_influenced": return "Worth exploring together" if source == "explore": return "Still unrated" values = sorted(reasons) if isinstance(reasons, set) else list(reasons or []) text = str(values[0] or "").strip() if values else "" if text: return text return "Near things you already like" def _apply_diversity_pool(ranked: list[dict[str, Any]], limit: int) -> list[dict[str, Any]]: if len(ranked) <= 1: return ranked pool_sz = min(_REC_POOL_CAP, len(ranked), max(_REC_DIV_POOL_MIN, limit * 5, 12)) pool = [] for item in ranked[:pool_sz]: if not isinstance(item, dict) or not isinstance(item.get("kink"), dict): pool.append(item) continue pool.append({**item, "kink": {**item["kink"], "diversity_lane": discovery_lane_for_kink(item["kink"])}}) rest = ranked[pool_sz:] diverse = mmr_reorder_items( pool, relevance_key="score", cluster_key_path=("kink", "diversity_lane"), lambda_param=_REC_MMR_LAMBDA, ) return diverse + rest def _rec_sampling_weight(agg: float, max_edge: float, *, temperature: float, agg_blend: float) -> float: """Blend total edge mass with strongest single edge so one hub path doesn't dominate; spread via temperature.""" a = max(float(agg), 1e-9) m = max(float(max_edge), 1e-9) mix = (a**agg_blend) * (m ** (1.0 - agg_blend)) t = max(float(temperature), 0.12) return max(mix, 1e-12) ** (1.0 / t) def _weighted_sample_without_replacement(items: list[Any], weights: list[float], k: int, rng: random.Random) -> list[Any]: """Weighted sampling without replacement (numpy ``Generator.choice``).""" if not items or k <= 0: return [] n = len(items) if k >= n: return list(items) w_arr = np.asarray(weights, dtype=np.float64) if w_arr.shape[0] != n: raise ValueError("weights length must match items") gen = np.random.default_rng(int.from_bytes(rng.randbytes(8), "little") & (2**63 - 1)) rem_idx = np.arange(n) picked: list[Any] = [] for _ in range(k): sub = w_arr[rem_idx] total = float(sub.sum()) if total <= 1e-18: pos = int(gen.integers(len(rem_idx))) else: p = sub / total pos = int(gen.choice(len(rem_idx), p=p)) chosen_i = int(rem_idx[pos]) rem_idx = np.delete(rem_idx, pos) picked.append(items[chosen_i]) return picked def _lane_balanced_items(items: list[dict[str, Any]], fallback_pool: list[dict[str, Any]], limit: int) -> list[dict[str, Any]]: if limit <= 0 or not items: return [] lane_cap = max(2, min(5, limit // 5 or 2)) generic_cap = max(2, lane_cap - 1) out: list[dict[str, Any]] = [] counts: Counter[str] = Counter() seen_ids: set[str] = set() def lane_for(item: dict[str, Any]) -> str: kink = item.get("kink", {}) if isinstance(item, dict) else {} return str(kink.get("diversity_lane") or discovery_lane_for_kink(kink)) def maybe_add(item: dict[str, Any], *, enforce_cap: bool) -> None: if len(out) >= limit: return kid = item.get("kink", {}).get("id") if not kid or kid in seen_ids: return lane = lane_for(item) cap = generic_cap if lane == "fetlife_fetish" else lane_cap if enforce_cap and counts[lane] >= cap: return seen_ids.add(kid) counts[lane] += 1 out.append(item) for item in items: maybe_add(item, enforce_cap=True) if len(out) < limit: for item in fallback_pool: maybe_add(item, enforce_cap=True) if len(out) < limit: for item in [*items, *fallback_pool]: maybe_add(item, enforce_cap=False) return out[:limit] def _recommendation_item( self, kink: dict[str, Any], score: float, reasons: list[str] | set[str], mode: str, support_count: int = 0, discovery_source: str | None = None, user_reason_label: str | None = None, ) -> dict[str, Any]: source = (discovery_source or "").strip() or ( "starter" if mode == "starter" else "partner_influenced" if mode in {"group", "partner_influenced"} else "explore" if mode == "explore" else "personalized" ) item: dict[str, Any] = { "kink": kink, "score": score, "reasons": sorted(reasons) if isinstance(reasons, set) else reasons, "recommendation_mode": mode, "discovery_source": source, "user_reason_label": user_reason_label or _default_user_reason_label(source, reasons), } if support_count: item["support_count"] = support_count return item def _finalize_reasons( self, reasons: list[str] | set[str], *, fallback: str, limit: int = 2, drop_prefixes: tuple[str, ...] = ("less like ",), ) -> list[str]: seen: set[str] = set() cleaned: list[str] = [] values = sorted(reasons) if isinstance(reasons, set) else list(reasons) for reason in values: text = str(reason or "").strip() if not text: continue lowered = text.lower() if any(lowered.startswith(prefix) for prefix in drop_prefixes): continue if text in seen: continue seen.add(text) cleaned.append(text) if len(cleaned) >= limit: break return cleaned or [fallback] def _strip_rec_sampler_meta(items: list[dict[str, Any]]) -> None: for it in items: it.pop("_rec_max_edge", None) def _stochastic_select_recommendations(self, ranked: list[dict[str, Any]], limit: int, rng: random.Random) -> list[dict[str, Any]]: """Pick ``limit`` items with score-biased randomness (pool already MMR-diversified).""" if len(ranked) <= limit: out = _lane_balanced_items(ranked[:limit], ranked, limit) _strip_rec_sampler_meta(out) return out pool = ranked[: min(len(ranked), _REC_POOL_CAP)] weights = [ _rec_sampling_weight( float(it.get("score", 0.0)), float(it.get("_rec_max_edge", it.get("score", 0.0))), temperature=_REC_SAMPLE_TEMPERATURE, agg_blend=_REC_AGG_BLEND, ) for it in pool ] picked = _weighted_sample_without_replacement(pool, weights, limit, rng) picked = _lane_balanced_items(picked, pool, limit) _strip_rec_sampler_meta(picked) return picked def _starter_recommendations(self, plays: dict[str, dict[str, Any]], limit: int, seed_count: int, mode: str) -> list[dict[str, Any]]: pool_n = min(max(limit * _STARTER_POOL_MULT, limit + 8), 96) raw = self._starter_candidates(plays, pool_n) rng = random.Random() items: list[dict[str, Any]] = [] for kink in raw: sc = self._starter_score(kink) it = self._recommendation_item( kink, sc, [f"popular starter ({min(seed_count, 12)}/12 saved)"], mode, discovery_source="starter", user_reason_label="Popular place to start", ) it["_rec_max_edge"] = float(sc) items.append(it) return self._stochastic_select_recommendations(items, limit, rng) def _main_recommendation_target( self, kink: dict[str, Any], plays: dict[str, dict[str, Any]], detail_by_id: dict[str, Any], suppressed_ids: set[str], ) -> tuple[dict[str, Any] | None, dict[str, Any] | None]: """Return the main Discover target, replacing scenario candidates with a parent play.""" if kink.get("title_surface_as_scenario") and not kink.get("is_scenario"): return None, kink if not kink.get("is_scenario"): return kink, None for parent_id in kink.get("scenario_parent_ids", []) or []: if parent_id in plays or parent_id in suppressed_ids: continue parent = detail_by_id.get(parent_id) if not parent or play_excluded_from_surfacing(parent): continue if self._content_kind(parent) != "play" or not parent.get("shared_eligible"): continue return parent, kink return None, kink def _top_edges_by_source( self, source_ids: list[str], *, per_source_limit: int = _EDGE_FETCH_LIMIT, ) -> dict[str, list[dict[str, Any]]]: if not source_ids: return {} cache = getattr(self, "_top_edges_by_source_cache", None) if cache is None: cache = {} self._top_edges_by_source_cache = cache missing = [source_id for source_id in dict.fromkeys(source_ids) if source_id not in cache] if not missing: return {source_id: cache[source_id] for source_id in source_ids} grouped: dict[str, list[dict[str, Any]]] = defaultdict(list) with self._sqlite() as conn: for start in range(0, len(missing), _EDGE_FETCH_CHUNK): chunk = missing[start : start + _EDGE_FETCH_CHUNK] placeholders = ",".join("?" for _ in chunk) rows = conn.execute( f""" SELECT left_kink_id, right_kink_id, score, similarity_type FROM similarityedge WHERE left_kink_id IN ({placeholders}) ORDER BY left_kink_id ASC, score DESC, right_kink_id ASC """, tuple(chunk), ).fetchall() for row in rows: source_id = row["left_kink_id"] if len(grouped[source_id]) >= per_source_limit: continue grouped[source_id].append( { "id": row["right_kink_id"], "score": float(row["score"]), "type": row["similarity_type"], } ) for source_id in missing: cache[source_id] = grouped.get(source_id, []) return {source_id: cache[source_id] for source_id in source_ids} def _legacy_edge_sum_scores( self, plays: dict[str, dict[str, Any]], detail_by_id: dict[str, Any], suppressed_ids: set[str], ) -> tuple[dict[str, float], dict[str, float], dict[str, set[str]]]: """Weighted sum over outgoing edges (original personalized recommender).""" candidate_scores: dict[str, float] = defaultdict(float) candidate_max_edge: dict[str, float] = defaultdict(float) candidate_reasons: dict[str, set[str]] = defaultdict(set) rated_sources: list[tuple[str, dict[str, Any], float, dict[str, Any]]] = [] for rated_kink_id, state in plays.items(): rating = state["interest_state"] if rating not in VALID_RATINGS: continue weight = RATING_WEIGHTS.get(rating, 0.0) if weight == 0: continue source = detail_by_id.get(rated_kink_id) if not source: continue rated_sources.append((rated_kink_id, state, weight, source)) edges_by_source = _top_edges_by_source(self, [item[0] for item in rated_sources]) for rated_kink_id, state, weight, source in rated_sources: rating = state["interest_state"] for edge in edges_by_source.get(rated_kink_id, []): candidate_id = edge["id"] if candidate_id in plays: continue target = detail_by_id.get(candidate_id) if not target or candidate_id in suppressed_ids: continue if self._content_kind(target) != "play" or not target.get("shared_eligible"): continue edge_weight = EDGE_TYPE_WEIGHTS.get(edge.get("type", ""), 0.8) contrib = edge["score"] * weight * edge_weight candidate_scores[candidate_id] += contrib if contrib > candidate_max_edge[candidate_id]: candidate_max_edge[candidate_id] = contrib if rating in POSITIVE_RATINGS: reason = f"because of {source['name']}" if edge.get("type") == "fetlife_collab": reason = f"people who like {source['name']} also like this" elif edge.get("type") == "fetlife_similar": reason = f"similar to {source['name']}" candidate_reasons[candidate_id].add(reason) return candidate_scores, candidate_max_edge, candidate_reasons def _negative_edge_adjustments( self, plays: dict[str, dict[str, Any]], detail_by_id: dict[str, Any], suppressed_ids: set[str], ) -> dict[str, float]: """Pull from neighbors of negative-rated seeds (e.g. hard_no); PPR only teleports on positives.""" adj: dict[str, float] = defaultdict(float) negative_sources: list[tuple[str, float]] = [] for rated_kink_id, state in plays.items(): weight = RATING_WEIGHTS.get(state["interest_state"], 0.0) if weight >= 0: continue source = detail_by_id.get(rated_kink_id) if not source: continue negative_sources.append((rated_kink_id, weight)) edges_by_source = _top_edges_by_source(self, [item[0] for item in negative_sources]) for rated_kink_id, weight in negative_sources: for edge in edges_by_source.get(rated_kink_id, []): candidate_id = edge["id"] if candidate_id in plays: continue target = detail_by_id.get(candidate_id) if not target or candidate_id in suppressed_ids: continue if self._content_kind(target) != "play" or not target.get("shared_eligible"): continue edge_weight = EDGE_TYPE_WEIGHTS.get(edge.get("type", ""), 0.8) adj[candidate_id] += edge["score"] * weight * edge_weight return dict(adj) def _positive_seed_reasons_only( self, plays: dict[str, dict[str, Any]], detail_by_id: dict[str, Any], suppressed_ids: set[str], ) -> dict[str, set[str]]: """Explainability strings from positive-rated seeds only (aligned with PPR teleport set).""" candidate_reasons: dict[str, set[str]] = defaultdict(set) positive_sources: list[tuple[str, str, dict[str, Any]]] = [] for rated_kink_id, state in plays.items(): rating = state["interest_state"] if rating not in POSITIVE_RATINGS: continue weight = RATING_WEIGHTS.get(rating, 0.0) if weight <= 0: continue source = detail_by_id.get(rated_kink_id) if not source: continue positive_sources.append((rated_kink_id, rating, source)) edges_by_source = _top_edges_by_source(self, [item[0] for item in positive_sources]) for rated_kink_id, rating, source in positive_sources: for edge in edges_by_source.get(rated_kink_id, []): candidate_id = edge["id"] if candidate_id in plays: continue target = detail_by_id.get(candidate_id) if not target or candidate_id in suppressed_ids: continue if self._content_kind(target) != "play" or not target.get("shared_eligible"): continue reason = f"because of {source['name']}" if edge.get("type") == "fetlife_collab": reason = f"people who like {source['name']} also like this" elif edge.get("type") == "fetlife_similar": reason = f"similar to {source['name']}" candidate_reasons[candidate_id].add(reason) return candidate_reasons def _personalized_recommendations( self, plays: dict[str, dict[str, Any]], limit: int, mode: str = "personalized", ) -> list[dict[str, Any]]: detail_by_id = self._catalog()["detail_by_id"] suppressed_ids = {kink_id for kink_id, state in plays.items() if state["interest_state"] in SUPPRESSED_RATINGS | BLOCKED_RATINGS} positive_seed_weights: dict[str, float] = {} for kid, state in plays.items(): if state["interest_state"] not in POSITIVE_RATINGS: continue w = RATING_WEIGHTS.get(state["interest_state"], 0.0) if w > 0: positive_seed_weights[kid] = positive_seed_weights.get(kid, 0.0) + w use_ppr = bool(positive_seed_weights) and ppr_transition_ready(self) ppr_raw: dict[str, float] = {} if use_ppr: ppr_raw = personalized_pagerank_scores(self.engine, positive_seed_weights, backend=self) if not ppr_raw: use_ppr = False neg_adj = _negative_edge_adjustments(self, plays, detail_by_id, suppressed_ids) if use_ppr: candidate_reasons = _positive_seed_reasons_only(self, plays, detail_by_id, suppressed_ids) mx = max(ppr_raw.values()) if ppr_raw else 1e-9 candidate_scores = { kid: (ppr_raw[kid] / mx) * _PPR_SCORE_SCALE + neg_adj.get(kid, 0.0) for kid in ppr_raw } candidate_max_edge = { kid: max(float(ppr_raw.get(kid, 0.0)), abs(neg_adj.get(kid, 0.0))) for kid in candidate_scores } else: candidate_scores, candidate_max_edge, candidate_reasons = _legacy_edge_sum_scores( self, plays, detail_by_id, suppressed_ids ) rng = random.Random() ranked = [] seen_targets: set[str] = set() for kink_id, raw_score in sorted(candidate_scores.items(), key=lambda item: item[1], reverse=True): if kink_id in plays: continue candidate = detail_by_id.get(kink_id) if not candidate: continue kink, scenario_source = self._main_recommendation_target(candidate, plays, detail_by_id, suppressed_ids) if not kink: continue target_id = kink["id"] if target_id in plays or target_id in suppressed_ids or target_id in seen_targets: continue if self._content_kind(kink) != "play": continue if not kink.get("shared_eligible"): continue if play_excluded_from_surfacing(kink): continue score = raw_score score += min(kink["popularity"] / 60000.0, 0.2) score -= self._discoverability_penalty(kink["cluster"]) if score <= 0: continue reasons = set(candidate_reasons[kink_id]) if scenario_source: reasons.add(f"{scenario_source['name']} belongs under {kink['name']}") rec = self._recommendation_item( kink, score, self._finalize_reasons(reasons, fallback="fits what you already like"), mode, discovery_source="personalized", user_reason_label="Near things you already like", ) rec["_rec_max_edge"] = float(candidate_max_edge.get(kink_id, score)) if scenario_source: rec["scenario_source_kink_id"] = scenario_source["id"] seen_targets.add(target_id) ranked.append(rec) ranked.sort(key=_recommendation_sort_key, reverse=True) if not ranked: liked_kinks = [detail_by_id.get(kink_id) for kink_id, state in plays.items() if state["interest_state"] in POSITIVE_RATINGS] query = " ".join(kink["name"] for kink in liked_kinks if kink) fallback = self.search_kinks(query, limit=limit + len(plays)) for item in fallback: kink = item["kink"] if kink["id"] in plays or kink["id"] in suppressed_ids: continue if play_excluded_from_surfacing(kink): continue if self._content_kind(kink) != "play": continue sc = float(item["score"]) rec = self._recommendation_item( kink, sc, ["matched against your selected plays"], mode, discovery_source="personalized", user_reason_label="Near things you already like", ) rec["_rec_max_edge"] = sc ranked.append(rec) ranked.sort(key=_recommendation_sort_key, reverse=True) ranked = _apply_diversity_pool(ranked, limit) return self._stochastic_select_recommendations(ranked, limit, rng) def _dedupe_recommendations(self, items: list[dict[str, Any]], limit: int) -> list[dict[str, Any]]: seen: set[str] = set() deduped: list[dict[str, Any]] = [] for item in items: kink_id = item.get("kink", {}).get("id") if not kink_id or kink_id in seen: continue seen.add(kink_id) deduped.append(item) if len(deduped) >= limit: break return deduped def _explore_backfill( self, plays: dict[str, dict[str, Any]], limit: int, excluded_ids: set[str], ) -> list[dict[str, Any]]: """When graph/starter pools run dry, sample unrated catalog plays — still biased toward popularity first. ``list_kink_summaries()`` is sorted by descending popularity, so index-based weights keep common kinks more likely while jitter explores the long tail. """ if limit <= 0: return [] suppressed_ids = {k for k, st in plays.items() if st["interest_state"] in SUPPRESSED_RATINGS | BLOCKED_RATINGS} detail_by_id = self._catalog()["detail_by_id"] pool: list[dict[str, Any]] = [] for kink in self.list_kink_summaries(): kid = kink["id"] if kid in excluded_ids or kid in suppressed_ids: continue if self._content_kind(kink) != "play": continue if play_excluded_from_surfacing(kink): continue if not kink.get("shared_eligible"): continue pool.append(kink) if not pool: return [] rng = random.Random() cap = min(len(pool), 520) head = pool[:cap] weights: list[float] = [] for i, k in enumerate(head): pop = max(float(k.get("popularity", 0.0) or 0.0), 1.0) sb = float(k.get("source_backed_popularity", 0.0) or 0.0) pos_w = 1.0 / (1.0 + (i / 42.0) ** 0.9) w = (pop**0.42) * (1.0 + min(sb, 2_000_000.0) / 100_000.0) * pos_w * rng.uniform(0.72, 1.28) weights.append(w) pick_n = min(limit, len(head)) picked = _weighted_sample_without_replacement(head, weights, pick_n, rng) out: list[dict[str, Any]] = [] for kink in picked: detail = detail_by_id[kink["id"]] pop = float(kink.get("popularity", 0.0) or 0.0) sb = float(kink.get("source_backed_popularity", 0.0) or 0.0) sc = min(pop / 800.0, 60.0) + min(sb / 60_000.0, 10.0) out.append( self._recommendation_item( detail, sc, ["still unrated — biased toward widely-known items"], "explore", discovery_source="explore", user_reason_label="Still unrated", ) ) return out def _take_unique_kink_items( items: list[dict[str, Any]], *, limit: int, excluded_ids: set[str], ) -> list[dict[str, Any]]: """First occurrence wins; skips ``excluded_ids`` (e.g. already in plays).""" seen: set[str] = set() out: list[dict[str, Any]] = [] for item in items: kid = item.get("kink", {}).get("id") if not kid or kid in excluded_ids or kid in seen: continue seen.add(kid) out.append(item) if len(out) >= limit: break return out def _blended_recommendations(self, plays: dict[str, dict[str, Any]], seed_count: int, limit: int) -> list[dict[str, Any]]: phase = _discovery_phase_for_seed_count(seed_count) if phase == "starter": return self._starter_recommendations(plays, limit, seed_count, "starter") if phase == "blended": personalized_limit = max(4, limit // 3) starters = self._starter_recommendations(plays, limit, seed_count, "blended") personalized = self._personalized_recommendations(plays, personalized_limit, mode="blended") return self._dedupe_recommendations(starters[:6] + personalized + starters[6:], limit) starter_safety_net = max(4, limit // 5) personalized = self._personalized_recommendations(plays, max(limit - starter_safety_net, limit // 2), mode="personalized") starters = self._starter_recommendations(plays, starter_safety_net, seed_count, "personalized") return self._dedupe_recommendations(personalized + starters, limit) def _recsys_source_mode(self) -> str: return str(self.recsys_settings.source) def _candidate_cache_path(self) -> Path: configured = self.recsys_settings.candidates_path if configured is not None: return Path(configured) return self.path.parent / "recsys" / "candidates.json" def _load_ots_candidate_cache(self) -> dict[str, Any] | None: path = _candidate_cache_path(self) if not path.exists(): return None mtime = path.stat().st_mtime cached = getattr(self, "_ots_candidate_cache", None) if cached and cached.get("path") == str(path) and cached.get("mtime") == mtime: return cached.get("payload") try: payload = read_candidate_cache(path) except (OSError, ValueError): return None self._ots_candidate_cache = {"path": str(path), "mtime": mtime, "payload": payload} return payload def _ots_recommendations(self, user: dict[str, Any], limit: int, *, mode: str = "ots") -> list[dict[str, Any]]: cache = _load_ots_candidate_cache(self) if not cache: return [] rows = cached_candidates_for_user(cache, user["id"]) if not rows: return [] plays = user["plays"] play_ids = set(plays) suppressed_ids = { kink_id for kink_id, state in plays.items() if state["interest_state"] in SUPPRESSED_RATINGS | BLOCKED_RATINGS } detail_by_id = self._catalog()["detail_by_id"] out: list[dict[str, Any]] = [] seen_targets: set[str] = set() pool_cap = max(limit * 5, limit) for row in rows: candidate_id = str(row.get("kink_id", "")) candidate = detail_by_id.get(candidate_id) if not candidate: continue kink, scenario_source = self._main_recommendation_target(candidate, plays, detail_by_id, suppressed_ids) if not kink: continue target_id = kink["id"] if target_id in play_ids or target_id in suppressed_ids or target_id in seen_targets: continue if self._content_kind(kink) != "play" or play_excluded_from_surfacing(kink) or not kink.get("shared_eligible"): continue score = float(row.get("score", 0.0) or 0.0) reasons = [f"OTS candidate via {row.get('model', cache.get('source', 'model'))}"] if scenario_source: reasons.append(f"{scenario_source['name']} belongs under {kink['name']}") rec = self._recommendation_item(kink, score, reasons, mode) if scenario_source: rec["scenario_source_kink_id"] = scenario_source["id"] seen_targets.add(target_id) out.append(rec) if len(out) >= pool_cap: break out = _apply_diversity_pool(out, limit) out = _lane_balanced_items(out, out, limit) return self._dedupe_recommendations(out, limit) def _partner_influenced_recommendations( self, user_id: str, *, limit: int, group_id: str | None = None, ) -> list[dict[str, Any]]: if limit <= 0: return [] candidates = self.prompt_candidates(user_id, limit=limit * 2, group_id=group_id) if not candidates: return [] items: list[dict[str, Any]] = [] for item in candidates: items.append( self._recommendation_item( item["kink"], float(item.get("score", 0.0) or 0.0), self._finalize_reasons(item.get("reasons", []), fallback="worth getting your take on"), "partner_influenced", int(item.get("support_count", 0) or 0), discovery_source="partner_influenced", user_reason_label="Worth exploring together", ) ) return self._dedupe_recommendations(items, limit) def _graph_recommend(self, user_id: str, limit: int = 5, group_id: str | None = None) -> list[dict]: limit_i = int(limit) user = self.get_user(user_id) if not user: return [] plays = user["plays"] phase = _discovery_phase_for_seed_count(len(plays)) play_ids = set(plays.keys()) if group_id and phase != "starter": if not self.can_access_partner_group(user_id, group_id): return [] group_items = self._group_recommendations(user_id, group_id, limit) if group_items: filtered = _take_unique_kink_items(group_items, limit=limit, excluded_ids=play_ids) if filtered: out = list(filtered) if len(out) < limit_i: need = limit_i - len(out) ex = play_ids | {x["kink"]["id"] for x in out} extra = self._explore_backfill(plays, need, ex) out = _take_unique_kink_items(out + extra, limit=limit_i, excluded_ids=play_ids) return list(out) blended = self._blended_recommendations(plays, len(plays), limit) if phase != "starter" and user.get("partners"): partner_items = self._partner_influenced_recommendations( user_id, limit=max(2, min(6, limit_i // 3 or 2)), ) if partner_items: head = max(2, min(8, limit_i // 4 or 2)) blended = self._dedupe_recommendations(blended[:head] + partner_items + blended[head:], limit_i) out = _take_unique_kink_items(blended, limit=limit, excluded_ids=play_ids) if len(out) < limit_i: need = limit_i - len(out) ex = play_ids | {x["kink"]["id"] for x in out} extra = self._explore_backfill(plays, need, ex) out = _take_unique_kink_items(out + extra, limit=limit_i, excluded_ids=play_ids) return list(out) def recommend(self, user_id: str, limit: int = 5, group_id: str | None = None) -> list[dict]: user = self.get_user(user_id) if not user: return [] source = _recsys_source_mode(self) phase = _discovery_phase_for_seed_count(len(user["plays"])) if source == "graph" or group_id or phase != "mature": return self._graph_recommend(user_id, limit=limit, group_id=group_id) limit_i = int(limit) play_ids = set(user["plays"]) ots_items = self._ots_recommendations(user, limit_i, mode="ots" if source == "ots" else "hybrid_ots") if source == "ots": if not ots_items: return self._graph_recommend(user_id, limit=limit, group_id=group_id) if len(ots_items) < limit_i: need = limit_i - len(ots_items) ex = play_ids | {item["kink"]["id"] for item in ots_items} extra = self._explore_backfill(user["plays"], need, ex) ots_items = _take_unique_kink_items(ots_items + extra, limit=limit_i, excluded_ids=play_ids) return list(ots_items) graph_items = self._graph_recommend(user_id, limit=limit_i, group_id=group_id) if not ots_items: return graph_items head = max(2, limit_i // 2) combined = _take_unique_kink_items( ots_items[:head] + graph_items + ots_items[head:], limit=limit_i, excluded_ids=play_ids, ) if len(combined) < limit_i: need = limit_i - len(combined) ex = play_ids | {item["kink"]["id"] for item in combined} extra = self._explore_backfill(user["plays"], need, ex) combined = _take_unique_kink_items(combined + extra, limit=limit_i, excluded_ids=play_ids) return list(combined) def recommendation_debug(self, user_id: str, limit: int = 10, group_id: str | None = None) -> dict[str, Any]: user = self.get_user(user_id) if not user: return {"items": [], "seed_count": 0, "group_id": group_id, "signals": {}} items = self.recommend(user_id, limit=limit, group_id=group_id) seed_ids = list(user["plays"].keys()) edge_counts: dict[str, int] = defaultdict(int) for kink_id in seed_ids: for edge in self._edges_for_kink(kink_id, limit=80): edge_counts[str(edge.get("type", "unknown"))] += 1 return { "seed_count": len(seed_ids), "discovery_phase": _discovery_phase_for_seed_count(len(seed_ids)), "group_id": group_id, "signals": { "seed_ratings": dict(Counter(state["interest_state"] for state in user["plays"].values())), "edge_candidates_by_type": dict(edge_counts), "active_group": bool(group_id), "recsys_source": _recsys_source_mode(self), "ots_candidate_cache": str(_candidate_cache_path(self)), "personalized_ppr_eligible": any( st["interest_state"] in POSITIVE_RATINGS and RATING_WEIGHTS.get(st["interest_state"], 0) > 0 for st in user["plays"].values() ), }, "items": [ { "kink_id": item["kink"]["id"], "name": item["kink"]["name"], "score": item["score"], "recommendation_mode": item.get("recommendation_mode", ""), "discovery_source": item.get("discovery_source", ""), "user_reason_label": item.get("user_reason_label", ""), "reasons": item.get("reasons", []), "starter_tier": item["kink"].get("starter_tier", ""), "popularity": item["kink"].get("popularity", 0), "source_backed_popularity": item["kink"].get("source_backed_popularity", 0), "similar_count": item["kink"].get("similar_count", 0), "image_relevance_score": item["kink"].get("image_relevance_score", 0), "is_scenario": bool(item["kink"].get("is_scenario")), "scenario_parent_ids": item["kink"].get("scenario_parent_ids", []), } for item in items ], } def _group_recommendations(self, owner_user_id: str, group_id: str, limit: int) -> list[dict[str, Any]]: owner = self.get_user(owner_user_id) users = self._group_participants(owner_user_id, group_id) if not owner or len(users) < 2: return [] owner_plays = owner["plays"] shared = self.shared_play_list_for_group(owner_user_id, group_id, limit=limit) shared_ids = {item["kink"]["id"] for item in shared["similar_matches"]} group_items = [ self._recommendation_item( item["kink"], float(item.get("score", 0.0) or 0.0), self._finalize_reasons(item.get("reasons", []), fallback="close to what you both already like"), "group", int(item.get("support_count", 0) or 0), discovery_source="partner_influenced", user_reason_label="Worth exploring together", ) for item in shared["similar_matches"] ] prompt_items = [ self._recommendation_item( item["kink"], float(item.get("score", 0.0) or 0.0), self._finalize_reasons(item.get("reasons", []), fallback="worth getting your take on"), "group", int(item.get("support_count", 0) or 0), discovery_source="partner_influenced", user_reason_label="Worth exploring together", ) for item in self.group_prompt_candidates(owner_user_id, group_id, limit=limit * 2) if item["kink"]["id"] not in shared_ids ] starter_items = self._starter_recommendations(owner_plays, max(4, limit // 4), len(owner_plays), "group") random.shuffle(prompt_items) return self._dedupe_recommendations(group_items + prompt_items + starter_items, limit)