danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2

Merge pull request #301 from kettui/fix/respect-metadata-provider-in-more-jobs

Browse source 
Respect the primary metadata provider in more jobs
This commit is contained in:
BoulderBadgeDad 2026-04-15 13:04:05 -07:00 committed by GitHub
commit 19a5256feb
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 1342 additions and 322 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

71
core/repair_jobs/library_reorganize.py
View file

@ -16,8 +16,8 @@ import os
import re
import shutil
import sys
import time
from core.metadata_service import get_client_for_source, get_primary_source, get_source_priority
from core.repair_jobs import register_job
from core.repair_jobs.base import JobContext, JobResult, RepairJob
from utils.logging_config import get_logger
@ -758,10 +758,10 @@ class LibraryReorganizeJob(RepairJob):
return years
def _lookup_years_from_api(self, context, missing_pairs) -> dict:
"""Batch-lookup release years from Spotify/iTunes/Deezer for albums not found in DB.
"""Batch-lookup release years from the configured metadata providers for albums not found in DB.
Args:
context: JobContext with spotify_client, config_manager
context: JobContext with config_manager
missing_pairs: set of (artist, album) tuples needing year lookup
Returns:
@ -771,37 +771,17 @@ class LibraryReorganizeJob(RepairJob):
if not missing_pairs:
return years
# Determine which client to use
search_client = None
source_name = 'unknown'
if context.spotify_client and hasattr(context.spotify_client, 'search_albums'):
try:
if context.spotify_client.is_spotify_authenticated():
search_client = context.spotify_client
source_name = 'Spotify'
except Exception:
pass
if not search_client:
# Try fallback (iTunes/Deezer)
try:
from core.metadata_service import get_primary_client
search_client = get_primary_client()
source_name = 'fallback'
except Exception:
pass
if not search_client or not hasattr(search_client, 'search_albums'):
return years
primary_source = get_primary_source()
source_priority = get_source_priority(primary_source)
# Cap lookups to avoid excessive API calls
max_lookups = 200
pairs_list = list(missing_pairs)[:max_lookups]
logger.info("Looking up %d album years from %s API", len(pairs_list), source_name)
logger.info("Looking up %d album years from configured metadata providers", len(pairs_list))
if context.report_progress:
context.report_progress(
phase=f'Looking up {len(pairs_list)} album years from {source_name}...',
phase=f'Looking up {len(pairs_list)} album years from metadata providers...',
log_line=f'Fetching release years for {len(pairs_list)} albums',
log_type='info'
)
@ -809,22 +789,27 @@ class LibraryReorganizeJob(RepairJob):
for artist, album in pairs_list:
if context.check_stop():
break
try:
results = search_client.search_albums(f"{artist} {album}", limit=3)
if results:
for r in results:
release_date = getattr(r, 'release_date', '') or ''
if release_date and len(release_date) >= 4:
year_str = release_date[:4]
if year_str.isdigit():
key = (artist.lower(), album.lower())
years[key] = year_str
break
import time
if context.sleep_or_stop(0.1): # Rate limit courtesy
break
except Exception as e:
logger.debug("API year lookup failed for %s - %s: %s", artist, album, e)
key = (artist.lower(), album.lower())
for source_name in source_priority:
try:
search_client = get_client_for_source(source_name)
if not search_client or not hasattr(search_client, 'search_albums'):
continue
results = search_client.search_albums(f"{artist} {album}", limit=3)
if results:
for r in results:
release_date = getattr(r, 'release_date', '') or ''
if release_date and len(release_date) >= 4:
year_str = release_date[:4]
if year_str.isdigit():
years[key] = year_str
break
if key in years:
break
if context.sleep_or_stop(0.1): # Rate limit courtesy
return years
except Exception as e:
logger.debug("API year lookup failed for %s - %s via %s: %s", artist, album, source_name, e)
logger.info("API year lookup: found %d/%d years", len(years), len(pairs_list))
return years

132
core/repair_jobs/metadata_gap_filler.py
View file

@ -1,8 +1,6 @@
"""Metadata Gap Filler Job — finds tracks missing key metadata and locates it from APIs."""
import time
from core.metadata_service import get_primary_source
from core.metadata_service import get_client_for_source, get_primary_source, get_source_priority
from core.repair_jobs import register_job
from core.repair_jobs.base import JobContext, JobResult, RepairJob
from utils.logging_config import get_logger
@ -40,6 +38,7 @@ class MetadataGapFillerJob(RepairJob):
settings = self._get_settings(context)
fill_isrc = settings.get('fill_isrc', True)
fill_mb_id = settings.get('fill_musicbrainz_id', True)
source_priority = get_source_priority(get_primary_source())
# Build WHERE clauses for missing fields (only columns that exist on tracks)
conditions = []
@ -53,27 +52,47 @@ class MetadataGapFillerJob(RepairJob):
where = " OR ".join(conditions)
# Fetch tracks with gaps, prioritizing those with spotify_track_id
# Fetch tracks with gaps, prioritizing those with source track IDs.
tracks = []
conn = None
try:
conn = context.db._get_connection()
cursor = conn.cursor()
cursor.execute("PRAGMA table_info(tracks)")
track_columns = {column[1] for column in cursor.fetchall()}
select_cols = [
"t.id",
"t.title",
"ar.name",
"al.title",
"t.isrc",
"t.musicbrainz_recording_id",
"al.thumb_url",
"ar.thumb_url",
]
column_map = [
("spotify_track_id", "t.spotify_track_id"),
("itunes_track_id", "t.itunes_track_id"),
("deezer_track_id", "t.deezer_track_id"),
]
column_index = {}
for alias, column in column_map:
if column.split('.', 1)[1] in track_columns:
column_index[alias] = len(select_cols)
select_cols.append(f"{column} AS {alias}")
cursor.execute(f"""
SELECT t.id, t.title, ar.name, al.title, t.spotify_track_id,
t.isrc, t.musicbrainz_recording_id,
al.thumb_url, ar.thumb_url
SELECT {', '.join(select_cols)}
FROM tracks t
LEFT JOIN artists ar ON ar.id = t.artist_id
LEFT JOIN albums al ON al.id = t.album_id
WHERE t.title IS NOT NULL AND t.title != ''
AND ({where})
ORDER BY
CASE WHEN t.spotify_track_id IS NOT NULL AND t.spotify_track_id != '' THEN 0 ELSE 1 END,
t.id
LIMIT 500
""")
tracks = cursor.fetchall()
tracks = sorted(tracks, key=lambda row: _track_row_priority(row, column_index, source_priority))
except Exception as e:
logger.error("Error fetching tracks with metadata gaps: %s", e, exc_info=True)
result.errors += 1
@ -97,7 +116,12 @@ class MetadataGapFillerJob(RepairJob):
if i % 20 == 0 and context.wait_if_paused():
return result
track_id, title, artist_name, album_title, spotify_track_id, isrc, mb_id, album_thumb, artist_thumb = row
track_id, title, artist_name, album_title, isrc, mb_id, album_thumb, artist_thumb = row[:8]
source_track_ids = {
'spotify': row[column_index['spotify_track_id']] if 'spotify_track_id' in column_index else None,
'itunes': row[column_index['itunes_track_id']] if 'itunes_track_id' in column_index else None,
'deezer': row[column_index['deezer_track_id']] if 'deezer_track_id' in column_index else None,
}
result.scanned += 1
if context.report_progress:
@ -108,20 +132,29 @@ class MetadataGapFillerJob(RepairJob):
log_type='info'
)
found_fields = {}
resolved_source = None
resolved_track_id = None
# Try Spotify enrichment for ISRC — only when Spotify is the configured primary source.
# If Deezer/iTunes is primary, Spotify may still be authenticated for playlist sync
# but should not be called here, as it burns API quota unnecessarily.
if spotify_track_id and context.spotify_client and not context.is_spotify_rate_limited() and get_primary_source() == 'spotify':
try:
track_data = context.spotify_client.get_track_details(spotify_track_id)
if track_data:
if fill_isrc and not isrc:
ext_ids = track_data.get('external_ids', {})
if ext_ids.get('isrc'):
found_fields['isrc'] = ext_ids['isrc']
except Exception as e:
logger.debug("Spotify enrichment failed for track %s: %s", track_id, e)
# Try source-aware track detail lookups only when ISRC enrichment is enabled.
if fill_isrc and not isrc:
for source in source_priority:
track_source_id = source_track_ids.get(source)
if not track_source_id:
continue
try:
client = get_client_for_source(source)
if not client or not hasattr(client, 'get_track_details'):
continue
track_data = client.get_track_details(track_source_id)
if track_data:
isrc_value = _extract_isrc(track_data)
if isrc_value:
found_fields['isrc'] = isrc_value
resolved_source = source
resolved_track_id = track_source_id
break
except Exception as e:
logger.debug("%s enrichment failed for track %s: %s", source.capitalize(), track_id, e)
# Try MusicBrainz for MB recording ID
if fill_mb_id and not mb_id and context.mb_client:
@ -161,7 +194,9 @@ class MetadataGapFillerJob(RepairJob):
'title': title,
'artist': artist_name,
'album': album_title,
'spotify_track_id': spotify_track_id,
'track_ids': source_track_ids,
'resolved_source': resolved_source,
'resolved_track_id': resolved_track_id,
'found_fields': found_fields,
'album_thumb_url': album_thumb or None,
'artist_thumb_url': artist_thumb or None,
@ -175,7 +210,7 @@ class MetadataGapFillerJob(RepairJob):
result.skipped += 1
# Rate limit API calls
if spotify_track_id:
if fill_isrc and any(source_track_ids.values()):
if context.sleep_or_stop(0.5):
return result
@ -215,3 +250,48 @@ class MetadataGapFillerJob(RepairJob):
finally:
if conn:
conn.close()
def _extract_isrc(track_data):
"""Extract ISRC from a track detail payload."""
if not track_data or not isinstance(track_data, dict):
return None
external_ids = track_data.get('external_ids')
if isinstance(external_ids, dict):
isrc = external_ids.get('isrc')
if isrc:
return isrc
isrc = track_data.get('isrc')
if isrc:
return isrc
raw_data = track_data.get('raw_data')
if isinstance(raw_data, dict):
external_ids = raw_data.get('external_ids')
if isinstance(external_ids, dict) and external_ids.get('isrc'):
return external_ids['isrc']
if raw_data.get('isrc'):
return raw_data['isrc']
return None
def _track_row_priority(row, column_index, source_priority):
"""Sort rows by the first source track ID available in priority order."""
source_columns = {
'spotify': 'spotify_track_id',
'itunes': 'itunes_track_id',
'deezer': 'deezer_track_id',
}
for idx, source in enumerate(source_priority):
column_name = source_columns.get(source)
if not column_name:
continue
column_pos = column_index.get(column_name)
if column_pos is not None and row[column_pos]:
return idx
return len(source_priority)

