fix: stop enrichment workers from re-processing rows forever
Four enrichment workers (Last.fm, MusicBrainz, Tidal, Qobuz) had a bug where every background loop re-processed the same rows because the existing-ID short-circuit path never set match_status, and two workers queried the wrong column when checking for an existing ID. lastfm_worker._get_existing_id queried a non-existent lastfm_id column; the real column is lastfm_url. The method now reads lastfm_url for all three entity types. musicbrainz_worker._get_existing_id queried musicbrainz_id for all entity types, but albums use musicbrainz_release_id and tracks use musicbrainz_recording_id. The method now uses a per-type column map. All four workers (lastfm, musicbrainz, tidal, qobuz) now write match_status='matched' when they short-circuit on an already-present external ID, so these rows are no longer re-selected on the next worker sweep. A new migration (_backfill_match_status_for_existing_ids) runs once on startup to retroactively set match_status='matched' for rows that already have an external ID but NULL match_status. This covers legacy data, manual matches, and rows populated from file tags outside the worker.
This commit is contained in:
JohnBaumb
parent
b1c2e89595
commit
f4c8c231a7
5 changed files with 106 additions and 10 deletions
|
|
@ -307,7 +307,12 @@ class LastFMWorker:
|
|||
logger.error(f"Error updating item status: {e2}")
|
||||
|
||||
def _get_existing_id(self, entity_type: str, entity_id: int) -> Optional[str]:
|
||||
"""Check if an entity already has a lastfm_id (e.g. from manual match)."""
|
||||
"""Check if an entity already has a lastfm_url (e.g. from manual match).
|
||||
|
||||
The Last.fm schema uses `lastfm_url` (not `lastfm_id`) on artists, albums,
|
||||
and tracks. This helper always returned None before because it queried a
|
||||
non-existent column and silently caught the exception.
|
||||
"""
|
||||
table_map = {'artist': 'artists', 'album': 'albums', 'track': 'tracks'}
|
||||
table = table_map.get(entity_type)
|
||||
if not table:
|
||||
|
|
@ -316,7 +321,7 @@ class LastFMWorker:
|
|||
try:
|
||||
conn = self.db._get_connection()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(f"SELECT lastfm_id FROM {table} WHERE id = ?", (entity_id,))
|
||||
cursor.execute(f"SELECT lastfm_url FROM {table} WHERE id = ?", (entity_id,))
|
||||
row = cursor.fetchone()
|
||||
return row[0] if row and row[0] else None
|
||||
except Exception:
|
||||
|
|
@ -329,7 +334,10 @@ class LastFMWorker:
|
|||
"""Process an artist: get full info from Last.fm"""
|
||||
existing_id = self._get_existing_id('artist', artist_id)
|
||||
if existing_id:
|
||||
logger.debug(f"Preserving existing Last.fm ID for artist '{artist_name}': {existing_id}")
|
||||
# Row already has a Last.fm URL but status is NULL (legacy/manual match).
|
||||
# Mark as matched so the worker does not re-select it forever.
|
||||
logger.debug(f"Preserving existing Last.fm URL for artist '{artist_name}': {existing_id}")
|
||||
self._mark_status('artist', artist_id, 'matched')
|
||||
return
|
||||
|
||||
# Use get_artist_info for detailed data (includes stats, bio, tags, similar)
|
||||
|
|
@ -353,7 +361,8 @@ class LastFMWorker:
|
|||
"""Process an album: get full info from Last.fm"""
|
||||
existing_id = self._get_existing_id('album', album_id)
|
||||
if existing_id:
|
||||
logger.debug(f"Preserving existing Last.fm ID for album '{album_name}': {existing_id}")
|
||||
logger.debug(f"Preserving existing Last.fm URL for album '{album_name}': {existing_id}")
|
||||
self._mark_status('album', album_id, 'matched')
|
||||
return
|
||||
|
||||
result = self.client.get_album_info(artist_name, album_name)
|
||||
|
|
@ -376,7 +385,8 @@ class LastFMWorker:
|
|||
"""Process a track: get full info from Last.fm"""
|
||||
existing_id = self._get_existing_id('track', track_id)
|
||||
if existing_id:
|
||||
logger.debug(f"Preserving existing Last.fm ID for track '{track_name}': {existing_id}")
|
||||
logger.debug(f"Preserving existing Last.fm URL for track '{track_name}': {existing_id}")
|
||||
self._mark_status('track', track_id, 'matched')
|
||||
return
|
||||
|
||||
result = self.client.get_track_info(artist_name, track_name)
|
||||
|
|
|
|||
|
|
@ -254,16 +254,27 @@ class MusicBrainzWorker:
|
|||
conn.close()
|
||||
|
||||
def _get_existing_id(self, entity_type: str, entity_id: int) -> Optional[str]:
|
||||
"""Check if an entity already has a musicbrainz_id (e.g. from manual match)."""
|
||||
table_map = {'artist': 'artists', 'album': 'albums', 'track': 'tracks'}
|
||||
table = table_map.get(entity_type)
|
||||
if not table:
|
||||
"""Check if an entity already has a MusicBrainz ID (e.g. from manual match).
|
||||
|
||||
MusicBrainz ID columns differ per entity type: artists use `musicbrainz_id`,
|
||||
albums use `musicbrainz_release_id`, and tracks use `musicbrainz_recording_id`.
|
||||
Before this fix, all three were queried as `musicbrainz_id`, so the
|
||||
existing-ID check silently failed for albums and tracks.
|
||||
"""
|
||||
table_config = {
|
||||
'artist': ('artists', 'musicbrainz_id'),
|
||||
'album': ('albums', 'musicbrainz_release_id'),
|
||||
'track': ('tracks', 'musicbrainz_recording_id'),
|
||||
}
|
||||
cfg = table_config.get(entity_type)
|
||||
if not cfg:
|
||||
return None
|
||||
table, column = cfg
|
||||
conn = None
|
||||
try:
|
||||
conn = self.db._get_connection()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(f"SELECT musicbrainz_id FROM {table} WHERE id = ?", (entity_id,))
|
||||
cursor.execute(f"SELECT {column} FROM {table} WHERE id = ?", (entity_id,))
|
||||
row = cursor.fetchone()
|
||||
return row[0] if row and row[0] else None
|
||||
except Exception:
|
||||
|
|
@ -285,6 +296,17 @@ class MusicBrainzWorker:
|
|||
existing_id = self._get_existing_id(item_type, item_id)
|
||||
if existing_id:
|
||||
logger.debug(f"Preserving existing MusicBrainz ID for {item_type} '{item_name}': {existing_id}")
|
||||
# Mark as matched so this row is not re-selected forever when
|
||||
# match_status is NULL but the MBID is already populated.
|
||||
try:
|
||||
if item_type == 'artist':
|
||||
self.mb_service.update_artist_mbid(item_id, existing_id, 'matched')
|
||||
elif item_type == 'album':
|
||||
self.mb_service.update_album_mbid(item_id, existing_id, 'matched')
|
||||
elif item_type == 'track':
|
||||
self.mb_service.update_track_mbid(item_id, existing_id, 'matched')
|
||||
except Exception as mark_err:
|
||||
logger.error(f"Error marking {item_type} #{item_id} matched: {mark_err}")
|
||||
return
|
||||
|
||||
if item_type == 'artist':
|
||||
|
|
|
|||
|
|
@ -388,6 +388,8 @@ class QobuzWorker:
|
|||
existing_id = self._get_existing_id('artist', artist_id)
|
||||
if existing_id:
|
||||
logger.debug(f"Preserving existing Qobuz ID for artist '{artist_name}': {existing_id}")
|
||||
# Mark as matched so this row is not re-selected on every loop.
|
||||
self._mark_status('artist', artist_id, 'matched')
|
||||
return
|
||||
|
||||
result = self.client.search_artist(artist_name)
|
||||
|
|
@ -429,6 +431,7 @@ class QobuzWorker:
|
|||
existing_id = self._get_existing_id('album', album_id)
|
||||
if existing_id:
|
||||
logger.debug(f"Preserving existing Qobuz ID for album '{album_name}': {existing_id}")
|
||||
self._mark_status('album', album_id, 'matched')
|
||||
return
|
||||
|
||||
result = self.client.search_album(artist_name, album_name)
|
||||
|
|
@ -484,6 +487,7 @@ class QobuzWorker:
|
|||
existing_id = self._get_existing_id('track', track_id)
|
||||
if existing_id:
|
||||
logger.debug(f"Preserving existing Qobuz ID for track '{track_name}': {existing_id}")
|
||||
self._mark_status('track', track_id, 'matched')
|
||||
return
|
||||
|
||||
result = self.client.search_track(artist_name, track_name)
|
||||
|
|
|
|||
|
|
@ -401,6 +401,8 @@ class TidalWorker:
|
|||
existing_id = self._get_existing_id('artist', artist_id)
|
||||
if existing_id:
|
||||
logger.debug(f"Preserving existing Tidal ID for artist '{artist_name}': {existing_id}")
|
||||
# Mark as matched so this row is not re-selected on every loop.
|
||||
self._mark_status('artist', artist_id, 'matched')
|
||||
return
|
||||
|
||||
result = self.client.search_artist(artist_name)
|
||||
|
|
@ -438,6 +440,7 @@ class TidalWorker:
|
|||
existing_id = self._get_existing_id('album', album_id)
|
||||
if existing_id:
|
||||
logger.debug(f"Preserving existing Tidal ID for album '{album_name}': {existing_id}")
|
||||
self._mark_status('album', album_id, 'matched')
|
||||
return
|
||||
|
||||
result = self.client.search_album(artist_name, album_name)
|
||||
|
|
@ -486,6 +489,7 @@ class TidalWorker:
|
|||
existing_id = self._get_existing_id('track', track_id)
|
||||
if existing_id:
|
||||
logger.debug(f"Preserving existing Tidal ID for track '{track_name}': {existing_id}")
|
||||
self._mark_status('track', track_id, 'matched')
|
||||
return
|
||||
|
||||
result = self.client.search_track(artist_name, track_name)
|
||||
|
|
|
|||
|
|
@ -409,6 +409,12 @@ class MusicDatabase:
|
|||
# Add Discogs enrichment columns (migration)
|
||||
self._add_discogs_columns(cursor)
|
||||
|
||||
# Backfill match_status for rows that already have an external ID but
|
||||
# NULL status. Prevents enrichment workers from re-processing these
|
||||
# rows forever. Must run AFTER all *_match_status columns have been
|
||||
# created by the migrations above.
|
||||
self._backfill_match_status_for_existing_ids(cursor)
|
||||
|
||||
# Bubble snapshots table for persisting UI state across page refreshes
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS bubble_snapshots (
|
||||
|
|
@ -1922,6 +1928,56 @@ class MusicDatabase:
|
|||
except Exception as e:
|
||||
logger.error(f"Error adding Discogs columns: {e}")
|
||||
|
||||
def _backfill_match_status_for_existing_ids(self, cursor):
|
||||
"""Set `<provider>_match_status = 'matched'` for rows that already have a
|
||||
populated external ID but NULL match_status.
|
||||
|
||||
Prevents enrichment workers from re-selecting the same rows forever when
|
||||
the ID was populated outside the worker (file tags, manual match,
|
||||
pre-migration legacy data) without a corresponding status update.
|
||||
|
||||
Only runs columns that actually exist, so pre-migration databases are
|
||||
handled safely. UPDATE statements are cheap no-ops when nothing matches.
|
||||
"""
|
||||
# (table, id_column, status_column)
|
||||
targets = [
|
||||
('artists', 'lastfm_url', 'lastfm_match_status'),
|
||||
('albums', 'lastfm_url', 'lastfm_match_status'),
|
||||
('tracks', 'lastfm_url', 'lastfm_match_status'),
|
||||
('artists', 'musicbrainz_id', 'musicbrainz_match_status'),
|
||||
('albums', 'musicbrainz_release_id', 'musicbrainz_match_status'),
|
||||
('tracks', 'musicbrainz_recording_id', 'musicbrainz_match_status'),
|
||||
('artists', 'tidal_id', 'tidal_match_status'),
|
||||
('albums', 'tidal_id', 'tidal_match_status'),
|
||||
('tracks', 'tidal_id', 'tidal_match_status'),
|
||||
('artists', 'qobuz_id', 'qobuz_match_status'),
|
||||
('albums', 'qobuz_id', 'qobuz_match_status'),
|
||||
('tracks', 'qobuz_id', 'qobuz_match_status'),
|
||||
]
|
||||
|
||||
total_backfilled = 0
|
||||
for table, id_col, status_col in targets:
|
||||
try:
|
||||
cursor.execute(f"PRAGMA table_info({table})")
|
||||
cols = {row[1] for row in cursor.fetchall()}
|
||||
if id_col not in cols or status_col not in cols:
|
||||
continue
|
||||
cursor.execute(
|
||||
f"UPDATE {table} SET {status_col} = 'matched' "
|
||||
f"WHERE {status_col} IS NULL AND {id_col} IS NOT NULL AND {id_col} != ''"
|
||||
)
|
||||
if cursor.rowcount and cursor.rowcount > 0:
|
||||
total_backfilled += cursor.rowcount
|
||||
logger.info(
|
||||
f"Backfilled {cursor.rowcount} rows in {table}.{status_col} "
|
||||
f"where {id_col} was already set."
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Error backfilling {table}.{status_col}: {e}")
|
||||
|
||||
if total_backfilled == 0:
|
||||
logger.debug("Match-status backfill: no rows needed updating.")
|
||||
|
||||
def _add_deezer_columns(self, cursor):
|
||||
"""Add Deezer tracking + generic metadata columns for enrichment (artists, albums, tracks)"""
|
||||
try:
|
||||
|
|
|
|||
Loading…
Reference in a new issue