from dataclasses import dataclass from datetime import date, datetime import psycopg from lyra_worker.browser import MbBrowser from lyra_worker.release_filter import is_core_release _TRUE = {"1", "true", "yes", "on"} @dataclass(frozen=True) class MonitorConfig: enabled: bool = False poll_interval_hours: int = 24 retry_interval_hours: int = 6 quality_cutoff: int = 2 upgrade_window_days: int = 14 @classmethod def from_config(cls, config: dict) -> "MonitorConfig": def _int(key: str, default: int) -> int: try: return int(config[key]) except (KeyError, TypeError, ValueError): return default return cls( enabled=str(config.get("monitor.enabled", "")).strip().lower() in _TRUE, poll_interval_hours=_int("monitor.pollIntervalHours", 24), retry_interval_hours=_int("monitor.retryIntervalHours", 6), quality_cutoff=_int("monitor.qualityCutoff", 2), upgrade_window_days=_int("monitor.upgradeWindowDays", 14), ) def _is_new(first_release_date: str, monitor_from: datetime) -> bool: """A release is 'new going forward' if its first-release date is after monitorFrom.""" if not first_release_date: return False try: frd = date.fromisoformat(first_release_date[:10]) except ValueError: return False return frd > monitor_from.date() def discover(conn: psycopg.Connection, browser: MbBrowser, cfg: MonitorConfig) -> int: """Poll each due watched artist; insert newly-seen release-groups. Returns #rows inserted.""" with conn.cursor() as cur: cur.execute( 'SELECT id, mbid, name, "autoMonitorFuture", "monitorFrom" FROM "WatchedArtist" ' 'WHERE "lastPolledAt" IS NULL ' ' OR "lastPolledAt" < now() - make_interval(hours => %s)', (cfg.poll_interval_hours,), ) artists = cur.fetchall() inserted = 0 for artist_id, mbid, name, auto_future, monitor_from in artists: for rg in browser.browse_release_groups(mbid): monitored = ( bool(auto_future) and _is_new(rg.first_release_date, monitor_from) and is_core_release(rg.primary_type, rg.secondary_types) ) with conn.cursor() as cur: cur.execute( 'INSERT INTO "MonitoredRelease" (id, "watchedArtistId", "artistMbid", ' '"artistName", "rgMbid", album, "primaryType", "secondaryTypes", ' '"firstReleaseDate", monitored, state, "createdAt") ' "VALUES (gen_random_uuid()::text, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'wanted', now()) " 'ON CONFLICT ("rgMbid") DO NOTHING', (artist_id, mbid, name, rg.rg_mbid, rg.title, rg.primary_type or None, list(rg.secondary_types), rg.first_release_date or None, monitored), ) inserted += cur.rowcount with conn.cursor() as cur: cur.execute('UPDATE "WatchedArtist" SET "lastPolledAt" = now() WHERE id = %s', (artist_id,)) conn.commit() return inserted def _set_state(conn: psycopg.Connection, rel_id: str, state: str) -> None: with conn.cursor() as cur: cur.execute('UPDATE "MonitoredRelease" SET state = %s WHERE id = %s', (state, rel_id)) def _in_library_at_cutoff( conn: psycopg.Connection, artist: str, album: str, rg_mbid: str, cutoff: int ) -> bool: """True if the album is owned at/above `cutoff`. Matches on the release-group MBID when the library row has one — robust against MusicBrainz artist-name canonicalization (e.g. the followed 'Kanye West' imports as 'Ye') — and falls back to exact artist+album for scanned rows that predate MBID capture.""" with conn.cursor() as cur: cur.execute( 'SELECT 1 FROM "LibraryItem" WHERE "qualityClass" >= %s AND (' ' ("rgMbid" IS NOT NULL AND "rgMbid" = %s) ' ' OR ("rgMbid" IS NULL AND artist = %s AND album = %s)) LIMIT 1', (cutoff, rg_mbid, artist, album), ) return cur.fetchone() is not None def _retire_failed_attempts(conn: psycopg.Connection, rel_id: str) -> None: """Delete prior failed (needs_attention) Request rows for this release (Job/Candidate cascade). A hard-to-match release is re-pressed every retry interval; each failed attempt used to leave a needs_attention Request behind, piling up phantom "needs attention" cards on the press even after a later attempt imported the album. Retiring them when the release is fulfilled or re-attempted keeps the press showing one live attempt at a time. Manual (non-monitor) requests have no monitoredReleaseId and are never touched.""" with conn.cursor() as cur: cur.execute( 'DELETE FROM "Request" r USING "Job" j ' 'WHERE j."requestId" = r.id AND r."monitoredReleaseId" = %s ' "AND j.state = 'needs_attention'", (rel_id,), ) def _enqueue_one(conn: psycopg.Connection, rel_id: str, artist: str, album: str) -> None: with conn.cursor() as cur: cur.execute( 'INSERT INTO "Request" (id, artist, album, status, "monitoredReleaseId", "createdAt") ' "VALUES (gen_random_uuid()::text, %s, %s, 'pending', %s, now()) RETURNING id", (artist, album, rel_id), ) request_id = cur.fetchone()[0] cur.execute( 'INSERT INTO "Job" (id, "requestId", state, "currentStage", attempts, "createdAt", "updatedAt") ' "VALUES (gen_random_uuid()::text, %s, 'requested', 'intake', 0, now(), now())", (request_id,), ) cur.execute('UPDATE "MonitoredRelease" SET "lastSearchedAt" = now() WHERE id = %s', (rel_id,)) def enqueue_due(conn: psycopg.Connection, cfg: MonitorConfig) -> int: """Enqueue due wanted/upgrade releases as pipeline jobs. Returns #jobs enqueued.""" with conn.cursor() as cur: cur.execute( 'SELECT mr.id, mr."artistName", mr.album, mr.state, mr."currentQualityClass", ' ' (mr."firstGrabbedAt" IS NOT NULL ' ' AND mr."firstGrabbedAt" < now() - make_interval(days => %s)) AS window_expired, ' ' mr."rgMbid" ' 'FROM "MonitoredRelease" mr ' 'WHERE mr.monitored = true ' ' AND NOT mr.ignored ' # user-ignored releases are never enqueued " AND mr.state <> 'fulfilled' " ' AND (mr."lastSearchedAt" IS NULL ' ' OR mr."lastSearchedAt" < now() - make_interval(hours => %s)) ' ' AND NOT EXISTS (' ' SELECT 1 FROM "Request" r JOIN "Job" j ON j."requestId" = r.id ' ' WHERE r."monitoredReleaseId" = mr.id ' " AND j.state NOT IN ('imported', 'needs_attention'))", (cfg.upgrade_window_days, cfg.retry_interval_hours), ) rows = cur.fetchall() enqueued = 0 for rel_id, artist, album, state, quality, window_expired, rg_mbid in rows: # A due release is about to be fulfilled or re-attempted; either way any earlier failed # attempt is superseded, so retire it before deciding. _retire_failed_attempts(conn, rel_id) if _in_library_at_cutoff(conn, artist, album, rg_mbid, cfg.quality_cutoff): _set_state(conn, rel_id, "fulfilled") continue if state == "grabbed" and (window_expired or (quality is not None and quality >= cfg.quality_cutoff)): _set_state(conn, rel_id, "fulfilled") continue _enqueue_one(conn, rel_id, artist, album) enqueued += 1 conn.commit() return enqueued def fulfill_owned_releases(conn: psycopg.Connection, cfg: MonitorConfig) -> int: """Mark fulfilled any monitored, not-yet-fulfilled release already owned at/above the quality cutoff. Run once at startup so releases pressed while the monitor was off (and therefore never reconciled) drop off the wanted list. Idempotent. Returns #rows fixed.""" with conn.cursor() as cur: cur.execute( 'UPDATE "MonitoredRelease" mr SET state = \'fulfilled\', ' ' "currentQualityClass" = owned.q, ' ' "firstGrabbedAt" = COALESCE(mr."firstGrabbedAt", now()) ' 'FROM (' ' SELECT mr2.id AS mid, max(li."qualityClass") AS q ' ' FROM "MonitoredRelease" mr2 JOIN "LibraryItem" li ON (' # match on rgMbid where the library row has one (survives artist-name canonicalization), # else fall back to exact artist+album for scanned rows lacking an MBID ' (li."rgMbid" IS NOT NULL AND li."rgMbid" = mr2."rgMbid") ' ' OR (li."rgMbid" IS NULL AND li.artist = mr2."artistName" AND li.album = mr2.album)) ' " WHERE mr2.monitored = true AND mr2.state <> 'fulfilled' " ' GROUP BY mr2.id HAVING max(li."qualityClass") >= %s' ') owned ' 'WHERE mr.id = owned.mid', (cfg.quality_cutoff,), ) fixed = cur.rowcount conn.commit() return fixed def reconcile(conn: psycopg.Connection, job_id: str, cfg: MonitorConfig) -> None: """Feed a finished job's outcome back into its MonitoredRelease (if any).""" with conn.cursor() as cur: cur.execute( 'SELECT j.state, r."monitoredReleaseId", r.artist, r.album, mr."rgMbid" ' 'FROM "Job" j JOIN "Request" r ON r.id = j."requestId" ' 'LEFT JOIN "MonitoredRelease" mr ON mr.id = r."monitoredReleaseId" ' 'WHERE j.id = %s', (job_id,), ) row = cur.fetchone() if row is None: return state, rel_id, artist, album, rg_mbid = row if rel_id is None or state != "imported": return # needs_attention (or in-flight) leaves the release wanted # A successful import supersedes any earlier failed attempts for this release. _retire_failed_attempts(conn, rel_id) with conn.cursor() as cur: cur.execute( 'SELECT max("qualityClass") FROM "LibraryItem" WHERE ' ' ("rgMbid" IS NOT NULL AND "rgMbid" = %s) ' ' OR ("rgMbid" IS NULL AND artist = %s AND album = %s)', (rg_mbid, artist, album), ) lib = cur.fetchone() quality = lib[0] if lib else None new_state = "fulfilled" if (quality is not None and quality >= cfg.quality_cutoff) else "grabbed" with conn.cursor() as cur: cur.execute( 'UPDATE "MonitoredRelease" ' 'SET "currentQualityClass" = %s, state = %s, ' ' "firstGrabbedAt" = COALESCE("firstGrabbedAt", now()) WHERE id = %s', (quality, new_state, rel_id), ) conn.commit() def sweep(conn: psycopg.Connection, browser: MbBrowser, cfg: MonitorConfig) -> None: """One monitor pass: discover new releases, then enqueue everything due.""" discover(conn, browser, cfg) enqueue_due(conn, cfg)