feat(discovery): run_discovery aggregates similar artists into suggestions

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

worker/lyra_worker/discovery.py

@@ -0,0 +1,138 @@
+from dataclasses import dataclass
+
+import psycopg
+
+from lyra_worker.browser import MbBrowser
+from lyra_worker.similarity.base import SimilaritySource
+
+_TRUE = {"1", "true", "yes", "on"}
+
+
+@dataclass(frozen=True)
+class DiscoveryConfig:
+    enabled: bool = False
+    interval_hours: int = 168
+    max_seeds: int = 50
+    similar_per_seed: int = 20
+    albums_per_artist: int = 1
+    min_score: float = 0.0
+
+    @classmethod
+    def from_config(cls, config: dict) -> "DiscoveryConfig":
+        def _int(key: str, default: int) -> int:
+            try:
+                return int(config[key])
+            except (KeyError, TypeError, ValueError):
+                return default
+
+        def _float(key: str, default: float) -> float:
+            try:
+                return float(config[key])
+            except (KeyError, TypeError, ValueError):
+                return default
+
+        return cls(
+            enabled=str(config.get("discover.enabled", "")).strip().lower() in _TRUE,
+            interval_hours=_int("discover.intervalHours", 168),
+            max_seeds=_int("discover.maxSeeds", 50),
+            similar_per_seed=_int("discover.similarPerSeed", 20),
+            albums_per_artist=_int("discover.albumsPerArtist", 1),
+            min_score=_float("discover.minScore", 0.0),
+        )
+
+
+@dataclass(frozen=True)
+class DiscoveryResult:
+    artists: int
+    albums: int
+
+
+def _seed_artists(conn: psycopg.Connection, cfg: DiscoveryConfig):
+    with conn.cursor() as cur:
+        cur.execute(
+            'SELECT mbid, name FROM "WatchedArtist" '
+            'WHERE "lastDiscoveredAt" IS NULL '
+            '   OR "lastDiscoveredAt" < now() - make_interval(hours => %s) '
+            'ORDER BY "lastDiscoveredAt" ASC NULLS FIRST LIMIT %s',
+            (cfg.interval_hours, cfg.max_seeds),
+        )
+        return cur.fetchall()
+
+
+def _followed_mbids(conn: psycopg.Connection) -> set[str]:
+    with conn.cursor() as cur:
+        cur.execute('SELECT mbid FROM "WatchedArtist"')
+        return {r[0] for r in cur.fetchall()}
+
+
+def _upsert_artist(conn: psycopg.Connection, cand: dict) -> int:
+    dedupe = f"artist:{cand['mbid']}:"
+    with conn.cursor() as cur:
+        cur.execute(
+            'INSERT INTO "DiscoverySuggestion" (id, kind, "artistMbid", "artistName", '
+            'score, "seedCount", sources, status, "dedupeKey", "createdAt", "updatedAt") '
+            "VALUES (gen_random_uuid()::text, 'artist', %s, %s, %s, %s, %s, 'pending', %s, now(), now()) "
+            'ON CONFLICT ("dedupeKey") DO UPDATE SET '
+            'score = EXCLUDED.score, "seedCount" = EXCLUDED."seedCount", '
+            'sources = EXCLUDED.sources, "updatedAt" = now() '
+            "WHERE \"DiscoverySuggestion\".status = 'pending'",
+            (cand["mbid"], cand["name"], cand["score"], cand["seed_count"],
+             sorted(cand["sources"]), dedupe),
+        )
+        return cur.rowcount
+
+
+def _derive_albums(conn: psycopg.Connection, browser: MbBrowser, surfaced: list[dict],
+                   cfg: DiscoveryConfig) -> int:
+    return 0  # replaced in the album-derivation task
+
+
+def run_discovery(conn: psycopg.Connection, sources: list[SimilaritySource],
+                  browser: MbBrowser, cfg: DiscoveryConfig) -> DiscoveryResult:
+    """Aggregate similar artists across the library's seed artists; upsert suggestions."""
+    seeds = _seed_artists(conn, cfg)
+    followed = _followed_mbids(conn)
+    live = [s for s in sources if s.health()]
+
+    agg: dict[str, dict] = {}
+    for seed_mbid, _seed_name in seeds:
+        for src in live:
+            try:
+                similar = src.similar_artists(seed_mbid)
+            except Exception as e:  # a bad source/seed must not abort the sweep
+                print(f"worker: discovery source {src.name} failed for {seed_mbid}: {e}", flush=True)
+                continue
+            for sa in sorted(similar, key=lambda a: a.score, reverse=True)[: cfg.similar_per_seed]:
+                if sa.mbid in followed:
+                    continue
+                c = agg.setdefault(sa.mbid, {"mbid": sa.mbid, "name": sa.name,
+                                             "score": 0.0, "seeds": set(), "sources": set()})
+                c["score"] += sa.score
+                c["seeds"].add(seed_mbid)
+                c["sources"].add(src.name)
+                if sa.name and not c["name"]:
+                    c["name"] = sa.name
+
+    artists = 0
+    surfaced: list[dict] = []
+    for c in sorted(agg.values(), key=lambda c: c["score"], reverse=True):
+        if c["score"] < cfg.min_score:
+            continue
+        cand = {"mbid": c["mbid"], "name": c["name"], "score": c["score"],
+                "seed_count": len(c["seeds"]), "sources": c["sources"]}
+        inserted = _upsert_artist(conn, cand)
+        artists += inserted
+        # Only derive albums for candidates that produced a live pending suggestion.
+        # A no-op upsert (rowcount 0) means the artist was already dismissed/wanted —
+        # don't surface their albums against a dismissal.
+        if inserted:
+            surfaced.append(cand)
+
+    albums = _derive_albums(conn, browser, surfaced, cfg)
+
+    with conn.cursor() as cur:
+        for seed_mbid, _seed_name in seeds:
+            cur.execute('UPDATE "WatchedArtist" SET "lastDiscoveredAt" = now() WHERE mbid = %s',
+                        (seed_mbid,))
+    conn.commit()
+    return DiscoveryResult(artists=artists, albums=albums)

