from dataclasses import dataclass import psycopg from lyra_worker.browser import MbBrowser from lyra_worker.release_filter import is_core_release from lyra_worker.similarity.base import SimilaritySource _TRUE = {"1", "true", "yes", "on"} @dataclass(frozen=True) class DiscoveryConfig: enabled: bool = False interval_hours: int = 168 chunk_size: int = 5 similar_per_seed: int = 20 albums_per_artist: int = 1 min_score: float = 0.0 albums_only: bool = True seed_from_library: bool = True @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 def _bool(key: str, default: bool) -> bool: raw = config.get(key) if raw is None: return default return str(raw).strip().lower() in _TRUE return cls( enabled=str(config.get("discover.enabled", "")).strip().lower() in _TRUE, interval_hours=_int("discover.intervalHours", 168), chunk_size=_int("discover.chunkSize", 5), similar_per_seed=_int("discover.similarPerSeed", 20), albums_per_artist=_int("discover.albumsPerArtist", 1), min_score=_float("discover.minScore", 0.0), albums_only=_bool("discover.albumsOnly", True), seed_from_library=_bool("discover.seedFromLibrary", True), ) @dataclass(frozen=True) class DiscoveryResult: artists: int albums: int seeds: int = 0 def _healthy_sources(sources): """Return sources whose health() is truthy; a source whose health() raises is logged and skipped, never aborting the sweep.""" live = [] for s in sources: try: if s.health(): live.append(s) except Exception as e: # a bad health check must not abort the sweep print(f"worker: discovery source {getattr(s, 'name', '?')} health check failed: {e}", flush=True) return live def _seed_artists(conn: psycopg.Connection, cfg: DiscoveryConfig): """Return up to chunk_size due seeds as (mbid, name, source) where source is 'watched' (a followed artist, throttled on WatchedArtist.lastDiscoveredAt) or 'library' (an artist you own albums by but don't follow, throttled on DiscoverySeed.lastDiscoveredAt). Both are ordered least-recently-swept first so a sweep drains all eligible seeds across chunks.""" interval = cfg.interval_hours with conn.cursor() as cur: cur.execute( 'SELECT mbid, name, \'watched\' AS source, "lastDiscoveredAt" FROM "WatchedArtist" ' 'WHERE "lastDiscoveredAt" IS NULL OR "lastDiscoveredAt" < now() - make_interval(hours => %s) ', (interval,), ) rows = cur.fetchall() if cfg.seed_from_library: # Distinct owned-album artists (with a captured MBID) that we don't already follow, # left-joined to their throttle row. A brand-new library seed (no DiscoverySeed row) # is immediately eligible via NULLS FIRST. cur.execute( 'SELECT DISTINCT ON (li."artistMbid") li."artistMbid", li.artist, \'library\', ds."lastDiscoveredAt" ' 'FROM "LibraryItem" li ' 'LEFT JOIN "DiscoverySeed" ds ON ds.mbid = li."artistMbid" ' 'WHERE li."artistMbid" IS NOT NULL ' ' AND li."artistMbid" NOT IN (SELECT mbid FROM "WatchedArtist") ' ' AND (ds."lastDiscoveredAt" IS NULL OR ds."lastDiscoveredAt" < now() - make_interval(hours => %s))', (interval,), ) rows += cur.fetchall() # Merge both sources, least-recently-swept first (NULL = never), cap at chunk_size. rows.sort(key=lambda r: (r[3] is not None, r[3])) return [(mbid, name, source) for mbid, name, source, _last in rows[: cfg.chunk_size]] 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 _owned_artist_mbids(conn: psycopg.Connection) -> set[str]: """MBIDs of artists you own albums by (captured on LibraryItem). Used both as extra discovery seeds and to keep already-owned artists out of the suggestion candidates.""" with conn.cursor() as cur: cur.execute('SELECT DISTINCT "artistMbid" FROM "LibraryItem" WHERE "artistMbid" IS NOT NULL') 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 _existing_rg_mbids(conn: psycopg.Connection) -> set[str]: with conn.cursor() as cur: cur.execute('SELECT "rgMbid" FROM "MonitoredRelease"') return {r[0] for r in cur.fetchall()} def _upsert_album(conn: psycopg.Connection, artist: dict, rg, reason: str) -> int: dedupe = f"album:{artist['mbid']}:{rg.rg_mbid}" with conn.cursor() as cur: cur.execute( 'INSERT INTO "DiscoverySuggestion" (id, kind, "artistMbid", "artistName", ' '"rgMbid", album, "primaryType", "secondaryTypes", "firstReleaseDate", ' 'score, "seedCount", sources, "albumReason", status, "dedupeKey", "createdAt", "updatedAt") ' "VALUES (gen_random_uuid()::text, 'album', %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, " "'pending', %s, now(), now()) " 'ON CONFLICT ("dedupeKey") DO UPDATE SET score = EXCLUDED.score, ' '"albumReason" = EXCLUDED."albumReason", "updatedAt" = now() ' "WHERE \"DiscoverySuggestion\".status = 'pending'", (artist["mbid"], artist["name"], rg.rg_mbid, rg.title, rg.primary_type or None, list(rg.secondary_types), rg.first_release_date or None, artist["score"], artist["seed_count"], sorted(artist["sources"]), reason, dedupe), ) return cur.rowcount def _album_kind_ok(g, albums_only: bool) -> bool: """Which release groups qualify as discovery album suggestions. Discovery-only policy: when albums_only (default), suggest full studio Albums only — no Singles/EPs, no secondary types (live/compilation/etc.). Falls back to the broader is_core_release (Album/Single/EP) when the toggle is off. This deliberately does NOT touch is_core_release, which the monitor/scan still use.""" if albums_only: return (g.primary_type or "") == "Album" and len(g.secondary_types or ()) == 0 return is_core_release(g.primary_type, g.secondary_types) def _norm_title(s: str) -> str: return (s or "").strip().lower() def _popular_pick(conn, popularity, artist_name: str, core: list) -> str | None: """The rg_mbid of the artist's most-played qualifying un-owned album (Last.fm), matched into the already-browsed `core` by MBID first then normalized title. None if unavailable.""" if popularity is None or not core: return None by_mbid = {g.rg_mbid: g for g in core} by_title = {_norm_title(g.title): g for g in core} for name, mbid, _plays in popularity.top_albums(conn, artist_name): g = (by_mbid.get(mbid) if mbid else None) or by_title.get(_norm_title(name)) if g is not None: return g.rg_mbid return None def _derive_albums(conn: psycopg.Connection, browser: MbBrowser, surfaced: list[dict], cfg: DiscoveryConfig, popularity=None) -> int: if cfg.albums_per_artist <= 0: return 0 have = _existing_rg_mbids(conn) albums = 0 for artist in surfaced: try: groups = browser.browse_release_groups(artist["mbid"]) except Exception as e: # one artist's browse failure must not abort the sweep print(f"worker: discovery album browse failed for {artist['mbid']}: {e}", flush=True) continue core = [g for g in groups if _album_kind_ok(g, cfg.albums_only) and g.rg_mbid not in have] core.sort(key=lambda g: g.first_release_date or "", reverse=True) # newest picks (existing behavior) + the single most-played album (Last.fm); merged so # an album that is both newest AND most-played is one row tagged "newest,popular". reasons: dict[str, str] = {} for rg in core[: cfg.albums_per_artist]: reasons[rg.rg_mbid] = "newest" popular = _popular_pick(conn, popularity, artist["name"], core) if popular is not None: reasons[popular] = "newest,popular" if reasons.get(popular) == "newest" else "popular" for rg in core: # preserve newest-first order among the chosen reason = reasons.get(rg.rg_mbid) if reason is not None: albums += _upsert_album(conn, artist, rg, reason) have.add(rg.rg_mbid) return albums def _record_contributions(conn: psycopg.Connection, seed_mbid: str, live, followed: set[str], cfg: DiscoveryConfig) -> set[str]: """Replace seed_mbid's contribution rows with its current similar-artist scores. Returns the set of candidate mbids whose aggregate may have changed (old contributors of this seed ∪ newly inserted), so callers recompute drop-offs too.""" with conn.cursor() as cur: cur.execute('SELECT "candidateMbid" FROM "DiscoverySeedContribution" WHERE "seedMbid" = %s', (seed_mbid,)) affected = {r[0] for r in cur.fetchall()} cur.execute('DELETE FROM "DiscoverySeedContribution" WHERE "seedMbid" = %s', (seed_mbid,)) per_cand: dict[str, dict] = {} 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 top = sorted(similar, key=lambda a: a.score, reverse=True)[: cfg.similar_per_seed] # Max-normalize each source to [0, 1] by its own top score before summing across # sources, so sources on wildly different scales contribute comparably (Last.fm's # `match` is 0–1; ListenBrainz co-occurrence counts run to the thousands — raw sums # let ListenBrainz drown Last.fm's ranking signal). max_score = max((sa.score for sa in top), default=0.0) for sa in top: if sa.mbid in followed: continue norm = sa.score / max_score if max_score > 0 else 0.0 c = per_cand.setdefault(sa.mbid, {"name": sa.name, "score": 0.0, "sources": set()}) c["score"] += norm c["sources"].add(src.name) if sa.name and not c["name"]: c["name"] = sa.name with conn.cursor() as cur: for mbid, c in per_cand.items(): cur.execute( 'INSERT INTO "DiscoverySeedContribution" (id, "candidateMbid", "candidateName", ' '"seedMbid", score, sources, "updatedAt") ' 'VALUES (gen_random_uuid()::text, %s, %s, %s, %s, %s, now())', (mbid, c["name"], seed_mbid, c["score"], sorted(c["sources"])), ) affected.add(mbid) return affected def _recompute_suggestions(conn: psycopg.Connection, affected: set[str], cfg: DiscoveryConfig) -> list[dict]: """Recompute each affected candidate's DiscoverySuggestion from the sum of its contributions. Returns the candidates that produced a live pending suggestion.""" if not affected: return [] with conn.cursor() as cur: cur.execute( 'SELECT "candidateMbid", "candidateName", score, "seedMbid", sources ' 'FROM "DiscoverySeedContribution" WHERE "candidateMbid" = ANY(%s)', (list(affected),)) rows = cur.fetchall() agg: dict[str, dict] = {} for mbid, name, score, seed, sources in rows: a = agg.setdefault(mbid, {"mbid": mbid, "name": name, "score": 0.0, "seeds": set(), "sources": set()}) a["score"] += score a["seeds"].add(seed) a["sources"].update(sources or []) if name and not a["name"]: a["name"] = name surfaced: list[dict] = [] for a in sorted(agg.values(), key=lambda a: a["score"], reverse=True): if a["score"] < cfg.min_score: continue cand = {"mbid": a["mbid"], "name": a["name"], "score": a["score"], "seed_count": len(a["seeds"]), "sources": a["sources"]} if _upsert_artist(conn, cand): # rowcount 1 => a live pending suggestion surfaced.append(cand) return surfaced def run_discovery(conn: psycopg.Connection, sources: list[SimilaritySource], browser: MbBrowser, cfg: DiscoveryConfig, popularity=None) -> DiscoveryResult: """Record each seed's similar-artist contributions, then recompute affected suggestions from the summed contributions — so a candidate's score is the sum of every seed's latest contribution, independent of chunk and sweep boundaries.""" seeds = _seed_artists(conn, cfg) followed = _followed_mbids(conn) owned = _owned_artist_mbids(conn) if cfg.seed_from_library else set() # Artists you already have (followed or owned) are seeds/inputs, never suggested as # candidates — discovery only surfaces genuinely new artists. known = followed | owned live = _healthy_sources(sources) # Evict contributions whose seed is no longer valid — an unfollowed artist AND (when # library seeding is on) an artist whose last owned album was deleted. Neither has an FK # cascade, so without this a stale seed's score would linger and inflate the aggregate. # Seed `affected` with the touched candidates so their scores recompute from live seeds. valid_seeds = list(known) with conn.cursor() as cur: cur.execute('SELECT DISTINCT "candidateMbid" FROM "DiscoverySeedContribution" ' 'WHERE NOT ("seedMbid" = ANY(%s))', (valid_seeds,)) affected = {r[0] for r in cur.fetchall()} cur.execute('DELETE FROM "DiscoverySeedContribution" WHERE NOT ("seedMbid" = ANY(%s))', (valid_seeds,)) for seed_mbid, _seed_name, _source in seeds: affected |= _record_contributions(conn, seed_mbid, live, known, cfg) surfaced = _recompute_suggestions(conn, affected, cfg) albums = _derive_albums(conn, browser, surfaced, cfg, popularity) with conn.cursor() as cur: for seed_mbid, _seed_name, source in seeds: if source == "library": cur.execute( 'INSERT INTO "DiscoverySeed" (mbid, "lastDiscoveredAt") VALUES (%s, now()) ' 'ON CONFLICT (mbid) DO UPDATE SET "lastDiscoveredAt" = now()', (seed_mbid,)) else: cur.execute('UPDATE "WatchedArtist" SET "lastDiscoveredAt" = now() WHERE mbid = %s', (seed_mbid,)) conn.commit() return DiscoveryResult(artists=len(surfaced), albums=albums, seeds=len(seeds))