164
core/repair_jobs/missing_cover_art.py
View file

@ -1,5 +1,6 @@
"""Missing Cover Art Filler Job — finds albums without artwork and locates art from APIs."""
from core.metadata_service import get_client_for_source, get_primary_source, get_source_priority
from core.repair_jobs import register_job
from core.repair_jobs.base import JobContext, JobResult, RepairJob
from utils.logging_config import get_logger
@ -11,29 +12,36 @@ logger = get_logger("repair_job.cover_art")
class MissingCoverArtJob(RepairJob):
job_id = 'missing_cover_art'
display_name = 'Cover Art Filler'
description = 'Finds albums missing artwork and locates art from Spotify/iTunes'
description = 'Finds albums missing artwork and locates art from metadata sources'
help_text = (
'Scans your library for albums that have no cover art stored in the database. '
'For each missing cover, it searches Spotify and iTunes APIs using the album name '
'and artist to find matching artwork.\n\n'
'For each missing cover, it searches the configured metadata sources using the '
'album name and artist to find matching artwork. If Prefer Source is set, that '
'source is tried first; otherwise the primary metadata source is used.\n\n'
'When artwork is found, a finding is created with the image URL so you can review '
'and apply it. The job does not download or embed artwork automatically.\n\n'
'Settings:\n'
'- Prefer Source: Which API to try first for artwork (spotify or itunes)'
'- Prefer Source: Optional source to try first; otherwise the primary metadata source is used'
)
icon = 'repair-icon-coverart'
default_enabled = True
default_interval_hours = 48
default_settings = {
'prefer_source': 'spotify',
}
default_settings = {}
auto_fix = False
def scan(self, context: JobContext) -> JobResult:
result = JobResult()
settings = self._get_settings(context)
prefer_source = settings.get('prefer_source', 'spotify')
primary_source = get_primary_source()
source_priority = get_source_priority(primary_source)
prefer_source = settings.get('prefer_source')
if prefer_source and prefer_source != primary_source and prefer_source in source_priority:
source_priority.remove(prefer_source)
source_priority.insert(0, prefer_source)
if primary_source in source_priority:
source_priority.remove(primary_source)
source_priority.insert(1, primary_source)
# Fetch albums with missing artwork
albums = []
@ -41,9 +49,31 @@ class MissingCoverArtJob(RepairJob):
try:
conn = context.db._get_connection()
cursor = conn.cursor()
cursor.execute("""
SELECT al.id, al.title, ar.name, al.spotify_album_id, al.thumb_url,
ar.thumb_url
cursor.execute("PRAGMA table_info(albums)")
album_columns = {column[1] for column in cursor.fetchall()}
select_cols = [
"al.id",
"al.title",
"ar.name",
"al.spotify_album_id",
"al.thumb_url",
"ar.thumb_url",
]
column_map = [
("itunes_album_id", "al.itunes_album_id"),
("deezer_album_id", "al.deezer_id"),
("discogs_album_id", "al.discogs_id"),
("hydrabase_album_id", "al.soul_id"),
]
column_index = {}
for alias, column in column_map:
if column.split('.', 1)[1] in album_columns:
column_index[alias] = len(select_cols)
select_cols.append(f"{column} AS {alias}")
cursor.execute(f"""
SELECT {', '.join(select_cols)}
FROM albums al
LEFT JOIN artists ar ON ar.id = al.artist_id
WHERE (al.thumb_url IS NULL OR al.thumb_url = '')
@ -67,15 +97,20 @@ class MissingCoverArtJob(RepairJob):
if context.report_progress:
context.report_progress(phase=f'Searching artwork for {total} albums...', total=total)
spotify_skipped = False # Track if we've logged the rate limit skip
for i, row in enumerate(albums):
if context.check_stop():
return result
if i % 10 == 0 and context.wait_if_paused():
return result
album_id, title, artist_name, spotify_album_id, _, artist_thumb = row
album_id, title, artist_name, spotify_album_id, _, artist_thumb = row[:6]
source_album_ids = {
'spotify': spotify_album_id,
'itunes': row[column_index['itunes_album_id']] if 'itunes_album_id' in column_index else None,
'deezer': row[column_index['deezer_album_id']] if 'deezer_album_id' in column_index else None,
'discogs': row[column_index['discogs_album_id']] if 'discogs_album_id' in column_index else None,
'hydrabase': row[column_index['hydrabase_album_id']] if 'hydrabase_album_id' in column_index else None,
}
result.scanned += 1
if context.report_progress:
@ -88,24 +123,11 @@ class MissingCoverArtJob(RepairJob):
artwork_url = None
# Check Spotify rate limit at the top level — skip Spotify entirely if banned
spotify_available = not context.is_spotify_rate_limited()
if not spotify_available and not spotify_skipped:
logger.info("Spotify rate limited — skipping Spotify artwork lookups, using fallback only")
spotify_skipped = True
# Try to find artwork URL from APIs
if prefer_source == 'spotify' and spotify_available:
artwork_url = self._try_spotify(spotify_album_id, title, artist_name, context)
if not artwork_url:
artwork_url = self._try_itunes(title, artist_name, context)
elif prefer_source == 'spotify' and not spotify_available:
# Spotify preferred but rate limited — use fallback only
artwork_url = self._try_itunes(title, artist_name, context)
else:
artwork_url = self._try_itunes(title, artist_name, context)
if not artwork_url and spotify_available:
artwork_url = self._try_spotify(spotify_album_id, title, artist_name, context)
# Try source-specific IDs first, then title/artist search, in priority order.
for source in source_priority:
artwork_url = self._try_source(source, source_album_ids.get(source), title, artist_name)
if artwork_url:
break
if artwork_url:
if context.report_progress:
@ -151,47 +173,63 @@ class MissingCoverArtJob(RepairJob):
result.scanned, result.findings_created, result.skipped)
return result
def _try_spotify(self, spotify_album_id, title, artist_name, context):
"""Try to get album art from Spotify."""
client = context.spotify_client
def _try_source(self, source, source_album_id, title, artist_name):
"""Try to get album art from a specific metadata source."""
client = get_client_for_source(source)
if not client:
return None
query = f"{artist_name} {title}" if artist_name else title
try:
if context.is_spotify_rate_limited():
return None
if source_album_id:
album_data = self._get_album_for_source(source, client, source_album_id)
artwork_url = self._extract_artwork_url(album_data)
if artwork_url:
return artwork_url
# If we have a Spotify album ID, fetch directly
if spotify_album_id and client.is_spotify_authenticated():
album_data = client.get_album(spotify_album_id)
if album_data:
images = album_data.get('images', [])
if images:
return images[0].get('url')
# Search by name
if title and client.is_spotify_authenticated():
query = f"{artist_name} {title}" if artist_name else title
if query and hasattr(client, 'search_albums'):
results = client.search_albums(query, limit=1)
if results and hasattr(results[0], 'image_url') and results[0].image_url:
return results[0].image_url
if results:
artwork_url = self._extract_artwork_url(results[0])
if artwork_url:
return artwork_url
candidate_id = self._extract_album_id(results[0])
if candidate_id:
album_data = self._get_album_for_source(source, client, candidate_id)
return self._extract_artwork_url(album_data)
except Exception as e:
logger.debug("Spotify art lookup failed for '%s': %s", title, e)
logger.debug("%s art lookup failed for '%s': %s", source.capitalize(), title, e)
return None
def _try_itunes(self, title, artist_name, context):
"""Try to get album art from iTunes."""
client = context.itunes_client
if not client:
return None
@staticmethod
def _get_album_for_source(source, client, album_id):
if source == 'spotify':
return client.get_album(album_id)
return client.get_album(album_id, include_tracks=False)
try:
query = f"{artist_name} {title}" if artist_name else title
results = client.search_albums(query, limit=1)
if results and hasattr(results[0], 'image_url') and results[0].image_url:
return results[0].image_url
except Exception as e:
logger.debug("iTunes art lookup failed for '%s': %s", title, e)
@staticmethod
def _extract_album_id(item):
if hasattr(item, 'id'):
return getattr(item, 'id', None)
if isinstance(item, dict):
return item.get('id')
return None
@staticmethod
def _extract_artwork_url(item):
if not item:
return None
if hasattr(item, 'image_url') and getattr(item, 'image_url', None):
return item.image_url
if isinstance(item, dict):
if item.get('image_url'):
return item['image_url']
images = item.get('images') or []
if images and isinstance(images, list):
first = images[0]
if isinstance(first, dict):
return first.get('url')
return None
def _get_settings(self, context: JobContext) -> dict:

