Quality Upgrade Finder: new findings-based repair job (replaces auto-acting Quality Scanner)
The old Quality Scanner tool judged quality by file EXTENSION only (a 128k and a 320k MP3 looked identical), ignored the bitrate-based quality profile, used min() of enabled tiers so the default profile flagged the ENTIRE non-lossless library, and auto-dumped every match into the wishlist with no review. This new repair job does it properly: - meets_preferred_quality(): pure, bitrate-AWARE decision honoring every enabled quality bucket (320 MP3 passes a FLAC+320+256 profile; 128 MP3 doesn't). Floor is the worst enabled bucket, not the best. - scans watchlist artists or whole library, finds below-quality tracks, matches a better version at scan time (reusing the existing tested match helpers), emits a FINDING showing the match + confidence. Off by default; nothing auto-queued. - _fix_quality_upgrade apply handler adds the matched track WITH album context to the wishlist — the user-approved version of what the old tool did silently. - Transcode/fake-lossless detection intentionally left to the existing Fake Lossless Detector job. 12 seam tests incl. a regression pinning the default-profile flooding bug. The old tool is still in place; removing it + rewiring its automation action is the next step.
This commit is contained in:
BoulderBadgeDad
parent
78f47f04d7
commit
69dd4e1792
4 changed files with 649 additions and 0 deletions
|
|
@ -50,6 +50,7 @@ _JOB_MODULES = [
|
|||
'core.repair_jobs.discography_backfill',
|
||||
'core.repair_jobs.canonical_version_resolve',
|
||||
'core.repair_jobs.library_retag',
|
||||
'core.repair_jobs.quality_upgrade',
|
||||
]
|
||||
|
||||
|
||||
|
|
|
|||
396
core/repair_jobs/quality_upgrade.py
Normal file
396
core/repair_jobs/quality_upgrade.py
Normal file
|
|
@ -0,0 +1,396 @@
|
|||
"""Quality Upgrade Finder maintenance job.
|
||||
|
||||
Replaces the old auto-acting "Quality Scanner" tool. That tool decided quality
|
||||
purely by file EXTENSION (so a 128 kbps MP3 and a 320 kbps MP3 looked identical),
|
||||
ignored the bitrate-based quality profile, and silently dumped every match
|
||||
straight into the wishlist with no review — which, on the default profile, meant
|
||||
flagging an entire non-lossless library at once.
|
||||
|
||||
This job does it the way the rest of the app works: it SCANS (watchlist artists
|
||||
or the whole library), judges each track against the user's quality profile using
|
||||
BOTH format and bitrate, and for anything below the preferred quality it searches
|
||||
the configured metadata source for a better version and emits a FINDING. Nothing
|
||||
is queued until you review and Apply the finding — at which point the matched
|
||||
track (carrying its album context) is added to the wishlist, exactly like every
|
||||
other acquisition path.
|
||||
|
||||
The quality decision (``meets_preferred_quality``) is a pure function so it can be
|
||||
unit-tested without a database or network. Transcode/"fake lossless" detection is
|
||||
intentionally NOT done here — that's the separate Fake Lossless Detector job.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import time
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
from core.metadata.registry 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
|
||||
# Reuse the (tested) provider search + result-normalization helpers from the old
|
||||
# scanner module so matching stays a single source of truth.
|
||||
from core.discovery.quality_scanner import (
|
||||
_extract_lookup_value,
|
||||
_normalize_track_match,
|
||||
_search_tracks_for_source,
|
||||
_track_artist_names,
|
||||
_track_name,
|
||||
)
|
||||
from utils.logging_config import get_logger
|
||||
|
||||
logger = get_logger("repair_jobs.quality_upgrade")
|
||||
|
||||
|
||||
# Quality ranks — higher is better. Lossless tops everything; lossy tiers fall out
|
||||
# of bitrate. 0 means "below the lowest tracked tier / unknown".
|
||||
RANK_LOSSLESS = 4
|
||||
RANK_320 = 3
|
||||
RANK_256 = 2
|
||||
RANK_192 = 1
|
||||
RANK_BELOW = 0
|
||||
|
||||
LOSSLESS_EXTENSIONS = {'.flac', '.alac', '.ape', '.wav', '.aiff', '.aif', '.dsf', '.dff', '.m4a'}
|
||||
# NB: .m4a is ambiguous (ALAC vs AAC); we treat the *format* as lossy-capable and
|
||||
# rely on bitrate below — a true ALAC .m4a reports a lossless-scale bitrate.
|
||||
|
||||
# Quality-profile bucket key -> rank.
|
||||
_PROFILE_KEY_RANK = {
|
||||
'flac': RANK_LOSSLESS,
|
||||
'mp3_320': RANK_320,
|
||||
'mp3_256': RANK_256,
|
||||
'mp3_192': RANK_192,
|
||||
}
|
||||
|
||||
|
||||
def _normalize_kbps(bitrate: Optional[int]) -> Optional[int]:
|
||||
"""Library bitrate may be stored in bps (e.g. 320000) or kbps (320).
|
||||
Normalize to kbps. Returns None when unknown/zero."""
|
||||
if not bitrate:
|
||||
return None
|
||||
try:
|
||||
b = int(bitrate)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
if b <= 0:
|
||||
return None
|
||||
return b // 1000 if b > 4000 else b
|
||||
|
||||
|
||||
def classify_track_quality(file_path: str, bitrate: Optional[int]) -> Optional[int]:
|
||||
"""Rank a file by format + bitrate. Returns a RANK_* value, or None when it
|
||||
can't be judged (a lossy file with no known bitrate)."""
|
||||
ext = os.path.splitext(file_path or '')[1].lower()
|
||||
kbps = _normalize_kbps(bitrate)
|
||||
|
||||
# Lossless containers: a real lossless file has a high bitrate; a low one is a
|
||||
# lossy stream in a lossless container — but flagging that is the Fake Lossless
|
||||
# Detector's job, so here we treat the lossless *format* as top rank.
|
||||
if ext in {'.flac', '.alac', '.ape', '.wav', '.aiff', '.aif', '.dsf', '.dff'}:
|
||||
return RANK_LOSSLESS
|
||||
# .m4a / lossy: judge purely by bitrate. A lossless-scale bitrate (ALAC in m4a,
|
||||
# or a mislabeled lossless) ranks as lossless.
|
||||
if kbps is None:
|
||||
return None
|
||||
if kbps >= 800:
|
||||
return RANK_LOSSLESS
|
||||
if kbps >= 280:
|
||||
return RANK_320
|
||||
if kbps >= 200:
|
||||
return RANK_256
|
||||
if kbps >= 150:
|
||||
return RANK_192
|
||||
return RANK_BELOW
|
||||
|
||||
|
||||
def preferred_quality_floor(quality_profile: Dict[str, Any]) -> Optional[int]:
|
||||
"""The lowest acceptable quality rank from the profile's ENABLED buckets — the
|
||||
floor a track must meet. Returns None when nothing is enabled (caller should
|
||||
then flag nothing, rather than flagging everything)."""
|
||||
qualities = (quality_profile or {}).get('qualities', {}) or {}
|
||||
enabled_ranks = [
|
||||
_PROFILE_KEY_RANK[key]
|
||||
for key, cfg in qualities.items()
|
||||
if isinstance(cfg, dict) and cfg.get('enabled') and key in _PROFILE_KEY_RANK
|
||||
]
|
||||
if not enabled_ranks:
|
||||
return None
|
||||
return min(enabled_ranks)
|
||||
|
||||
|
||||
def meets_preferred_quality(file_path: str, bitrate: Optional[int],
|
||||
quality_profile: Dict[str, Any]) -> bool:
|
||||
"""Pure decision: does this track already meet the user's preferred quality?
|
||||
|
||||
A track meets quality when its format+bitrate rank is at least the profile's
|
||||
floor (the worst quality the user still accepts). This honors a profile that
|
||||
enables, say, FLAC *and* MP3-320: a 320 kbps MP3 passes, a 128 kbps MP3 does
|
||||
not. With nothing enabled, everything passes (we never flag the whole library
|
||||
on an empty profile)."""
|
||||
floor = preferred_quality_floor(quality_profile)
|
||||
if floor is None:
|
||||
return True
|
||||
|
||||
file_rank = classify_track_quality(file_path, bitrate)
|
||||
if file_rank is None:
|
||||
# Lossy file with unknown bitrate: only judgeable when the floor is
|
||||
# lossless (then any lossy file is below it). Otherwise don't flag.
|
||||
ext = os.path.splitext(file_path or '')[1].lower()
|
||||
if floor == RANK_LOSSLESS and ext not in LOSSLESS_EXTENSIONS:
|
||||
return False
|
||||
return True
|
||||
|
||||
return file_rank >= floor
|
||||
|
||||
|
||||
def _rank_label(rank: Optional[int]) -> str:
|
||||
return {
|
||||
RANK_LOSSLESS: 'Lossless', RANK_320: 'MP3 320', RANK_256: 'MP3 256',
|
||||
RANK_192: 'MP3 192', RANK_BELOW: 'low bitrate',
|
||||
}.get(rank, 'unknown')
|
||||
|
||||
|
||||
def _find_best_match(engine: Any, source_priority: List[str], title: str, artist: str,
|
||||
album: str, min_confidence: float) -> Tuple[Optional[Any], float, Optional[str], bool]:
|
||||
"""Search the configured metadata sources for the best replacement match.
|
||||
Returns (best_track, confidence, source, attempted_any_provider)."""
|
||||
temp_track = type('TempTrack', (), {'name': title, 'artists': [artist], 'album': album})()
|
||||
queries = engine.generate_download_queries(temp_track)
|
||||
|
||||
best, best_conf, best_src = None, 0.0, None
|
||||
attempted = False
|
||||
for query in queries:
|
||||
for source in source_priority:
|
||||
client = get_client_for_source(source)
|
||||
if not client or not hasattr(client, 'search_tracks'):
|
||||
continue
|
||||
attempted = True
|
||||
matches = _search_tracks_for_source(source, query, limit=5, client=client)
|
||||
time.sleep(0.5) # be gentle on metadata APIs
|
||||
for cand in matches or []:
|
||||
cand_artists = _track_artist_names(cand)
|
||||
artist_conf = max(
|
||||
(engine.similarity_score(engine.normalize_string(artist),
|
||||
engine.normalize_string(n)) for n in cand_artists),
|
||||
default=0.0,
|
||||
)
|
||||
title_conf = engine.similarity_score(
|
||||
engine.normalize_string(title), engine.normalize_string(_track_name(cand)))
|
||||
conf = artist_conf * 0.5 + title_conf * 0.5
|
||||
album_type = _extract_lookup_value(cand, 'album_type', default='') or ''
|
||||
if album_type == 'album':
|
||||
conf += 0.02
|
||||
elif album_type == 'ep':
|
||||
conf += 0.01
|
||||
if conf > best_conf and conf >= min_confidence:
|
||||
best, best_conf, best_src = cand, conf, source
|
||||
if best_conf >= 0.9:
|
||||
break
|
||||
if best_conf >= 0.9:
|
||||
break
|
||||
return best, best_conf, best_src, attempted
|
||||
|
||||
|
||||
@register_job
|
||||
class QualityUpgradeJob(RepairJob):
|
||||
job_id = 'quality_upgrade'
|
||||
display_name = 'Quality Upgrade Finder'
|
||||
description = 'Finds library tracks below your preferred quality and proposes a better version'
|
||||
help_text = (
|
||||
'Scans your library (or just your watchlist artists) and compares each '
|
||||
"track against your Quality Profile using BOTH the file format and its "
|
||||
'bitrate — so a 128 kbps MP3 is no longer treated the same as a 320 kbps '
|
||||
'one, and enabling MP3-320/256 in your profile actually counts.\n\n'
|
||||
'For every track below your preferred quality, it searches your configured '
|
||||
'metadata source for a better version and creates a finding showing the '
|
||||
'match and a confidence score. Nothing is queued automatically: applying a '
|
||||
'finding adds that matched track — with its album context — to the wishlist, '
|
||||
'the same as any other download.\n\n'
|
||||
'Settings:\n'
|
||||
'- Scope: "watchlist" (watchlisted artists only) or "all" (whole library)\n'
|
||||
'- Min confidence: minimum match confidence (0-1) to surface a finding\n\n'
|
||||
'Note: detecting fake/transcoded lossless files is handled by the separate '
|
||||
'Fake Lossless Detector job.'
|
||||
)
|
||||
icon = 'repair-icon-lossy'
|
||||
default_enabled = False
|
||||
default_interval_hours = 168
|
||||
default_settings = {'scope': 'watchlist', 'min_confidence': 0.7}
|
||||
setting_options = {'scope': ['watchlist', 'all']}
|
||||
auto_fix = False
|
||||
|
||||
def _get_settings(self, context: JobContext) -> Dict[str, Any]:
|
||||
cfg = context.config_manager
|
||||
scope = 'watchlist'
|
||||
min_conf = 0.7
|
||||
if cfg:
|
||||
scope = cfg.get(self.get_config_key('settings.scope'), 'watchlist') or 'watchlist'
|
||||
try:
|
||||
min_conf = float(cfg.get(self.get_config_key('settings.min_confidence'), 0.7))
|
||||
except (TypeError, ValueError):
|
||||
min_conf = 0.7
|
||||
return {'scope': scope, 'min_confidence': min_conf}
|
||||
|
||||
def _load_tracks(self, db: Any, scope: str) -> List[tuple]:
|
||||
conn = db._get_connection()
|
||||
try:
|
||||
base = (
|
||||
"SELECT t.id, t.title, t.file_path, t.bitrate, a.name AS artist_name, "
|
||||
"al.title AS album_title, t.album_id "
|
||||
"FROM tracks t "
|
||||
"JOIN artists a ON t.artist_id = a.id "
|
||||
"JOIN albums al ON t.album_id = al.id "
|
||||
"WHERE t.file_path IS NOT NULL AND t.file_path != ''"
|
||||
)
|
||||
if scope == 'watchlist':
|
||||
artists = db.get_watchlist_artists(profile_id=1)
|
||||
names = [getattr(ar, 'artist_name', None) for ar in artists]
|
||||
names = [n for n in names if n]
|
||||
if not names:
|
||||
return []
|
||||
placeholders = ','.join('?' for _ in names)
|
||||
rows = conn.execute(
|
||||
base + f" AND a.name IN ({placeholders})", names).fetchall()
|
||||
else:
|
||||
rows = conn.execute(base).fetchall()
|
||||
return rows
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def estimate_scope(self, context: JobContext) -> int:
|
||||
try:
|
||||
return len(self._load_tracks(context.db, self._get_settings(context)['scope']))
|
||||
except Exception:
|
||||
return 0
|
||||
|
||||
def scan(self, context: JobContext) -> JobResult:
|
||||
result = JobResult()
|
||||
settings = self._get_settings(context)
|
||||
scope = settings['scope']
|
||||
min_conf = settings['min_confidence']
|
||||
|
||||
db = context.db
|
||||
quality_profile = db.get_quality_profile()
|
||||
if preferred_quality_floor(quality_profile) is None:
|
||||
logger.info("[Quality Upgrade] No quality buckets enabled in profile — nothing to flag")
|
||||
return result
|
||||
|
||||
try:
|
||||
tracks = self._load_tracks(db, scope)
|
||||
except Exception as e:
|
||||
logger.error("[Quality Upgrade] Error loading tracks: %s", e, exc_info=True)
|
||||
result.errors += 1
|
||||
return result
|
||||
|
||||
total = len(tracks)
|
||||
if context.update_progress:
|
||||
context.update_progress(0, total)
|
||||
if context.report_progress:
|
||||
context.report_progress(phase=f'Checking quality on {total} tracks...', total=total)
|
||||
|
||||
# Metadata source for matching — resolved lazily so we only fail if we
|
||||
# actually find a low-quality track that needs a match.
|
||||
engine = None
|
||||
source_priority: List[str] = []
|
||||
|
||||
for i, row in enumerate(tracks):
|
||||
if context.check_stop():
|
||||
return result
|
||||
if i % 10 == 0 and context.wait_if_paused():
|
||||
return result
|
||||
|
||||
track_id, title, file_path, bitrate, artist_name, album_title, album_id = (
|
||||
row[0], row[1], row[2], row[3], row[4], row[5], row[6])
|
||||
result.scanned += 1
|
||||
|
||||
if meets_preferred_quality(file_path, bitrate, quality_profile):
|
||||
result.skipped += 1
|
||||
if context.update_progress and (i + 1) % 25 == 0:
|
||||
context.update_progress(i + 1, total)
|
||||
continue
|
||||
|
||||
# Below preferred quality — find a better version to propose.
|
||||
if engine is None:
|
||||
from core.matching_engine import MusicMatchingEngine
|
||||
engine = MusicMatchingEngine()
|
||||
source_priority = get_source_priority(get_primary_source()) or []
|
||||
if not source_priority:
|
||||
logger.warning("[Quality Upgrade] No metadata provider available — cannot propose upgrades")
|
||||
return result
|
||||
|
||||
if context.is_spotify_rate_limited():
|
||||
logger.info("[Quality Upgrade] Spotify rate-limited — stopping scan early")
|
||||
return result
|
||||
|
||||
current_rank = classify_track_quality(file_path, bitrate)
|
||||
current_label = _rank_label(current_rank)
|
||||
if context.report_progress:
|
||||
context.report_progress(
|
||||
scanned=i + 1, total=total,
|
||||
log_line=f'Low quality ({current_label}): {artist_name} - {title}',
|
||||
log_type='info')
|
||||
|
||||
try:
|
||||
best, conf, source, attempted = _find_best_match(
|
||||
engine, source_priority, title, artist_name or '', album_title or '', min_conf)
|
||||
except Exception as e:
|
||||
logger.debug("[Quality Upgrade] Match error for %s - %s: %s", artist_name, title, e)
|
||||
result.errors += 1
|
||||
continue
|
||||
|
||||
if not attempted:
|
||||
logger.warning("[Quality Upgrade] No metadata provider responded — stopping")
|
||||
return result
|
||||
if not best:
|
||||
result.skipped += 1
|
||||
continue
|
||||
|
||||
matched = _normalize_track_match(best, source or 'metadata')
|
||||
# Carry album context: prefer the matched album, fall back to the
|
||||
# library album the low-quality track came from.
|
||||
alb = matched.get('album')
|
||||
if (not isinstance(alb, dict) or not alb.get('name')) and album_title:
|
||||
matched['album'] = {'name': album_title, 'images': (alb or {}).get('images', []) if isinstance(alb, dict) else []}
|
||||
|
||||
if context.create_finding:
|
||||
try:
|
||||
inserted = context.create_finding(
|
||||
job_id=self.job_id,
|
||||
finding_type='quality_upgrade',
|
||||
severity='info',
|
||||
entity_type='track',
|
||||
entity_id=str(track_id),
|
||||
file_path=file_path,
|
||||
title=f'Upgrade: {artist_name} - {title} ({current_label})',
|
||||
description=(
|
||||
f'"{title}" by {artist_name} is {current_label}, below your preferred '
|
||||
f'quality. Best match: "{_track_name(best)}" via {source} '
|
||||
f'(confidence {conf:.0%}). Apply to add it to the wishlist.'),
|
||||
details={
|
||||
'track_id': track_id,
|
||||
'track_title': title,
|
||||
'artist': artist_name,
|
||||
'album_id': album_id,
|
||||
'album_title': album_title,
|
||||
'current_format': current_label,
|
||||
'current_bitrate': bitrate,
|
||||
'match_confidence': conf,
|
||||
'provider': source,
|
||||
'matched_track_data': matched,
|
||||
})
|
||||
if inserted:
|
||||
result.findings_created += 1
|
||||
else:
|
||||
result.findings_skipped_dedup += 1
|
||||
except Exception as e:
|
||||
logger.debug("[Quality Upgrade] create finding failed for track %s: %s", track_id, e)
|
||||
result.errors += 1
|
||||
|
||||
if context.update_progress and (i + 1) % 10 == 0:
|
||||
context.update_progress(i + 1, total)
|
||||
|
||||
if context.update_progress:
|
||||
context.update_progress(total, total)
|
||||
logger.info("[Quality Upgrade] %d scanned, %d upgrades found, %d met/skip",
|
||||
result.scanned, result.findings_created, result.skipped)
|
||||
return result
|
||||
|
|
@ -981,6 +981,7 @@ class RepairWorker:
|
|||
'acoustid_mismatch': self._fix_acoustid_mismatch,
|
||||
'missing_discography_track': self._fix_discography_backfill,
|
||||
'library_retag': self._fix_library_retag,
|
||||
'quality_upgrade': self._fix_quality_upgrade,
|
||||
}
|
||||
handler = handlers.get(finding_type)
|
||||
if not handler:
|
||||
|
|
@ -1007,6 +1008,36 @@ class RepairWorker:
|
|||
except Exception as e:
|
||||
return {'success': False, 'error': str(e)}
|
||||
|
||||
def _fix_quality_upgrade(self, entity_type, entity_id, file_path, details):
|
||||
"""Add the matched higher-quality version to the wishlist (with album
|
||||
context). Applying a Quality Upgrade finding is the user-approved step
|
||||
that the old auto-acting Quality Scanner did without review."""
|
||||
track_data = details.get('matched_track_data')
|
||||
if not track_data:
|
||||
return {'success': False, 'error': 'No matched track in finding'}
|
||||
try:
|
||||
success = self.db.add_to_wishlist(
|
||||
spotify_track_data=track_data,
|
||||
failure_reason=f"Quality upgrade — current file is {details.get('current_format', 'low quality')}",
|
||||
source_type='repair',
|
||||
source_info={
|
||||
'job': 'quality_upgrade',
|
||||
'original_file_path': file_path,
|
||||
'original_format': details.get('current_format'),
|
||||
'original_bitrate': details.get('current_bitrate'),
|
||||
'album_title': details.get('album_title'),
|
||||
'match_confidence': details.get('match_confidence'),
|
||||
'provider': details.get('provider'),
|
||||
},
|
||||
)
|
||||
track_name = track_data.get('name', '?')
|
||||
if success:
|
||||
return {'success': True, 'action': 'added_to_wishlist',
|
||||
'message': f"Added '{track_name}' to wishlist for re-download"}
|
||||
return {'success': False, 'error': f"Could not add '{track_name}' to wishlist (may already exist or be blocklisted)"}
|
||||
except Exception as e:
|
||||
return {'success': False, 'error': str(e)}
|
||||
|
||||
def _fix_dead_file(self, entity_type, entity_id, file_path, details):
|
||||
"""Fix a dead file reference. Action depends on details['_fix_action']:
|
||||
'redownload' (default) — add to wishlist + remove DB entry
|
||||
|
|
|
|||
221
tests/repair_jobs/test_quality_upgrade.py
Normal file
221
tests/repair_jobs/test_quality_upgrade.py
Normal file
|
|
@ -0,0 +1,221 @@
|
|||
"""Quality Upgrade Finder job — the findings-based replacement for the old
|
||||
auto-acting Quality Scanner.
|
||||
|
||||
The old tool judged quality by file EXTENSION only and used min() of the enabled
|
||||
tiers, so with the default profile (FLAC + MP3-320 + MP3-256 enabled) it flagged
|
||||
EVERY non-lossless file — a 320 kbps MP3 included — and dumped them all into the
|
||||
wishlist with no review. These tests pin the corrected behavior: bitrate-aware,
|
||||
honors every enabled bucket, and only proposes (findings) rather than auto-acting.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import types
|
||||
|
||||
import core.repair_jobs.quality_upgrade as qu
|
||||
from core.repair_jobs.base import JobContext, JobResult
|
||||
|
||||
|
||||
# Profiles ------------------------------------------------------------------
|
||||
|
||||
BALANCED = { # default: FLAC + MP3-320 + MP3-256 enabled, MP3-192 off
|
||||
'qualities': {
|
||||
'flac': {'enabled': True, 'min_kbps': 500},
|
||||
'mp3_320': {'enabled': True, 'min_kbps': 280},
|
||||
'mp3_256': {'enabled': True, 'min_kbps': 200},
|
||||
'mp3_192': {'enabled': False, 'min_kbps': 150},
|
||||
}
|
||||
}
|
||||
LOSSLESS_ONLY = {
|
||||
'qualities': {
|
||||
'flac': {'enabled': True, 'min_kbps': 500},
|
||||
'mp3_320': {'enabled': False, 'min_kbps': 280},
|
||||
'mp3_256': {'enabled': False, 'min_kbps': 200},
|
||||
'mp3_192': {'enabled': False, 'min_kbps': 150},
|
||||
}
|
||||
}
|
||||
NOTHING_ENABLED = {'qualities': {'flac': {'enabled': False}, 'mp3_320': {'enabled': False}}}
|
||||
|
||||
|
||||
# --- pure quality decision -------------------------------------------------
|
||||
|
||||
def test_balanced_profile_accepts_320_mp3_REGRESSION():
|
||||
"""The headline bug: with FLAC+320+256 enabled, a 320 kbps MP3 is acceptable.
|
||||
The old min()-tier logic flagged it (and every other MP3) for re-download."""
|
||||
assert meets('song.mp3', 320, BALANCED) is True
|
||||
|
||||
|
||||
def test_balanced_profile_accepts_256_mp3():
|
||||
assert meets('song.mp3', 256, BALANCED) is True
|
||||
|
||||
|
||||
def test_balanced_profile_flags_low_bitrate_mp3():
|
||||
assert meets('song.mp3', 128, BALANCED) is False
|
||||
assert meets('song.mp3', 192, BALANCED) is False # below the 256 floor
|
||||
|
||||
|
||||
def test_flac_always_meets_when_flac_enabled():
|
||||
assert meets('song.flac', 900, BALANCED) is True
|
||||
assert meets('song.flac', 900, LOSSLESS_ONLY) is True
|
||||
|
||||
|
||||
def test_lossless_only_flags_every_lossy_regardless_of_bitrate():
|
||||
assert meets('song.mp3', 320, LOSSLESS_ONLY) is False
|
||||
assert meets('song.m4a', 256, LOSSLESS_ONLY) is False
|
||||
|
||||
|
||||
def test_nothing_enabled_flags_nothing():
|
||||
"""Empty/disabled profile must NOT flag the whole library."""
|
||||
assert meets('song.mp3', 64, NOTHING_ENABLED) is True
|
||||
|
||||
|
||||
def test_bitrate_in_bps_is_normalized():
|
||||
"""Library bitrate stored as bps (320000) classifies the same as 320 kbps."""
|
||||
assert qu.classify_track_quality('song.mp3', 320000) == qu.RANK_320
|
||||
assert meets('song.mp3', 320000, BALANCED) is True
|
||||
|
||||
|
||||
def test_unknown_lossy_bitrate_not_flagged_under_lossy_floor():
|
||||
"""A lossy file with no bitrate can't be judged against a lossy floor → don't
|
||||
flag (avoid false positives); but under a lossless floor it's clearly below."""
|
||||
assert meets('song.mp3', None, BALANCED) is True
|
||||
assert meets('song.mp3', None, LOSSLESS_ONLY) is False
|
||||
|
||||
|
||||
def test_floor_is_worst_enabled_not_best():
|
||||
# FLAC+320+256 enabled → floor is MP3-256 (rank 2), not FLAC.
|
||||
assert qu.preferred_quality_floor(BALANCED) == qu.RANK_256
|
||||
assert qu.preferred_quality_floor(LOSSLESS_ONLY) == qu.RANK_LOSSLESS
|
||||
assert qu.preferred_quality_floor(NOTHING_ENABLED) is None
|
||||
|
||||
|
||||
def meets(path, bitrate, profile):
|
||||
return qu.meets_preferred_quality(path, bitrate, profile)
|
||||
|
||||
|
||||
# --- scan produces a finding (seam) ----------------------------------------
|
||||
|
||||
class _FakeConn:
|
||||
def __init__(self, rows):
|
||||
self._rows = rows
|
||||
|
||||
def execute(self, *a, **k):
|
||||
return self
|
||||
|
||||
def fetchall(self):
|
||||
return self._rows
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
|
||||
class _FakeDB:
|
||||
def __init__(self, rows, profile):
|
||||
self._rows = rows
|
||||
self._profile = profile
|
||||
|
||||
def get_quality_profile(self):
|
||||
return self._profile
|
||||
|
||||
def _get_connection(self):
|
||||
return _FakeConn(self._rows)
|
||||
|
||||
def get_watchlist_artists(self, profile_id=1):
|
||||
return [types.SimpleNamespace(artist_name='Artist A')]
|
||||
|
||||
|
||||
def _ctx(db, findings):
|
||||
return JobContext(
|
||||
db=db,
|
||||
transfer_folder='/tmp',
|
||||
config_manager=None,
|
||||
create_finding=lambda **kw: findings.append(kw) or True,
|
||||
should_stop=lambda: False,
|
||||
is_paused=lambda: False,
|
||||
)
|
||||
|
||||
|
||||
def test_scan_creates_finding_for_low_quality_track(monkeypatch):
|
||||
# One 128 kbps MP3 (below the balanced floor) for Artist A.
|
||||
rows = [(1, 'Song One', '/music/a.mp3', 128, 'Artist A', 'Album X', 10)]
|
||||
db = _FakeDB(rows, BALANCED)
|
||||
|
||||
# Stub the metadata side so the test stays offline.
|
||||
monkeypatch.setattr(qu, 'get_primary_source', lambda: 'spotify')
|
||||
monkeypatch.setattr(qu, 'get_source_priority', lambda src: ['spotify'])
|
||||
monkeypatch.setattr(
|
||||
'core.matching_engine.MusicMatchingEngine',
|
||||
lambda: types.SimpleNamespace(
|
||||
generate_download_queries=lambda t: ['q'],
|
||||
similarity_score=lambda a, b: 1.0,
|
||||
normalize_string=lambda s: s,
|
||||
),
|
||||
)
|
||||
fake_match = {'id': 'sp1', 'name': 'Song One', 'artists': ['Artist A'],
|
||||
'album': {'name': 'Album X', 'images': []}}
|
||||
monkeypatch.setattr(qu, '_find_best_match',
|
||||
lambda *a, **k: (fake_match, 0.95, 'spotify', True))
|
||||
monkeypatch.setattr(qu, '_normalize_track_match', lambda track, src: dict(fake_match))
|
||||
monkeypatch.setattr(qu, '_track_name', lambda t: 'Song One')
|
||||
|
||||
findings = []
|
||||
job = qu.QualityUpgradeJob()
|
||||
# default scope 'watchlist'; config_manager None → defaults used
|
||||
result = job.scan(_ctx(db, findings))
|
||||
|
||||
assert result.findings_created == 1
|
||||
assert len(findings) == 1
|
||||
f = findings[0]
|
||||
assert f['finding_type'] == 'quality_upgrade'
|
||||
assert f['entity_id'] == '1'
|
||||
# Album context + matched track carried for the apply step.
|
||||
assert f['details']['matched_track_data']['id'] == 'sp1'
|
||||
assert f['details']['album_title'] == 'Album X'
|
||||
assert f['details']['provider'] == 'spotify'
|
||||
|
||||
|
||||
def test_scan_skips_tracks_meeting_quality(monkeypatch):
|
||||
# A 320 kbps MP3 meets the balanced profile → no finding, no metadata calls.
|
||||
rows = [(2, 'Good Song', '/music/b.mp3', 320, 'Artist A', 'Album Y', 11)]
|
||||
db = _FakeDB(rows, BALANCED)
|
||||
|
||||
def _boom(*a, **k): # must never be called for an acceptable track
|
||||
raise AssertionError("matching should not run for an acceptable track")
|
||||
|
||||
monkeypatch.setattr(qu, '_find_best_match', _boom)
|
||||
|
||||
findings = []
|
||||
result = qu.QualityUpgradeJob().scan(_ctx(db, findings))
|
||||
assert result.findings_created == 0
|
||||
assert result.skipped == 1
|
||||
assert findings == []
|
||||
|
||||
|
||||
# --- fix handler adds to wishlist ------------------------------------------
|
||||
|
||||
def test_fix_handler_adds_matched_track_to_wishlist():
|
||||
from core.repair_worker import RepairWorker
|
||||
|
||||
captured = {}
|
||||
|
||||
class _DB:
|
||||
def add_to_wishlist(self, **kw):
|
||||
captured.update(kw)
|
||||
return True
|
||||
|
||||
worker = object.__new__(RepairWorker)
|
||||
worker.db = _DB()
|
||||
|
||||
details = {
|
||||
'matched_track_data': {'id': 'sp1', 'name': 'Song One',
|
||||
'album': {'name': 'Album X'}},
|
||||
'current_format': 'MP3 192', 'current_bitrate': 192,
|
||||
'album_title': 'Album X', 'provider': 'spotify', 'match_confidence': 0.9,
|
||||
}
|
||||
res = worker._fix_quality_upgrade('track', '1', '/music/a.mp3', details)
|
||||
|
||||
assert res['success'] is True
|
||||
assert captured['spotify_track_data']['id'] == 'sp1'
|
||||
assert captured['source_type'] == 'repair'
|
||||
assert captured['source_info']['job'] == 'quality_upgrade'
|
||||
assert captured['source_info']['album_title'] == 'Album X'
|
||||
Loading…
Reference in a new issue