From ccf3fb59ab033051122d51d2fa9808241aa58c99 Mon Sep 17 00:00:00 2001 From: Jonathan Date: Sun, 12 Jul 2026 00:07:09 +0200 Subject: [PATCH] feat(discovery): run_discovery aggregates similar artists into suggestions Co-Authored-By: Claude Opus 4.8 (1M context) --- worker/lyra_worker/discovery.py | 138 ++++++++++++++++++++++++++++++++ worker/tests/test_discovery.py | 81 +++++++++++++++++++ 2 files changed, 219 insertions(+) create mode 100644 worker/lyra_worker/discovery.py create mode 100644 worker/tests/test_discovery.py diff --git a/worker/lyra_worker/discovery.py b/worker/lyra_worker/discovery.py new file mode 100644 index 0000000..dbff2bd --- /dev/null +++ b/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) diff --git a/worker/tests/test_discovery.py b/worker/tests/test_discovery.py new file mode 100644 index 0000000..47ae187 --- /dev/null +++ b/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)