From ac2c710a1e21f5ebfe7c8137ba1802d44fc0e688 Mon Sep 17 00:00:00 2001
From: Broque Thomas <26755000+Nezreka@users.noreply.github.com>
Date: Wed, 11 Mar 2026 21:26:20 -0700
Subject: [PATCH] Tidal & Qobuz Background Enrichment Workers
---
core/qobuz_client.py | 141 +++++++
core/qobuz_worker.py | 811 ++++++++++++++++++++++++++++++++++++
core/tidal_client.py | 229 ++++++++++-
core/tidal_worker.py | 818 +++++++++++++++++++++++++++++++++++++
database/music_database.py | 87 ++++
web_server.py | 311 ++++++++++++--
webui/index.html | 44 ++
webui/static/script.js | 274 ++++++++++++-
webui/static/style.css | 454 ++++++++++++++++++++
9 files changed, 3123 insertions(+), 46 deletions(-)
create mode 100644 core/qobuz_worker.py
create mode 100644 core/tidal_worker.py
diff --git a/core/qobuz_client.py b/core/qobuz_client.py
index e91e77b2..8c696d51 100644
--- a/core/qobuz_client.py
+++ b/core/qobuz_client.py
@@ -34,6 +34,40 @@ logger = get_logger("qobuz_client")
QOBUZ_API_BASE = "https://www.qobuz.com/api.json/0.2/"
+# ── Module-level rate limiting (shared across ALL QobuzClient instances) ──
+_qobuz_api_lock = threading.Lock()
+_qobuz_last_api_call = 0.0
+_QOBUZ_MIN_INTERVAL = 1.0 # 1 request/sec (60/min, matches streamrip default)
+
+# Global rate limit ban state (like Spotify's pattern)
+_qobuz_rate_limit_until = 0.0
+_qobuz_rate_limit_lock = threading.Lock()
+
+
+def _qobuz_throttle():
+ """Enforce minimum interval between Qobuz API calls across all instances."""
+ global _qobuz_last_api_call
+ with _qobuz_api_lock:
+ now = time.time()
+ elapsed = now - _qobuz_last_api_call
+ if elapsed < _QOBUZ_MIN_INTERVAL:
+ time.sleep(_QOBUZ_MIN_INTERVAL - elapsed)
+ _qobuz_last_api_call = time.time()
+
+
+def _qobuz_set_rate_limit(retry_after: float = 60.0):
+ """Set a global rate limit ban for all Qobuz instances."""
+ global _qobuz_rate_limit_until
+ with _qobuz_rate_limit_lock:
+ _qobuz_rate_limit_until = time.time() + retry_after
+ logger.warning(f"Qobuz global rate limit set for {retry_after}s")
+
+
+def _qobuz_is_rate_limited() -> bool:
+ """Check if Qobuz is currently rate limited."""
+ with _qobuz_rate_limit_lock:
+ return time.time() < _qobuz_rate_limit_until
+
# Quality tier definitions (format_id values)
QOBUZ_QUALITY_MAP = {
'mp3': {
@@ -450,6 +484,12 @@ class QobuzClient:
logger.warning("Qobuz not authenticated")
return None
+ if _qobuz_is_rate_limited():
+ logger.debug(f"Qobuz rate limited, skipping {endpoint}")
+ return None
+
+ _qobuz_throttle()
+
try:
resp = self.session.get(
QOBUZ_API_BASE + endpoint,
@@ -461,6 +501,10 @@ class QobuzClient:
logger.warning("Qobuz auth token expired")
self.user_auth_token = None
return None
+ elif resp.status_code == 429:
+ retry_after = float(resp.headers.get('Retry-After', 60))
+ _qobuz_set_rate_limit(retry_after)
+ return None
elif resp.status_code != 200:
logger.warning(f"Qobuz API error: {endpoint} returned HTTP {resp.status_code}")
return None
@@ -471,6 +515,93 @@ class QobuzClient:
logger.error(f"Qobuz API request failed ({endpoint}): {e}")
return None
+ # ── Enrichment API Methods ──
+
+ def search_artist(self, name: str):
+ """Search for an artist by name. Returns first result as raw dict or None."""
+ try:
+ data = self._api_request('artist/search', {
+ 'query': name,
+ 'limit': 1,
+ })
+ if data and 'artists' in data:
+ items = data['artists'].get('items', [])
+ if items:
+ return items[0]
+ return None
+ except Exception as e:
+ logger.error(f"Error searching Qobuz artist: {e}")
+ return None
+
+ def search_album(self, artist: str, title: str):
+ """Search for an album by artist + title. Returns first result as raw dict or None."""
+ try:
+ query = f"{artist} {title}" if artist else title
+ data = self._api_request('album/search', {
+ 'query': query,
+ 'limit': 1,
+ })
+ if data and 'albums' in data:
+ items = data['albums'].get('items', [])
+ if items:
+ return items[0]
+ return None
+ except Exception as e:
+ logger.error(f"Error searching Qobuz album: {e}")
+ return None
+
+ def search_track(self, artist: str, title: str):
+ """Search for a track by artist + title. Returns first result as raw dict or None."""
+ try:
+ query = f"{artist} {title}" if artist else title
+ data = self._api_request('track/search', {
+ 'query': query,
+ 'limit': 1,
+ })
+ if data and 'tracks' in data:
+ items = data['tracks'].get('items', [])
+ if items:
+ return items[0]
+ return None
+ except Exception as e:
+ logger.error(f"Error searching Qobuz track: {e}")
+ return None
+
+ def get_artist(self, artist_id):
+ """Get full artist details by Qobuz ID."""
+ try:
+ data = self._api_request('artist/get', {
+ 'artist_id': artist_id,
+ 'extra': 'albums',
+ })
+ return data
+ except Exception as e:
+ logger.error(f"Error getting Qobuz artist {artist_id}: {e}")
+ return None
+
+ def get_album(self, album_id):
+ """Get full album details by Qobuz ID."""
+ try:
+ data = self._api_request('album/get', {
+ 'album_id': album_id,
+ 'extra': 'tracks',
+ })
+ return data
+ except Exception as e:
+ logger.error(f"Error getting Qobuz album {album_id}: {e}")
+ return None
+
+ def get_track(self, track_id):
+ """Get full track details by Qobuz ID."""
+ try:
+ data = self._api_request('track/get', {
+ 'track_id': track_id,
+ })
+ return data
+ except Exception as e:
+ logger.error(f"Error getting Qobuz track {track_id}: {e}")
+ return None
+
async def search(self, query: str, timeout: int = None, progress_callback=None) -> Tuple[List[TrackResult], List[AlbumResult]]:
"""
Search Qobuz for tracks (async, Soulseek-compatible interface).
@@ -603,6 +734,12 @@ class QobuzClient:
logger.error("No app_secret available for stream URL signing")
return None
+ if _qobuz_is_rate_limited():
+ logger.debug("Qobuz rate limited, skipping stream URL request")
+ return None
+
+ _qobuz_throttle()
+
ts = str(int(time.time()))
sig_raw = f"trackgetFileUrlformat_id{format_id}intentstreamtrack_id{track_id}{ts}{self.app_secret}"
sig = hashlib.md5(sig_raw.encode()).hexdigest()
@@ -623,6 +760,10 @@ class QobuzClient:
if resp.status_code == 401:
logger.warning("Qobuz stream URL auth failed — token may be expired")
return None
+ elif resp.status_code == 429:
+ retry_after = float(resp.headers.get('Retry-After', 60))
+ _qobuz_set_rate_limit(retry_after)
+ return None
elif resp.status_code == 400:
data = resp.json() if resp.text else {}
logger.warning(f"Qobuz stream URL rejected: {data.get('message', 'unknown error')}")
diff --git a/core/qobuz_worker.py b/core/qobuz_worker.py
new file mode 100644
index 00000000..b5b351cd
--- /dev/null
+++ b/core/qobuz_worker.py
@@ -0,0 +1,811 @@
+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.qobuz_client import _qobuz_is_rate_limited
+
+logger = get_logger("qobuz_worker")
+
+
+class QobuzWorker:
+ """Background worker for enriching library artists, albums, and tracks with Qobuz metadata"""
+
+ def __init__(self, database: MusicDatabase, client=None):
+ self.db = database
+ self.client = client # Set externally or created during init in web_server
+
+ # Worker state
+ self.running = False
+ self.paused = False
+ self.should_stop = False
+ self.thread = None
+
+ # 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
+ self.error_retry_days = 7
+
+ # Name matching threshold
+ self.name_similarity_threshold = 0.80
+
+ logger.info("Qobuz background worker initialized")
+
+ def start(self):
+ """Start the background worker"""
+ if self.running:
+ logger.warning("Worker already running")
+ return
+
+ self.running = True
+ self.should_stop = False
+ self.thread = threading.Thread(target=self._run, daemon=True)
+ self.thread.start()
+ logger.info("Qobuz background worker started")
+
+ def stop(self):
+ """Stop the background worker"""
+ if not self.running:
+ return
+
+ logger.info("Stopping Qobuz worker...")
+ self.should_stop = True
+ self.running = False
+
+ if self.thread:
+ self.thread.join(timeout=5)
+
+ logger.info("Qobuz worker stopped")
+
+ def pause(self):
+ """Pause the worker"""
+ if not self.running:
+ logger.warning("Worker not running, cannot pause")
+ return
+ self.paused = True
+ logger.info("Qobuz worker paused")
+
+ def resume(self):
+ """Resume the worker"""
+ if not self.running:
+ logger.warning("Worker not running, start it first")
+ return
+ self.paused = False
+ logger.info("Qobuz worker resumed")
+
+ def get_stats(self) -> Dict[str, Any]:
+ """Get current statistics"""
+ self.stats['pending'] = self._count_pending_items()
+
+ progress = self._get_progress_breakdown()
+
+ 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
+
+ authenticated = False
+ try:
+ if self.client:
+ authenticated = self.client.is_authenticated()
+ except Exception:
+ pass
+
+ return {
+ 'enabled': True,
+ 'running': is_actually_running and not self.paused,
+ 'paused': self.paused,
+ 'idle': is_idle,
+ 'authenticated': authenticated,
+ 'current_item': self.current_item,
+ 'stats': self.stats.copy(),
+ 'progress': progress
+ }
+
+ def _run(self):
+ """Main worker loop"""
+ logger.info("Qobuz worker thread started")
+
+ while not self.should_stop:
+ try:
+ if self.paused:
+ time.sleep(1)
+ continue
+
+ # Auth guard: sleep if not authenticated
+ try:
+ if not self.client or not self.client.is_authenticated():
+ self.current_item = None
+ time.sleep(30)
+ continue
+ except Exception:
+ time.sleep(30)
+ continue
+
+ # Rate limit guard: back off if globally rate limited
+ if _qobuz_is_rate_limited():
+ self.current_item = None
+ logger.debug("Qobuz rate limited, backing off...")
+ time.sleep(10)
+ continue
+
+ self.current_item = None
+
+ item = self._get_next_item()
+
+ if not item:
+ logger.debug("No pending items, sleeping...")
+ time.sleep(10)
+ continue
+
+ self.current_item = item
+
+ self._process_item(item)
+
+ # Throttle between API calls
+ time.sleep(2)
+
+ except Exception as e:
+ logger.error(f"Error in worker loop: {e}")
+ time.sleep(5)
+
+ logger.info("Qobuz worker thread finished")
+
+ def _get_next_item(self) -> Optional[Dict[str, Any]]:
+ """Get next item to process from priority queue (artists -> albums -> tracks)"""
+ conn = None
+ try:
+ conn = self.db._get_connection()
+ cursor = conn.cursor()
+
+ # Priority 1: Unattempted artists
+ cursor.execute("""
+ SELECT id, name
+ FROM artists
+ WHERE qobuz_match_status IS 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.qobuz_id AS artist_qobuz_id
+ FROM albums a
+ JOIN artists ar ON a.artist_id = ar.id
+ WHERE a.qobuz_match_status IS 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_qobuz_id': row[3]}
+
+ # Priority 3: Unattempted tracks
+ cursor.execute("""
+ SELECT t.id, t.title, ar.name AS artist_name, ar.qobuz_id AS artist_qobuz_id
+ FROM tracks t
+ JOIN artists ar ON t.artist_id = ar.id
+ WHERE t.qobuz_match_status IS NULL
+ ORDER BY t.id ASC
+ LIMIT 1
+ """)
+ row = cursor.fetchone()
+ if row:
+ return {'type': 'track', 'id': row[0], 'name': row[1], 'artist': row[2], 'artist_qobuz_id': row[3]}
+
+ # Priority 4: Retry 'not_found' or 'error' artists
+ not_found_cutoff = datetime.now() - timedelta(days=self.retry_days)
+ error_cutoff = datetime.now() - timedelta(days=self.error_retry_days)
+ cursor.execute("""
+ SELECT id, name
+ FROM artists
+ WHERE (qobuz_match_status = 'not_found' AND qobuz_last_attempted < ?)
+ OR (qobuz_match_status = 'error' AND qobuz_last_attempted < ?)
+ ORDER BY qobuz_last_attempted ASC
+ LIMIT 1
+ """, (not_found_cutoff, error_cutoff))
+ row = cursor.fetchone()
+ if row:
+ logger.info(f"Retrying artist '{row[1]}' (last attempted before cutoff)")
+ return {'type': 'artist', 'id': row[0], 'name': row[1]}
+
+ # Priority 5: Retry 'not_found' or 'error' albums
+ cursor.execute("""
+ SELECT a.id, a.title, ar.name AS artist_name, ar.qobuz_id AS artist_qobuz_id
+ FROM albums a
+ JOIN artists ar ON a.artist_id = ar.id
+ WHERE (a.qobuz_match_status = 'not_found' AND a.qobuz_last_attempted < ?)
+ OR (a.qobuz_match_status = 'error' AND a.qobuz_last_attempted < ?)
+ ORDER BY a.qobuz_last_attempted ASC
+ LIMIT 1
+ """, (not_found_cutoff, error_cutoff))
+ row = cursor.fetchone()
+ if row:
+ return {'type': 'album', 'id': row[0], 'name': row[1], 'artist': row[2], 'artist_qobuz_id': row[3]}
+
+ # Priority 6: Retry 'not_found' or 'error' tracks
+ cursor.execute("""
+ SELECT t.id, t.title, ar.name AS artist_name, ar.qobuz_id AS artist_qobuz_id
+ FROM tracks t
+ JOIN artists ar ON t.artist_id = ar.id
+ WHERE (t.qobuz_match_status = 'not_found' AND t.qobuz_last_attempted < ?)
+ OR (t.qobuz_match_status = 'error' AND t.qobuz_last_attempted < ?)
+ ORDER BY t.qobuz_last_attempted ASC
+ LIMIT 1
+ """, (not_found_cutoff, error_cutoff))
+ row = cursor.fetchone()
+ if row:
+ return {'type': 'track', 'id': row[0], 'name': row[1], 'artist': row[2], 'artist_qobuz_id': row[3]}
+
+ return None
+
+ except Exception as e:
+ logger.error(f"Error getting next item: {e}")
+ return None
+ finally:
+ if conn:
+ conn.close()
+
+ 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'[^\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 Qobuz 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()
+ logger.debug(f"Name similarity: '{query_name}' vs '{result_name}' = {similarity:.2f}")
+ return similarity >= self.name_similarity_threshold
+
+ def _verify_artist_id(self, item: Dict[str, Any], result_artist_id) -> bool:
+ """Verify/correct parent artist's Qobuz ID based on album/track match"""
+ parent_qobuz_id = item.get('artist_qobuz_id')
+ if not parent_qobuz_id or not result_artist_id:
+ return True
+
+ if str(result_artist_id) != str(parent_qobuz_id):
+ logger.info(
+ f"Artist ID correction from {item['type']} '{item['name']}': "
+ f"updating parent artist Qobuz ID from {parent_qobuz_id} to {result_artist_id}"
+ )
+ self._correct_artist_qobuz_id(item, str(result_artist_id))
+
+ return True
+
+ def _correct_artist_qobuz_id(self, item: Dict[str, Any], correct_qobuz_id: str):
+ """Correct the parent artist's qobuz_id based on a more specific album/track match"""
+ conn = None
+ try:
+ conn = self.db._get_connection()
+ cursor = conn.cursor()
+
+ table = 'albums' if item['type'] == 'album' else 'tracks'
+ cursor.execute(f"SELECT artist_id FROM {table} WHERE id = ?", (item['id'],))
+ row = cursor.fetchone()
+ if not row:
+ return
+
+ artist_id = row[0]
+ cursor.execute("""
+ UPDATE artists SET
+ qobuz_id = ?,
+ updated_at = CURRENT_TIMESTAMP
+ WHERE id = ?
+ """, (correct_qobuz_id, artist_id))
+ conn.commit()
+
+ logger.info(f"Corrected artist #{artist_id} Qobuz ID to {correct_qobuz_id}")
+
+ except Exception as e:
+ logger.error(f"Error correcting artist Qobuz ID: {e}")
+ finally:
+ if conn:
+ conn.close()
+
+ def _process_item(self, item: Dict[str, Any]):
+ """Process a single item (artist, album, or track)"""
+ try:
+ item_type = item['type']
+ item_id = item['id']
+ item_name = item['name']
+
+ logger.debug(f"Processing {item_type} #{item_id}: {item_name}")
+
+ if item_type == 'artist':
+ self._process_artist(item_id, item_name)
+ elif item_type == 'album':
+ self._process_album(item_id, item_name, item.get('artist', ''), item)
+ elif item_type == 'track':
+ self._process_track(item_id, item_name, item.get('artist', ''), item)
+
+ except Exception as e:
+ logger.error(f"Error processing {item['type']} #{item['id']}: {e}")
+ self.stats['errors'] += 1
+ try:
+ self._mark_status(item['type'], item['id'], 'error')
+ except Exception as e2:
+ logger.error(f"Error updating item status: {e2}")
+
+ def _process_artist(self, artist_id: int, artist_name: str):
+ """Process an artist: search Qobuz, verify, store metadata"""
+ result = self.client.search_artist(artist_name)
+
+ if result:
+ result_name = result.get('name', '')
+ if self._name_matches(artist_name, result_name):
+ qobuz_artist_id = result.get('id')
+ if not qobuz_artist_id:
+ self._mark_status('artist', artist_id, 'error')
+ self.stats['errors'] += 1
+ logger.warning(f"Qobuz search result for '{artist_name}' has no ID")
+ return
+
+ # Fetch full artist details
+ full_artist = None
+ try:
+ full_artist = self.client.get_artist(qobuz_artist_id)
+ except Exception as e:
+ logger.warning(f"Failed to fetch full artist details for '{artist_name}': {e}")
+
+ self._update_artist(artist_id, result, full_artist)
+ self.stats['matched'] += 1
+ logger.info(f"Matched artist '{artist_name}' -> Qobuz ID: {qobuz_artist_id}")
+ else:
+ self._mark_status('artist', artist_id, 'not_found')
+ self.stats['not_found'] += 1
+ logger.debug(f"Name mismatch for artist '{artist_name}' (got '{result_name}')")
+ else:
+ if _qobuz_is_rate_limited():
+ logger.warning(f"Rate limited while searching artist '{artist_name}', will retry")
+ return
+ self._mark_status('artist', artist_id, 'not_found')
+ self.stats['not_found'] += 1
+ logger.debug(f"No match for artist '{artist_name}'")
+
+ def _process_album(self, album_id: int, album_name: str, artist_name: str, item: Dict[str, Any]):
+ """Process an album: search Qobuz, verify, fetch full details, store metadata"""
+ result = self.client.search_album(artist_name, album_name)
+
+ if result:
+ result_name = result.get('title', '')
+ if self._name_matches(album_name, result_name):
+ # Verify artist ID
+ result_artist = result.get('artist', {})
+ result_artist_id = result_artist.get('id') if result_artist else None
+ self._verify_artist_id(item, result_artist_id)
+
+ # Fetch full album details
+ qobuz_album_id = result.get('id')
+ if not qobuz_album_id:
+ self._mark_status('album', album_id, 'error')
+ self.stats['errors'] += 1
+ logger.warning(f"Qobuz search result for album '{album_name}' has no ID")
+ return
+
+ full_album = None
+ try:
+ full_album = self.client.get_album(qobuz_album_id)
+ except Exception as e:
+ logger.warning(f"Failed to fetch full album details for '{album_name}': {e}")
+
+ if full_album is None:
+ if _qobuz_is_rate_limited():
+ logger.warning(f"Rate limited while fetching album '{album_name}', will retry")
+ return
+ self._mark_status('album', album_id, 'error')
+ self.stats['errors'] += 1
+ logger.warning(f"Album '{album_name}' matched but full details unavailable, will retry")
+ return
+
+ self._update_album(album_id, result, full_album)
+ self.stats['matched'] += 1
+ logger.info(f"Matched album '{album_name}' -> Qobuz ID: {qobuz_album_id}")
+ else:
+ self._mark_status('album', album_id, 'not_found')
+ self.stats['not_found'] += 1
+ logger.debug(f"Name mismatch for album '{album_name}' (got '{result_name}')")
+ else:
+ if _qobuz_is_rate_limited():
+ logger.warning(f"Rate limited while searching album '{album_name}', will retry")
+ return
+ self._mark_status('album', album_id, 'not_found')
+ self.stats['not_found'] += 1
+ logger.debug(f"No match for album '{album_name}'")
+
+ def _process_track(self, track_id: int, track_name: str, artist_name: str, item: Dict[str, Any]):
+ """Process a track: search Qobuz, verify, fetch full details, store metadata"""
+ result = self.client.search_track(artist_name, track_name)
+
+ if result:
+ result_name = result.get('title', '')
+ if self._name_matches(track_name, result_name):
+ # Verify artist ID
+ result_artist = result.get('artist', result.get('performer', {}))
+ result_artist_id = result_artist.get('id') if result_artist else None
+ self._verify_artist_id(item, result_artist_id)
+
+ # Fetch full track details
+ qobuz_track_id = result.get('id')
+ if not qobuz_track_id:
+ self._mark_status('track', track_id, 'error')
+ self.stats['errors'] += 1
+ logger.warning(f"Qobuz search result for track '{track_name}' has no ID")
+ return
+
+ full_track = None
+ try:
+ full_track = self.client.get_track(qobuz_track_id)
+ except Exception as e:
+ logger.warning(f"Failed to fetch full track details for '{track_name}': {e}")
+
+ if full_track is None:
+ if _qobuz_is_rate_limited():
+ logger.warning(f"Rate limited while fetching track '{track_name}', will retry")
+ return
+ self._mark_status('track', track_id, 'error')
+ self.stats['errors'] += 1
+ logger.warning(f"Track '{track_name}' matched but full details unavailable, will retry")
+ return
+
+ self._update_track(track_id, result, full_track)
+ self.stats['matched'] += 1
+ logger.info(f"Matched track '{track_name}' -> Qobuz ID: {qobuz_track_id}")
+ else:
+ self._mark_status('track', track_id, 'not_found')
+ self.stats['not_found'] += 1
+ logger.debug(f"Name mismatch for track '{track_name}' (got '{result_name}')")
+ else:
+ if _qobuz_is_rate_limited():
+ logger.warning(f"Rate limited while searching track '{track_name}', will retry")
+ return
+ self._mark_status('track', track_id, 'not_found')
+ self.stats['not_found'] += 1
+ logger.debug(f"No match for track '{track_name}'")
+
+ def _update_artist(self, artist_id: int, data: Dict[str, Any], full_data: Optional[Dict[str, Any]] = None):
+ """Store Qobuz metadata for an artist"""
+ conn = None
+ try:
+ conn = self.db._get_connection()
+ cursor = conn.cursor()
+
+ cursor.execute("""
+ UPDATE artists SET
+ qobuz_id = ?,
+ qobuz_match_status = 'matched',
+ qobuz_last_attempted = CURRENT_TIMESTAMP,
+ updated_at = CURRENT_TIMESTAMP
+ WHERE id = ?
+ """, (
+ str(data.get('id')),
+ artist_id
+ ))
+ conn.commit()
+
+ # Backfill optional metadata (failures here won't lose the match)
+ try:
+ src = full_data or data
+ thumb_url = None
+ image = src.get('image', {})
+ if isinstance(image, dict):
+ thumb_url = image.get('large', image.get('medium', image.get('small', image.get('thumbnail', ''))))
+ elif isinstance(image, str):
+ thumb_url = image
+ # Also check picture field
+ if not thumb_url:
+ thumb_url = src.get('picture', '')
+
+ if thumb_url:
+ cursor.execute("""
+ UPDATE artists SET thumb_url = ?
+ WHERE id = ? AND (thumb_url IS NULL OR thumb_url = '')
+ """, (thumb_url, artist_id))
+
+ conn.commit()
+ except Exception as e:
+ logger.warning(f"Backfill failed for artist #{artist_id} (match preserved): {e}")
+
+ except Exception as e:
+ logger.error(f"Error updating artist #{artist_id} with Qobuz data: {e}")
+ raise
+ finally:
+ if conn:
+ conn.close()
+
+ def _update_album(self, album_id: int, search_data: Dict[str, Any], full_data: Optional[Dict[str, Any]]):
+ """Store Qobuz metadata for an album"""
+ conn = None
+ try:
+ conn = self.db._get_connection()
+ cursor = conn.cursor()
+
+ data = full_data or search_data
+
+ cursor.execute("""
+ UPDATE albums SET
+ qobuz_id = ?,
+ qobuz_match_status = 'matched',
+ qobuz_last_attempted = CURRENT_TIMESTAMP,
+ updated_at = CURRENT_TIMESTAMP
+ WHERE id = ?
+ """, (
+ str(search_data.get('id')),
+ album_id
+ ))
+ conn.commit()
+
+ # Backfill optional metadata (failures here won't lose the match)
+ try:
+ label = data.get('label', {})
+ label_name = label.get('name', '') if isinstance(label, dict) else str(label) if label else ''
+ if label_name:
+ cursor.execute("""
+ UPDATE albums SET label = ?
+ WHERE id = ? AND (label IS NULL OR label = '')
+ """, (label_name, album_id))
+
+ parental = data.get('parental_warning')
+ if parental is not None:
+ cursor.execute("""
+ UPDATE albums SET explicit = ?
+ WHERE id = ? AND explicit IS NULL
+ """, (1 if parental else 0, album_id))
+
+ genre = data.get('genre', {})
+ genre_name = genre.get('name', '') if isinstance(genre, dict) else str(genre) if genre else ''
+ if genre_name:
+ cursor.execute("""
+ UPDATE albums SET genres = ?
+ WHERE id = ? AND (genres IS NULL OR genres = '' OR genres = '[]')
+ """, (json.dumps([genre_name]), album_id))
+
+ upc = data.get('upc')
+ if upc:
+ cursor.execute("""
+ UPDATE albums SET upc = ?
+ WHERE id = ? AND (upc IS NULL OR upc = '')
+ """, (str(upc), album_id))
+
+ tracks_count = data.get('tracks_count')
+ if tracks_count and isinstance(tracks_count, int) and tracks_count > 0:
+ cursor.execute("""
+ UPDATE albums SET track_count = ?
+ WHERE id = ? AND track_count IS NULL
+ """, (tracks_count, album_id))
+
+ duration = data.get('duration')
+ if duration and isinstance(duration, (int, float)) and duration > 0:
+ duration_ms = int(duration * 1000)
+ cursor.execute("""
+ UPDATE albums SET duration = ?
+ WHERE id = ? AND duration IS NULL
+ """, (duration_ms, album_id))
+
+ copyright_text = data.get('copyright')
+ if copyright_text:
+ cursor.execute("""
+ UPDATE albums SET copyright = ?
+ WHERE id = ? AND (copyright IS NULL OR copyright = '')
+ """, (copyright_text, album_id))
+
+ thumb_url = None
+ image = data.get('image', {})
+ if isinstance(image, dict):
+ thumb_url = image.get('large', image.get('medium', image.get('small', image.get('thumbnail', ''))))
+ elif isinstance(image, str):
+ thumb_url = image
+ if thumb_url:
+ cursor.execute("""
+ UPDATE albums SET thumb_url = ?
+ WHERE id = ? AND (thumb_url IS NULL OR thumb_url = '')
+ """, (thumb_url, album_id))
+
+ conn.commit()
+ except Exception as e:
+ logger.warning(f"Backfill failed for album #{album_id} (match preserved): {e}")
+
+ except Exception as e:
+ logger.error(f"Error updating album #{album_id} with Qobuz data: {e}")
+ raise
+ finally:
+ if conn:
+ conn.close()
+
+ def _update_track(self, track_id: int, search_data: Dict[str, Any], full_data: Optional[Dict[str, Any]]):
+ """Store Qobuz metadata for a track"""
+ conn = None
+ try:
+ conn = self.db._get_connection()
+ cursor = conn.cursor()
+
+ data = full_data or search_data
+
+ cursor.execute("""
+ UPDATE tracks SET
+ qobuz_id = ?,
+ qobuz_match_status = 'matched',
+ qobuz_last_attempted = CURRENT_TIMESTAMP,
+ updated_at = CURRENT_TIMESTAMP
+ WHERE id = ?
+ """, (
+ str(search_data.get('id')),
+ track_id
+ ))
+ conn.commit()
+
+ # Backfill optional metadata (failures here won't lose the match)
+ try:
+ parental = data.get('parental_warning')
+ if parental is not None:
+ cursor.execute("""
+ UPDATE tracks SET explicit = ?
+ WHERE id = ? AND explicit IS NULL
+ """, (1 if parental else 0, track_id))
+
+ isrc = data.get('isrc')
+ if isrc:
+ cursor.execute("""
+ UPDATE tracks SET isrc = ?
+ WHERE id = ? AND (isrc IS NULL OR isrc = '')
+ """, (isrc, track_id))
+
+ duration = data.get('duration')
+ if duration and isinstance(duration, (int, float)) and duration > 0:
+ duration_ms = int(duration * 1000)
+ cursor.execute("""
+ UPDATE tracks SET duration = ?
+ WHERE id = ? AND duration IS NULL
+ """, (duration_ms, track_id))
+
+ copyright_text = data.get('copyright')
+ if copyright_text:
+ cursor.execute("""
+ UPDATE tracks SET copyright = ?
+ WHERE id = ? AND (copyright IS NULL OR copyright = '')
+ """, (copyright_text, track_id))
+
+ conn.commit()
+ except Exception as e:
+ logger.warning(f"Backfill failed for track #{track_id} (match preserved): {e}")
+
+ except Exception as e:
+ logger.error(f"Error updating track #{track_id} with Qobuz data: {e}")
+ raise
+ finally:
+ if conn:
+ conn.close()
+
+ def _mark_status(self, entity_type: str, entity_id: int, status: str):
+ """Mark an entity with a match status"""
+ table_map = {'artist': 'artists', 'album': 'albums', 'track': 'tracks'}
+ table = table_map.get(entity_type)
+ if not table:
+ logger.error(f"Unknown entity type: {entity_type}")
+ return
+
+ conn = None
+ try:
+ conn = self.db._get_connection()
+ cursor = conn.cursor()
+ cursor.execute(f"""
+ UPDATE {table} SET
+ qobuz_match_status = ?,
+ qobuz_last_attempted = CURRENT_TIMESTAMP,
+ updated_at = CURRENT_TIMESTAMP
+ WHERE id = ?
+ """, (status, entity_id))
+ conn.commit()
+ except Exception as e:
+ logger.error(f"Error marking {entity_type} #{entity_id} status: {e}")
+ finally:
+ if conn:
+ conn.close()
+
+ def _count_pending_items(self) -> int:
+ """Count how many items still need processing"""
+ conn = None
+ try:
+ conn = self.db._get_connection()
+ cursor = conn.cursor()
+
+ cursor.execute("""
+ SELECT
+ (SELECT COUNT(*) FROM artists WHERE qobuz_match_status IS NULL) +
+ (SELECT COUNT(*) FROM albums WHERE qobuz_match_status IS NULL) +
+ (SELECT COUNT(*) FROM tracks WHERE qobuz_match_status IS NULL)
+ AS pending
+ """)
+
+ row = cursor.fetchone()
+ return row[0] if row else 0
+
+ except Exception as e:
+ logger.error(f"Error counting pending items: {e}")
+ return 0
+ finally:
+ if conn:
+ conn.close()
+
+ def _get_progress_breakdown(self) -> Dict[str, Dict[str, int]]:
+ """Get progress breakdown by entity type"""
+ conn = None
+ try:
+ conn = self.db._get_connection()
+ cursor = conn.cursor()
+
+ progress = {}
+
+ cursor.execute("""
+ SELECT
+ COUNT(*) AS total,
+ SUM(CASE WHEN qobuz_match_status IS NOT NULL THEN 1 ELSE 0 END) AS processed
+ FROM artists
+ """)
+ row = cursor.fetchone()
+ if row:
+ total, processed = row[0], row[1] or 0
+ progress['artists'] = {
+ 'matched': processed,
+ 'total': total,
+ 'percent': int((processed / total * 100) if total > 0 else 0)
+ }
+
+ cursor.execute("""
+ SELECT
+ COUNT(*) AS total,
+ SUM(CASE WHEN qobuz_match_status IS NOT NULL THEN 1 ELSE 0 END) AS processed
+ FROM albums
+ """)
+ row = cursor.fetchone()
+ if row:
+ total, processed = row[0], row[1] or 0
+ progress['albums'] = {
+ 'matched': processed,
+ 'total': total,
+ 'percent': int((processed / total * 100) if total > 0 else 0)
+ }
+
+ cursor.execute("""
+ SELECT
+ COUNT(*) AS total,
+ SUM(CASE WHEN qobuz_match_status IS NOT NULL THEN 1 ELSE 0 END) AS processed
+ FROM tracks
+ """)
+ row = cursor.fetchone()
+ if row:
+ total, processed = row[0], row[1] or 0
+ progress['tracks'] = {
+ 'matched': processed,
+ 'total': total,
+ 'percent': int((processed / total * 100) if total > 0 else 0)
+ }
+
+ return progress
+
+ except Exception as e:
+ logger.error(f"Error getting progress breakdown: {e}")
+ return {}
+ finally:
+ if conn:
+ conn.close()
diff --git a/core/tidal_client.py b/core/tidal_client.py
index 78758bb3..eca0c904 100644
--- a/core/tidal_client.py
+++ b/core/tidal_client.py
@@ -714,7 +714,234 @@ class TidalClient:
except Exception as e:
logger.error(f"Error searching Tidal tracks: {e}")
return []
-
+
+ # ── Enrichment API Methods ──
+
+ @rate_limited
+ def search_artist(self, name: str) -> Optional[Dict]:
+ """Search for an artist by name. Returns first result as raw dict or None."""
+ try:
+ if not self._ensure_valid_token():
+ return None
+
+ params = {
+ 'query': name,
+ 'type': 'artists',
+ 'limit': 1,
+ 'countryCode': 'US'
+ }
+
+ response = self.session.get(
+ f"{self.base_url}/searchresults",
+ params=params,
+ timeout=10
+ )
+
+ if response.status_code == 429:
+ raise Exception(f"Rate limited (429) on search_artist")
+ if response.status_code == 200:
+ data = response.json()
+ items = []
+ if 'artists' in data and 'items' in data['artists']:
+ items = data['artists']['items']
+ elif 'artists' in data and isinstance(data['artists'], list):
+ items = data['artists']
+ if items:
+ return items[0]
+ else:
+ logger.debug(f"Tidal artist search failed: {response.status_code}")
+ return None
+
+ except Exception as e:
+ if "429" in str(e):
+ raise # Let rate_limited decorator handle retry
+ logger.error(f"Error searching Tidal artist: {e}")
+ return None
+
+ @rate_limited
+ def search_album(self, artist: str, title: str) -> Optional[Dict]:
+ """Search for an album by artist + title. Returns first result as raw dict or None."""
+ try:
+ if not self._ensure_valid_token():
+ return None
+
+ query = f"{artist} {title}" if artist else title
+ params = {
+ 'query': query,
+ 'type': 'albums',
+ 'limit': 1,
+ 'countryCode': 'US'
+ }
+
+ response = self.session.get(
+ f"{self.base_url}/searchresults",
+ params=params,
+ timeout=10
+ )
+
+ if response.status_code == 429:
+ raise Exception(f"Rate limited (429) on search_album")
+ if response.status_code == 200:
+ data = response.json()
+ items = []
+ if 'albums' in data and 'items' in data['albums']:
+ items = data['albums']['items']
+ elif 'albums' in data and isinstance(data['albums'], list):
+ items = data['albums']
+ if items:
+ return items[0]
+ else:
+ logger.debug(f"Tidal album search failed: {response.status_code}")
+ return None
+
+ except Exception as e:
+ if "429" in str(e):
+ raise # Let rate_limited decorator handle retry
+ logger.error(f"Error searching Tidal album: {e}")
+ return None
+
+ @rate_limited
+ def search_track(self, artist: str, title: str) -> Optional[Dict]:
+ """Search for a track by artist + title. Returns first result as raw dict or None."""
+ try:
+ if not self._ensure_valid_token():
+ return None
+
+ query = f"{artist} {title}" if artist else title
+ params = {
+ 'query': query,
+ 'type': 'tracks',
+ 'limit': 1,
+ 'countryCode': 'US'
+ }
+
+ response = self.session.get(
+ f"{self.base_url}/searchresults",
+ params=params,
+ timeout=10
+ )
+
+ if response.status_code == 429:
+ raise Exception(f"Rate limited (429) on search_track")
+ if response.status_code == 200:
+ data = response.json()
+ items = []
+ if 'tracks' in data and 'items' in data['tracks']:
+ items = data['tracks']['items']
+ elif 'tracks' in data and isinstance(data['tracks'], list):
+ items = data['tracks']
+ if items:
+ return items[0]
+ else:
+ logger.debug(f"Tidal track search failed: {response.status_code}")
+ return None
+
+ except Exception as e:
+ if "429" in str(e):
+ raise # Let rate_limited decorator handle retry
+ logger.error(f"Error searching Tidal track: {e}")
+ return None
+
+ @rate_limited
+ def get_artist(self, artist_id: str) -> Optional[Dict]:
+ """Get full artist details by Tidal ID."""
+ try:
+ if not self._ensure_valid_token():
+ return None
+
+ response = self.session.get(
+ f"{self.base_url}/artists/{artist_id}",
+ params={'countryCode': 'US'},
+ headers={'accept': 'application/vnd.api+json'},
+ timeout=10
+ )
+
+ if response.status_code == 429:
+ raise Exception(f"Rate limited (429) on get_artist")
+ if response.status_code == 200:
+ data = response.json()
+ # Handle JSON:API format
+ if 'data' in data and 'attributes' in data.get('data', {}):
+ result = dict(data['data'].get('attributes', {}))
+ result['id'] = data['data'].get('id', artist_id)
+ return result
+ return data
+ else:
+ logger.debug(f"Tidal get_artist failed: {response.status_code}")
+ return None
+
+ except Exception as e:
+ if "429" in str(e):
+ raise # Let rate_limited decorator handle retry
+ logger.error(f"Error getting Tidal artist {artist_id}: {e}")
+ return None
+
+ @rate_limited
+ def get_album(self, album_id: str) -> Optional[Dict]:
+ """Get full album details by Tidal ID."""
+ try:
+ if not self._ensure_valid_token():
+ return None
+
+ response = self.session.get(
+ f"{self.base_url}/albums/{album_id}",
+ params={'countryCode': 'US'},
+ headers={'accept': 'application/vnd.api+json'},
+ timeout=10
+ )
+
+ if response.status_code == 429:
+ raise Exception(f"Rate limited (429) on get_album")
+ if response.status_code == 200:
+ data = response.json()
+ if 'data' in data and 'attributes' in data.get('data', {}):
+ result = dict(data['data'].get('attributes', {}))
+ result['id'] = data['data'].get('id', album_id)
+ return result
+ return data
+ else:
+ logger.debug(f"Tidal get_album failed: {response.status_code}")
+ return None
+
+ except Exception as e:
+ if "429" in str(e):
+ raise # Let rate_limited decorator handle retry
+ logger.error(f"Error getting Tidal album {album_id}: {e}")
+ return None
+
+ @rate_limited
+ def get_track(self, track_id: str) -> Optional[Dict]:
+ """Get full track details by Tidal ID."""
+ try:
+ if not self._ensure_valid_token():
+ return None
+
+ response = self.session.get(
+ f"{self.base_url}/tracks/{track_id}",
+ params={'countryCode': 'US'},
+ headers={'accept': 'application/vnd.api+json'},
+ timeout=10
+ )
+
+ if response.status_code == 429:
+ raise Exception(f"Rate limited (429) on get_track")
+ if response.status_code == 200:
+ data = response.json()
+ if 'data' in data and 'attributes' in data.get('data', {}):
+ result = dict(data['data'].get('attributes', {}))
+ result['id'] = data['data'].get('id', track_id)
+ return result
+ return data
+ else:
+ logger.debug(f"Tidal get_track failed: {response.status_code}")
+ return None
+
+ except Exception as e:
+ if "429" in str(e):
+ raise # Let rate_limited decorator handle retry
+ logger.error(f"Error getting Tidal track {track_id}: {e}")
+ return None
+
@rate_limited
def get_playlist(self, playlist_id: str) -> Optional[Playlist]:
"""Get playlist details including tracks using JSON:API format"""
diff --git a/core/tidal_worker.py b/core/tidal_worker.py
new file mode 100644
index 00000000..9dcfa8a9
--- /dev/null
+++ b/core/tidal_worker.py
@@ -0,0 +1,818 @@
+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.tidal_client import TidalClient
+
+logger = get_logger("tidal_worker")
+
+
+class TidalWorker:
+ """Background worker for enriching library artists, albums, and tracks with Tidal metadata"""
+
+ def __init__(self, database: MusicDatabase, client: TidalClient = None):
+ self.db = database
+ self.client = client or TidalClient()
+
+ # Worker state
+ self.running = False
+ self.paused = False
+ self.should_stop = False
+ self.thread = None
+
+ # 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
+ self.error_retry_days = 7
+
+ # Name matching threshold
+ self.name_similarity_threshold = 0.80
+
+ logger.info("Tidal background worker initialized")
+
+ def start(self):
+ """Start the background worker"""
+ if self.running:
+ logger.warning("Worker already running")
+ return
+
+ self.running = True
+ self.should_stop = False
+ self.thread = threading.Thread(target=self._run, daemon=True)
+ self.thread.start()
+ logger.info("Tidal background worker started")
+
+ def stop(self):
+ """Stop the background worker"""
+ if not self.running:
+ return
+
+ logger.info("Stopping Tidal worker...")
+ self.should_stop = True
+ self.running = False
+
+ if self.thread:
+ self.thread.join(timeout=5)
+
+ logger.info("Tidal worker stopped")
+
+ def pause(self):
+ """Pause the worker"""
+ if not self.running:
+ logger.warning("Worker not running, cannot pause")
+ return
+ self.paused = True
+ logger.info("Tidal worker paused")
+
+ def resume(self):
+ """Resume the worker"""
+ if not self.running:
+ logger.warning("Worker not running, start it first")
+ return
+ self.paused = False
+ logger.info("Tidal worker resumed")
+
+ def get_stats(self) -> Dict[str, Any]:
+ """Get current statistics"""
+ self.stats['pending'] = self._count_pending_items()
+
+ progress = self._get_progress_breakdown()
+
+ 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
+
+ authenticated = False
+ try:
+ authenticated = self.client.is_authenticated()
+ except Exception:
+ pass
+
+ return {
+ 'enabled': True,
+ 'running': is_actually_running and not self.paused,
+ 'paused': self.paused,
+ 'idle': is_idle,
+ 'authenticated': authenticated,
+ 'current_item': self.current_item,
+ 'stats': self.stats.copy(),
+ 'progress': progress
+ }
+
+ def _run(self):
+ """Main worker loop"""
+ logger.info("Tidal worker thread started")
+
+ while not self.should_stop:
+ try:
+ if self.paused:
+ time.sleep(1)
+ continue
+
+ # Auth guard: sleep if not authenticated
+ try:
+ if not self.client.is_authenticated():
+ self.current_item = None
+ time.sleep(30)
+ continue
+ except Exception:
+ time.sleep(30)
+ continue
+
+ self.current_item = None
+
+ item = self._get_next_item()
+
+ if not item:
+ logger.debug("No pending items, sleeping...")
+ time.sleep(10)
+ continue
+
+ self.current_item = item
+
+ self._process_item(item)
+
+ time.sleep(2)
+
+ except Exception as e:
+ logger.error(f"Error in worker loop: {e}")
+ time.sleep(5)
+
+ logger.info("Tidal worker thread finished")
+
+ def _get_next_item(self) -> Optional[Dict[str, Any]]:
+ """Get next item to process from priority queue (artists -> albums -> tracks)"""
+ conn = None
+ try:
+ conn = self.db._get_connection()
+ cursor = conn.cursor()
+
+ # Priority 1: Unattempted artists
+ cursor.execute("""
+ SELECT id, name
+ FROM artists
+ WHERE tidal_match_status IS 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.tidal_id AS artist_tidal_id
+ FROM albums a
+ JOIN artists ar ON a.artist_id = ar.id
+ WHERE a.tidal_match_status IS 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_tidal_id': row[3]}
+
+ # Priority 3: Unattempted tracks
+ cursor.execute("""
+ SELECT t.id, t.title, ar.name AS artist_name, ar.tidal_id AS artist_tidal_id
+ FROM tracks t
+ JOIN artists ar ON t.artist_id = ar.id
+ WHERE t.tidal_match_status IS NULL
+ ORDER BY t.id ASC
+ LIMIT 1
+ """)
+ row = cursor.fetchone()
+ if row:
+ return {'type': 'track', 'id': row[0], 'name': row[1], 'artist': row[2], 'artist_tidal_id': row[3]}
+
+ # Priority 4: Retry 'not_found' or 'error' artists
+ not_found_cutoff = datetime.now() - timedelta(days=self.retry_days)
+ error_cutoff = datetime.now() - timedelta(days=self.error_retry_days)
+ cursor.execute("""
+ SELECT id, name
+ FROM artists
+ WHERE (tidal_match_status = 'not_found' AND tidal_last_attempted < ?)
+ OR (tidal_match_status = 'error' AND tidal_last_attempted < ?)
+ ORDER BY tidal_last_attempted ASC
+ LIMIT 1
+ """, (not_found_cutoff, error_cutoff))
+ row = cursor.fetchone()
+ if row:
+ logger.info(f"Retrying artist '{row[1]}' (last attempted before cutoff)")
+ return {'type': 'artist', 'id': row[0], 'name': row[1]}
+
+ # Priority 5: Retry 'not_found' or 'error' albums
+ cursor.execute("""
+ SELECT a.id, a.title, ar.name AS artist_name, ar.tidal_id AS artist_tidal_id
+ FROM albums a
+ JOIN artists ar ON a.artist_id = ar.id
+ WHERE (a.tidal_match_status = 'not_found' AND a.tidal_last_attempted < ?)
+ OR (a.tidal_match_status = 'error' AND a.tidal_last_attempted < ?)
+ ORDER BY a.tidal_last_attempted ASC
+ LIMIT 1
+ """, (not_found_cutoff, error_cutoff))
+ row = cursor.fetchone()
+ if row:
+ return {'type': 'album', 'id': row[0], 'name': row[1], 'artist': row[2], 'artist_tidal_id': row[3]}
+
+ # Priority 6: Retry 'not_found' or 'error' tracks
+ cursor.execute("""
+ SELECT t.id, t.title, ar.name AS artist_name, ar.tidal_id AS artist_tidal_id
+ FROM tracks t
+ JOIN artists ar ON t.artist_id = ar.id
+ WHERE (t.tidal_match_status = 'not_found' AND t.tidal_last_attempted < ?)
+ OR (t.tidal_match_status = 'error' AND t.tidal_last_attempted < ?)
+ ORDER BY t.tidal_last_attempted ASC
+ LIMIT 1
+ """, (not_found_cutoff, error_cutoff))
+ row = cursor.fetchone()
+ if row:
+ return {'type': 'track', 'id': row[0], 'name': row[1], 'artist': row[2], 'artist_tidal_id': row[3]}
+
+ return None
+
+ except Exception as e:
+ logger.error(f"Error getting next item: {e}")
+ return None
+ finally:
+ if conn:
+ conn.close()
+
+ 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'[^\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 Tidal 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()
+ logger.debug(f"Name similarity: '{query_name}' vs '{result_name}' = {similarity:.2f}")
+ return similarity >= self.name_similarity_threshold
+
+ def _verify_artist_id(self, item: Dict[str, Any], result_artist_id) -> bool:
+ """Verify/correct parent artist's Tidal ID based on album/track match"""
+ parent_tidal_id = item.get('artist_tidal_id')
+ if not parent_tidal_id or not result_artist_id:
+ return True
+
+ if str(result_artist_id) != str(parent_tidal_id):
+ logger.info(
+ f"Artist ID correction from {item['type']} '{item['name']}': "
+ f"updating parent artist Tidal ID from {parent_tidal_id} to {result_artist_id}"
+ )
+ self._correct_artist_tidal_id(item, str(result_artist_id))
+
+ return True
+
+ def _correct_artist_tidal_id(self, item: Dict[str, Any], correct_tidal_id: str):
+ """Correct the parent artist's tidal_id based on a more specific album/track match"""
+ conn = None
+ try:
+ conn = self.db._get_connection()
+ cursor = conn.cursor()
+
+ table = 'albums' if item['type'] == 'album' else 'tracks'
+ cursor.execute(f"SELECT artist_id FROM {table} WHERE id = ?", (item['id'],))
+ row = cursor.fetchone()
+ if not row:
+ return
+
+ artist_id = row[0]
+ cursor.execute("""
+ UPDATE artists SET
+ tidal_id = ?,
+ updated_at = CURRENT_TIMESTAMP
+ WHERE id = ?
+ """, (correct_tidal_id, artist_id))
+ conn.commit()
+
+ logger.info(f"Corrected artist #{artist_id} Tidal ID to {correct_tidal_id}")
+
+ except Exception as e:
+ logger.error(f"Error correcting artist Tidal ID: {e}")
+ finally:
+ if conn:
+ conn.close()
+
+ def _process_item(self, item: Dict[str, Any]):
+ """Process a single item (artist, album, or track)"""
+ try:
+ item_type = item['type']
+ item_id = item['id']
+ item_name = item['name']
+
+ logger.debug(f"Processing {item_type} #{item_id}: {item_name}")
+
+ if item_type == 'artist':
+ self._process_artist(item_id, item_name)
+ elif item_type == 'album':
+ self._process_album(item_id, item_name, item.get('artist', ''), item)
+ elif item_type == 'track':
+ self._process_track(item_id, item_name, item.get('artist', ''), item)
+
+ except Exception as e:
+ error_str = str(e).lower()
+ if '429' in error_str or 'rate limit' in error_str:
+ # Rate limit — don't mark as error, leave for retry on next loop
+ logger.warning(f"Rate limited while processing {item['type']} #{item['id']}, will retry")
+ return
+ logger.error(f"Error processing {item['type']} #{item['id']}: {e}")
+ self.stats['errors'] += 1
+ try:
+ self._mark_status(item['type'], item['id'], 'error')
+ except Exception as e2:
+ logger.error(f"Error updating item status: {e2}")
+
+ def _process_artist(self, artist_id: int, artist_name: str):
+ """Process an artist: search Tidal, verify, store metadata"""
+ result = self.client.search_artist(artist_name)
+ if result:
+ result_name = result.get('name', '')
+ if self._name_matches(artist_name, result_name):
+ tidal_artist_id = result.get('id')
+ if not tidal_artist_id:
+ self._mark_status('artist', artist_id, 'error')
+ self.stats['errors'] += 1
+ logger.warning(f"Tidal search result for '{artist_name}' has no ID")
+ return
+
+ # Fetch full artist details for image
+ full_artist = None
+ try:
+ full_artist = self.client.get_artist(tidal_artist_id)
+ except Exception as e:
+ logger.warning(f"Failed to fetch full artist details for '{artist_name}': {e}")
+
+ self._update_artist(artist_id, result, full_artist)
+ self.stats['matched'] += 1
+ logger.info(f"Matched artist '{artist_name}' -> Tidal ID: {tidal_artist_id}")
+ else:
+ self._mark_status('artist', artist_id, 'not_found')
+ self.stats['not_found'] += 1
+ logger.debug(f"Name mismatch for artist '{artist_name}' (got '{result_name}')")
+ else:
+ self._mark_status('artist', artist_id, 'not_found')
+ self.stats['not_found'] += 1
+ logger.debug(f"No match for artist '{artist_name}'")
+
+ def _process_album(self, album_id: int, album_name: str, artist_name: str, item: Dict[str, Any]):
+ """Process an album: search Tidal, verify, fetch full details, store metadata"""
+ result = self.client.search_album(artist_name, album_name)
+ if result:
+ result_name = result.get('title', '')
+ if self._name_matches(album_name, result_name):
+ # Verify artist ID
+ result_artist = result.get('artist', {})
+ result_artist_id = result_artist.get('id') if result_artist else None
+ self._verify_artist_id(item, result_artist_id)
+
+ # Fetch full album details
+ tidal_album_id = result.get('id')
+ if not tidal_album_id:
+ self._mark_status('album', album_id, 'error')
+ self.stats['errors'] += 1
+ logger.warning(f"Tidal search result for album '{album_name}' has no ID")
+ return
+
+ full_album = None
+ try:
+ full_album = self.client.get_album(tidal_album_id)
+ except Exception as e:
+ logger.warning(f"Failed to fetch full album details for '{album_name}': {e}")
+
+ if full_album is None:
+ self._mark_status('album', album_id, 'error')
+ self.stats['errors'] += 1
+ logger.warning(f"Album '{album_name}' matched but full details unavailable, will retry")
+ return
+
+ self._update_album(album_id, result, full_album)
+ self.stats['matched'] += 1
+ logger.info(f"Matched album '{album_name}' -> Tidal ID: {tidal_album_id}")
+ else:
+ self._mark_status('album', album_id, 'not_found')
+ self.stats['not_found'] += 1
+ logger.debug(f"Name mismatch for album '{album_name}' (got '{result_name}')")
+ else:
+ self._mark_status('album', album_id, 'not_found')
+ self.stats['not_found'] += 1
+ logger.debug(f"No match for album '{album_name}'")
+
+ def _process_track(self, track_id: int, track_name: str, artist_name: str, item: Dict[str, Any]):
+ """Process a track: search Tidal, verify, fetch full details, store metadata"""
+ result = self.client.search_track(artist_name, track_name)
+ if result:
+ result_name = result.get('title', '')
+ if self._name_matches(track_name, result_name):
+ # Verify artist ID
+ result_artist = result.get('artist', {})
+ result_artist_id = result_artist.get('id') if result_artist else None
+ self._verify_artist_id(item, result_artist_id)
+
+ # Fetch full track details
+ tidal_track_id = result.get('id')
+ if not tidal_track_id:
+ self._mark_status('track', track_id, 'error')
+ self.stats['errors'] += 1
+ logger.warning(f"Tidal search result for track '{track_name}' has no ID")
+ return
+
+ full_track = None
+ try:
+ full_track = self.client.get_track(tidal_track_id)
+ except Exception as e:
+ logger.warning(f"Failed to fetch full track details for '{track_name}': {e}")
+
+ if full_track is None:
+ self._mark_status('track', track_id, 'error')
+ self.stats['errors'] += 1
+ logger.warning(f"Track '{track_name}' matched but full details unavailable, will retry")
+ return
+
+ self._update_track(track_id, result, full_track)
+ self.stats['matched'] += 1
+ logger.info(f"Matched track '{track_name}' -> Tidal ID: {tidal_track_id}")
+ else:
+ self._mark_status('track', track_id, 'not_found')
+ self.stats['not_found'] += 1
+ logger.debug(f"Name mismatch for track '{track_name}' (got '{result_name}')")
+ else:
+ self._mark_status('track', track_id, 'not_found')
+ self.stats['not_found'] += 1
+ logger.debug(f"No match for track '{track_name}'")
+
+ def _update_artist(self, artist_id: int, data: Dict[str, Any], full_data: Optional[Dict[str, Any]] = None):
+ """Store Tidal metadata for an artist"""
+ conn = None
+ try:
+ conn = self.db._get_connection()
+ cursor = conn.cursor()
+
+ cursor.execute("""
+ UPDATE artists SET
+ tidal_id = ?,
+ tidal_match_status = 'matched',
+ tidal_last_attempted = CURRENT_TIMESTAMP,
+ updated_at = CURRENT_TIMESTAMP
+ WHERE id = ?
+ """, (
+ str(data.get('id')),
+ artist_id
+ ))
+ conn.commit()
+
+ # Backfill optional metadata (failures here won't lose the match)
+ try:
+ thumb_url = None
+ if full_data:
+ # V2 detail may have picture array or picture URL
+ pictures = full_data.get('picture', [])
+ if isinstance(pictures, list) and pictures:
+ # Pick largest available
+ for size in ['1080x1080', '750x750', '480x480', '320x320']:
+ for pic in pictures:
+ if isinstance(pic, dict) and size in pic.get('url', ''):
+ thumb_url = pic['url']
+ break
+ if thumb_url:
+ break
+ if not thumb_url and isinstance(pictures[0], dict):
+ thumb_url = pictures[0].get('url')
+ elif not thumb_url and isinstance(pictures[0], str):
+ thumb_url = pictures[0]
+ elif isinstance(pictures, str):
+ thumb_url = pictures
+ # Also check imageLinks (JSON:API attributes are flattened to top level)
+ if not thumb_url:
+ pic_links = full_data.get('imageLinks', [])
+ if pic_links:
+ for pl in pic_links:
+ if isinstance(pl, dict):
+ thumb_url = pl.get('href', '')
+ break
+
+ if not thumb_url:
+ thumb_url = data.get('picture', data.get('image', ''))
+ if isinstance(thumb_url, list) and thumb_url:
+ thumb_url = thumb_url[0].get('url', '') if isinstance(thumb_url[0], dict) else str(thumb_url[0])
+
+ if thumb_url:
+ cursor.execute("""
+ UPDATE artists SET thumb_url = ?
+ WHERE id = ? AND (thumb_url IS NULL OR thumb_url = '')
+ """, (thumb_url, artist_id))
+
+ conn.commit()
+ except Exception as e:
+ logger.warning(f"Backfill failed for artist #{artist_id} (match preserved): {e}")
+
+ except Exception as e:
+ logger.error(f"Error updating artist #{artist_id} with Tidal data: {e}")
+ raise
+ finally:
+ if conn:
+ conn.close()
+
+ def _update_album(self, album_id: int, search_data: Dict[str, Any], full_data: Optional[Dict[str, Any]]):
+ """Store Tidal metadata for an album"""
+ conn = None
+ try:
+ conn = self.db._get_connection()
+ cursor = conn.cursor()
+
+ data = full_data or search_data
+
+ cursor.execute("""
+ UPDATE albums SET
+ tidal_id = ?,
+ tidal_match_status = 'matched',
+ tidal_last_attempted = CURRENT_TIMESTAMP,
+ updated_at = CURRENT_TIMESTAMP
+ WHERE id = ?
+ """, (
+ str(search_data.get('id')),
+ album_id
+ ))
+ conn.commit()
+
+ # Backfill optional metadata (failures here won't lose the match)
+ try:
+ # Backfill label (can be string or dict with 'name' key in JSON:API)
+ label = data.get('label')
+ if isinstance(label, dict):
+ label = label.get('name', '')
+ if label:
+ cursor.execute("""
+ UPDATE albums SET label = ?
+ WHERE id = ? AND (label IS NULL OR label = '')
+ """, (str(label), album_id))
+
+ # Backfill explicit flag
+ explicit = data.get('explicit')
+ if explicit is not None:
+ cursor.execute("""
+ UPDATE albums SET explicit = ?
+ WHERE id = ? AND explicit IS NULL
+ """, (1 if explicit else 0, album_id))
+
+ # Backfill UPC
+ upc = data.get('upc', data.get('barcodeId', ''))
+ if upc:
+ cursor.execute("""
+ UPDATE albums SET upc = ?
+ WHERE id = ? AND (upc IS NULL OR upc = '')
+ """, (str(upc), album_id))
+
+ # Backfill track_count
+ num_tracks = data.get('numberOfTracks', data.get('numberOfItems'))
+ if num_tracks and isinstance(num_tracks, int) and num_tracks > 0:
+ cursor.execute("""
+ UPDATE albums SET track_count = ?
+ WHERE id = ? AND track_count IS NULL
+ """, (num_tracks, album_id))
+
+ # Backfill duration (Tidal returns seconds, DB stores milliseconds)
+ duration = data.get('duration')
+ if duration and isinstance(duration, (int, float)) and duration > 0:
+ duration_ms = int(duration * 1000)
+ cursor.execute("""
+ UPDATE albums SET duration = ?
+ WHERE id = ? AND duration IS NULL
+ """, (duration_ms, album_id))
+
+ # Backfill copyright
+ copyright_text = data.get('copyright')
+ if copyright_text:
+ cursor.execute("""
+ UPDATE albums SET copyright = ?
+ WHERE id = ? AND (copyright IS NULL OR copyright = '')
+ """, (copyright_text, album_id))
+
+ # Backfill thumb_url
+ thumb_url = None
+ cover = data.get('cover', data.get('image', ''))
+ if isinstance(cover, list) and cover:
+ thumb_url = cover[0].get('url', '') if isinstance(cover[0], dict) else str(cover[0])
+ elif isinstance(cover, str) and cover:
+ thumb_url = cover
+ # JSON:API imageLinks (attributes are flattened to top level)
+ if not thumb_url:
+ img_links = data.get('imageLinks', [])
+ if img_links:
+ for il in img_links:
+ if isinstance(il, dict):
+ thumb_url = il.get('href', '')
+ break
+ if thumb_url:
+ cursor.execute("""
+ UPDATE albums SET thumb_url = ?
+ WHERE id = ? AND (thumb_url IS NULL OR thumb_url = '')
+ """, (thumb_url, album_id))
+
+ conn.commit()
+ except Exception as e:
+ logger.warning(f"Backfill failed for album #{album_id} (match preserved): {e}")
+
+ except Exception as e:
+ logger.error(f"Error updating album #{album_id} with Tidal data: {e}")
+ raise
+ finally:
+ if conn:
+ conn.close()
+
+ def _update_track(self, track_id: int, search_data: Dict[str, Any], full_data: Optional[Dict[str, Any]]):
+ """Store Tidal metadata for a track"""
+ conn = None
+ try:
+ conn = self.db._get_connection()
+ cursor = conn.cursor()
+
+ data = full_data or search_data
+
+ cursor.execute("""
+ UPDATE tracks SET
+ tidal_id = ?,
+ tidal_match_status = 'matched',
+ tidal_last_attempted = CURRENT_TIMESTAMP,
+ updated_at = CURRENT_TIMESTAMP
+ WHERE id = ?
+ """, (
+ str(search_data.get('id')),
+ track_id
+ ))
+ conn.commit()
+
+ # Backfill optional metadata (failures here won't lose the match)
+ try:
+ explicit = data.get('explicit')
+ if explicit is not None:
+ cursor.execute("""
+ UPDATE tracks SET explicit = ?
+ WHERE id = ? AND explicit IS NULL
+ """, (1 if explicit else 0, track_id))
+
+ isrc = data.get('isrc')
+ if isrc:
+ cursor.execute("""
+ UPDATE tracks SET isrc = ?
+ WHERE id = ? AND (isrc IS NULL OR isrc = '')
+ """, (isrc, track_id))
+
+ duration = data.get('duration')
+ if duration and isinstance(duration, (int, float)) and duration > 0:
+ duration_ms = int(duration * 1000)
+ cursor.execute("""
+ UPDATE tracks SET duration = ?
+ WHERE id = ? AND duration IS NULL
+ """, (duration_ms, track_id))
+
+ copyright_text = data.get('copyright')
+ if copyright_text:
+ cursor.execute("""
+ UPDATE tracks SET copyright = ?
+ WHERE id = ? AND (copyright IS NULL OR copyright = '')
+ """, (copyright_text, track_id))
+
+ conn.commit()
+ except Exception as e:
+ logger.warning(f"Backfill failed for track #{track_id} (match preserved): {e}")
+
+ except Exception as e:
+ logger.error(f"Error updating track #{track_id} with Tidal data: {e}")
+ raise
+ finally:
+ if conn:
+ conn.close()
+
+ def _mark_status(self, entity_type: str, entity_id: int, status: str):
+ """Mark an entity with a match status"""
+ table_map = {'artist': 'artists', 'album': 'albums', 'track': 'tracks'}
+ table = table_map.get(entity_type)
+ if not table:
+ logger.error(f"Unknown entity type: {entity_type}")
+ return
+
+ conn = None
+ try:
+ conn = self.db._get_connection()
+ cursor = conn.cursor()
+ cursor.execute(f"""
+ UPDATE {table} SET
+ tidal_match_status = ?,
+ tidal_last_attempted = CURRENT_TIMESTAMP,
+ updated_at = CURRENT_TIMESTAMP
+ WHERE id = ?
+ """, (status, entity_id))
+ conn.commit()
+ except Exception as e:
+ logger.error(f"Error marking {entity_type} #{entity_id} status: {e}")
+ finally:
+ if conn:
+ conn.close()
+
+ def _count_pending_items(self) -> int:
+ """Count how many items still need processing"""
+ conn = None
+ try:
+ conn = self.db._get_connection()
+ cursor = conn.cursor()
+
+ cursor.execute("""
+ SELECT
+ (SELECT COUNT(*) FROM artists WHERE tidal_match_status IS NULL) +
+ (SELECT COUNT(*) FROM albums WHERE tidal_match_status IS NULL) +
+ (SELECT COUNT(*) FROM tracks WHERE tidal_match_status IS NULL)
+ AS pending
+ """)
+
+ row = cursor.fetchone()
+ return row[0] if row else 0
+
+ except Exception as e:
+ logger.error(f"Error counting pending items: {e}")
+ return 0
+ finally:
+ if conn:
+ conn.close()
+
+ def _get_progress_breakdown(self) -> Dict[str, Dict[str, int]]:
+ """Get progress breakdown by entity type"""
+ conn = None
+ try:
+ conn = self.db._get_connection()
+ cursor = conn.cursor()
+
+ progress = {}
+
+ cursor.execute("""
+ SELECT
+ COUNT(*) AS total,
+ SUM(CASE WHEN tidal_match_status IS NOT NULL THEN 1 ELSE 0 END) AS processed
+ FROM artists
+ """)
+ row = cursor.fetchone()
+ if row:
+ total, processed = row[0], row[1] or 0
+ progress['artists'] = {
+ 'matched': processed,
+ 'total': total,
+ 'percent': int((processed / total * 100) if total > 0 else 0)
+ }
+
+ cursor.execute("""
+ SELECT
+ COUNT(*) AS total,
+ SUM(CASE WHEN tidal_match_status IS NOT NULL THEN 1 ELSE 0 END) AS processed
+ FROM albums
+ """)
+ row = cursor.fetchone()
+ if row:
+ total, processed = row[0], row[1] or 0
+ progress['albums'] = {
+ 'matched': processed,
+ 'total': total,
+ 'percent': int((processed / total * 100) if total > 0 else 0)
+ }
+
+ cursor.execute("""
+ SELECT
+ COUNT(*) AS total,
+ SUM(CASE WHEN tidal_match_status IS NOT NULL THEN 1 ELSE 0 END) AS processed
+ FROM tracks
+ """)
+ row = cursor.fetchone()
+ if row:
+ total, processed = row[0], row[1] or 0
+ progress['tracks'] = {
+ 'matched': processed,
+ 'total': total,
+ 'percent': int((processed / total * 100) if total > 0 else 0)
+ }
+
+ return progress
+
+ except Exception as e:
+ logger.error(f"Error getting progress breakdown: {e}")
+ return {}
+ finally:
+ if conn:
+ conn.close()
diff --git a/database/music_database.py b/database/music_database.py
index 080b8258..fa5dbe61 100644
--- a/database/music_database.py
+++ b/database/music_database.py
@@ -319,6 +319,9 @@ class MusicDatabase:
# Add Last.fm and Genius enrichment columns (migration)
self._add_lastfm_genius_columns(cursor)
+ # Add Tidal and Qobuz enrichment columns (migration)
+ self._add_tidal_qobuz_enrichment_columns(cursor)
+
# Bubble snapshots table for persisting UI state across page refreshes
cursor.execute("""
CREATE TABLE IF NOT EXISTS bubble_snapshots (
@@ -1571,6 +1574,90 @@ class MusicDatabase:
logger.error(f"Error adding Last.fm/Genius enrichment columns: {e}")
# Don't raise - this is a migration, database can still function
+ def _add_tidal_qobuz_enrichment_columns(self, cursor):
+ """Add Tidal and Qobuz enrichment tracking columns to artists, albums, tracks"""
+ try:
+ # --- Artists ---
+ cursor.execute("PRAGMA table_info(artists)")
+ artists_columns = [column[1] for column in cursor.fetchall()]
+
+ if 'tidal_id' not in artists_columns:
+ cursor.execute("ALTER TABLE artists ADD COLUMN tidal_id TEXT")
+ if 'tidal_match_status' not in artists_columns:
+ cursor.execute("ALTER TABLE artists ADD COLUMN tidal_match_status TEXT")
+ if 'tidal_last_attempted' not in artists_columns:
+ cursor.execute("ALTER TABLE artists ADD COLUMN tidal_last_attempted TIMESTAMP")
+ if 'qobuz_id' not in artists_columns:
+ cursor.execute("ALTER TABLE artists ADD COLUMN qobuz_id TEXT")
+ if 'qobuz_match_status' not in artists_columns:
+ cursor.execute("ALTER TABLE artists ADD COLUMN qobuz_match_status TEXT")
+ if 'qobuz_last_attempted' not in artists_columns:
+ cursor.execute("ALTER TABLE artists ADD COLUMN qobuz_last_attempted TIMESTAMP")
+
+ cursor.execute("CREATE INDEX IF NOT EXISTS idx_artists_tidal_id ON artists (tidal_id)")
+ cursor.execute("CREATE INDEX IF NOT EXISTS idx_artists_tidal_status ON artists (tidal_match_status)")
+ cursor.execute("CREATE INDEX IF NOT EXISTS idx_artists_qobuz_id ON artists (qobuz_id)")
+ cursor.execute("CREATE INDEX IF NOT EXISTS idx_artists_qobuz_status ON artists (qobuz_match_status)")
+
+ # --- Albums ---
+ cursor.execute("PRAGMA table_info(albums)")
+ albums_columns = [column[1] for column in cursor.fetchall()]
+
+ if 'tidal_id' not in albums_columns:
+ cursor.execute("ALTER TABLE albums ADD COLUMN tidal_id TEXT")
+ if 'tidal_match_status' not in albums_columns:
+ cursor.execute("ALTER TABLE albums ADD COLUMN tidal_match_status TEXT")
+ if 'tidal_last_attempted' not in albums_columns:
+ cursor.execute("ALTER TABLE albums ADD COLUMN tidal_last_attempted TIMESTAMP")
+ if 'qobuz_id' not in albums_columns:
+ cursor.execute("ALTER TABLE albums ADD COLUMN qobuz_id TEXT")
+ if 'qobuz_match_status' not in albums_columns:
+ cursor.execute("ALTER TABLE albums ADD COLUMN qobuz_match_status TEXT")
+ if 'qobuz_last_attempted' not in albums_columns:
+ cursor.execute("ALTER TABLE albums ADD COLUMN qobuz_last_attempted TIMESTAMP")
+
+ cursor.execute("CREATE INDEX IF NOT EXISTS idx_albums_tidal_id ON albums (tidal_id)")
+ cursor.execute("CREATE INDEX IF NOT EXISTS idx_albums_tidal_status ON albums (tidal_match_status)")
+ cursor.execute("CREATE INDEX IF NOT EXISTS idx_albums_qobuz_id ON albums (qobuz_id)")
+ cursor.execute("CREATE INDEX IF NOT EXISTS idx_albums_qobuz_status ON albums (qobuz_match_status)")
+
+ # --- Albums (extra metadata columns) ---
+ if 'upc' not in albums_columns:
+ cursor.execute("ALTER TABLE albums ADD COLUMN upc TEXT")
+ if 'copyright' not in albums_columns:
+ cursor.execute("ALTER TABLE albums ADD COLUMN copyright TEXT")
+
+ # --- Tracks ---
+ cursor.execute("PRAGMA table_info(tracks)")
+ tracks_columns = [column[1] for column in cursor.fetchall()]
+
+ if 'tidal_id' not in tracks_columns:
+ cursor.execute("ALTER TABLE tracks ADD COLUMN tidal_id TEXT")
+ if 'tidal_match_status' not in tracks_columns:
+ cursor.execute("ALTER TABLE tracks ADD COLUMN tidal_match_status TEXT")
+ if 'tidal_last_attempted' not in tracks_columns:
+ cursor.execute("ALTER TABLE tracks ADD COLUMN tidal_last_attempted TIMESTAMP")
+ if 'qobuz_id' not in tracks_columns:
+ cursor.execute("ALTER TABLE tracks ADD COLUMN qobuz_id TEXT")
+ if 'qobuz_match_status' not in tracks_columns:
+ cursor.execute("ALTER TABLE tracks ADD COLUMN qobuz_match_status TEXT")
+ if 'qobuz_last_attempted' not in tracks_columns:
+ cursor.execute("ALTER TABLE tracks ADD COLUMN qobuz_last_attempted TIMESTAMP")
+ if 'isrc' not in tracks_columns:
+ cursor.execute("ALTER TABLE tracks ADD COLUMN isrc TEXT")
+ if 'copyright' not in tracks_columns:
+ cursor.execute("ALTER TABLE tracks ADD COLUMN copyright TEXT")
+
+ cursor.execute("CREATE INDEX IF NOT EXISTS idx_tracks_tidal_id ON tracks (tidal_id)")
+ cursor.execute("CREATE INDEX IF NOT EXISTS idx_tracks_tidal_status ON tracks (tidal_match_status)")
+ cursor.execute("CREATE INDEX IF NOT EXISTS idx_tracks_qobuz_id ON tracks (qobuz_id)")
+ cursor.execute("CREATE INDEX IF NOT EXISTS idx_tracks_qobuz_status ON tracks (qobuz_match_status)")
+ cursor.execute("CREATE INDEX IF NOT EXISTS idx_tracks_isrc ON tracks (isrc)")
+
+ except Exception as e:
+ logger.error(f"Error adding Tidal/Qobuz enrichment columns: {e}")
+ # Don't raise - this is a migration, database can still function
+
def _add_retag_tables(self, cursor):
"""Add retag tool tables for tracking processed downloads"""
try:
diff --git a/web_server.py b/web_server.py
index 2e8b8d1f..01f9450f 100644
--- a/web_server.py
+++ b/web_server.py
@@ -95,6 +95,8 @@ from core.spotify_worker import SpotifyWorker
from core.itunes_worker import iTunesWorker
from core.lastfm_worker import LastFMWorker
from core.genius_worker import GeniusWorker
+from core.tidal_worker import TidalWorker
+from core.qobuz_worker import QobuzWorker
from core.hydrabase_worker import HydrabaseWorker
from core.hydrabase_client import HydrabaseClient
from core.automation_engine import AutomationEngine
@@ -4038,6 +4040,8 @@ def handle_settings():
lastfm_worker._init_client()
if genius_worker:
genius_worker._init_client()
+ if tidal_enrichment_worker:
+ tidal_enrichment_worker.client = tidal_client
print("✅ Service clients re-initialized with new settings.")
return jsonify({"success": True, "message": "Settings saved successfully."})
except Exception as e:
@@ -5626,6 +5630,8 @@ def tidal_callback():
if success:
# Re-initialize the main global tidal_client instance with the new token
tidal_client = TidalClient()
+ if tidal_enrichment_worker:
+ tidal_enrichment_worker.client = tidal_client
return "✅ Tidal Authentication Successful!
You can now close this window and return to the SoulSync application.
"
else:
return "❌ Tidal Authentication Failed
Could not exchange authorization code for a token. Please try again.
", 400
@@ -9641,14 +9647,14 @@ def library_play_track():
print(f"❌ Error playing library track: {e}")
return jsonify({"success": False, "error": str(e)}), 500
-_enrichment_locks = {svc: threading.Lock() for svc in ('audiodb', 'deezer', 'musicbrainz', 'spotify', 'itunes', 'lastfm', 'genius')}
+_enrichment_locks = {svc: threading.Lock() for svc in ('audiodb', 'deezer', 'musicbrainz', 'spotify', 'itunes', 'lastfm', 'genius', 'tidal', 'qobuz')}
@app.route('/api/library/enrich', methods=['POST'])
def library_enrich_entity():
"""Trigger enrichment of a specific entity from a single service.
Body: { entity_type: 'artist'|'album'|'track', entity_id: str, service: str,
name: str, artist_name: str? }
- service: 'audiodb', 'deezer', 'musicbrainz', 'spotify', 'itunes', 'lastfm', 'genius'
+ service: 'audiodb', 'deezer', 'musicbrainz', 'spotify', 'itunes', 'lastfm', 'genius', 'tidal', 'qobuz'
"""
try:
data = request.get_json()
@@ -9667,7 +9673,7 @@ def library_enrich_entity():
if entity_type not in ('artist', 'album', 'track'):
return jsonify({"success": False, "error": "entity_type must be artist, album, or track"}), 400
- valid_services = ('audiodb', 'deezer', 'musicbrainz', 'spotify', 'itunes', 'lastfm', 'genius')
+ valid_services = ('audiodb', 'deezer', 'musicbrainz', 'spotify', 'itunes', 'lastfm', 'genius', 'tidal', 'qobuz')
if service not in valid_services:
return jsonify({"success": False, "error": f"service must be one of: {', '.join(valid_services)}"}), 400
@@ -9809,6 +9815,30 @@ def _run_single_enrichment(service, entity_type, entity_id, name, artist_name):
genius_worker._process_item(item)
return {"success": True, "message": f"Genius lookup complete for {entity_type}"}
+ elif service == 'tidal':
+ if not tidal_enrichment_worker:
+ return {"success": False, "error": "Tidal worker not initialized"}
+ item = {'type': entity_type, 'id': entity_id, 'name': name, 'artist': artist_name}
+ if entity_type == 'artist':
+ tidal_enrichment_worker._process_artist(entity_id, name)
+ elif entity_type == 'album':
+ tidal_enrichment_worker._process_album(entity_id, name, artist_name, item)
+ elif entity_type == 'track':
+ tidal_enrichment_worker._process_track(entity_id, name, artist_name, item)
+ return {"success": True, "message": f"Tidal lookup complete for {entity_type}"}
+
+ elif service == 'qobuz':
+ if not qobuz_enrichment_worker:
+ return {"success": False, "error": "Qobuz worker not initialized"}
+ item = {'type': entity_type, 'id': entity_id, 'name': name, 'artist': artist_name}
+ if entity_type == 'artist':
+ qobuz_enrichment_worker._process_artist(entity_id, name)
+ elif entity_type == 'album':
+ qobuz_enrichment_worker._process_album(entity_id, name, artist_name, item)
+ elif entity_type == 'track':
+ qobuz_enrichment_worker._process_track(entity_id, name, artist_name, item)
+ return {"success": True, "message": f"Qobuz lookup complete for {entity_type}"}
+
else:
return {"success": False, "error": f"Unknown service: {service}"}
@@ -9967,6 +9997,60 @@ def _search_service(service, entity_type, query):
'image': result.get('song_art_image_url'), 'extra': result.get('artist_names', '')}]
return []
+ elif service == 'tidal':
+ if not tidal_enrichment_worker or not tidal_enrichment_worker.client:
+ raise ValueError("Tidal worker not initialized")
+ client = tidal_enrichment_worker.client
+ if entity_type == 'artist':
+ result = client.search_artist(query)
+ if result:
+ thumb = result.get('picture', '')
+ if isinstance(thumb, list) and thumb:
+ thumb = thumb[0].get('url', '') if isinstance(thumb[0], dict) else str(thumb[0])
+ return [{'id': str(result.get('id', '')), 'name': result.get('name', ''),
+ 'image': thumb if isinstance(thumb, str) else None, 'extra': ''}]
+ elif entity_type == 'album':
+ result = client.search_album('', query)
+ if result:
+ return [{'id': str(result.get('id', '')), 'name': result.get('title', ''),
+ 'image': None, 'extra': result.get('artist', {}).get('name', '') if isinstance(result.get('artist'), dict) else ''}]
+ elif entity_type == 'track':
+ result = client.search_track('', query)
+ if result:
+ artist_name = result.get('artist', {}).get('name', '') if isinstance(result.get('artist'), dict) else ''
+ return [{'id': str(result.get('id', '')), 'name': result.get('title', ''),
+ 'image': None, 'extra': artist_name}]
+ return []
+
+ elif service == 'qobuz':
+ if not qobuz_enrichment_worker or not qobuz_enrichment_worker.client:
+ raise ValueError("Qobuz worker not initialized")
+ client = qobuz_enrichment_worker.client
+ if entity_type == 'artist':
+ result = client.search_artist(query)
+ if result:
+ image = result.get('image', {})
+ thumb = image.get('large', image.get('medium', '')) if isinstance(image, dict) else ''
+ return [{'id': str(result.get('id', '')), 'name': result.get('name', ''),
+ 'image': thumb, 'extra': ''}]
+ elif entity_type == 'album':
+ result = client.search_album('', query)
+ if result:
+ artist_name = result.get('artist', {}).get('name', '') if isinstance(result.get('artist'), dict) else ''
+ image = result.get('image', {})
+ thumb = image.get('large', image.get('medium', '')) if isinstance(image, dict) else ''
+ return [{'id': str(result.get('id', '')), 'name': result.get('title', ''),
+ 'image': thumb, 'extra': artist_name}]
+ elif entity_type == 'track':
+ result = client.search_track('', query)
+ if result:
+ artist_name = result.get('performer', {}).get('name', '') if isinstance(result.get('performer'), dict) else ''
+ if not artist_name:
+ artist_name = result.get('artist', {}).get('name', '') if isinstance(result.get('artist'), dict) else ''
+ return [{'id': str(result.get('id', '')), 'name': result.get('title', ''),
+ 'image': None, 'extra': artist_name}]
+ return []
+
elif service == 'audiodb':
if not audiodb_worker or not audiodb_worker.client:
raise ValueError("AudioDB worker not initialized")
@@ -10005,6 +10089,8 @@ _SERVICE_ID_COLUMNS = {
'itunes': {'artist': 'itunes_artist_id', 'album': 'itunes_album_id', 'track': 'itunes_track_id'},
'lastfm': {'artist': 'lastfm_url', 'album': 'lastfm_url', 'track': 'lastfm_url'},
'genius': {'artist': 'genius_id', 'track': 'genius_id'},
+ 'tidal': {'artist': 'tidal_id', 'album': 'tidal_id', 'track': 'tidal_id'},
+ 'qobuz': {'artist': 'qobuz_id', 'album': 'qobuz_id', 'track': 'qobuz_id'},
}
@app.route('/api/library/manual-match', methods=['PUT'])
@@ -22606,47 +22692,46 @@ _playlist_discovery_cancelled = set() # Set of automation_ids that have been ca
def _pause_enrichment_workers(label='discovery'):
"""Pause enrichment workers during discovery to reduce API contention.
- Returns tuple of (spotify_was_running, itunes_was_running) for resume."""
- s_was = False
- i_was = False
- try:
- if spotify_enrichment_worker and not spotify_enrichment_worker.paused:
- spotify_enrichment_worker.pause()
- s_was = True
- print(f"⏸️ Paused Spotify enrichment worker during {label}")
- except Exception:
- pass
- try:
- if itunes_enrichment_worker and not itunes_enrichment_worker.paused:
- itunes_enrichment_worker.pause()
- i_was = True
- print(f"⏸️ Paused iTunes enrichment worker during {label}")
- except Exception:
- pass
- return s_was, i_was
+ Returns dict of {name: was_running} for resume."""
+ was_running = {}
+ workers = {
+ 'Spotify': spotify_enrichment_worker,
+ 'iTunes': itunes_enrichment_worker,
+ 'Tidal': tidal_enrichment_worker if 'tidal_enrichment_worker' in globals() else None,
+ 'Qobuz': qobuz_enrichment_worker if 'qobuz_enrichment_worker' in globals() else None,
+ }
+ for name, worker in workers.items():
+ try:
+ if worker and not worker.paused:
+ worker.pause()
+ was_running[name] = True
+ print(f"⏸️ Paused {name} enrichment worker during {label}")
+ except Exception:
+ pass
+ return was_running
def _resume_enrichment_workers(was_running, label='discovery'):
"""Resume enrichment workers that were paused by _pause_enrichment_workers."""
- s_was, i_was = was_running
- try:
- if s_was and spotify_enrichment_worker:
- spotify_enrichment_worker.resume()
- print(f"▶️ Resumed Spotify enrichment worker after {label}")
- except Exception:
- pass
- try:
- if i_was and itunes_enrichment_worker:
- itunes_enrichment_worker.resume()
- print(f"▶️ Resumed iTunes enrichment worker after {label}")
- except Exception:
- pass
+ workers = {
+ 'Spotify': spotify_enrichment_worker,
+ 'iTunes': itunes_enrichment_worker,
+ 'Tidal': tidal_enrichment_worker if 'tidal_enrichment_worker' in globals() else None,
+ 'Qobuz': qobuz_enrichment_worker if 'qobuz_enrichment_worker' in globals() else None,
+ }
+ for name, worker in workers.items():
+ try:
+ if was_running.get(name) and worker:
+ worker.resume()
+ print(f"▶️ Resumed {name} enrichment worker after {label}")
+ except Exception:
+ pass
def _run_playlist_discovery_worker(playlists, automation_id=None):
"""Background worker that discovers Spotify/iTunes metadata for undiscovered
mirrored playlist tracks. Stores results in extra_data for use by sync."""
- _ew_state = (False, False)
+ _ew_state = {}
try:
_ew_state = _pause_enrichment_workers('mirrored playlist discovery')
use_spotify = spotify_client and spotify_client.is_spotify_authenticated()
@@ -23054,7 +23139,7 @@ def _discovery_score_candidates(source_title, source_artist, source_duration_ms,
def _run_tidal_discovery_worker(playlist_id):
"""Background worker for Tidal discovery process (Spotify preferred, iTunes fallback)"""
- _ew_state = (False, False)
+ _ew_state = {}
try:
_ew_state = _pause_enrichment_workers('Tidal discovery')
state = tidal_discovery_states[playlist_id]
@@ -23783,7 +23868,7 @@ def update_youtube_discovery_match():
def _run_youtube_discovery_worker(url_hash):
"""Background worker for YouTube music discovery process (Spotify preferred, iTunes fallback)"""
- _ew_state = (False, False)
+ _ew_state = {}
try:
_ew_state = _pause_enrichment_workers('YouTube discovery')
state = youtube_playlist_states[url_hash]
@@ -24094,7 +24179,7 @@ def _run_youtube_discovery_worker(url_hash):
def _run_listenbrainz_discovery_worker(playlist_mbid):
"""Background worker for ListenBrainz music discovery process (Spotify preferred, iTunes fallback)"""
- _ew_state = (False, False)
+ _ew_state = {}
try:
_ew_state = _pause_enrichment_workers('ListenBrainz discovery')
state = listenbrainz_playlist_states[playlist_mbid]
@@ -26312,7 +26397,7 @@ def start_watchlist_scan():
# Start the scan in a background thread
scan_profile_id = get_current_profile_id()
def run_scan():
- _ew_state = (False, False)
+ _ew_state = {}
try:
global watchlist_scan_state, watchlist_auto_scanning, watchlist_auto_scanning_timestamp
from core.watchlist_scanner import WatchlistScanner
@@ -27143,7 +27228,7 @@ def _process_watchlist_scan_automatically(automation_id=None):
print("🤖 [Auto-Watchlist] Timer triggered - starting automatic watchlist scan...")
- _ew_state = (False, False)
+ _ew_state = {}
try:
# CRITICAL FIX: Use smart stuck detection BEFORE acquiring lock
@@ -31271,7 +31356,7 @@ def clean_beatport_text(text):
def _run_beatport_discovery_worker(url_hash):
"""Background worker for Beatport discovery process (Spotify preferred, iTunes fallback)"""
- _ew_state = (False, False)
+ _ew_state = {}
try:
_ew_state = _pause_enrichment_workers('Beatport discovery')
state = beatport_chart_states[url_hash]
@@ -33142,7 +33227,9 @@ def start_oauth_callback_servers():
# Reinitialize the global tidal client with new tokens
global tidal_client
tidal_client = TidalClient()
-
+ if tidal_enrichment_worker:
+ tidal_enrichment_worker.client = tidal_client
+
add_activity_item("✅", "Tidal Auth Complete", "Successfully authenticated with Tidal", "Now")
self.send_response(200)
self.send_header('Content-type', 'text/html')
@@ -33871,6 +33958,144 @@ def genius_enrichment_resume():
# END GENIUS ENRICHMENT INTEGRATION
# ================================================================================================
+# ================================================================================================
+# TIDAL ENRICHMENT WORKER
+# ================================================================================================
+
+tidal_enrichment_worker = None
+try:
+ from database.music_database import MusicDatabase
+ tidal_enrich_db = MusicDatabase()
+ tidal_enrichment_worker = TidalWorker(database=tidal_enrich_db, client=tidal_client)
+ tidal_enrichment_worker.start()
+ print("✅ Tidal enrichment worker initialized and started")
+except Exception as e:
+ print(f"⚠️ Tidal worker initialization failed: {e}")
+ tidal_enrichment_worker = None
+
+# --- Tidal Enrichment API Endpoints ---
+
+@app.route('/api/tidal-enrichment/status', methods=['GET'])
+def tidal_enrichment_status():
+ """Get Tidal enrichment status for UI polling"""
+ try:
+ if tidal_enrichment_worker is None:
+ return jsonify({
+ 'enabled': False,
+ 'running': False,
+ 'paused': False,
+ 'authenticated': False,
+ 'current_item': None,
+ 'stats': {'matched': 0, 'not_found': 0, 'pending': 0, 'errors': 0},
+ 'progress': {}
+ }), 200
+
+ status = tidal_enrichment_worker.get_stats()
+ return jsonify(status), 200
+ except Exception as e:
+ logger.error(f"Error getting Tidal enrichment status: {e}")
+ return jsonify({'error': str(e)}), 500
+
+@app.route('/api/tidal-enrichment/pause', methods=['POST'])
+def tidal_enrichment_pause():
+ """Pause Tidal enrichment worker"""
+ try:
+ if tidal_enrichment_worker is None:
+ return jsonify({'error': 'Tidal worker not initialized'}), 400
+
+ tidal_enrichment_worker.pause()
+ logger.info("Tidal worker paused via UI")
+ return jsonify({'status': 'paused'}), 200
+ except Exception as e:
+ logger.error(f"Error pausing Tidal worker: {e}")
+ return jsonify({'error': str(e)}), 500
+
+@app.route('/api/tidal-enrichment/resume', methods=['POST'])
+def tidal_enrichment_resume():
+ """Resume Tidal enrichment worker"""
+ try:
+ if tidal_enrichment_worker is None:
+ return jsonify({'error': 'Tidal worker not initialized'}), 400
+
+ tidal_enrichment_worker.resume()
+ logger.info("Tidal worker resumed via UI")
+ return jsonify({'status': 'running'}), 200
+ except Exception as e:
+ logger.error(f"Error resuming Tidal worker: {e}")
+ return jsonify({'error': str(e)}), 500
+
+# ================================================================================================
+# QOBUZ ENRICHMENT WORKER
+# ================================================================================================
+
+qobuz_enrichment_worker = None
+try:
+ from database.music_database import MusicDatabase
+ from core.qobuz_client import QobuzClient
+ qobuz_enrich_db = MusicDatabase()
+ qobuz_enrich_client = QobuzClient() # Separate client instance for thread safety
+ qobuz_enrichment_worker = QobuzWorker(database=qobuz_enrich_db, client=qobuz_enrich_client)
+ qobuz_enrichment_worker.start()
+ print("✅ Qobuz enrichment worker initialized and started")
+except Exception as e:
+ print(f"⚠️ Qobuz worker initialization failed: {e}")
+ qobuz_enrichment_worker = None
+
+# --- Qobuz Enrichment API Endpoints ---
+
+@app.route('/api/qobuz-enrichment/status', methods=['GET'])
+def qobuz_enrichment_status():
+ """Get Qobuz enrichment status for UI polling"""
+ try:
+ if qobuz_enrichment_worker is None:
+ return jsonify({
+ 'enabled': False,
+ 'running': False,
+ 'paused': False,
+ 'authenticated': False,
+ 'current_item': None,
+ 'stats': {'matched': 0, 'not_found': 0, 'pending': 0, 'errors': 0},
+ 'progress': {}
+ }), 200
+
+ status = qobuz_enrichment_worker.get_stats()
+ return jsonify(status), 200
+ except Exception as e:
+ logger.error(f"Error getting Qobuz enrichment status: {e}")
+ return jsonify({'error': str(e)}), 500
+
+@app.route('/api/qobuz-enrichment/pause', methods=['POST'])
+def qobuz_enrichment_pause():
+ """Pause Qobuz enrichment worker"""
+ try:
+ if qobuz_enrichment_worker is None:
+ return jsonify({'error': 'Qobuz worker not initialized'}), 400
+
+ qobuz_enrichment_worker.pause()
+ logger.info("Qobuz worker paused via UI")
+ return jsonify({'status': 'paused'}), 200
+ except Exception as e:
+ logger.error(f"Error pausing Qobuz worker: {e}")
+ return jsonify({'error': str(e)}), 500
+
+@app.route('/api/qobuz-enrichment/resume', methods=['POST'])
+def qobuz_enrichment_resume():
+ """Resume Qobuz enrichment worker"""
+ try:
+ if qobuz_enrichment_worker is None:
+ return jsonify({'error': 'Qobuz worker not initialized'}), 400
+
+ qobuz_enrichment_worker.resume()
+ logger.info("Qobuz worker resumed via UI")
+ return jsonify({'status': 'running'}), 200
+ except Exception as e:
+ logger.error(f"Error resuming Qobuz worker: {e}")
+ return jsonify({'error': str(e)}), 500
+
+# ================================================================================================
+# END TIDAL/QOBUZ ENRICHMENT INTEGRATION
+# ================================================================================================
+
# ================================================================================================
# HYDRABASE P2P MIRROR WORKER
@@ -35114,6 +35339,8 @@ def _emit_enrichment_status_loop():
'itunes-enrichment': lambda: itunes_enrichment_worker,
'lastfm-enrichment': lambda: lastfm_worker,
'genius-enrichment': lambda: genius_worker,
+ 'tidal-enrichment': lambda: tidal_enrichment_worker,
+ 'qobuz-enrichment': lambda: qobuz_enrichment_worker,
'hydrabase': lambda: hydrabase_worker,
'repair': lambda: repair_worker,
}
diff --git a/webui/index.html b/webui/index.html
index e269655d..fd5baa32 100644
--- a/webui/index.html
+++ b/webui/index.html
@@ -439,6 +439,50 @@
+
+
+
+