#913: add group_similars_by_seed assembly helper (pure, tested)

The stored similar_artists rows key the similar artist by the SEED's source/db id, not its name,
so rank_recommended_artists can't consume them directly. group_similars_by_seed resolves each
row's source id to a seed name via a caller-supplied id_to_name map and reshapes to the
{seed_name: [{'name': similar}]} the ranker wants — the fragile id->name join, now pure + tested
(dataclass + dict rows, unknown-id drop, non-seed drop, group->rank end-to-end). 15 tests total.

core/discovery/listening_recommendations.py

@@ -39,6 +39,44 @@ def _positive_float(value: object, default: float = 1.0) -> float:
     return f if f > 0 else default
 
 
+def _get(row: object, attr: str):
+    """Read a field from a dataclass row or a dict row."""
+    if isinstance(row, dict):
+        return row.get(attr)
+    return getattr(row, attr, None)
+
+
+def group_similars_by_seed(
+    seeds: Sequence[dict],
+    similar_rows: Sequence,
+    id_to_name: Dict[str, str],
+    *,
+    source_id_attr: str = "source_artist_id",
+    similar_name_attr: str = "similar_artist_name",
+) -> Dict[str, List[dict]]:
+    """Reshape flat ``similar_artists`` rows into ``{seed_name_lower: [{'name': similar}]}``.
+
+    The stored rows key the similar artist by the SEED's source id (``source_artist_id``),
+    not its name, so :func:`rank_recommended_artists` can't consume them directly. This
+    resolves each row's source id to a name via ``id_to_name`` (``{source_artist_id:
+    artist_name}`` for the library, built by the caller) and keeps only rows that resolve
+    to one of the ``seeds``. Rows may be dataclass objects or dicts. Pure — no I/O.
+    """
+    seed_names = {_norm(s.get("name")) for s in seeds}
+    seed_names.discard("")
+    id_to_norm = {str(k): _norm(v) for k, v in (id_to_name or {}).items()}
+
+    out: Dict[str, List[dict]] = {}
+    for row in similar_rows or ():
+        seed_name = id_to_norm.get(str(_get(row, source_id_attr) or ""), "")
+        if not seed_name or seed_name not in seed_names:
+            continue
+        sim_name = str(_get(row, similar_name_attr) or "").strip()
+        if sim_name:
+            out.setdefault(seed_name, []).append({"name": sim_name})
+    return out
+
+
 @dataclass
 class RecommendedArtist:
     """One artist recommended from your listening, with the why."""
@@ -161,6 +199,7 @@ def aggregate_candidate_tracks(
 
 __all__ = [
     "RecommendedArtist",
+    "group_similars_by_seed",
     "rank_recommended_artists",
     "aggregate_candidate_tracks",
 ]

tests/discovery/test_listening_recommendations.py

@@ -115,3 +115,55 @@ 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
+
+
+# ── group_similars_by_seed (id->name join) ───────────────────────────────────
+from dataclasses import dataclass as _dc  # noqa: E402
+
+from core.discovery.listening_recommendations import group_similars_by_seed  # noqa: E402
+
+
+@_dc
+class _Row:
+    source_artist_id: str
+    similar_artist_name: str
+
+
+def test_group_resolves_source_id_to_seed_name():
+    seeds = [_seed("Radiohead"), _seed("Bjork")]
+    rows = [
+        _Row("id-rh", "Muse"),
+        _Row("id-rh", "Coldplay"),
+        _Row("id-bj", "Portishead"),
+        _Row("id-unknown", "Nobody"),   # id not in map -> dropped
+    ]
+    id_to_name = {"id-rh": "Radiohead", "id-bj": "Bjork"}
+    out = group_similars_by_seed(seeds, rows, id_to_name)
+    assert {n["name"] for n in out["radiohead"]} == {"Muse", "Coldplay"}
+    assert [n["name"] for n in out["bjork"]] == ["Portishead"]
+    assert "id-unknown" not in out and "Nobody" not in str(out)
+
+
+def test_group_keeps_only_rows_for_actual_seeds():
+    # id resolves to a name, but that name isn't a seed -> dropped.
+    seeds = [_seed("A")]
+    rows = [_Row("id-a", "SimA"), _Row("id-x", "SimX")]
+    out = group_similars_by_seed(seeds, rows, {"id-a": "A", "id-x": "X"})
+    assert list(out.keys()) == ["a"]
+
+
+def test_group_accepts_dict_rows():
+    seeds = [_seed("A")]
+    rows = [{"source_artist_id": "id-a", "similar_artist_name": "SimA"}]
+    out = group_similars_by_seed(seeds, rows, {"id-a": "A"})
+    assert out["a"] == [{"name": "SimA"}]
+
+
+def test_group_then_rank_end_to_end():
+    # The two-step the scanner will run: group rows, then rank.
+    seeds = [_seed("A", weight=2), _seed("B", weight=1)]
+    rows = [_Row("ia", "Common"), _Row("ia", "Solo"), _Row("ib", "Common")]
+    grouped = group_similars_by_seed(seeds, rows, {"ia": "A", "ib": "B"})
+    ranked = rank_recommended_artists(seeds, grouped, owned_artist_names={"solo"})
+    assert ranked[0].name == "Common" and ranked[0].seed_count == 2
+    assert all(r.name != "Solo" for r in ranked)   # owned excluded