worker/tests/test_discovery.py

@@ -0,0 +1,81 @@
+from lyra_worker.adapters.fakes import FakeMbBrowser, FakeSimilaritySource
+from lyra_worker.discovery import DiscoveryConfig, run_discovery
+from lyra_worker.similarity.base import SimilarArtist
+from tests.conftest import insert_discovery_suggestion, insert_watched_artist
+
+
+def _artist_rows(conn):
+    with conn.cursor() as cur:
+        cur.execute('SELECT "artistMbid", "artistName", score, "seedCount", sources, status::text '
+                    'FROM "DiscoverySuggestion" WHERE kind = \'artist\' ORDER BY score DESC')
+        return cur.fetchall()
+
+
+def test_aggregates_scores_across_seeds(conn):
+    insert_watched_artist(conn, mbid="s1", name="Seed One")
+    insert_watched_artist(conn, mbid="s2", name="Seed Two")
+    src = FakeSimilaritySource(similar={
+        "s1": [SimilarArtist("c1", "Shared", 0.6), SimilarArtist("c2", "Only1", 0.4)],
+        "s2": [SimilarArtist("c1", "Shared", 0.5)],
+    })
+    result = run_discovery(conn, [src], FakeMbBrowser(), DiscoveryConfig())
+    assert result.artists == 2
+    rows = _artist_rows(conn)
+    # c1 recommended by both seeds: score 1.1, seedCount 2; c2 by one seed
+    assert rows[0][0] == "c1" and abs(rows[0][2] - 1.1) < 1e-6 and rows[0][3] == 2
+    assert rows[0][4] == ["fake"] and rows[0][5] == "pending"
+    assert rows[1][0] == "c2" and rows[1][3] == 1
+
+
+def test_skips_already_followed_artists(conn):
+    insert_watched_artist(conn, mbid="s1", name="Seed")
+    insert_watched_artist(conn, mbid="c1", name="Already Followed")
+    src = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "X", 0.9),
+                                                SimilarArtist("c2", "New", 0.8)]})
+    run_discovery(conn, [src], FakeMbBrowser(), DiscoveryConfig())
+    assert [r[0] for r in _artist_rows(conn)] == ["c2"]  # c1 filtered (followed)
+
+
+def test_min_score_filters_weak_candidates(conn):
+    insert_watched_artist(conn, mbid="s1", name="Seed")
+    src = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "Strong", 0.9),
+                                                SimilarArtist("c2", "Weak", 0.1)]})
+    run_discovery(conn, [src], FakeMbBrowser(), DiscoveryConfig(min_score=0.5))
+    assert [r[0] for r in _artist_rows(conn)] == ["c1"]
+
+
+def test_dismissed_suggestion_is_never_revived(conn):
+    insert_watched_artist(conn, mbid="s1", name="Seed")
+    insert_discovery_suggestion(conn, kind="artist", artist_mbid="c1", status="dismissed", score=0.0)
+    src = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "X", 0.9)]})
+    run_discovery(conn, [src], FakeMbBrowser(), DiscoveryConfig())
+    with conn.cursor() as cur:
+        cur.execute('SELECT status::text, score FROM "DiscoverySuggestion" WHERE "artistMbid" = %s', ("c1",))
+        assert cur.fetchone() == ("dismissed", 0.0)  # untouched — WHERE status='pending' blocked the update
+
+
+def test_stamps_last_discovered_and_throttles(conn):
+    insert_watched_artist(conn, mbid="s1", name="Seed")
+    src = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "X", 0.9)]})
+    assert run_discovery(conn, [src], FakeMbBrowser(), DiscoveryConfig()).artists == 1
+    with conn.cursor() as cur:
+        cur.execute('SELECT "lastDiscoveredAt" IS NOT NULL FROM "WatchedArtist" WHERE mbid = %s', ("s1",))
+        assert cur.fetchone()[0] is True
+    # second run: seed not due (lastDiscoveredAt recent) -> no seeds -> nothing new
+    assert run_discovery(conn, [src], FakeMbBrowser(), DiscoveryConfig()).artists == 0
+
+
+def test_unhealthy_source_contributes_nothing(conn):
+    insert_watched_artist(conn, mbid="s1", name="Seed")
+    src = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "X", 0.9)]}, healthy=False)
+    assert run_discovery(conn, [src], FakeMbBrowser(), DiscoveryConfig()).artists == 0
+    assert _artist_rows(conn) == []
+
+
+def test_config_from_config_parses_and_defaults():
+    cfg = DiscoveryConfig.from_config({"discover.enabled": "true", "discover.maxSeeds": "10",
+                                       "discover.minScore": "0.3"})
+    assert cfg.enabled is True and cfg.max_seeds == 10 and cfg.min_score == 0.3
+    d = DiscoveryConfig.from_config({})
+    assert (d.enabled, d.interval_hours, d.max_seeds, d.similar_per_seed,
+            d.albums_per_artist, d.min_score) == (False, 168, 50, 20, 1, 0.0)