232
core/repair_jobs/track_number_repair.py
View file

@ -1,8 +1,9 @@
"""Track Number Repair Job — fixes embedded track number tags and filename prefixes.
Detects albums where 3+ files share the same track number (the "all tracks = 01"
bug pattern), then uses cascading API lookups (Spotify → iTunes → MusicBrainz →
AudioDB) to resolve the correct tracklist and repair each file.
bug pattern), then uses cascading API lookups in metadata-source priority order
before falling back to MusicBrainz and AudioDB to resolve the correct tracklist
and repair each file.
"""
import os
@ -10,6 +11,12 @@ import re
from difflib import SequenceMatcher
from typing import Any, Dict, List, Optional, Tuple
from core.metadata_service import (
get_album_tracks_for_source,
get_client_for_source,
get_primary_source,
get_source_priority,
)
from core.repair_jobs import register_job
from core.repair_jobs.base import JobContext, JobResult, RepairJob
from utils.logging_config import get_logger
@ -24,6 +31,14 @@ _PLACEHOLDER_IDS = {
'unknown', 'none', 'null', '',
}
_SOURCE_ALBUM_ID_COLUMNS = (
('spotify', 'spotify_album_id'),
('itunes', 'itunes_album_id'),
('deezer', 'deezer_id'),
('discogs', 'discogs_id'),
('hydrabase', 'soul_id'),
)
@register_job
class TrackNumberRepairJob(RepairJob):
@ -32,8 +47,9 @@ class TrackNumberRepairJob(RepairJob):
description = 'Detects mismatched track numbers using API lookups (dry run by default)'
help_text = (
'Scans album folders and compares each file\'s track number against the correct '
'tracklist from Spotify or iTunes. If a file\'s embedded track number doesn\'t match '
'the API data, the job creates a finding showing what needs to change.\n\n'
'tracklist from the configured metadata sources. If a file\'s embedded track '
'number doesn\'t match the API data, the job creates a finding showing what '
'needs to change.\n\n'
'In dry run mode (default), no files are modified — you review each proposed change '
'in the Findings tab and decide what to approve. Disable dry run in settings to let '
'the job automatically rename and re-number files.\n\n'
@ -187,7 +203,7 @@ class TrackNumberRepairJob(RepairJob):
logger.info("Anomaly detected in %s — %d files share track number(s): %s",
os.path.basename(folder_path), sum(duped.values()), duped)
# Resolve album tracklist via cascading fallbacks
# Resolve album tracklist via source-aware cascading fallbacks
api_tracks = self._resolve_album_tracklist(file_track_data, folder_path, context, scan_state)
if not api_tracks:
result.skipped += len(filenames)
@ -253,38 +269,24 @@ class TrackNumberRepairJob(RepairJob):
cache = scan_state['album_tracks_cache']
folder_name = os.path.basename(folder_path)
primary_source = get_primary_source()
source_priority = get_source_priority(primary_source)
# Fallback 0: Check DB first — if these files are tracked and their album
# has a spotify_album_id, use that directly without reading file tags
db_album_id, db_spotify_album_id, db_itunes_album_id = _lookup_album_ids_from_db(
file_track_data, context
)
if db_spotify_album_id and _is_valid_album_id(db_spotify_album_id):
tracks = _get_album_tracklist(db_spotify_album_id, context, cache)
if tracks:
logger.info("[Repair] %s — resolved via DB spotify_album_id: %s", folder_name, db_spotify_album_id)
return tracks
if db_itunes_album_id and _is_valid_album_id(db_itunes_album_id):
tracks = _get_album_tracklist(db_itunes_album_id, context, cache)
if tracks:
logger.info("[Repair] %s — resolved via DB itunes_album_id: %s", folder_name, db_itunes_album_id)
return tracks
# Fallback 0: Check DB first. If any tracked file already has source IDs,
# prefer the configured source order and use the first available album ID.
source_album_ids = _lookup_album_ids_from_db(file_track_data, context)
# Collect available IDs from file tags (fallback when DB has no IDs)
spotify_album_id = None
itunes_album_id = None
spotify_track_id = None
mb_album_id = None
album_name = None
artist_name = None
for fpath, fname, _ in file_track_data:
if not spotify_album_id or not itunes_album_id:
if 'spotify' not in source_album_ids or 'itunes' not in source_album_ids:
aid, source = _read_album_id_from_file(fpath)
if aid and source == 'spotify' and not spotify_album_id:
spotify_album_id = aid
elif aid and source == 'itunes' and not itunes_album_id:
itunes_album_id = aid
if aid and source in ('spotify', 'itunes') and source not in source_album_ids:
source_album_ids[source] = aid
if not spotify_track_id:
spotify_track_id = _read_spotify_track_id_from_file(fpath)
@ -295,31 +297,27 @@ class TrackNumberRepairJob(RepairJob):
if not album_name:
album_name, artist_name = _read_album_artist_from_file(fpath)
if spotify_album_id and itunes_album_id and spotify_track_id and mb_album_id and album_name:
if source_album_ids and spotify_track_id and mb_album_id and album_name:
break
# Fallback 1: Spotify album ID from file tags
if spotify_album_id and _is_valid_album_id(spotify_album_id):
tracks = _get_album_tracklist(spotify_album_id, context, cache)
if tracks:
logger.info("[Repair] %s — resolved via Spotify album ID: %s", folder_name, spotify_album_id)
return tracks
# Fallback 1: Album IDs from DB / file tags, using source priority
for source in source_priority:
album_id = source_album_ids.get(source)
if album_id and _is_valid_album_id(album_id):
tracks = _get_album_tracklist(source, album_id, cache)
if tracks:
logger.info("[Repair] %s — resolved via %s album ID: %s",
folder_name, source, album_id)
return tracks
# Fallback 2: iTunes album ID
if itunes_album_id and _is_valid_album_id(itunes_album_id):
tracks = _get_album_tracklist(itunes_album_id, context, cache)
if tracks:
logger.info("[Repair] %s — resolved via iTunes album ID: %s", folder_name, itunes_album_id)
return tracks
# Fallback 3: Spotify track ID → discover album ID
client = context.spotify_client
if spotify_track_id and client and client.is_spotify_authenticated() and not context.is_spotify_rate_limited():
# Fallback 2: Spotify track ID → discover album ID
client = get_client_for_source('spotify')
if spotify_track_id and client:
try:
track_details = client.get_track_details(spotify_track_id)
if track_details and track_details.get('album', {}).get('id'):
real_album_id = track_details['album']['id']
tracks = _get_album_tracklist(real_album_id, context, cache)
tracks = _get_album_tracklist('spotify', real_album_id, cache)
if tracks:
logger.info("[Repair] %s — resolved via Spotify track ID %s → album %s",
folder_name, spotify_track_id, real_album_id)
@ -327,29 +325,35 @@ class TrackNumberRepairJob(RepairJob):
except Exception as e:
logger.debug("Spotify track lookup failed for %s: %s", spotify_track_id, e)
# Fallback 4: Search Spotify/iTunes by album name + artist
if album_name and client and not context.is_spotify_rate_limited():
try:
query = f"{artist_name} {album_name}" if artist_name else album_name
results = client.search_albums(query, limit=5)
if results:
best = results[0]
tracks = _get_album_tracklist(best.id, context, cache)
if tracks:
logger.info("[Repair] %s — resolved via album search: '%s' → %s",
folder_name, query, best.id)
return tracks
except Exception as e:
logger.debug("Album search failed for '%s': %s", album_name, e)
# Fallback 3: Search metadata sources by album name + artist
if album_name:
query = f"{artist_name} {album_name}" if artist_name else album_name
for source in source_priority:
client = get_client_for_source(source)
if not client or not hasattr(client, 'search_albums'):
continue
try:
results = client.search_albums(query, limit=5)
if results:
best = results[0]
best_album_id = getattr(best, 'id', None) if not isinstance(best, dict) else best.get('id')
if best_album_id:
tracks = _get_album_tracklist(source, str(best_album_id), cache)
if tracks:
logger.info("[Repair] %s — resolved via %s album search: '%s' → %s",
folder_name, source, query, best_album_id)
return tracks
except Exception as e:
logger.debug("%s album search failed for '%s': %s", source.capitalize(), album_name, e)
# Fallback 5: MusicBrainz album ID from tags
# Fallback 4: MusicBrainz album ID from tags
if mb_album_id:
tracks = _get_tracklist_from_musicbrainz(mb_album_id, context, cache)
if tracks:
logger.info("[Repair] %s — resolved via MusicBrainz album ID: %s", folder_name, mb_album_id)
return tracks
# Fallback 6: AudioDB → MusicBrainz
# Fallback 5: AudioDB → MusicBrainz
if album_name and artist_name:
adb_mb_id = _get_musicbrainz_id_via_audiodb(artist_name, album_name, context)
if adb_mb_id and adb_mb_id != mb_album_id:
@ -800,25 +804,36 @@ def _update_db_file_path(db, old_path: str, new_path: str):
def _lookup_album_ids_from_db(file_track_data: List[Tuple[str, str, Any]],
context: JobContext) -> Tuple[Optional[str], Optional[str], Optional[str]]:
context: JobContext) -> Dict[str, Optional[str]]:
"""Look up album IDs from the database using file paths.
Checks if any of the files in this folder are tracked in the DB, and if so,
returns the album's (album_id, spotify_album_id, itunes_album_id).
returns a mapping of metadata source -> album ID.
This avoids expensive file tag reads and API calls when the DB already knows.
"""
if not context.db:
return None, None, None
return {}
conn = None
try:
conn = context.db._get_connection()
cursor = conn.cursor()
cursor.execute("PRAGMA table_info(albums)")
album_columns = {row[1] for row in cursor.fetchall()}
selected_sources = [
(source, column)
for source, column in _SOURCE_ALBUM_ID_COLUMNS
if column in album_columns
]
if not selected_sources:
return {}
# Try each file path until we find one tracked in the DB
for fpath, _, _ in file_track_data:
cursor.execute("""
SELECT t.album_id, al.spotify_album_id, al.itunes_album_id
select_cols = ", ".join(f"al.{column}" for _source, column in selected_sources)
cursor.execute(f"""
SELECT {select_cols}
FROM tracks t
JOIN albums al ON al.id = t.album_id
WHERE t.file_path = ?
@ -826,7 +841,11 @@ def _lookup_album_ids_from_db(file_track_data: List[Tuple[str, str, Any]],
""", (fpath,))
row = cursor.fetchone()
if row:
return row[0], row[1], row[2]
return {
source: str(row[idx])
for idx, (source, _column) in enumerate(selected_sources)
if row[idx]
}
except Exception as e:
logger.debug("Error looking up album IDs from DB: %s", e)
@ -834,7 +853,7 @@ def _lookup_album_ids_from_db(file_track_data: List[Tuple[str, str, Any]],
if conn:
conn.close()
return None, None, None
return {}
def _lookup_album_artist_art(file_track_data: List[Tuple[str, str, Any]],
@ -1018,52 +1037,53 @@ def _repair_single_track(file_path: str, filename: str, api_tracks: List[Dict],
return True
def _get_album_tracklist(album_id: str, context: JobContext,
cache: dict) -> Optional[List[Dict]]:
"""Fetch an album tracklist from Spotify or iTunes, with per-scan caching.
def _normalize_album_track_items(data) -> List[Dict[str, Any]]:
"""Normalize album track payloads to a list of dicts."""
if isinstance(data, list):
return data
if not isinstance(data, dict):
return []
items = data.get('items')
if isinstance(items, list):
return items
tracks = data.get('tracks')
if isinstance(tracks, list):
return tracks
if isinstance(tracks, dict):
nested_items = tracks.get('items')
if isinstance(nested_items, list):
return nested_items
return []
def _get_album_tracklist(source: str, album_id: str, cache: dict) -> Optional[List[Dict]]:
"""Fetch an album tracklist from a specific source, with per-scan caching.
Returns a list of dicts with at least 'name' and 'track_number' keys,
or None if lookup fails.
"""
if album_id in cache:
return cache[album_id]
cache_key = f"{source}:{album_id}"
if cache_key in cache:
return cache[cache_key]
result = None
# Try Spotify first
client = context.spotify_client
if client and client.is_spotify_authenticated() and not context.is_spotify_rate_limited():
try:
data = client.get_album_tracks(album_id)
if data and 'items' in data and data['items']:
result = [
{
'name': item.get('name', ''),
'track_number': item.get('track_number'),
'disc_number': item.get('disc_number', 1),
}
for item in data['items']
]
except Exception as e:
logger.debug("Spotify get_album_tracks failed for %s: %s", album_id, e)
try:
data = get_album_tracks_for_source(source, album_id)
items = _normalize_album_track_items(data)
if items:
result = [
{
'name': item.get('name', '') if isinstance(item, dict) else getattr(item, 'name', ''),
'track_number': item.get('track_number') if isinstance(item, dict) else getattr(item, 'track_number', None),
'disc_number': item.get('disc_number', 1) if isinstance(item, dict) else getattr(item, 'disc_number', 1),
}
for item in items
]
except Exception as e:
logger.debug("%s get_album_tracks failed for %s: %s", source.capitalize(), album_id, e)
# Fallback to iTunes
if not result and context.itunes_client:
try:
data = context.itunes_client.get_album_tracks(album_id)
if data and 'items' in data and data['items']:
result = [
{
'name': item.get('name', ''),
'track_number': item.get('track_number'),
'disc_number': item.get('disc_number', 1),
}
for item in data['items']
]
except Exception as e:
logger.debug("iTunes get_album_tracks failed for %s: %s", album_id, e)
cache[album_id] = result
cache[cache_key] = result
return result

234
core/repair_jobs/unknown_artist_fixer.py
View file

@ -8,9 +8,8 @@ import os
import re
import shutil
import sys
import time
from core.metadata_service import get_client_for_source, get_primary_client, get_primary_source
from core.metadata_service import get_client_for_source, get_primary_source, get_source_priority
from core.repair_jobs import register_job
from core.repair_jobs.base import JobContext, JobResult, RepairJob
from utils.logging_config import get_logger
@ -18,6 +17,8 @@ from utils.logging_config import get_logger
logger = get_logger("repair_job.unknown_artist_fixer")
_UNKNOWN_NAMES = {'unknown artist', 'unknown', ''}
_TRACK_ID_SOURCES = {'spotify', 'deezer', 'itunes'}
_TITLE_SEARCH_SOURCES = {'spotify', 'deezer', 'itunes', 'hydrabase'}
# Sidecar extensions to move alongside audio files
_SIDECAR_EXTS = {'.lrc', '.jpg', '.jpeg', '.png', '.nfo', '.txt', '.cue'}
@ -249,6 +250,7 @@ class UnknownArtistFixerJob(RepairJob):
Returns dict with artist, album, track_number, year, etc. or None."""
title = track['title'] or ''
primary_source = get_primary_source()
# Priority 1: Read embedded file tags
try:
@ -269,96 +271,160 @@ class UnknownArtistFixerJob(RepairJob):
except Exception as e:
logger.debug(f"Failed to read tags from {resolved_path}: {e}")
# Priority 2: Look up by source track ID using the appropriate client.
# Try the primary source's ID first, then fall back to any available ID
# with its matching client so we never pass a Deezer/iTunes ID to Spotify
# (or vice-versa).
_primary = get_primary_source()
_id_candidates = []
for _src in [_primary] + [s for s in ('spotify', 'deezer', 'itunes') if s != _primary]:
_tid = track.get(f'{_src}_track_id')
if _tid:
_id_candidates.append((_src, _tid))
source_id = None
_lookup_client = None
for _src, _tid in _id_candidates:
_c = get_client_for_source(_src)
if _c:
source_id = _tid
_lookup_client = _c
break
if source_id and _lookup_client:
# Priority 2: Look up by source track ID
for source, source_id in self._iter_source_track_ids(track, primary_source):
client = get_client_for_source(source)
if not client or not hasattr(client, 'get_track_details'):
continue
try:
details = _lookup_client.get_track_details(str(source_id))
if details and details.get('primary_artist'):
artist = details['primary_artist']
if artist.lower() not in _UNKNOWN_NAMES:
album = details.get('album', {})
album_name = album.get('name', '') if isinstance(album, dict) else str(album)
return {
'artist': artist,
'album': album_name,
'title': details.get('name', title),
'track_number': details.get('track_number'),
'disc_number': details.get('disc_number', 1),
'year': (album.get('release_date', '') or '')[:4] if isinstance(album, dict) else '',
'image_url': album.get('images', [{}])[0].get('url', '') if isinstance(album, dict) and album.get('images') else '',
'source': 'track_id_lookup',
'confidence': 0.95,
}
details = client.get_track_details(str(source_id))
corrected = self._build_corrected_metadata(
details,
fallback_title=title,
source=f"{source}_track_id_lookup",
confidence=0.95,
)
if corrected:
return corrected
except Exception as e:
logger.debug(f"Track ID lookup failed for {source_id}: {e}")
logger.debug(f"Track ID lookup failed for {source} {source_id}: {e}")
# Priority 3: Search by title using the configured primary metadata source
_search_client = get_primary_client()
if title and _search_client:
try:
results = _search_client.search_tracks(title, limit=5)
if results:
# Score candidates
from difflib import SequenceMatcher
best = None
best_score = 0
for r in results:
name_sim = SequenceMatcher(None, title.lower(), r.name.lower()).ratio()
# Boost if album matches
album_name = r.album if hasattr(r, 'album') else ''
if album_name and track.get('album_title'):
album_sim = SequenceMatcher(None, track['album_title'].lower(), album_name.lower()).ratio()
name_sim = (name_sim * 0.7) + (album_sim * 0.3)
if name_sim > best_score:
best_score = name_sim
best = r
# Priority 3: Search by title
if title:
for source in self._iter_source_priority(primary_source, _TITLE_SEARCH_SOURCES):
client = get_client_for_source(source)
if not client or not hasattr(client, 'search_tracks'):
continue
try:
results = client.search_tracks(title, limit=5)
if not results:
continue
if best and best_score >= 0.7:
artist = best.artists[0] if best.artists else ''
if artist and artist.lower() not in _UNKNOWN_NAMES:
# Get full details for track_number
best, best_score = self._pick_best_track_candidate(title, track.get('album_title'), results)
if not best or best_score < 0.7:
continue
full_details = None
if hasattr(client, 'get_track_details') and getattr(best, 'id', None):
try:
full_details = client.get_track_details(str(best.id))
except Exception:
full_details = None
try:
full_details = _search_client.get_track_details(best.id)
except Exception:
pass
album_data = full_details.get('album', {}) if full_details else {}
return {
'artist': artist,
'album': best.album if hasattr(best, 'album') else '',
'title': best.name,
'track_number': full_details.get('track_number') if full_details else None,
'disc_number': full_details.get('disc_number', 1) if full_details else 1,
'year': (album_data.get('release_date', '') or '')[:4] if isinstance(album_data, dict) else '',
'image_url': getattr(best, 'image_url', '') or '',
'source': 'title_search',
'confidence': round(best_score, 3),
}
except Exception as e:
logger.debug(f"Title search failed for '{title}': {e}")
# Rate limit courtesy
if context.sleep_or_stop(0.2):
return None
corrected = self._build_corrected_metadata(
full_details or best,
fallback_title=title,
source=f"{source}_title_search",
confidence=round(best_score, 3),
)
if corrected:
return corrected
except Exception as e:
logger.debug(f"Title search failed for '{title}' via {source}: {e}")
# Rate limit courtesy
if context.sleep_or_stop(0.2):
return None
return None
@staticmethod
def _get_track_value(payload, key, default=None):
if isinstance(payload, dict):
return payload.get(key, default)
return getattr(payload, key, default)
def _iter_source_track_ids(self, track: dict, primary_source: str):
source_fields = {
'spotify': 'spotify_track_id',
'deezer': 'deezer_track_id',
'itunes': 'itunes_track_id',
}
ordered_sources = [source for source in self._iter_source_priority(primary_source, _TRACK_ID_SOURCES) if source in source_fields]
for source in ordered_sources:
source_id = track.get(source_fields[source])
if source_id:
yield source, source_id
@staticmethod
def _iter_source_priority(primary_source: str, allowed_sources: set[str]):
return [source for source in get_source_priority(primary_source) if source in allowed_sources]
def _pick_best_track_candidate(self, title: str, album_title: str, results):
from difflib import SequenceMatcher
best = None
best_score = 0.0
title_lower = title.lower()
album_lower = album_title.lower() if album_title else ''
for candidate in results:
candidate_name = self._get_track_value(candidate, 'name', '') or ''
if not candidate_name:
continue
name_sim = SequenceMatcher(None, title_lower, candidate_name.lower()).ratio()
candidate_album = self._get_track_value(candidate, 'album', '') or ''
if album_lower and candidate_album:
if isinstance(candidate_album, dict):
candidate_album = candidate_album.get('name') or candidate_album.get('title') or ''
album_sim = SequenceMatcher(None, album_lower, str(candidate_album).lower()).ratio()
name_sim = (name_sim * 0.7) + (album_sim * 0.3)
if name_sim > best_score:
best_score = name_sim
best = candidate
return best, best_score
def _build_corrected_metadata(self, payload, fallback_title: str, source: str, confidence: float):
if not payload:
return None
artist = self._get_track_value(payload, 'primary_artist', '') or ''
artists = self._get_track_value(payload, 'artists', []) or []
if not artist and artists:
if isinstance(artists, list):
first_artist = artists[0]
if isinstance(first_artist, dict):
artist = first_artist.get('name', '')
else:
artist = str(first_artist)
artist = (artist or '').strip()
if not artist or artist.lower() in _UNKNOWN_NAMES:
return None
album = self._get_track_value(payload, 'album', {}) or {}
if isinstance(album, dict):
album_name = album.get('name', '') or album.get('title', '') or ''
year = (album.get('release_date', '') or '')[:4]
image_url = ''
images = album.get('images') or []
if images:
first_image = images[0]
if isinstance(first_image, dict):
image_url = first_image.get('url', '') or ''
else:
album_name = str(album)
year = ''
image_url = ''
image_url = self._get_track_value(payload, 'image_url', image_url) or image_url
title = self._get_track_value(payload, 'name', fallback_title) or fallback_title
return {
'artist': artist,
'album': album_name,
'title': title,
'track_number': self._get_track_value(payload, 'track_number'),
'disc_number': self._get_track_value(payload, 'disc_number', 1) or 1,
'year': year,
'image_url': image_url,
'source': source,
'confidence': confidence,
}
def _apply_fix(self, context, track, corrected, resolved_path,
expected_rel, transfer, fix_tags, reorganize_files):
"""Apply the fix: re-tag file, move to correct path, update DB."""

129
tests/test_library_reorganize.py Normal file
View file

@ -0,0 +1,129 @@
import sys
import types
from types import SimpleNamespace
# Stub optional Spotify dependency so metadata_service can import in tests.
if 'spotipy' not in sys.modules:
spotipy = types.ModuleType('spotipy')
oauth2 = types.ModuleType('spotipy.oauth2')
class _DummySpotify:
pass
class _DummyOAuth:
pass
spotipy.Spotify = _DummySpotify
oauth2.SpotifyOAuth = _DummyOAuth
oauth2.SpotifyClientCredentials = _DummyOAuth
spotipy.oauth2 = oauth2
sys.modules['spotipy'] = spotipy
sys.modules['spotipy.oauth2'] = oauth2
if 'config.settings' not in sys.modules:
config_mod = types.ModuleType('config')
settings_mod = types.ModuleType('config.settings')
class _DummyConfigManager:
def get(self, key, default=None):
return default
settings_mod.config_manager = _DummyConfigManager()
config_mod.settings = settings_mod
sys.modules['config'] = config_mod
sys.modules['config.settings'] = settings_mod
from core.repair_jobs import library_reorganize as lr
class _FakeSearchClient:
def __init__(self, source_name, year=None):
self.source_name = source_name
self.year = year
self.calls = []
def search_albums(self, query, limit=3):
self.calls.append((query, limit))
if self.year is None:
return []
return [SimpleNamespace(release_date=f"{self.year}-01-01")]
def test_lookup_years_prefers_primary_source(monkeypatch):
deezer_client = _FakeSearchClient('deezer', '2022')
spotify_client = _FakeSearchClient('spotify', '1999')
monkeypatch.setattr(lr, 'get_primary_source', lambda: 'deezer')
monkeypatch.setattr(lr, 'get_source_priority', lambda primary: [primary, 'itunes', 'spotify'])
monkeypatch.setattr(
lr,
'get_client_for_source',
lambda source: {'deezer': deezer_client, 'spotify': spotify_client}.get(source),
)
job = lr.LibraryReorganizeJob()
result = job._lookup_years_from_api(SimpleNamespace(report_progress=None, check_stop=lambda: False, sleep_or_stop=lambda *_: False), {('Artist', 'Album')})
assert result == {('artist', 'album'): '2022'}
assert deezer_client.calls == [('Artist Album', 3)]
assert spotify_client.calls == []
def test_lookup_years_falls_through_to_later_source(monkeypatch):
deezer_client = _FakeSearchClient('deezer', None)
spotify_client = _FakeSearchClient('spotify', '1999')
monkeypatch.setattr(lr, 'get_primary_source', lambda: 'deezer')
monkeypatch.setattr(lr, 'get_source_priority', lambda primary: [primary, 'itunes', 'spotify'])
monkeypatch.setattr(
lr,
'get_client_for_source',
lambda source: {'deezer': deezer_client, 'spotify': spotify_client}.get(source),
)
job = lr.LibraryReorganizeJob()
result = job._lookup_years_from_api(
SimpleNamespace(report_progress=None, check_stop=lambda: False, sleep_or_stop=lambda *_: False),
{('Artist', 'Album')},
)
assert result == {('artist', 'album'): '1999'}
assert deezer_client.calls == [('Artist Album', 3)]
assert spotify_client.calls == [('Artist Album', 3)]
def test_lookup_years_rechecks_client_availability_per_album(monkeypatch):
availability = {'spotify': True}
class _SpotifyClient(_FakeSearchClient):
def search_albums(self, query, limit=3):
self.calls.append((query, limit))
availability['spotify'] = False
return []
spotify_client = _SpotifyClient('spotify', None)
itunes_client = _FakeSearchClient('itunes', '2002')
helper_calls = []
def fake_get_client_for_source(source):
helper_calls.append(source)
if source == 'spotify' and not availability['spotify']:
return None
return {'spotify': spotify_client, 'itunes': itunes_client}.get(source)
monkeypatch.setattr(lr, 'get_primary_source', lambda: 'spotify')
monkeypatch.setattr(lr, 'get_source_priority', lambda primary: [primary, 'itunes'])
monkeypatch.setattr(lr, 'get_client_for_source', fake_get_client_for_source)
job = lr.LibraryReorganizeJob()
result = job._lookup_years_from_api(
SimpleNamespace(report_progress=None, check_stop=lambda: False, sleep_or_stop=lambda *_: False),
{('Artist A', 'Album A'), ('Artist B', 'Album B')},
)
assert result == {('artist a', 'album a'): '2002', ('artist b', 'album b'): '2002'}
assert helper_calls.count('spotify') == 2
assert helper_calls.count('itunes') == 2
assert len(spotify_client.calls) == 1
assert spotify_client.calls[0] in [('Artist A Album A', 3), ('Artist B Album B', 3)]
assert set(itunes_client.calls) == {('Artist A Album A', 3), ('Artist B Album B', 3)}

179
tests/test_metadata_gap_filler.py Normal file
View file

@ -0,0 +1,179 @@
import sqlite3
import sys
import types
from types import SimpleNamespace
# Stub optional Spotify dependency so metadata_service can import in tests.
if 'spotipy' not in sys.modules:
spotipy = types.ModuleType('spotipy')
oauth2 = types.ModuleType('spotipy.oauth2')
class _DummySpotify:
pass
class _DummyOAuth:
pass
spotipy.Spotify = _DummySpotify
oauth2.SpotifyOAuth = _DummyOAuth
oauth2.SpotifyClientCredentials = _DummyOAuth
spotipy.oauth2 = oauth2
sys.modules['spotipy'] = spotipy
sys.modules['spotipy.oauth2'] = oauth2
if 'config.settings' not in sys.modules:
config_mod = types.ModuleType('config')
settings_mod = types.ModuleType('config.settings')
class _DummyConfigManager:
def get(self, key, default=None):
return default
settings_mod.config_manager = _DummyConfigManager()
config_mod.settings = settings_mod
sys.modules['config'] = config_mod
sys.modules['config.settings'] = settings_mod
from core.repair_jobs import metadata_gap_filler as mgf
class _FakeTrackClient:
def __init__(self, source_name, isrc=None):
self.source_name = source_name
self.isrc = isrc
self.calls = []
def get_track_details(self, track_id):
self.calls.append(track_id)
if self.isrc is None:
return None
return {
'id': track_id,
'external_ids': {'isrc': self.isrc},
}
class _FakeMBClient:
def __init__(self):
self.calls = []
def search_recording(self, title, artist_name=None, limit=1):
self.calls.append((title, artist_name, limit))
return [{'id': 'mb-recording'}]
def _make_db():
conn = sqlite3.connect(':memory:')
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute(
"""
CREATE TABLE artists (
id INTEGER PRIMARY KEY,
name TEXT,
thumb_url TEXT
)
"""
)
cursor.execute(
"""
CREATE TABLE albums (
id INTEGER PRIMARY KEY,
title TEXT,
thumb_url TEXT
)
"""
)
cursor.execute(
"""
CREATE TABLE tracks (
id INTEGER PRIMARY KEY,
title TEXT,
artist_id INTEGER,
album_id INTEGER,
spotify_track_id TEXT,
itunes_track_id TEXT,
deezer_track_id TEXT,
isrc TEXT,
musicbrainz_recording_id TEXT
)
"""
)
cursor.execute("INSERT INTO artists (id, name, thumb_url) VALUES (1, 'Artist', '')")
cursor.execute("INSERT INTO albums (id, title, thumb_url) VALUES (1, 'Album', '')")
cursor.execute(
"""
INSERT INTO tracks
(id, title, artist_id, album_id, spotify_track_id, itunes_track_id, deezer_track_id, isrc, musicbrainz_recording_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(1, 'Track Title', 1, 1, 'sp-1', None, 'dz-1', '', ''),
)
conn.commit()
return conn
def _make_context(conn):
findings = []
return SimpleNamespace(
db=SimpleNamespace(_get_connection=lambda: conn),
config_manager=SimpleNamespace(get=lambda key, default=None: default),
check_stop=lambda: False,
wait_if_paused=lambda: False,
update_progress=lambda *args, **kwargs: None,
report_progress=lambda *args, **kwargs: None,
sleep_or_stop=lambda seconds: False,
mb_client=_FakeMBClient(),
create_finding=lambda **kwargs: findings.append(kwargs),
findings=findings,
)
def test_metadata_gap_filler_prefers_primary_track_source(monkeypatch):
conn = _make_db()
context = _make_context(conn)
spotify_client = _FakeTrackClient('spotify', isrc='SP-ISRC')
deezer_client = _FakeTrackClient('deezer', isrc='DZ-ISRC')
itunes_client = _FakeTrackClient('itunes', isrc=None)
monkeypatch.setattr(mgf, 'get_primary_source', lambda: 'deezer')
monkeypatch.setattr(
mgf,
'get_client_for_source',
lambda source: {'spotify': spotify_client, 'deezer': deezer_client, 'itunes': itunes_client}.get(source),
)
result = mgf.MetadataGapFillerJob().scan(context)
assert result.findings_created == 1
assert deezer_client.calls == ['dz-1']
assert spotify_client.calls == []
assert context.findings[0]['details']['found_fields']['isrc'] == 'DZ-ISRC'
assert context.findings[0]['details']['resolved_source'] == 'deezer'
assert context.findings[0]['details']['resolved_track_id'] == 'dz-1'
def test_metadata_gap_filler_skips_track_detail_lookup_when_isrc_disabled(monkeypatch):
conn = _make_db()
context = _make_context(conn)
spotify_client = _FakeTrackClient('spotify', isrc='SP-ISRC')
deezer_client = _FakeTrackClient('deezer', isrc='DZ-ISRC')
monkeypatch.setattr(mgf, 'get_primary_source', lambda: 'deezer')
monkeypatch.setattr(
mgf,
'get_client_for_source',
lambda source: {'spotify': spotify_client, 'deezer': deezer_client}.get(source),
)
job = mgf.MetadataGapFillerJob()
monkeypatch.setattr(job, '_get_settings', lambda context: {'fill_isrc': False, 'fill_musicbrainz_id': True})
result = job.scan(context)
assert result.findings_created == 1
assert spotify_client.calls == []
assert deezer_client.calls == []
assert context.findings[0]['details']['found_fields']['musicbrainz_recording_id'] == 'mb-recording'

167
tests/test_missing_cover_art.py Normal file
View file

@ -0,0 +1,167 @@
import sqlite3
import sys
import types
from types import SimpleNamespace
# Stub optional Spotify dependency so metadata_service can import in tests.
if 'spotipy' not in sys.modules:
spotipy = types.ModuleType('spotipy')
oauth2 = types.ModuleType('spotipy.oauth2')
class _DummySpotify:
pass
class _DummyOAuth:
pass
spotipy.Spotify = _DummySpotify
oauth2.SpotifyOAuth = _DummyOAuth
oauth2.SpotifyClientCredentials = _DummyOAuth
spotipy.oauth2 = oauth2
sys.modules['spotipy'] = spotipy
sys.modules['spotipy.oauth2'] = oauth2
if 'config.settings' not in sys.modules:
config_mod = types.ModuleType('config')
settings_mod = types.ModuleType('config.settings')
class _DummyConfigManager:
def get(self, key, default=None):
return default
settings_mod.config_manager = _DummyConfigManager()
config_mod.settings = settings_mod
sys.modules['config'] = config_mod
sys.modules['config.settings'] = settings_mod
from core.repair_jobs import missing_cover_art as mca
class _FakeClient:
def __init__(self, album_image=None, search_image=None):
self.album_image = album_image
self.search_image = search_image
self.get_album_calls = []
self.search_calls = []
def get_album(self, album_id, include_tracks=False):
self.get_album_calls.append((album_id, include_tracks))
if self.album_image is None:
return None
if isinstance(self.album_image, str):
return {'images': [{'url': self.album_image}]}
return self.album_image
def search_albums(self, query, limit=1):
self.search_calls.append((query, limit))
if self.search_image is None:
return []
return [SimpleNamespace(id='search-album', image_url=self.search_image)]
def _make_db(album_row):
conn = sqlite3.connect(':memory:')
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute(
"""
CREATE TABLE artists (
id INTEGER PRIMARY KEY,
name TEXT,
thumb_url TEXT
)
"""
)
cursor.execute(
"""
CREATE TABLE albums (
id INTEGER PRIMARY KEY,
title TEXT,
artist_id INTEGER,
thumb_url TEXT,
spotify_album_id TEXT,
itunes_album_id TEXT,
deezer_id TEXT,
discogs_id TEXT,
soul_id TEXT
)
"""
)
cursor.execute(
"INSERT INTO artists (id, name, thumb_url) VALUES (?, ?, ?)",
(1, 'Artist', 'https://artist/thumb'),
)
cursor.execute(
"""
INSERT INTO albums
(id, title, artist_id, thumb_url, spotify_album_id, itunes_album_id, deezer_id, discogs_id, soul_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
album_row,
)
conn.commit()
return conn
def _make_context(conn, prefer_source=None):
job_settings = {}
if prefer_source is not None:
job_settings['prefer_source'] = prefer_source
settings = {'repair.jobs.missing_cover_art.settings': job_settings}
findings = []
return SimpleNamespace(
db=SimpleNamespace(_get_connection=lambda: conn),
config_manager=SimpleNamespace(get=lambda key, default=None: settings.get(key, default)),
check_stop=lambda: False,
wait_if_paused=lambda: False,
update_progress=lambda *args, **kwargs: None,
report_progress=lambda *args, **kwargs: None,
create_finding=lambda **kwargs: findings.append(kwargs),
findings=findings,
)
def test_missing_cover_art_prefers_explicit_source_over_primary(monkeypatch):
conn = _make_db((1, 'Album', 1, '', 'sp-album', 'it-album', 'dz-album', 'dg-album', 'hy-album'))
context = _make_context(conn, prefer_source='spotify')
deezer_client = _FakeClient(album_image='https://img/deezer-direct')
spotify_client = _FakeClient(album_image='https://img/spotify-direct')
monkeypatch.setattr(mca, 'get_primary_source', lambda: 'deezer')
monkeypatch.setattr(
mca,
'get_client_for_source',
lambda source: {'deezer': deezer_client, 'spotify': spotify_client}.get(source),
)
result = mca.MissingCoverArtJob().scan(context)
assert result.findings_created == 1
assert spotify_client.get_album_calls == [('sp-album', False)]
assert deezer_client.get_album_calls == []
assert context.findings[0]['details']['found_artwork_url'] == 'https://img/spotify-direct'
def test_missing_cover_art_uses_primary_when_prefer_unset(monkeypatch):
conn = _make_db((1, 'Album', 1, '', None, None, None, None, None))
context = _make_context(conn)
discogs_client = _FakeClient(search_image='https://img/discogs-search')
spotify_client = _FakeClient(search_image='https://img/spotify-search')
itunes_client = _FakeClient(search_image='https://img/itunes-search')
monkeypatch.setattr(mca, 'get_primary_source', lambda: 'discogs')
monkeypatch.setattr(
mca,
'get_client_for_source',
lambda source: {'discogs': discogs_client, 'spotify': spotify_client, 'itunes': itunes_client}.get(source),
)
result = mca.MissingCoverArtJob().scan(context)
assert result.findings_created == 1
assert discogs_client.search_calls == [('Artist Album', 1)]
assert spotify_client.search_calls == []
assert itunes_client.search_calls == []
assert context.findings[0]['details']['found_artwork_url'] == 'https://img/discogs-search'

116
tests/test_track_number_repair.py Normal file
View file

@ -0,0 +1,116 @@
import sys
import types
from types import SimpleNamespace
# Stub optional Spotify dependency so metadata_service can import in tests.
if 'spotipy' not in sys.modules:
spotipy = types.ModuleType('spotipy')
oauth2 = types.ModuleType('spotipy.oauth2')
class _DummySpotify:
pass
class _DummyOAuth:
pass
spotipy.Spotify = _DummySpotify
oauth2.SpotifyOAuth = _DummyOAuth
oauth2.SpotifyClientCredentials = _DummyOAuth
spotipy.oauth2 = oauth2
sys.modules['spotipy'] = spotipy
sys.modules['spotipy.oauth2'] = oauth2
if 'config.settings' not in sys.modules:
config_mod = types.ModuleType('config')
settings_mod = types.ModuleType('config.settings')
class _DummyConfigManager:
def get(self, key, default=None):
return default
settings_mod.config_manager = _DummyConfigManager()
config_mod.settings = settings_mod
sys.modules['config'] = config_mod
sys.modules['config.settings'] = settings_mod
from core.repair_jobs import track_number_repair as tnr
class _FakeSearchClient:
def __init__(self, search_results=None):
self.search_results = search_results or []
self.search_calls = []
def search_albums(self, query, limit=10):
self.search_calls.append((query, limit))
return self.search_results
def _make_context():
return SimpleNamespace(db=None, mb_client=None)
def test_resolve_album_tracklist_prefers_primary_source_album_id(monkeypatch):
context = _make_context()
file_track_data = [('/music/test.flac', '01 - Track.flac', None)]
cache = {'album_tracks_cache': {}, 'title_similarity': 0.8}
calls = []
monkeypatch.setattr(tnr, 'get_primary_source', lambda: 'deezer')
monkeypatch.setattr(
tnr,
'_lookup_album_ids_from_db',
lambda *_args, **_kwargs: {'spotify': 'sp-album', 'itunes': 'it-album', 'deezer': 'dz-album'},
)
monkeypatch.setattr(tnr, '_read_album_id_from_file', lambda *_args, **_kwargs: (None, None))
monkeypatch.setattr(tnr, '_read_spotify_track_id_from_file', lambda *_args, **_kwargs: None)
monkeypatch.setattr(tnr, '_read_musicbrainz_album_id_from_file', lambda *_args, **_kwargs: None)
monkeypatch.setattr(tnr, '_read_album_artist_from_file', lambda *_args, **_kwargs: (None, None))
def fake_get_album_tracks_for_source(source, album_id):
calls.append((source, album_id))
return {'items': [{'name': f'{source} track', 'track_number': 1, 'disc_number': 1}]}
monkeypatch.setattr(tnr, 'get_album_tracks_for_source', fake_get_album_tracks_for_source)
result = tnr.TrackNumberRepairJob()._resolve_album_tracklist(file_track_data, '/music', context, cache)
assert result is not None
assert result[0]['name'] == 'deezer track'
assert calls == [('deezer', 'dz-album')]
def test_resolve_album_tracklist_uses_source_priority_for_search(monkeypatch):
context = _make_context()
file_track_data = [('/music/test.flac', 'Track.flac', None)]
cache = {'album_tracks_cache': {}, 'title_similarity': 0.8}
track_calls = []
discogs_client = _FakeSearchClient([SimpleNamespace(id='dg-album')])
spotify_client = _FakeSearchClient([SimpleNamespace(id='sp-album')])
monkeypatch.setattr(tnr, 'get_primary_source', lambda: 'discogs')
monkeypatch.setattr(tnr, '_lookup_album_ids_from_db', lambda *_args, **_kwargs: {})
monkeypatch.setattr(tnr, '_read_album_id_from_file', lambda *_args, **_kwargs: (None, None))
monkeypatch.setattr(tnr, '_read_spotify_track_id_from_file', lambda *_args, **_kwargs: None)
monkeypatch.setattr(tnr, '_read_musicbrainz_album_id_from_file', lambda *_args, **_kwargs: None)
monkeypatch.setattr(tnr, '_read_album_artist_from_file', lambda *_args, **_kwargs: ('Album Title', 'Artist Name'))
monkeypatch.setattr(
tnr,
'get_client_for_source',
lambda source: {'discogs': discogs_client, 'spotify': spotify_client}.get(source),
)
def fake_get_album_tracks_for_source(source, album_id):
track_calls.append((source, album_id))
return {'items': [{'name': f'{source} track', 'track_number': 1, 'disc_number': 1}]}
monkeypatch.setattr(tnr, 'get_album_tracks_for_source', fake_get_album_tracks_for_source)
result = tnr.TrackNumberRepairJob()._resolve_album_tracklist(file_track_data, '/music', context, cache)
assert result is not None
assert result[0]['name'] == 'discogs track'
assert discogs_client.search_calls == [('Artist Name Album Title', 5)]
assert spotify_client.search_calls == []
assert track_calls == [('discogs', 'dg-album')]

238
tests/test_unknown_artist_fixer.py Normal file
View file

@ -0,0 +1,238 @@
import sys
import types
from types import SimpleNamespace
if "spotipy" not in sys.modules:
spotipy = types.ModuleType("spotipy")
class _DummySpotify:
def __init__(self, *args, **kwargs):
pass
oauth2 = types.ModuleType("spotipy.oauth2")
class _DummyOAuth:
def __init__(self, *args, **kwargs):
pass
spotipy.Spotify = _DummySpotify
oauth2.SpotifyOAuth = _DummyOAuth
oauth2.SpotifyClientCredentials = _DummyOAuth
spotipy.oauth2 = oauth2
sys.modules["spotipy"] = spotipy
sys.modules["spotipy.oauth2"] = oauth2
if "config.settings" not in sys.modules:
config_pkg = types.ModuleType("config")
settings_mod = types.ModuleType("config.settings")
class _DummyConfigManager:
def get(self, key, default=None):
return default
settings_mod.config_manager = _DummyConfigManager()
config_pkg.settings = settings_mod
sys.modules["config"] = config_pkg
sys.modules["config.settings"] = settings_mod
from core.repair_jobs.unknown_artist_fixer import UnknownArtistFixerJob
import core.repair_jobs.unknown_artist_fixer as unknown_artist_fixer_module
class _FakeClient:
def __init__(self, track_details=None, search_results=None):
self.track_details = track_details or {}
self.search_results = search_results or {}
self.get_calls = []
self.search_calls = []
def get_track_details(self, track_id):
self.get_calls.append(track_id)
return self.track_details.get(track_id)
def search_tracks(self, query, limit=5):
self.search_calls.append((query, limit))
return self.search_results.get(query, [])
def _install_tag_reader(monkeypatch, tags=None):
fake_module = types.ModuleType("core.tag_writer")
fake_module.read_file_tags = lambda path: tags or {}
monkeypatch.setitem(sys.modules, "core.tag_writer", fake_module)
def test_unknown_artist_fixer_uses_primary_source_track_id_first(monkeypatch):
job = UnknownArtistFixerJob()
_install_tag_reader(monkeypatch)
deezer_client = _FakeClient(
track_details={
"dz-1": {
"primary_artist": "Deezer Artist",
"album": {
"name": "Deezer Album",
"release_date": "2024-02-01",
"images": [{"url": "https://img/deezer"}],
},
"name": "Deezer Song",
"track_number": 7,
"disc_number": 1,
}
}
)
spotify_client = _FakeClient(
track_details={
"sp-1": {
"primary_artist": "Spotify Artist",
"album": {
"name": "Spotify Album",
"release_date": "2023-01-01",
"images": [{"url": "https://img/spotify"}],
},
"name": "Spotify Song",
"track_number": 1,
"disc_number": 1,
}
}
)
monkeypatch.setattr(unknown_artist_fixer_module, "get_primary_source", lambda: "deezer")
monkeypatch.setattr(
unknown_artist_fixer_module,
"get_client_for_source",
lambda source: {"deezer": deezer_client, "spotify": spotify_client}.get(source),
)
track = {
"title": "Unknown Title",
"album_title": "Unknown Album",
"spotify_track_id": "sp-1",
"deezer_track_id": "dz-1",
"itunes_track_id": "",
}
result = job._resolve_metadata(SimpleNamespace(), track, "/tmp/track.flac")
assert result["artist"] == "Deezer Artist"
assert result["album"] == "Deezer Album"
assert result["source"] == "deezer_track_id_lookup"
assert deezer_client.get_calls == ["dz-1"]
assert spotify_client.get_calls == []
def test_unknown_artist_fixer_searches_primary_source_first(monkeypatch):
job = UnknownArtistFixerJob()
_install_tag_reader(monkeypatch)
candidate = SimpleNamespace(
id="dz-song-1",
name="Matching Title",
album="Matching Album",
artists=["Deezer Artist"],
image_url="https://img/deezer-search",
)
deezer_client = _FakeClient(
track_details={
"dz-song-1": {
"primary_artist": "Deezer Artist",
"album": {
"name": "Matching Album",
"release_date": "2024-03-02",
"images": [{"url": "https://img/deezer-full"}],
},
"name": "Matching Title",
"track_number": 4,
"disc_number": 1,
}
},
search_results={"Matching Title": [candidate]},
)
spotify_client = _FakeClient(
search_results={
"Matching Title": [
SimpleNamespace(
id="sp-song-1",
name="Matching Title",
album="Matching Album",
artists=["Spotify Artist"],
image_url="https://img/spotify-search",
)
]
}
)
monkeypatch.setattr(unknown_artist_fixer_module, "get_primary_source", lambda: "deezer")
monkeypatch.setattr(
unknown_artist_fixer_module,
"get_client_for_source",
lambda source: {"deezer": deezer_client, "spotify": spotify_client}.get(source),
)
track = {
"title": "Matching Title",
"album_title": "Matching Album",
"spotify_track_id": "",
"deezer_track_id": "",
"itunes_track_id": "",
}
result = job._resolve_metadata(SimpleNamespace(sleep_or_stop=lambda seconds: False), track, "/tmp/track.flac")
assert result["artist"] == "Deezer Artist"
assert result["album"] == "Matching Album"
assert result["source"] == "deezer_title_search"
assert deezer_client.search_calls == [("Matching Title", 5)]
assert spotify_client.search_calls == []
def test_unknown_artist_fixer_supports_hydrabase_title_search(monkeypatch):
job = UnknownArtistFixerJob()
_install_tag_reader(monkeypatch)
hydrabase_candidate = SimpleNamespace(
id="hy-song-1",
name="Hydra Match",
album="Hydra Album",
artists=["Hydra Artist"],
image_url="https://img/hydra-search",
)
hydrabase_client = _FakeClient(
track_details={
"hy-song-1": {
"primary_artist": "Hydra Artist",
"album": {
"name": "Hydra Album",
"release_date": "2024-04-03",
"images": [{"url": "https://img/hydra-full"}],
},
"name": "Hydra Match",
"track_number": 2,
"disc_number": 1,
}
},
search_results={"Hydra Match": [hydrabase_candidate]},
)
spotify_client = _FakeClient()
monkeypatch.setattr(unknown_artist_fixer_module, "get_primary_source", lambda: "hydrabase")
monkeypatch.setattr(
unknown_artist_fixer_module,
"get_client_for_source",
lambda source: {"hydrabase": hydrabase_client, "spotify": spotify_client}.get(source),
)
track = {
"title": "Hydra Match",
"album_title": "Hydra Album",
"spotify_track_id": "",
"deezer_track_id": "",
"itunes_track_id": "",
}
result = job._resolve_metadata(SimpleNamespace(sleep_or_stop=lambda seconds: False), track, "/tmp/track.flac")
assert result["artist"] == "Hydra Artist"
assert result["source"] == "hydrabase_title_search"
assert hydrabase_client.search_calls == [("Hydra Match", 5)]
assert spotify_client.search_calls == []

2
webui/static/script.js
View file

@ -64089,6 +64089,8 @@ function _renderFindingDetail(f) {
if (d.album) rows.push(['Album', d.album]);
if (d.title) rows.push(['Title', d.title]);
if (d.spotify_track_id) rows.push(['Spotify ID', d.spotify_track_id]);
if (d.resolved_source) rows.push(['Resolved Source', d.resolved_source]);
if (d.resolved_track_id) rows.push(['Resolved Track ID', d.resolved_track_id]);
if (d.found_fields && typeof d.found_fields === 'object') {
Object.entries(d.found_fields).forEach(([k, v]) => {
rows.push([`Found: ${k}`, String(v), 'success']);