Jonathan
e8483ee22b
Two related monitor bugs surfaced by owned albums that never leave the press:
1. Ownership was matched on (artist, album) strings, but MusicBrainz canonicalizes
the artist on import ("Kanye West"->"Ye", "Ne-Yo"->"Ne‐Yo", "The Goo Goo Dolls"
->"Goo Goo Dolls"). The strings no longer matched, so reconcile/enqueue/backfill
never recognized the album as owned — it stayed "wanted" and re-pressed every
retry interval. Match on the release-group MBID when the library row has one
(falling back to artist+album for scanned rows without an MBID) across
_in_library_at_cutoff, fulfill_owned_releases, and reconcile.
2. Each failed attempt left a needs_attention Request behind, so a hard-to-match
album piled up phantom "needs attention" cards on the press even after a later
attempt imported it (e.g. Maroon 5 "V": 3 stale cards + 1 success). Retire prior
failed attempts when a release is fulfilled or re-attempted (_retire_failed_attempts,
monitor-owned requests only — manual requests have no monitoredReleaseId).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
249 lines
11 KiB
Python
249 lines
11 KiB
Python
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)
|