Jonathan
9915bfbda2
POST /api/discover/reset sets discover.resetRequested; the worker, at the start of the next sweep, clears all PENDING suggestions + contributions + seed throttles and nulls WatchedArtist.lastDiscoveredAt (keeping followed/wanted/dismissed decisions), then regenerates everything with current logic. A 'Reset' button on /discover (2-step confirm) triggers it. worker 247, web 190 tests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
373 lines
17 KiB
Python
373 lines
17 KiB
Python
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"
|
||
|
||
# Re-deriving replaces this artist's album set: drop any pending album suggestion of
|
||
# theirs that is no longer a pick (stale singles/EPs from older sweeps, or an album now
|
||
# owned), keeping wanted/dismissed rows untouched. Then upsert the current picks.
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"DELETE FROM \"DiscoverySuggestion\" WHERE kind = 'album' AND status = 'pending' "
|
||
'AND "artistMbid" = %s AND NOT ("rgMbid" = ANY(%s))',
|
||
(artist["mbid"], list(reasons.keys())),
|
||
)
|
||
|
||
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 reset_pending(conn: psycopg.Connection) -> None:
|
||
"""Clear discovery state so the next sweep rebuilds from scratch with current logic. Drops
|
||
all PENDING suggestions (followed/wanted/dismissed user decisions are kept), every seed
|
||
contribution + library-seed throttle, and marks each watched artist due again."""
|
||
with conn.cursor() as cur:
|
||
cur.execute("DELETE FROM \"DiscoverySuggestion\" WHERE status = 'pending'")
|
||
cur.execute('DELETE FROM "DiscoverySeedContribution"')
|
||
cur.execute('DELETE FROM "DiscoverySeed"')
|
||
cur.execute('UPDATE "WatchedArtist" SET "lastDiscoveredAt" = NULL')
|
||
conn.commit()
|
||
|
||
|
||
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))
|