#913: listening-driven recommendation core (pure, tested)

New, fully-additive module — the heart of the 'expand Because You Listen To into a real
listening-driven block' plan. Two pure functions, no DB/network/config:

- rank_recommended_artists(seeds, similars_by_seed, owned): consensus-ranked artists you'd love
  but don't own. Score = Σ over endorsing seeds of (play_weight × similarity) — rewards consensus,
  play weight and similarity strength in one sum. Excludes owned + seeds; min_seed_count is the
  adventurousness dial's lever; exposes seed_count + which seeds ('because you like A, B, C').
- aggregate_candidate_tracks(recs, top_tracks_by_artist, owned): per-artist-capped, deduped,
  rank-ordered candidate list for the generated playlist; exclude_owned toggles discovery vs replay.

11 tests (consensus vs single, play-weight, similarity, owned/seed exclusion, min_seed_count,
case-insensitive dedup, per-artist cap, owned exclusion, total limit, empty-artist skip). Nothing
existing touched — wiring into the watchlist scan + playlist sync comes next.

core/discovery/listening_recommendations.py

@@ -0,0 +1,166 @@
+"""Listening-driven recommendation core (#913).
+
+PURE, side-effect-free ranking that turns "the artists you listen to most" plus
+"who's similar to each" into:
+
+  1. a consensus-ranked list of artists you'd probably love but don't own, and
+  2. an aggregated candidate-track list for a generated playlist.
+
+No DB / network / config here. The caller (the watchlist scanner) supplies the
+seeds (top-played artists), the ``similar_artists`` rows per seed, and the
+owned-artist set, then fetches top tracks for the winners. Keeping the decision
+logic in one pure place makes it fully unit-testable without the live stack and
+keeps the scan wiring thin — and additive, so it can't disturb existing flows.
+
+Scoring rationale (the "best in class" bit): a recommended artist's score is
+``Σ over the seeds that recommend it of (seed_weight × similarity)``. That single
+sum rewards all three signals at once — **consensus** (an artist endorsed by many
+of your seeds accumulates more terms), your **play weight** (heavier seeds push
+harder), and **similarity strength** — instead of a flat "appears in N lists".
+``seed_count`` is exposed separately for display ("because you like A, B, C") and
+as the adventurousness dial's lever (``min_seed_count``).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional, Sequence, Set
+
+
+def _norm(name: object) -> str:
+    return str(name or "").strip().lower()
+
+
+def _positive_float(value: object, default: float = 1.0) -> float:
+    try:
+        f = float(value)  # type: ignore[arg-type]
+    except (TypeError, ValueError):
+        return default
+    return f if f > 0 else default
+
+
+@dataclass
+class RecommendedArtist:
+    """One artist recommended from your listening, with the why."""
+    name: str                                       # display name (first-seen casing)
+    score: float                                    # Σ seed_weight × similarity
+    seed_count: int                                 # distinct seeds endorsing it (consensus)
+    seeds: List[str] = field(default_factory=list)  # display names of those seeds
+
+
+def rank_recommended_artists(
+    seeds: Sequence[dict],
+    similars_by_seed: Dict[str, Sequence[dict]],
+    owned_artist_names: Optional[Set[str]] = None,
+    *,
+    limit: int = 30,
+    min_seed_count: int = 1,
+) -> List[RecommendedArtist]:
+    """Rank artists similar to your most-played by consensus + play weight + similarity.
+
+    Args:
+        seeds: ``[{'name': str, 'weight': float}]`` — your top-played artists.
+            ``weight`` (play count or any positive number) defaults to 1.0.
+        similars_by_seed: ``{seed_name_lower: [{'name': str, 'score': float}]}`` — the
+            similar-artist rows for each seed. ``score`` is optional (defaults 1.0).
+        owned_artist_names: lowercased names already in the library — excluded so the
+            result is artists you DON'T have. The seeds themselves are always excluded.
+        limit: max results.
+        min_seed_count: drop recommendations endorsed by fewer than N seeds — the
+            adventurousness dial's "Safer" end raises this for higher-confidence picks.
+
+    Returns up to ``limit`` :class:`RecommendedArtist`, highest score first.
+    """
+    owned = {_norm(a) for a in (owned_artist_names or set())}
+    seed_norms = {_norm(s.get("name")) for s in seeds}
+    seed_norms.discard("")
+    exclude = owned | seed_norms
+
+    acc: Dict[str, dict] = {}
+    for seed in seeds:
+        s_name = _norm(seed.get("name"))
+        if not s_name:
+            continue
+        s_display = str(seed.get("name") or "").strip()
+        weight = _positive_float(seed.get("weight", 1.0))
+        for sim in similars_by_seed.get(s_name, ()) or ():
+            a_norm = _norm(sim.get("name"))
+            if not a_norm or a_norm in exclude:
+                continue
+            sim_score = _positive_float(sim.get("score", 1.0))
+            row = acc.setdefault(
+                a_norm, {"name": str(sim.get("name") or "").strip(), "score": 0.0, "seeds": {}}
+            )
+            row["score"] += weight * sim_score
+            row["seeds"].setdefault(s_name, s_display)   # one seed counts once
+
+    out: List[RecommendedArtist] = []
+    floor = max(1, int(min_seed_count))
+    for row in acc.values():
+        seed_count = len(row["seeds"])
+        if seed_count < floor:
+            continue
+        out.append(RecommendedArtist(
+            name=row["name"],
+            score=round(row["score"], 6),
+            seed_count=seed_count,
+            seeds=list(row["seeds"].values()),
+        ))
+    out.sort(key=lambda r: (-r.score, -r.seed_count, r.name.lower()))
+    return out[:limit]
+
+
+def aggregate_candidate_tracks(
+    recommended_artists: Sequence[RecommendedArtist],
+    top_tracks_by_artist: Dict[str, Sequence[dict]],
+    owned_track_keys: Optional[Set] = None,
+    *,
+    per_artist: int = 3,
+    limit: int = 50,
+    exclude_owned: bool = True,
+) -> List[dict]:
+    """Build the candidate track list for the generated playlist.
+
+    Takes the top ``per_artist`` tracks from each recommended artist **in artist-rank
+    order**, dedups by ``(artist, title)``, optionally drops owned tracks (the
+    "discovery" flavor) and caps at ``limit``. Each returned track dict is the source
+    track plus ``_seed_artist`` (which recommended artist it came from).
+
+    Args:
+        recommended_artists: ranked output of :func:`rank_recommended_artists`.
+        top_tracks_by_artist: ``{artist_name_lower: [track_dict, ...]}`` — fetched by
+            the caller (Last.fm / source top tracks), NOT limited to a curated pool.
+        owned_track_keys: set of ``(artist_lower, title_lower)`` already in the library.
+        exclude_owned: drop tracks in ``owned_track_keys`` (discovery flavor). Set False
+            for a "replay" playlist of tracks you already own.
+    """
+    owned = owned_track_keys or set()
+    seen: Set = set()
+    out: List[dict] = []
+    for art in recommended_artists:
+        tracks = top_tracks_by_artist.get(_norm(art.name), ()) or ()
+        taken = 0
+        for t in tracks:
+            if taken >= per_artist:
+                break
+            title = str(t.get("name") or t.get("title") or "").strip()
+            if not title:
+                continue
+            key = (_norm(art.name), _norm(title))
+            if key in seen:
+                continue
+            if exclude_owned and key in owned:
+                continue
+            seen.add(key)
+            out.append({**t, "_seed_artist": art.name})
+            taken += 1
+        if len(out) >= limit:
+            break
+    return out[:limit]
+
+
+__all__ = [
+    "RecommendedArtist",
+    "rank_recommended_artists",
+    "aggregate_candidate_tracks",
+]

tests/discovery/test_listening_recommendations.py

@@ -0,0 +1,117 @@
+"""Listening-driven recommendation core (#913) — pure ranking + candidate aggregation."""
+
+from __future__ import annotations
+
+from core.discovery.listening_recommendations import (
+    aggregate_candidate_tracks,
+    rank_recommended_artists,
+)
+
+
+def _seed(name, weight=1.0):
+    return {"name": name, "weight": weight}
+
+
+# ── rank_recommended_artists ─────────────────────────────────────────────────
+def test_consensus_outranks_single_endorsement():
+    # 'Common' is similar to BOTH seeds; 'Solo' to one. Equal weights/scores.
+    seeds = [_seed("A"), _seed("B")]
+    sims = {
+        "a": [{"name": "Common"}, {"name": "Solo"}],
+        "b": [{"name": "Common"}],
+    }
+    out = rank_recommended_artists(seeds, sims)
+    assert [r.name for r in out] == ["Common", "Solo"]
+    assert out[0].seed_count == 2
+    assert sorted(out[0].seeds) == ["A", "B"]
+    assert out[1].seed_count == 1
+
+
+def test_play_weight_boosts_a_seeds_similars():
+    seeds = [_seed("Fav", weight=100), _seed("Minor", weight=1)]
+    sims = {"fav": [{"name": "FromFav"}], "minor": [{"name": "FromMinor"}]}
+    out = rank_recommended_artists(seeds, sims)
+    assert out[0].name == "FromFav"          # heavier seed's similar wins
+
+
+def test_similarity_score_weights_within_a_seed():
+    seeds = [_seed("A")]
+    sims = {"a": [{"name": "Close", "score": 0.9}, {"name": "Far", "score": 0.1}]}
+    out = rank_recommended_artists(seeds, sims)
+    assert [r.name for r in out] == ["Close", "Far"]
+
+
+def test_owned_and_seed_artists_are_excluded():
+    seeds = [_seed("A"), _seed("B")]
+    sims = {"a": [{"name": "Owned"}, {"name": "B"}, {"name": "New"}]}  # 'B' is a seed
+    out = rank_recommended_artists(seeds, sims, owned_artist_names={"owned"})
+    assert [r.name for r in out] == ["New"]   # Owned dropped, seed B dropped
+
+
+def test_min_seed_count_filters_low_consensus():
+    seeds = [_seed("A"), _seed("B")]
+    sims = {"a": [{"name": "Common"}, {"name": "Solo"}], "b": [{"name": "Common"}]}
+    out = rank_recommended_artists(seeds, sims, min_seed_count=2)
+    assert [r.name for r in out] == ["Common"]   # 'Solo' (1 seed) dropped
+
+
+def test_case_insensitive_dedup_and_matching():
+    seeds = [_seed("Radiohead")]
+    sims = {"radiohead": [{"name": "Muse"}, {"name": "MUSE"}]}  # same artist twice
+    out = rank_recommended_artists(seeds, sims)
+    assert len(out) == 1 and out[0].name in ("Muse", "MUSE")
+    assert out[0].score == 2.0                # accumulated (still one seed)
+    assert out[0].seed_count == 1
+
+
+def test_empty_and_limit():
+    assert rank_recommended_artists([], {}) == []
+    seeds = [_seed("A")]
+    sims = {"a": [{"name": f"S{i}"} for i in range(10)]}
+    assert len(rank_recommended_artists(seeds, sims, limit=3)) == 3
+
+
+# ── aggregate_candidate_tracks ───────────────────────────────────────────────
+def _recs(*names):
+    return rank_recommended_artists(
+        [_seed(n) for n in names],
+        {n.lower(): [{"name": f"sim-{n}"}] for n in names},
+    )
+
+
+def test_aggregate_caps_per_artist_and_total_in_rank_order():
+    recs = _recs("A", "B")  # -> recommended sim-A, sim-B
+    tracks = {
+        "sim-a": [{"name": "a1"}, {"name": "a2"}, {"name": "a3"}],
+        "sim-b": [{"name": "b1"}, {"name": "b2"}],
+    }
+    out = aggregate_candidate_tracks(recs, tracks, per_artist=2, limit=10)
+    names = [t["name"] for t in out]
+    assert names == ["a1", "a2", "b1", "b2"]          # per_artist=2, rank order
+    assert all(t["_seed_artist"].startswith("sim-") for t in out)
+
+
+def test_aggregate_excludes_owned_when_requested():
+    recs = _recs("A")
+    tracks = {"sim-a": [{"name": "Owned Song"}, {"name": "New Song"}]}
+    owned = {("sim-a", "owned song")}
+    out = aggregate_candidate_tracks(recs, tracks, owned, per_artist=5, exclude_owned=True)
+    assert [t["name"] for t in out] == ["New Song"]
+    # replay flavor keeps owned
+    keep = aggregate_candidate_tracks(recs, tracks, owned, per_artist=5, exclude_owned=False)
+    assert [t["name"] for t in keep] == ["Owned Song", "New Song"]
+
+
+def test_aggregate_dedups_and_respects_total_limit():
+    recs = _recs("A", "B")
+    tracks = {"sim-a": [{"name": "dup"}], "sim-b": [{"name": "dup"}, {"name": "x"}]}
+    # 'dup' under sim-a and sim-b are different (artist,title) keys -> both kept;
+    # within an artist a repeat would dedup. Here check the total limit instead.
+    out = aggregate_candidate_tracks(recs, tracks, per_artist=5, limit=2)
+    assert len(out) == 2
+
+
+def test_aggregate_skips_artist_with_no_tracks():
+    recs = _recs("A", "B")
+    out = aggregate_candidate_tracks(recs, {"sim-a": [{"name": "only"}]}, per_artist=5)
+    assert [t["name"] for t in out] == ["only"]   # sim-b had no tracks -> skipped