a60546929e
Reported by sassmastawillis: the Album Completeness maintenance job
scans 3127 albums in 0.1 seconds and reports 0 findings — for every
user, regardless of whether their library is actually complete.
Restoring an older DB surfaced 7 correct findings, so the code logic
works; the DB state is what's making everything look complete.
Root cause: `albums.track_count` is only ever written by server-sync
paths — Plex's `leafCount`/`childCount` and SoulSync standalone's
`len(tracks)`. It's the OBSERVED count of tracks SoulSync has indexed,
which is always exactly what `COUNT(tracks)` returns for that album.
The completeness job treated it as the EXPECTED total and compared it
against the observed count. They're equal by construction, so
`actual >= expected` is always true: skip, 0.1s scan, 0 findings.
Fix: new `api_track_count INTEGER` column on `albums`, written only by
metadata-source code paths. Populated in two places so the scan is
fast and the fallback is robust.
1. Enrichment workers — shared helper `set_album_api_track_count`
in `core/worker_utils.py`. Called by each worker's existing
`_update_album` method alongside its other album-column UPDATEs:
- spotify_worker: `album_obj.total_tracks` from the Spotify Album
dataclass (already in hand, zero new API calls)
- itunes_worker: same, from the iTunes Album dataclass
- deezer_worker: `nb_tracks` from full_data, falling back to
search_data when the full lookup didn't run
- discogs_worker: count of tracklist rows where `type_=='track'`
(Discogs tracklists interleave heading and index rows that
shouldn't count as songs)
Helper skips the write on zero/None/negative/non-numeric inputs
so a source lacking track info can't clobber a good value a
different source already wrote. Caller owns the transaction —
helper just queues an UPDATE on the caller's cursor without
committing, so it batches cleanly with each worker's existing
multi-UPDATE pattern.
Hydrabase worker deliberately not touched — it's a P2P mirror
that doesn't write album metadata to the local DB. Hydrabase-
primary users hit the fallback path below.
2. Album Completeness repair job — new `al.api_track_count` column
in the SELECT, read first in the scan loop. On miss (album never
enriched, or enrichment workers haven't run yet on a fresh
install), falls through to the existing `_get_expected_total()`
API lookup and persists the result via the same shared helper
(wrapped in connection/commit management since the repair job
runs outside a worker's batched transaction).
Also removed `al.track_count` from the scan's SELECT — now unused
since the observed count was the whole source of this bug, and
leaving a dead SELECT would invite a future engineer to re-introduce
the same comparison.
Help text on the job card was reworded so it honestly describes
current behavior ("counts cached during normal enrichment are used
when available; otherwise the job queries a metadata source
directly") rather than the old "active provider first, then others
as fallback" phrasing, which doesn't match how the cache actually
fills — any enrichment worker that runs can populate it, and the
last writer wins. Document-only follow-up if this edge case ever
bites in practice: add a `api_track_count_source` column so the
scan can prefer the configured primary source's count over others
(e.g. deluxe vs. standard edition mismatches). Not worth the
complexity today.
For existing users, the first completeness scan after upgrade is
fast to the extent their library is already enriched: the workers
already ran and populated `api_track_count` on their normal schedule.
For brand-new installs, the scan's fallback path handles the cold
start — slower, but correct, and subsequent scans are fast.
Does NOT affect:
- Download / post-processing / wishlist / sync code paths — none
of them read `track_count` for completeness semantics.
- Plex / Jellyfin / Navidrome / standalone sync — still write
`track_count` exactly as before; `api_track_count` is a separate
column they never touch.
- Other repair jobs.
- Any UI path — same finding schema, just correct counts now.
Files:
- database/music_database.py — idempotent migration adding
`api_track_count INTEGER DEFAULT NULL` to the existing album-column
check block.
- core/worker_utils.py — new `set_album_api_track_count` helper with
the documented skip-on-bad-input contract.
- core/spotify_worker.py, itunes_worker.py, deezer_worker.py,
discogs_worker.py — one-liner call from each `_update_album`.
- core/repair_jobs/album_completeness.py — scan uses the cache;
fallback path persists API-lookup results via the shared helper;
help text updated to match actual behavior.
- tests/test_worker_utils_album_track_count.py — 9 tests covering
the helper's write/skip contract + no-commit invariant.
- tests/test_album_completeness_job.py — 2 tests for the repair
job's fallback-path wrapper.
- webui/static/helper.js — WHATS_NEW entry.
Credit: sassmastawillis spotted the bug; the "restored older DB
finds 7 albums" signal pinpointed DB state over code logic and
made the diagnosis tractable.
495 lines
19 KiB
Python
495 lines
19 KiB
Python
"""
|
|
Discogs background enrichment worker.
|
|
|
|
Enriches library artists and albums with Discogs metadata:
|
|
- Artists: discogs_id, bio, members, genres, styles, URLs, images
|
|
- Albums: discogs_id, genres, styles, label, catalog number, country, community rating
|
|
|
|
Follows the exact same pattern as AudioDBWorker.
|
|
"""
|
|
|
|
import json
|
|
import re
|
|
import threading
|
|
import time
|
|
from difflib import SequenceMatcher
|
|
from typing import Optional, Dict, Any
|
|
from datetime import datetime, timedelta
|
|
from utils.logging_config import get_logger
|
|
from database.music_database import MusicDatabase
|
|
from core.discogs_client import DiscogsClient
|
|
from core.worker_utils import interruptible_sleep, set_album_api_track_count
|
|
|
|
logger = get_logger("discogs_worker")
|
|
|
|
|
|
class DiscogsWorker:
|
|
"""Background worker for enriching library artists and albums with Discogs metadata."""
|
|
|
|
def __init__(self, database: MusicDatabase):
|
|
self.db = database
|
|
self.client = DiscogsClient()
|
|
|
|
# Worker state
|
|
self.running = False
|
|
self.paused = False
|
|
self.should_stop = False
|
|
self.thread = None
|
|
self._stop_event = threading.Event()
|
|
|
|
# Current item being processed (for UI tooltip)
|
|
self.current_item = None
|
|
|
|
# Statistics
|
|
self.stats = {
|
|
'matched': 0,
|
|
'not_found': 0,
|
|
'pending': 0,
|
|
'errors': 0
|
|
}
|
|
|
|
# Retry configuration
|
|
self.retry_days = 30
|
|
|
|
# Name matching threshold
|
|
self.name_similarity_threshold = 0.80
|
|
|
|
logger.info(f"Discogs background worker initialized (authenticated: {self.client.is_authenticated()})")
|
|
|
|
def start(self):
|
|
"""Start the background worker."""
|
|
if self.running:
|
|
logger.warning("Discogs worker already running")
|
|
return
|
|
|
|
self.running = True
|
|
self.should_stop = False
|
|
self._stop_event.clear()
|
|
self.thread = threading.Thread(target=self._run, daemon=True)
|
|
self.thread.start()
|
|
logger.info("Discogs background worker started")
|
|
|
|
def stop(self):
|
|
"""Stop the background worker."""
|
|
if not self.running:
|
|
return
|
|
logger.info("Stopping Discogs worker...")
|
|
self.should_stop = True
|
|
self.running = False
|
|
self._stop_event.set()
|
|
if self.thread:
|
|
self.thread.join(timeout=1)
|
|
logger.info("Discogs worker stopped")
|
|
|
|
def pause(self):
|
|
"""Pause the worker."""
|
|
if not self.running:
|
|
return
|
|
self.paused = True
|
|
logger.info("Discogs worker paused")
|
|
|
|
def resume(self):
|
|
"""Resume the worker."""
|
|
if not self.running:
|
|
return
|
|
self.paused = False
|
|
logger.info("Discogs worker resumed")
|
|
|
|
def get_stats(self) -> Dict[str, Any]:
|
|
"""Get current statistics."""
|
|
self.stats['pending'] = self._count_pending_items()
|
|
is_actually_running = self.running and (self.thread is not None and self.thread.is_alive())
|
|
is_idle = is_actually_running and not self.paused and self.stats['pending'] == 0 and self.current_item is None
|
|
|
|
return {
|
|
'enabled': True,
|
|
'running': is_actually_running and not self.paused,
|
|
'paused': self.paused,
|
|
'idle': is_idle,
|
|
'current_item': self.current_item,
|
|
'stats': self.stats.copy(),
|
|
}
|
|
|
|
def _run(self):
|
|
"""Main worker loop."""
|
|
logger.info("Discogs worker thread started")
|
|
|
|
while not self.should_stop:
|
|
try:
|
|
if self.paused:
|
|
interruptible_sleep(self._stop_event, 1)
|
|
continue
|
|
|
|
self.current_item = None
|
|
item = self._get_next_item()
|
|
|
|
if not item:
|
|
interruptible_sleep(self._stop_event, 10)
|
|
continue
|
|
|
|
self.current_item = item.get('name', '')
|
|
|
|
# Guard: skip items with None/NULL IDs
|
|
item_id = item.get('id')
|
|
if item_id is None:
|
|
logger.warning(f"Skipping {item.get('type', 'unknown')} with NULL id")
|
|
continue
|
|
|
|
self._process_item(item)
|
|
interruptible_sleep(self._stop_event, 2)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in Discogs worker loop: {e}")
|
|
interruptible_sleep(self._stop_event, 5)
|
|
|
|
logger.info("Discogs worker thread finished")
|
|
|
|
def _get_next_item(self) -> Optional[Dict[str, Any]]:
|
|
"""Get next item to process (artists → albums → retries)."""
|
|
conn = None
|
|
try:
|
|
conn = self.db._get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
# Priority 1: Unattempted artists
|
|
cursor.execute("""
|
|
SELECT id, name FROM artists
|
|
WHERE discogs_match_status IS NULL AND id IS NOT NULL
|
|
ORDER BY id ASC LIMIT 1
|
|
""")
|
|
row = cursor.fetchone()
|
|
if row:
|
|
return {'type': 'artist', 'id': row[0], 'name': row[1]}
|
|
|
|
# Priority 2: Unattempted albums
|
|
cursor.execute("""
|
|
SELECT a.id, a.title, ar.name AS artist_name, ar.discogs_id AS artist_discogs_id
|
|
FROM albums a
|
|
JOIN artists ar ON a.artist_id = ar.id
|
|
WHERE a.discogs_match_status IS NULL AND a.id IS NOT NULL
|
|
ORDER BY a.id ASC LIMIT 1
|
|
""")
|
|
row = cursor.fetchone()
|
|
if row:
|
|
return {'type': 'album', 'id': row[0], 'name': row[1], 'artist': row[2], 'artist_discogs_id': row[3]}
|
|
|
|
# Priority 3: Retry 'not_found' artists after retry_days
|
|
not_found_cutoff = datetime.now() - timedelta(days=self.retry_days)
|
|
cursor.execute("""
|
|
SELECT id, name FROM artists
|
|
WHERE discogs_match_status = 'not_found' AND discogs_last_attempted < ?
|
|
ORDER BY discogs_last_attempted ASC LIMIT 1
|
|
""", (not_found_cutoff,))
|
|
row = cursor.fetchone()
|
|
if row:
|
|
return {'type': 'artist', 'id': row[0], 'name': row[1]}
|
|
|
|
# Priority 4: Retry 'not_found' albums
|
|
cursor.execute("""
|
|
SELECT a.id, a.title, ar.name AS artist_name, ar.discogs_id AS artist_discogs_id
|
|
FROM albums a
|
|
JOIN artists ar ON a.artist_id = ar.id
|
|
WHERE a.discogs_match_status = 'not_found' AND a.discogs_last_attempted < ?
|
|
ORDER BY a.discogs_last_attempted ASC LIMIT 1
|
|
""", (not_found_cutoff,))
|
|
row = cursor.fetchone()
|
|
if row:
|
|
return {'type': 'album', 'id': row[0], 'name': row[1], 'artist': row[2], 'artist_discogs_id': row[3]}
|
|
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error getting next Discogs item: {e}")
|
|
return None
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
def _count_pending_items(self) -> int:
|
|
"""Count items still needing Discogs enrichment."""
|
|
try:
|
|
conn = self.db._get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT COUNT(*) FROM artists WHERE discogs_match_status IS NULL AND id IS NOT NULL")
|
|
artists = cursor.fetchone()[0]
|
|
cursor.execute("SELECT COUNT(*) FROM albums WHERE discogs_match_status IS NULL AND id IS NOT NULL")
|
|
albums = cursor.fetchone()[0]
|
|
conn.close()
|
|
return artists + albums
|
|
except Exception:
|
|
return 0
|
|
|
|
def _normalize_name(self, name: str) -> str:
|
|
"""Normalize name for comparison."""
|
|
name = name.lower().strip()
|
|
name = re.sub(r'\s+[-–—]\s+.*$', '', name)
|
|
name = re.sub(r'\s*\(.*?\)\s*', ' ', name)
|
|
name = re.sub(r'[^\w\s]', '', name)
|
|
name = re.sub(r'\s+', ' ', name).strip()
|
|
return name
|
|
|
|
def _name_matches(self, query_name: str, result_name: str) -> bool:
|
|
"""Check if Discogs result name matches our query with fuzzy matching."""
|
|
norm_query = self._normalize_name(query_name)
|
|
norm_result = self._normalize_name(result_name)
|
|
similarity = SequenceMatcher(None, norm_query, norm_result).ratio()
|
|
return similarity >= self.name_similarity_threshold
|
|
|
|
def _process_item(self, item: Dict[str, Any]):
|
|
"""Process a single artist or album."""
|
|
try:
|
|
item_type = item['type']
|
|
item_id = item['id']
|
|
item_name = item['name']
|
|
|
|
logger.debug(f"Processing {item_type} #{item_id}: {item_name}")
|
|
|
|
# Check for existing discogs_id (manual match) — use direct lookup
|
|
existing_id = self._get_existing_id(item_type, item_id)
|
|
if existing_id:
|
|
try:
|
|
if item_type == 'artist':
|
|
data = self.client._fetch_and_cache_artist(existing_id)
|
|
if data:
|
|
self._update_artist(item_id, data)
|
|
self.stats['matched'] += 1
|
|
logger.info(f"Enriched artist '{item_name}' from existing Discogs ID: {existing_id}")
|
|
return
|
|
elif item_type == 'album':
|
|
data = self.client._fetch_and_cache_album(existing_id)
|
|
if data:
|
|
self._update_album(item_id, data)
|
|
self.stats['matched'] += 1
|
|
logger.info(f"Enriched album '{item_name}' from existing Discogs ID: {existing_id}")
|
|
return
|
|
except Exception as e:
|
|
logger.warning(f"Direct Discogs lookup failed for ID {existing_id}: {e}")
|
|
return # Preserve manual match, don't search
|
|
|
|
if item_type == 'artist':
|
|
self._search_and_match_artist(item_id, item_name)
|
|
elif item_type == 'album':
|
|
self._search_and_match_album(item_id, item_name, item.get('artist', ''), item.get('artist_discogs_id'))
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error processing {item.get('type')} #{item.get('id')}: {e}")
|
|
self.stats['errors'] += 1
|
|
try:
|
|
self._mark_status(item['type'], item['id'], 'error')
|
|
except Exception:
|
|
pass
|
|
|
|
def _get_existing_id(self, entity_type: str, entity_id) -> Optional[str]:
|
|
"""Check if entity already has a discogs_id."""
|
|
table = 'artists' if entity_type == 'artist' else 'albums'
|
|
try:
|
|
conn = self.db._get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute(f"SELECT discogs_id FROM {table} WHERE id = ?", (entity_id,))
|
|
row = cursor.fetchone()
|
|
conn.close()
|
|
return row[0] if row and row[0] else None
|
|
except Exception:
|
|
return None
|
|
|
|
def _search_and_match_artist(self, artist_id, artist_name: str):
|
|
"""Search Discogs for an artist and store metadata if matched."""
|
|
results = self.client.search_artists(artist_name, limit=5)
|
|
if not results:
|
|
self._mark_status('artist', artist_id, 'not_found')
|
|
self.stats['not_found'] += 1
|
|
return
|
|
|
|
# Find best match by name similarity
|
|
for result in results:
|
|
if self._name_matches(artist_name, result.name):
|
|
# Fetch full artist detail (uses cache)
|
|
data = self.client._fetch_and_cache_artist(result.id)
|
|
if data:
|
|
self._update_artist(artist_id, data)
|
|
self.stats['matched'] += 1
|
|
logger.info(f"Matched artist '{artist_name}' -> Discogs ID: {result.id}")
|
|
return
|
|
|
|
self._mark_status('artist', artist_id, 'not_found')
|
|
self.stats['not_found'] += 1
|
|
logger.debug(f"No confident match for artist '{artist_name}'")
|
|
|
|
def _search_and_match_album(self, album_id, album_name: str, artist_name: str, artist_discogs_id: str = None):
|
|
"""Search Discogs for an album and store metadata if matched."""
|
|
# Search with artist + album for better precision
|
|
query = f"{artist_name} {album_name}" if artist_name else album_name
|
|
results = self.client.search_albums(query, limit=5)
|
|
if not results:
|
|
self._mark_status('album', album_id, 'not_found')
|
|
self.stats['not_found'] += 1
|
|
return
|
|
|
|
for result in results:
|
|
if self._name_matches(album_name, result.name):
|
|
# Fetch full release detail (uses cache)
|
|
data = self.client._fetch_and_cache_album(result.id)
|
|
if data:
|
|
self._update_album(album_id, data)
|
|
self.stats['matched'] += 1
|
|
logger.info(f"Matched album '{album_name}' -> Discogs ID: {result.id}")
|
|
return
|
|
|
|
self._mark_status('album', album_id, 'not_found')
|
|
self.stats['not_found'] += 1
|
|
logger.debug(f"No confident match for album '{album_name}'")
|
|
|
|
def _update_artist(self, artist_id, data: Dict[str, Any]):
|
|
"""Store Discogs metadata for an artist."""
|
|
conn = None
|
|
try:
|
|
conn = self.db._get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
discogs_id = str(data.get('id', ''))
|
|
bio = data.get('profile', '')
|
|
members = json.dumps([m.get('name', '') for m in data.get('members', [])]) if data.get('members') else None
|
|
urls = json.dumps(data.get('urls', [])) if data.get('urls') else None
|
|
|
|
# Get image
|
|
image_url = None
|
|
images = data.get('images', [])
|
|
if images:
|
|
primary = next((img for img in images if img.get('type') == 'primary'), None)
|
|
image_url = (primary or images[0]).get('uri')
|
|
|
|
cursor.execute("""
|
|
UPDATE artists SET
|
|
discogs_id = ?,
|
|
discogs_match_status = 'matched',
|
|
discogs_last_attempted = CURRENT_TIMESTAMP,
|
|
discogs_bio = ?,
|
|
discogs_members = ?,
|
|
discogs_urls = ?,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ?
|
|
""", (discogs_id, bio, members, urls, artist_id))
|
|
|
|
# Backfill summary/bio if empty (AudioDB backfill)
|
|
if bio:
|
|
cursor.execute("""
|
|
UPDATE artists SET summary = ?
|
|
WHERE id = ? AND (summary IS NULL OR summary = '')
|
|
""", (bio, artist_id))
|
|
|
|
# Backfill thumb_url if empty
|
|
if image_url:
|
|
cursor.execute("""
|
|
UPDATE artists SET thumb_url = ?
|
|
WHERE id = ? AND (thumb_url IS NULL OR thumb_url = '')
|
|
""", (image_url, artist_id))
|
|
|
|
conn.commit()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error updating artist #{artist_id} with Discogs data: {e}")
|
|
raise
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
def _update_album(self, album_id, data: Dict[str, Any]):
|
|
"""Store Discogs metadata for an album."""
|
|
conn = None
|
|
try:
|
|
conn = self.db._get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
discogs_id = str(data.get('id', ''))
|
|
genres = json.dumps(data.get('genres', []))
|
|
styles = json.dumps(data.get('styles', []))
|
|
labels = data.get('labels', [])
|
|
label = labels[0].get('name', '') if labels else ''
|
|
catno = labels[0].get('catno', '') if labels else ''
|
|
country = data.get('country', '')
|
|
|
|
# Community rating
|
|
community = data.get('community', {})
|
|
rating = community.get('rating', {})
|
|
rating_avg = rating.get('average', 0)
|
|
rating_count = rating.get('count', 0)
|
|
|
|
# Image
|
|
image_url = None
|
|
images = data.get('images', [])
|
|
if images:
|
|
primary = next((img for img in images if img.get('type') == 'primary'), None)
|
|
image_url = (primary or images[0]).get('uri')
|
|
|
|
cursor.execute("""
|
|
UPDATE albums SET
|
|
discogs_id = ?,
|
|
discogs_match_status = 'matched',
|
|
discogs_last_attempted = CURRENT_TIMESTAMP,
|
|
discogs_genres = ?,
|
|
discogs_styles = ?,
|
|
discogs_label = ?,
|
|
discogs_catno = ?,
|
|
discogs_country = ?,
|
|
discogs_rating = ?,
|
|
discogs_rating_count = ?,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ?
|
|
""", (discogs_id, genres, styles, label, catno, country, rating_avg, rating_count, album_id))
|
|
|
|
# Backfill genres if empty
|
|
if data.get('genres'):
|
|
from core.genre_filter import filter_genres
|
|
from config.settings import config_manager as _cfg
|
|
_filtered = filter_genres(data.get('genres', []), _cfg)
|
|
if _filtered:
|
|
cursor.execute("""
|
|
UPDATE albums SET genres = ?
|
|
WHERE id = ? AND (genres IS NULL OR genres = '' OR genres = '[]')
|
|
""", (json.dumps(_filtered), album_id))
|
|
|
|
# Backfill thumb_url if empty
|
|
if image_url:
|
|
cursor.execute("""
|
|
UPDATE albums SET thumb_url = ?
|
|
WHERE id = ? AND (thumb_url IS NULL OR thumb_url = '')
|
|
""", (image_url, album_id))
|
|
|
|
# Cache the authoritative expected track count for the Album
|
|
# Completeness repair job. Discogs tracklists mix actual tracks
|
|
# with section headings and index entries — only type_=='track'
|
|
# rows are real songs, so filter before counting. (Helper can't
|
|
# do this filter for us; it's Discogs-specific shape.)
|
|
tracklist = data.get('tracklist') or []
|
|
set_album_api_track_count(
|
|
cursor,
|
|
album_id,
|
|
sum(1 for t in tracklist if t.get('type_') == 'track'),
|
|
)
|
|
|
|
conn.commit()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error updating album #{album_id} with Discogs data: {e}")
|
|
raise
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
def _mark_status(self, entity_type: str, entity_id, status: str):
|
|
"""Mark entity's Discogs match status."""
|
|
table = {'artist': 'artists', 'album': 'albums'}.get(entity_type)
|
|
if not table:
|
|
return
|
|
try:
|
|
conn = self.db._get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute(f"""
|
|
UPDATE {table} SET
|
|
discogs_match_status = ?,
|
|
discogs_last_attempted = CURRENT_TIMESTAMP
|
|
WHERE id = ?
|
|
""", (status, entity_id))
|
|
conn.commit()
|
|
conn.close()
|
|
except Exception as e:
|
|
logger.error(f"Error marking {entity_type} #{entity_id} status: {e}")
|