backend/recommendations.py

"""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)