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

Merge pull request #777 from Nezreka/canonical-album-version

Browse source 
Canonical album version
This commit is contained in:
BoulderBadgeDad 2026-06-02 15:48:19 -07:00 committed by GitHub
commit e27e436465
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
19 changed files with 1903 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

144
SPEC_canonical_album_version.md Normal file
View file

@ -0,0 +1,144 @@
# Spec: Canonical Album Version (fixes #765 + #767-Bug2)
**Status:** design only — no code yet.
**Goal:** Pin ONE canonical `(source, album_id)` per album, chosen by best-fit to
the user's actual files, so the Library Reorganizer, Track Number Repair, and
tagging/enrichment all agree on the same release. Today each re-resolves
independently and they contradict each other (Spotify Believer=4 vs MusicBrainz
Believer=3; standard album mislinked to a deluxe release).
**Canonical-selection rule (decided):** *match the user's actual files.* The
canonical release is the candidate whose track count + per-track durations +
titles best fit what's on disk. Self-correcting: picks standard when you own the
standard, deluxe when you own the deluxe.
---
## Hard requirement: don't disrupt the running app
Every stage below is **additive and dormant until explicitly consumed**, and
every consumer **falls back to today's behavior when no canonical is set**. So:
- albums with no resolved canonical behave EXACTLY as they do now;
- each stage is independently shippable and reversible;
- nothing big-bangs.
---
## Stage 1 — Schema + pure scorer (ships dormant, zero behavior change)
### Schema (additive, nullable → migration-safe)
Add to `albums` (guarded `ALTER TABLE ... ADD COLUMN`, idempotent — mirror the
existing column-exists checks; see [[db-schema-review]] migration-safety notes):
- `canonical_source TEXT` — e.g. 'spotify' / 'itunes' / 'musicbrainz'
- `canonical_album_id TEXT`
- `canonical_score REAL` — best-fit score (for transparency / re-resolve gating)
- `canonical_resolved_at TIMESTAMP`
All nullable. Existing rows = NULL → "unresolved" → consumers fall back. No
backfill in this stage. No reads in this stage.
### Pure core helper (the testable heart) — `core/metadata/canonical_version.py`
```
score_release_against_files(file_tracks, release_tracks) -> float
pick_canonical_release(file_tracks, candidates) -> (best, score) | (None, 0)
```
- `file_tracks`: list of {duration_ms, title, track_number?} read from disk.
- `release_tracks`: a candidate release's tracklist (same shape).
- Scoring (tunable weights):
- **track-count fit** — exact match strongly preferred; |Δcount| penalized.
- **duration alignment** — greedily match each file to its closest release
track by duration (within a tolerance, e.g. ±3s); reward coverage.
- **title overlap** — token/fuzzy overlap as a tiebreaker.
- **graceful degradation** — if a source gives no per-track durations, fall
back to count + title only (never crash, never force-pick).
- Returns the best candidate + score, or (None, 0) when nothing clears a floor
(so we never pin a bad guess — leave it unresolved, consumers fall back).
### Tests (extreme, like the rest of this codebase)
- standard (11) vs deluxe (17) with 11 files on disk → picks standard.
- same album, 17 files → picks deluxe.
- duration disambiguation when track counts tie (e.g. radio edit vs album).
- missing-duration source → count+title fallback still picks sanely.
- no candidate clears the floor → (None, 0).
- "Believer" standard(=track 3 listing) vs Spotify(=4) with the user's files →
whichever the files actually match.
**End of Stage 1: scorer exists + tested, columns exist, NOTHING reads/writes
them yet. Provably zero behavior change.**
---
## Stage 2 — Resolver populates canonical (writes, still no consumers)
A function `resolve_canonical_for_album(album_id, db, ...)`:
1. Gather on-disk file metadata for the album (durations/titles) via the
library's known file paths.
2. Gather candidate releases: every source the album has an ID for
(spotify/itunes/deezer/discogs/soul/musicbrainz) AND — for the deluxe/standard
case — sibling editions discoverable from those. Fetch each tracklist
(cached, rate-limited).
3. `pick_canonical_release(files, candidates)` → store `(source, album_id, score)`
on the album row if it clears the floor.
Wiring: a small **backfill repair job** (dry-run-capable) + a hook in enrichment
when an album is (re)enriched. Still **no tool READS canonical**, so behavior is
unchanged — this stage only populates the new columns. Reversible: clearing the
columns reverts to unresolved.
Tests: resolver picks the right release for the standard/deluxe fixtures; stores
nothing when below floor; idempotent re-resolve.
Cost note: fetching multiple candidate releases = more API calls. Mitigate via
cache + only-on-(re)enrich + the existing rate trackers. Surface in the job's
progress so it's not silent.
---
## Stage 3 — Reorganizer reads canonical (first real behavior change, gated)
In `library_reorganize._resolve_source`: if the album has
`canonical_source`/`canonical_album_id`, use THAT first; else fall back to the
current `get_source_priority` walk. One-line precedence change, fully gated on
non-NULL.
Tests: with canonical set → resolves to it; with canonical NULL → byte-identical
to today. Re-run the existing reorganize battery (148 tests) — must stay green.
**This alone fixes #767-Bug2** (a standard album whose files match the standard
release pins the standard, so reorganize stops targeting the deluxe folder).
---
## Stage 4 — Track Number Repair reads canonical (closes #765)
In `track_number_repair._resolve_album_tracklist`: add **Fallback -1** (before
everything) — if the album has a canonical `(source, album_id)`, use it. The
existing 6-level cascade stays as the fallback for albums with no canonical
(preserves its all-01-album rescue ability — the regression risk we refused to
take in the reactive fix).
Now both tools resolve the SAME release → same track numbers → no contradiction.
Tests: canonical present → both tools agree (shared-release test); canonical
NULL → existing cascade unchanged.
---
## Risks & mitigations
- **Extra API calls** (Stage 2 fetches multiple releases) → cache, rate-limit,
only-on-(re)enrich, progress-logged.
- **Sources without per-track durations** → scorer degrades to count+title.
- **Schema migration** → additive nullable columns only; idempotent guards.
- **Wrong pick** → floor gate (never pin a low-confidence guess); `canonical_score`
stored for inspection/re-resolve; manual override possible later.
- **Backward-compat** → every consumer falls back to today's path when NULL, so
un-resolved albums (incl. all existing albums until backfilled) are unaffected.
## Out of scope (for now)
- Per-album manual version override UI (can layer on later — the columns support it).
- Merging the two tools into one (the reporter's alt suggestion) — unnecessary
once they share the canonical.
## Suggested order to build
1, then 2, then 3, then 4 — each shippable and verifiable on its own. We can stop
after any stage and the app is consistent (just with fewer consumers wired).

19
core/library_reorganize.py
View file

@ -227,6 +227,25 @@ def _resolve_source(album_data: dict, primary_source: str, strict_source: bool =
"""
source_ids = _extract_source_ids(album_data)
# #765: if a canonical release was pinned for this album (best-fit to the
# user's actual files), prefer it — so reorganize agrees with Track Number
# Repair and stops mislabelling standard albums as deluxe (#767-Bug2). Gated
# on the album row carrying a canonical, and skipped when the user explicitly
# picked a source in the modal (strict_source) — their choice wins. Falls
# through to the priority walk if the canonical fetch fails.
if not strict_source:
c_source = album_data.get('canonical_source')
c_id = album_data.get('canonical_album_id')
if c_source and c_id:
try:
api_album = get_album_for_source(c_source, c_id)
api_tracks = get_album_tracks_for_source(c_source, c_id)
items = _normalize_album_tracks(api_tracks)
if items and api_album:
return c_source, api_album, items
except Exception as e:
logger.warning(f"[Reorganize] canonical {c_source} lookup raised: {e}")
if strict_source:
sources_to_try = [primary_source] if primary_source else []
else:

281
core/metadata/canonical_resolver.py Normal file
View file

@ -0,0 +1,281 @@
"""Resolve (and persist) the canonical release for an album — Stage 2 of #765.
Stage 1 gave us the pure scorer (``core.metadata.canonical_version``). This
module turns it into an end-to-end resolver: gather the album's candidate
releases (one per metadata-source ID it has), score each against the on-disk
files, and return the best fit. Wiring (backfill job / enrichment hook) and the
DB store live alongside; the decision logic here is kept dependency-injected
(``fetch_tracklist`` is passed in) so it's fully unit-testable without live APIs
or real files.
Still NO consumer reads the result in Stage 2 — populating the columns is
behavior-neutral. Stages 3-4 wire the Reorganizer and Track Number Repair to
read it.
"""
from __future__ import annotations
from typing import Any, Callable, Dict, List, Optional
from core.metadata.canonical_version import (
score_release_against_files,
score_release_detail,
)
# Source-selection modes (a per-job setting). See resolve_canonical_for_album.
MODE_ACTIVE_PREFERRED = "active_preferred" # default: use the active source if it fits, else best-fit
MODE_ACTIVE_ONLY = "active_only" # only ever the active source
MODE_BEST_FIT = "best_fit" # whichever source fits the files best
VALID_MODES = (MODE_ACTIVE_PREFERRED, MODE_ACTIVE_ONLY, MODE_BEST_FIT)
def resolve_canonical_for_album(
*,
album_source_ids: Dict[str, str],
file_tracks: List[Dict[str, Any]],
fetch_tracklist: Callable[[str, str], Optional[List[Dict[str, Any]]]],
source_priority: List[str],
min_score: float = 0.5,
mode: str = MODE_ACTIVE_PREFERRED,
primary_source: Optional[str] = None,
) -> Optional[Dict[str, Any]]:
"""Pick the canonical release for one album, honoring the source-selection mode.
``album_source_ids``: ``{source: album_id}`` the album is linked to.
``file_tracks``: on-disk track metadata (``{duration_ms, title}``).
``fetch_tracklist(source, album_id)``: returns that release's tracklist (or
None/[] on miss); injected so callers supply ``get_album_tracks_for_source``
while tests supply a fake.
``source_priority``: source order; ties break toward the earlier source.
``primary_source``: the user's active metadata source (defaults to the first
of ``source_priority``).
Modes:
- ``active_preferred`` (default): use the active source's release when the
album has an ID for it AND it clears ``min_score``; otherwise fall back
to the best-fit among the remaining sources. So it normally respects the
user's configured source but self-heals when that link is clearly wrong.
- ``active_only``: only ever the active source (pinned if it clears the
floor; never considers other sources).
- ``best_fit``: whichever source's release best matches the files.
Returns an enriched dict for the chosen release — ``source``, ``album_id``,
``score``, the per-signal breakdown (``count_fit``/``duration_fit``/
``title_fit``), ``file_track_count`` vs ``release_track_count``, and a
``candidates`` list of everything it scored (so a finding can show WHY the
pick won and what it beat). ``None`` when there are no files, no resolvable
candidates, or nothing clears ``min_score``."""
if not file_tracks:
return None
primary = primary_source or (source_priority[0] if source_priority else None)
scored: List[Dict[str, Any]] = [] # every source we actually scored
def _score(source: Optional[str]) -> Optional[Dict[str, Any]]:
if not source or any(e['source'] == source for e in scored):
return next((e for e in scored if e['source'] == source), None)
album_id = album_source_ids.get(source)
if not album_id:
return None
try:
tracks = fetch_tracklist(source, str(album_id))
except Exception:
tracks = None
if not tracks:
return None
entry = {
'source': source, 'album_id': str(album_id),
'track_count': len(tracks), 'score': round(score_release_against_files(file_tracks, tracks), 4),
'_tracks': tracks,
}
scored.append(entry)
return entry
winner: Optional[Dict[str, Any]] = None
# Active-source modes: try the primary first.
if mode in (MODE_ACTIVE_ONLY, MODE_ACTIVE_PREFERRED):
p = _score(primary)
if p and p['score'] >= min_score:
winner = p
elif mode == MODE_ACTIVE_ONLY:
return None # never consider other sources
# best_fit, or active_preferred fallback: score the rest and pick the best.
if winner is None:
for source in source_priority:
_score(source)
best = None
for e in scored: # source_priority order -> strictly-greater = priority tiebreak
if best is None or e['score'] > best['score'] + 1e-9:
best = e
if best and best['score'] >= min_score:
winner = best
if winner is None:
return None
detail = score_release_detail(file_tracks, winner['_tracks'])
# Pinned-release track titles — already fetched, so free. Capped so a giant
# box set can't bloat the finding's details_json.
release_titles = [
(t.get('title') or t.get('name') or '') for t in winner['_tracks']
][:60]
return {
'source': winner['source'],
'album_id': winner['album_id'],
'score': winner['score'],
'file_track_count': detail['file_track_count'],
'release_track_count': detail['release_track_count'],
'count_fit': detail['count_fit'],
'duration_fit': detail['duration_fit'],
'title_fit': detail['title_fit'],
'release_track_titles': release_titles,
'candidates': [
{'source': e['source'], 'album_id': e['album_id'],
'track_count': e['track_count'], 'score': e['score']}
for e in scored
],
}
def _item_get(item: Any, key: str, default: Any = None) -> Any:
"""Read ``key`` from a track item that may be a dict or an object."""
return item.get(key, default) if isinstance(item, dict) else getattr(item, key, default)
def default_fetch_tracklist(source: str, album_id: str) -> Optional[List[Dict[str, Any]]]:
"""Production ``fetch_tracklist``: pull a release's tracklist from a metadata
source and normalise to ``{title, track_number, duration_ms}``. Duration is
best-effort (not every source exposes it); when absent the scorer just leans
on track-count + title. Returns None on any failure."""
try:
from core.metadata_service import get_album_tracks_for_source
data = get_album_tracks_for_source(source, album_id)
except Exception:
return None
items = data if isinstance(data, list) else (
(data.get('items') or data.get('tracks') or []) if isinstance(data, dict) else []
)
if isinstance(items, dict): # {'tracks': {'items': [...]}}
items = items.get('items') or []
out: List[Dict[str, Any]] = []
for it in items:
dur = _item_get(it, 'duration_ms')
if dur is None:
secs = _item_get(it, 'duration') # some sources give seconds
dur = int(secs * 1000) if isinstance(secs, (int, float)) and secs else None
out.append({
'title': _item_get(it, 'name') or _item_get(it, 'title') or '',
'track_number': _item_get(it, 'track_number'),
'duration_ms': dur,
})
return out or None
def _lookup_artist_thumb(db, artist_id) -> Optional[str]:
"""Best-effort artist thumb URL by id. Returns None on missing column / any
error (the artists table doesn't have thumb_url in every schema)."""
if not artist_id:
return None
conn = None
try:
conn = db._get_connection()
cursor = conn.cursor()
cursor.execute("PRAGMA table_info(artists)")
if 'thumb_url' not in {r[1] for r in cursor.fetchall()}:
return None
cursor.execute("SELECT thumb_url FROM artists WHERE id = ?", (str(artist_id),))
row = cursor.fetchone()
return (row[0] or None) if row else None
except Exception:
return None
finally:
if conn:
conn.close()
def resolve_and_store_canonical_for_album(
db,
album_id,
*,
fetch_tracklist: Optional[Callable[[str, str], Any]] = None,
source_priority: Optional[List[str]] = None,
min_score: float = 0.5,
store: bool = True,
mode: str = MODE_ACTIVE_PREFERRED,
) -> Optional[Dict[str, Any]]:
"""Gather an album's source IDs + its tracks' (duration, title) from the DB,
resolve the best-fit canonical release, and (when ``store``) persist it.
Returns the resolved ``{source, album_id, score}`` or None when unresolved.
``store=False`` resolves without writing — used by the backfill job's dry run.
Uses the SAME album/source-id loader the Reorganizer uses
(``load_album_and_tracks`` + ``_extract_source_ids``) so the canonical is
chosen over exactly the source IDs the reorganizer sees. Scores off the DB
track rows' ``duration`` (stored in ms) + ``title`` — the library's view of
the files — so no per-file disk reads are needed."""
from core.library_reorganize import _extract_source_ids, load_album_and_tracks
album_data, tracks = load_album_and_tracks(db, album_id)
if not album_data or not tracks:
return None
source_ids = {s: v for s, v in _extract_source_ids(album_data).items() if v}
if not source_ids:
return None
file_tracks = [
{'duration_ms': t.get('duration') or 0, 'title': t.get('title') or ''}
for t in tracks
]
if fetch_tracklist is None:
fetch_tracklist = default_fetch_tracklist
primary_source = None
if source_priority is None:
try:
from core.metadata_service import get_primary_source, get_source_priority
primary_source = get_primary_source()
source_priority = get_source_priority(primary_source)
except Exception:
source_priority = list(source_ids.keys())
result = resolve_canonical_for_album(
album_source_ids=source_ids,
file_tracks=file_tracks,
fetch_tracklist=fetch_tracklist,
source_priority=source_priority,
min_score=min_score,
mode=mode,
primary_source=primary_source,
)
if result:
# Album/artist/art context for richer findings (read from the row we
# already loaded — no extra query). Storage only uses source/id/score.
result['album_title'] = album_data.get('title') or ''
result['artist_name'] = album_data.get('artist_name') or ''
# Free context off the album row + the data we already gathered.
if album_data.get('year'):
result['year'] = album_data['year']
result['db_track_count'] = album_data.get('track_count') or len(file_tracks)
if album_data.get('duration'):
result['db_duration_ms'] = album_data['duration']
result['linked_sources'] = source_ids # {source: album_id} the album points at now
result['file_track_titles'] = [ft.get('title') or '' for ft in file_tracks][:60]
if album_data.get('thumb_url'):
result['album_thumb_url'] = album_data['thumb_url']
# Artist thumb via a guarded lookup (not the shared album loader — some
# schemas have no artists.thumb_url column). Only runs for resolved
# albums, so no cost on the no-source-id short-circuit majority.
artist_thumb = _lookup_artist_thumb(db, album_data.get('artist_id'))
if artist_thumb:
result['artist_thumb_url'] = artist_thumb
if store:
db.set_album_canonical(album_id, result['source'], result['album_id'], result['score'])
return result
__all__ = [
"resolve_canonical_for_album",
"resolve_and_store_canonical_for_album",
"default_fetch_tracklist",
]

210
core/metadata/canonical_version.py Normal file
View file

@ -0,0 +1,210 @@
"""Pick the canonical album release by best-fit to the user's actual files.
Issue #765 / #767-Bug2: SoulSync never pins ONE canonical album version per
album, so the Library Reorganizer, Track Number Repair, and tagging each
re-resolve independently and can land on different releases (standard vs
deluxe; Spotify vs MusicBrainz track numbering) and contradict each other.
This module is the pure, testable heart of the fix: given the metadata of the
files actually on disk and a set of candidate releases, score each release by
how well it FITS those files and pick the best. "Best-fit to the files" means:
- track-count fit — a 17-track deluxe is a poor fit for 11 files on disk
- duration alignment — each file should line up with a release track by length
- title overlap — a tiebreaker / sanity check
What this does and does NOT solve:
- It DOES pick the right EDITION (standard vs deluxe) — the discriminating
signal is track count + durations.
- It does NOT (and cannot) decide which of two listings of the SAME album is
"more correct" when they differ only in track numbering (same files match
both equally). Instead ``pick_canonical_release`` is DETERMINISTIC and
breaks ties toward the earlier candidate — so the caller passes candidates
in source-priority order and every tool that reads the pinned result agrees
on the same release. Agreement is what resolves #765, not picking a
"winner" of the numbering disagreement.
Pure, no I/O. Callers fetch candidate tracklists and read on-disk file metadata;
this module only scores.
"""
from __future__ import annotations
import re
from difflib import SequenceMatcher
from typing import Any, Dict, List, Optional, Tuple
# Weights for the three fit signals. Count + duration dominate because "matches
# my files" is fundamentally about having the right NUMBER of the right-LENGTH
# tracks; title is a tiebreaker. Missing signals are dropped and the present
# ones renormalized (see _combine).
_W_COUNT = 0.4
_W_DURATION = 0.4
_W_TITLE = 0.2
_DEFAULT_DURATION_TOLERANCE_MS = 3000 # ±3s — covers encode/version length jitter
_DEFAULT_MIN_SCORE = 0.5 # never pin below this — leave unresolved
_TITLE_FUZZY_THRESHOLD = 0.85
def _norm_title(text: str) -> str:
"""Lowercase, drop bracketed qualifiers ((feat. …), [Remastered]), strip
punctuation, collapse whitespace."""
if not text:
return ""
t = str(text).lower()
t = re.sub(r"[\(\[].*?[\)\]]", "", t)
t = re.sub(r"[^a-z0-9 ]", " ", t)
return " ".join(t.split())
def _count_fit(n_files: int, n_release: int) -> float:
"""1.0 when track counts match; decays with the relative difference."""
if n_files <= 0 or n_release <= 0:
return 0.0
return 1.0 - min(1.0, abs(n_files - n_release) / max(n_files, n_release))
def _duration_fit(
file_tracks: List[Dict[str, Any]],
release_tracks: List[Dict[str, Any]],
tolerance_ms: int,
) -> Optional[float]:
"""Fraction of tracks that line up by duration (greedy nearest match within
tolerance), over the larger of the two track counts — so missing or extra
tracks are penalised. Returns ``None`` when neither side has durations."""
f_durs = [int(f["duration_ms"]) for f in file_tracks if f.get("duration_ms")]
r_durs = [int(r["duration_ms"]) for r in release_tracks if r.get("duration_ms")]
if not f_durs or not r_durs:
return None
used = [False] * len(r_durs)
matched = 0
for fd in f_durs:
best_j, best_diff = -1, tolerance_ms + 1
for j, rd in enumerate(r_durs):
if used[j]:
continue
diff = abs(fd - rd)
if diff <= tolerance_ms and diff < best_diff:
best_diff, best_j = diff, j
if best_j >= 0:
used[best_j] = True
matched += 1
denom = max(len(file_tracks), len(release_tracks))
return matched / denom if denom else 0.0
def _title_fit(
file_tracks: List[Dict[str, Any]],
release_tracks: List[Dict[str, Any]],
) -> Optional[float]:
"""Fraction of files whose title matches some release title (exact-normalised
or fuzzy), over the larger track count. ``None`` when titles are absent."""
f_titles = [_norm_title(f.get("title", "")) for f in file_tracks]
f_titles = [t for t in f_titles if t]
r_titles = [_norm_title(r.get("title", "")) for r in release_tracks]
r_titles = [t for t in r_titles if t]
if not f_titles or not r_titles:
return None
r_set = set(r_titles)
matched = 0
for ft in f_titles:
if ft in r_set or any(
SequenceMatcher(None, ft, rt).ratio() >= _TITLE_FUZZY_THRESHOLD
for rt in r_titles
):
matched += 1
denom = max(len(file_tracks), len(release_tracks))
return matched / denom if denom else 0.0
def _combine(parts: List[Tuple[Optional[float], float]]) -> float:
"""Weighted mean over present (non-None) components, renormalising weights."""
present = [(v, w) for v, w in parts if v is not None]
total_w = sum(w for _, w in present)
if total_w <= 0:
return 0.0
return sum(v * w for v, w in present) / total_w
def score_release_against_files(
file_tracks: List[Dict[str, Any]],
release_tracks: List[Dict[str, Any]],
*,
duration_tolerance_ms: int = _DEFAULT_DURATION_TOLERANCE_MS,
) -> float:
"""Score 0.0–1.0 of how well ``release_tracks`` fits the on-disk
``file_tracks``. Each track dict may carry ``duration_ms`` and ``title``;
missing signals are dropped and the rest renormalised so the function never
crashes on sparse metadata (it just leans on what's available)."""
if not file_tracks or not release_tracks:
return 0.0
count = _count_fit(len(file_tracks), len(release_tracks))
dur = _duration_fit(file_tracks, release_tracks, duration_tolerance_ms)
title = _title_fit(file_tracks, release_tracks)
return _combine([(count, _W_COUNT), (dur, _W_DURATION), (title, _W_TITLE)])
def score_release_detail(
file_tracks: List[Dict[str, Any]],
release_tracks: List[Dict[str, Any]],
*,
duration_tolerance_ms: int = _DEFAULT_DURATION_TOLERANCE_MS,
) -> Dict[str, Any]:
"""Like ``score_release_against_files`` but returns the per-signal breakdown
so a UI can show WHY a release scored the way it did. ``duration_fit`` /
``title_fit`` are ``None`` when that signal was absent."""
if not file_tracks or not release_tracks:
return {
'score': 0.0, 'count_fit': 0.0, 'duration_fit': None, 'title_fit': None,
'release_track_count': len(release_tracks), 'file_track_count': len(file_tracks),
}
count = _count_fit(len(file_tracks), len(release_tracks))
dur = _duration_fit(file_tracks, release_tracks, duration_tolerance_ms)
title = _title_fit(file_tracks, release_tracks)
score = _combine([(count, _W_COUNT), (dur, _W_DURATION), (title, _W_TITLE)])
return {
'score': round(score, 4),
'count_fit': round(count, 3),
'duration_fit': round(dur, 3) if dur is not None else None,
'title_fit': round(title, 3) if title is not None else None,
'release_track_count': len(release_tracks),
'file_track_count': len(file_tracks),
}
def pick_canonical_release(
file_tracks: List[Dict[str, Any]],
candidates: List[Dict[str, Any]],
*,
min_score: float = _DEFAULT_MIN_SCORE,
duration_tolerance_ms: int = _DEFAULT_DURATION_TOLERANCE_MS,
) -> Tuple[Optional[Dict[str, Any]], float]:
"""Choose the best-fit candidate release for the on-disk files.
``candidates`` is a list of dicts each with a ``'tracks'`` list (plus any
caller fields like ``source``/``album_id``, returned untouched). **Pass
candidates in source-priority order** — ties break toward the EARLIER one,
so the choice is deterministic and priority-respecting (this is what makes
every tool agree, #765).
Returns ``(best_candidate, score)``, or ``(None, best_score)`` when nothing
clears ``min_score`` — so a low-confidence guess is never pinned (the caller
leaves the album unresolved and falls back to today's behaviour)."""
best: Optional[Dict[str, Any]] = None
best_score = 0.0
for cand in candidates:
score = score_release_against_files(
file_tracks, cand.get("tracks") or [],
duration_tolerance_ms=duration_tolerance_ms,
)
# Strictly-greater so equal scores keep the earlier (higher-priority)
# candidate — deterministic tiebreak.
if score > best_score + 1e-9:
best, best_score = cand, score
if best is None or best_score < min_score:
return None, best_score
return best, best_score
__all__ = ["score_release_against_files", "pick_canonical_release"]

1
core/repair_jobs/__init__.py
View file

@ -44,6 +44,7 @@ _JOB_MODULES = [
'core.repair_jobs.live_commentary_cleaner',
'core.repair_jobs.unknown_artist_fixer',
'core.repair_jobs.discography_backfill',
'core.repair_jobs.canonical_version_resolve',
]

3
core/repair_jobs/base.py
View file

@ -100,6 +100,9 @@ class RepairJob(ABC):
default_enabled: bool = False
default_interval_hours: int = 24
default_settings: Dict[str, Any] = {}
# Optional {setting_key: [allowed values]} — the UI renders a dropdown for
# these instead of a free-text box. Keys not listed render by value type.
setting_options: Dict[str, list] = {}
auto_fix: bool = False
@abstractmethod

221
core/repair_jobs/canonical_version_resolve.py Normal file
View file

@ -0,0 +1,221 @@
"""Resolve Canonical Album Versions — backfill job (#765 Stage 2 trigger).
Pins each album's canonical release (best-fit to its files) so the Library
Reorganizer (Stage 3) and Track Number Repair (Stage 4) resolve the SAME
release and stop contradicting each other. The resolution logic lives in the
tested core.metadata.canonical_resolver; this job is the opt-in, rate-limited,
progress-reported bulk runner.
Opt-in (``default_enabled = False``) because resolving compares an album's
candidate releases across sources, which costs metadata-source API calls — done
once per album, then stored. Albums that already have a canonical are skipped.
"""
import os
from typing import Optional
from core.metadata.canonical_resolver import resolve_and_store_canonical_for_album
from core.repair_jobs import register_job
from core.repair_jobs.base import JobContext, JobResult, RepairJob
from utils.logging_config import get_logger
logger = get_logger("repair_job.canonical_version")
def _pct(v) -> str:
return f"{round(v * 100)}%" if isinstance(v, (int, float)) else "n/a"
def _describe_pin(resolved: dict) -> str:
"""Human-readable, judge-able explanation of WHY this release was chosen."""
artist = resolved.get('artist_name') or ''
album = resolved.get('album_title') or ''
head = f"{artist} — {album}".strip(" —") or resolved.get('album_id', '')
year = resolved.get('year')
if year:
head += f" ({year})"
lines = [
f"{head}" if head else "",
f"Pin {resolved['source']} release {resolved['album_id']} "
f"(confidence {_pct(resolved.get('score'))}).",
f"Fit to your library: {resolved.get('file_track_count', '?')} files vs "
f"{resolved.get('release_track_count', '?')} tracks on this release — "
f"track count {_pct(resolved.get('count_fit'))}, "
f"durations {_pct(resolved.get('duration_fit'))}, "
f"titles {_pct(resolved.get('title_fit'))}.",
]
# What the album is currently linked to vs what we'd pin.
linked = resolved.get('linked_sources') or {}
if linked:
linked_str = ", ".join(f"{s}={i}" for s, i in linked.items())
lines.append(f"Currently linked: {linked_str} → pinning {resolved['source']}.")
others = [c for c in resolved.get('candidates', []) if c.get('source') != resolved.get('source')]
if others:
comp = ", ".join(
f"{c['source']} {_pct(c['score'])} ({c['track_count']} tk)" for c in others
)
lines.append(f"Beat: {comp}.")
elif len(resolved.get('candidates', [])) == 1:
lines.append("Only this source had a release linked for this album.")
# Track listing of the pinned release (so you can eyeball the actual songs).
titles = resolved.get('release_track_titles') or []
if titles:
shown = "; ".join(f"{i+1}. {t}" for i, t in enumerate(titles[:25]))
more = f" (+{len(titles) - 25} more)" if len(titles) > 25 else ""
lines.append(f"Release tracks: {shown}{more}")
return "\n".join(l for l in lines if l)
@register_job
class CanonicalVersionResolveJob(RepairJob):
job_id = 'canonical_version_resolve'
display_name = 'Resolve Canonical Album Versions'
description = (
'Pins the best-fit release per album (by track count + durations) so '
'reorganize and track-number repair agree (dry run by default)'
)
help_text = (
'For each album, compares the releases its linked metadata sources point '
'at and pins the one that best matches the files you actually have '
'(track count + durations + titles). The Library Reorganizer and Track '
'Number Repair then both use that pinned release, so they stop '
'contradicting each other (e.g. standard vs deluxe, or Spotify vs '
'MusicBrainz track numbering).\n\n'
'In dry run mode (default) it reports what it would pin without saving. '
'Disable dry run to store the pins. Albums already pinned are skipped.\n\n'
'Opt-in: resolving costs metadata-source API calls (once per album).'
)
icon = 'repair-icon-tracknumber'
default_enabled = False
default_interval_hours = 168 # weekly, but disabled by default
default_settings = {
'dry_run': True,
'min_score': 0.5,
# Which source's release to pin: 'active_preferred' (default — use your
# active metadata source when it fits, else best-fit fallback),
# 'active_only' (only ever the active source), or 'best_fit' (whichever
# source matches the files best, regardless of which it is).
'source_selection': 'active_preferred',
}
# Render source_selection as a dropdown (not a text box) in the settings UI.
setting_options = {
'source_selection': ['active_preferred', 'active_only', 'best_fit'],
}
auto_fix = True
def _get_settings(self, context: JobContext) -> dict:
merged = dict(self.default_settings)
if context.config_manager:
merged.update(context.config_manager.get(f'repair.jobs.{self.job_id}.settings', {}) or {})
return merged
def _load_album_ids(self, db, active_server: Optional[str]) -> list:
conn = None
try:
conn = db._get_connection()
cursor = conn.cursor()
if active_server:
cursor.execute(
"SELECT al.id, al.title FROM albums al WHERE al.server_source = ? ORDER BY al.id",
(active_server,),
)
else:
cursor.execute("SELECT al.id, al.title FROM albums al ORDER BY al.id")
return [(row[0], row[1]) for row in cursor.fetchall()]
except Exception as e:
logger.error("Error loading albums for canonical resolve: %s", e)
return []
finally:
if conn:
conn.close()
def scan(self, context: JobContext) -> JobResult:
result = JobResult()
settings = self._get_settings(context)
dry_run = settings.get('dry_run', True)
min_score = settings.get('min_score', 0.5)
mode = settings.get('source_selection', 'active_preferred')
active_server = None
if context.config_manager:
try:
active_server = context.config_manager.get_active_media_server()
except Exception as e:
logger.warning("Couldn't read active media server: %s", e)
albums = self._load_album_ids(context.db, active_server)
total = len(albums)
if context.report_progress:
mode = 'DRY RUN' if dry_run else 'LIVE'
context.report_progress(
phase=f'Resolving canonical versions for {total} albums ({mode})...',
total=total, scanned=0, log_type='info',
)
for i, (album_id, album_title) in enumerate(albums):
if context.check_stop():
return result
if i % 20 == 0 and context.wait_if_paused():
return result
# Skip albums already pinned — one-time cost per album.
try:
if context.db.get_album_canonical(album_id):
result.skipped += 1
result.scanned += 1
continue
except Exception: # noqa: S110 — best-effort skip check; on read error just resolve it
pass
try:
resolved = resolve_and_store_canonical_for_album(
context.db, album_id, min_score=min_score, store=not dry_run, mode=mode,
)
except Exception as e:
logger.warning("Canonical resolve failed for album %s ('%s'): %s",
album_id, album_title, e)
result.errors += 1
result.scanned += 1
continue
result.scanned += 1
if resolved:
if dry_run and context.create_finding:
artist = resolved.get('artist_name') or ''
label = f"{artist} — {album_title}" if artist else (album_title or str(album_id))
inserted = context.create_finding(
job_id=self.job_id,
finding_type='canonical_version',
severity='info',
entity_type='album',
entity_id=str(album_id),
file_path=None,
title=f'Pin {resolved["source"]} as canonical: {label}',
description=_describe_pin(resolved),
details={'album_id': str(album_id), **resolved},
)
if inserted:
result.findings_created += 1
else:
result.findings_skipped_dedup += 1
elif not dry_run:
result.auto_fixed += 1
if context.report_progress and (i + 1) % 25 == 0:
context.report_progress(scanned=i + 1, total=total,
phase=f'Resolving ({i+1}/{total})...')
return result
def estimate_scope(self, context: JobContext) -> int:
active_server = None
if context.config_manager:
try:
active_server = context.config_manager.get_active_media_server()
except Exception: # noqa: S110 — best-effort; fall back to no server filter
pass
return len(self._load_album_ids(context.db, active_server))

56
core/repair_jobs/track_number_repair.py
View file

@ -275,6 +275,22 @@ class TrackNumberRepairJob(RepairJob):
primary_source = get_primary_source()
source_priority = get_source_priority(primary_source)
# Fallback -1 (#765): a pinned canonical release wins over the whole
# cascade below — so Track Number Repair resolves the SAME release the
# Reorganizer does (Stage 3) and the two stop contradicting each other.
# Gated on the album carrying a canonical; everything below is untouched
# for albums without one (preserving the all-01-album rescue this job
# exists for — the regression we refused to take in a reactive fix).
canonical = _lookup_canonical_from_db(file_track_data, context)
if canonical:
c_source, c_id = canonical
if _is_valid_album_id(c_id):
tracks = _get_album_tracklist(c_source, c_id, cache)
if tracks:
logger.info("[Repair] %s — resolved via canonical %s album ID: %s",
folder_name, c_source, c_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)
@ -806,6 +822,46 @@ def _update_db_file_path(db, old_path: str, new_path: str):
conn.close()
def _lookup_canonical_from_db(file_track_data: List[Tuple[str, str, Any]],
context: JobContext) -> Optional[Tuple[str, str]]:
"""Return the album's pinned canonical ``(source, album_id)`` or None.
#765: when the album this folder's files belong to has a canonical release
pinned (best-fit to the files), Track Number Repair uses it first so it
agrees with the Reorganizer. Resolves by matching a file path to its DB
track row. None when no DB, no match, columns absent, or unresolved."""
if not context.db:
return None
conn = None
try:
conn = context.db._get_connection()
cursor = conn.cursor()
cursor.execute("PRAGMA table_info(albums)")
cols = {row[1] for row in cursor.fetchall()}
if 'canonical_source' not in cols or 'canonical_album_id' not in cols:
return None
for fpath, _, _ in file_track_data:
cursor.execute(
"""
SELECT al.canonical_source, al.canonical_album_id
FROM tracks t
JOIN albums al ON al.id = t.album_id
WHERE t.file_path = ?
LIMIT 1
""",
(fpath,),
)
row = cursor.fetchone()
if row and row[0] and row[1]:
return (str(row[0]), str(row[1]))
except Exception as e:
logger.debug("Error looking up canonical from DB: %s", e)
finally:
if conn:
conn.close()
return None
def _lookup_album_ids_from_db(file_track_data: List[Tuple[str, str, Any]],
context: JobContext) -> Dict[str, Optional[str]]:
"""Look up album IDs from the database using file paths.

3
core/repair_worker.py
View file

@ -371,6 +371,9 @@ class RepairWorker:
'interval_hours': config['interval_hours'],
'settings': config['settings'],
'default_settings': job.default_settings.copy(),
# Per-setting choice lists so the UI can render a dropdown
# instead of a free-text box (e.g. canonical source_selection).
'setting_options': dict(getattr(job, 'setting_options', {}) or {}),
'last_run': last_run,
'next_run': next_run,
'is_running': self._current_job_id == job_id,

62
database/music_database.py
View file

@ -956,9 +956,71 @@ class MusicDatabase:
if album_cols and 'api_track_count' not in album_cols:
cursor.execute("ALTER TABLE albums ADD COLUMN api_track_count INTEGER DEFAULT NULL")
logger.info("Repaired missing api_track_count column on albums table")
# Canonical album version (#765 / #767-Bug2). Additive + nullable:
# a NULL canonical means "unresolved" and every tool falls back to
# today's behavior, so this is safe to ship dormant. Columns are
# populated/consumed in later stages.
_canonical_cols = {
'canonical_source': 'TEXT DEFAULT NULL',
'canonical_album_id': 'TEXT DEFAULT NULL',
'canonical_score': 'REAL DEFAULT NULL',
'canonical_resolved_at': 'TIMESTAMP DEFAULT NULL',
}
for _col, _typedef in _canonical_cols.items():
if album_cols and _col not in album_cols:
cursor.execute(f"ALTER TABLE albums ADD COLUMN {_col} {_typedef}")
logger.info("Added %s column to albums table (canonical version)", _col)
except Exception as e:
logger.error("Error repairing core media schema columns: %s", e)
def set_album_canonical(self, album_id, source: str, canonical_album_id: str, score: float) -> bool:
"""Persist the resolved canonical (source, album_id, score) for an album
(#765 Stage 2). Returns True if a row was updated."""
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute(
"UPDATE albums SET canonical_source = ?, canonical_album_id = ?, "
"canonical_score = ?, canonical_resolved_at = CURRENT_TIMESTAMP "
"WHERE id = ?",
(source, str(canonical_album_id), float(score), album_id),
)
conn.commit()
return cursor.rowcount > 0
except Exception as e:
logger.error("Error setting album canonical for %s: %s", album_id, e)
return False
finally:
conn.close()
def get_album_canonical(self, album_id) -> Optional[dict]:
"""Return ``{'source','album_id','score','resolved_at'}`` for an album's
pinned canonical release, or ``None`` when unresolved (#765 Stage 2).
Consumers treat ``None`` as 'fall back to today's behavior'."""
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute(
"SELECT canonical_source, canonical_album_id, canonical_score, "
"canonical_resolved_at FROM albums WHERE id = ?",
(album_id,),
)
row = cursor.fetchone()
if not row or not row[0] or not row[1]:
return None
return {
'source': row[0],
'album_id': row[1],
'score': row[2],
'resolved_at': row[3],
}
except Exception as e:
logger.error("Error reading album canonical for %s: %s", album_id, e)
return None
finally:
conn.close()
def _add_mirrored_playlist_explored_column(self, cursor):
"""Add explored_at column to mirrored_playlists to persist explore badge."""
try:

69
tests/test_canonical_columns_migration.py Normal file
View file

@ -0,0 +1,69 @@
"""Migration test for the canonical-album-version columns (#765 Stage 1).
Additive + nullable, so it must: appear on a fresh DB, be idempotent (re-running
the migration is a no-op, not an error), and ALTER them onto an older albums
table that lacks them. NULL = unresolved = tools fall back to today's behavior.
"""
from __future__ import annotations
import sqlite3
from database.music_database import MusicDatabase
_CANONICAL_COLS = {
'canonical_source', 'canonical_album_id', 'canonical_score', 'canonical_resolved_at',
}
def _album_cols(cur):
cur.execute("PRAGMA table_info(albums)")
return {c[1] for c in cur.fetchall()}
def test_fresh_db_has_canonical_columns(tmp_path):
db = MusicDatabase(str(tmp_path / "m.db"))
cur = db._get_connection().cursor()
assert _CANONICAL_COLS <= _album_cols(cur)
def test_canonical_columns_default_null(tmp_path):
# Unresolved by default -> every consumer falls back. Verify each canonical
# column declares DEFAULT NULL and is nullable (notnull flag == 0).
db = MusicDatabase(str(tmp_path / "m.db"))
cur = db._get_connection().cursor()
cur.execute("PRAGMA table_info(albums)")
info = {c[1]: c for c in cur.fetchall()} # name -> (cid, name, type, notnull, dflt, pk)
for col in _CANONICAL_COLS:
assert col in info, f"{col} missing"
assert info[col][3] == 0, f"{col} must be nullable"
dflt = info[col][4]
assert dflt is None or str(dflt).upper() == 'NULL', f"{col} default should be NULL"
def test_migration_is_idempotent(tmp_path):
db = MusicDatabase(str(tmp_path / "m.db"))
cur = db._get_connection().cursor()
before = _album_cols(cur)
# Re-running must not raise (the PRAGMA guard skips existing columns).
db._ensure_core_media_schema_columns(cur)
db._ensure_core_media_schema_columns(cur)
assert _album_cols(cur) == before
assert _CANONICAL_COLS <= _album_cols(cur)
def test_migration_adds_columns_to_old_albums_table(tmp_path):
# Simulate an upgraded DB whose albums table predates these columns.
path = str(tmp_path / "old.db")
conn = sqlite3.connect(path)
conn.execute("CREATE TABLE albums (id INTEGER PRIMARY KEY, title TEXT)")
conn.commit()
cur = conn.cursor()
assert not (_CANONICAL_COLS & _album_cols(cur)) # none present yet
# Run the real migration against this old cursor.
db = MusicDatabase(str(tmp_path / "scratch.db"))
db._ensure_core_media_schema_columns(cur)
conn.commit()
assert _CANONICAL_COLS <= _album_cols(cur)

54
tests/test_canonical_db.py Normal file
View file

@ -0,0 +1,54 @@
"""DB persistence for canonical album version (#765 Stage 2)."""
from __future__ import annotations
from database.music_database import MusicDatabase
def _album(db, album_id="alb_evolve"):
# id columns are TEXT (GUID) post-migration, so insert explicit ids and a
# valid FK rather than relying on integer rowids.
conn = db._get_connection()
cur = conn.cursor()
cur.execute("INSERT INTO artists (id, name) VALUES ('art_id', 'Imagine Dragons')")
cur.execute(
"INSERT INTO albums (id, title, artist_id) VALUES (?, 'Evolve', 'art_id')",
(album_id,),
)
conn.commit()
conn.close()
return album_id
def test_set_then_get_roundtrip(tmp_path):
db = MusicDatabase(str(tmp_path / "m.db"))
album_id = _album(db)
assert db.get_album_canonical(album_id) is None # unresolved by default
assert db.set_album_canonical(album_id, "spotify", "sp_evolve_123", 0.97) is True
got = db.get_album_canonical(album_id)
assert got["source"] == "spotify"
assert got["album_id"] == "sp_evolve_123"
assert abs(got["score"] - 0.97) < 1e-6
assert got["resolved_at"] # timestamp populated
def test_get_unresolved_returns_none(tmp_path):
db = MusicDatabase(str(tmp_path / "m.db"))
album_id = _album(db)
assert db.get_album_canonical(album_id) is None
def test_set_overwrites_previous(tmp_path):
db = MusicDatabase(str(tmp_path / "m.db"))
album_id = _album(db)
db.set_album_canonical(album_id, "spotify", "old", 0.6)
db.set_album_canonical(album_id, "musicbrainz", "new", 0.95)
got = db.get_album_canonical(album_id)
assert got["source"] == "musicbrainz" and got["album_id"] == "new"
def test_set_on_missing_album_returns_false(tmp_path):
db = MusicDatabase(str(tmp_path / "m.db"))
assert db.set_album_canonical(999999, "spotify", "x", 0.9) is False

143
tests/test_canonical_orchestration.py Normal file
View file

@ -0,0 +1,143 @@
"""End-to-end orchestration for canonical resolve+store (#765 Stage 2 trigger).
Uses a real temp DB (album + tracks + source IDs) and an INJECTED fetcher, so
the DB gathering + persistence are exercised for real without live APIs.
"""
from __future__ import annotations
from core.metadata.canonical_resolver import (
default_fetch_tracklist,
resolve_and_store_canonical_for_album,
)
from database.music_database import MusicDatabase
STD = [{"duration_ms": 180_000 + i * 10_000, "title": f"Song {i+1}", "track_number": i + 1} for i in range(11)]
DLX = STD + [{"duration_ms": 320_000 + i * 10_000, "title": f"Bonus {i+1}", "track_number": 12 + i} for i in range(6)]
def _seed(db, *, spotify=None, deezer=None, n_files=11):
"""Insert an album (with given source IDs) + n_files tracks whose
durations/titles match the STANDARD release."""
conn = db._get_connection()
cur = conn.cursor()
cur.execute("INSERT INTO artists (id, name) VALUES ('art1', 'Imagine Dragons')")
cur.execute(
"INSERT INTO albums (id, title, artist_id, spotify_album_id, deezer_id) "
"VALUES ('alb1', 'Evolve', 'art1', ?, ?)",
(spotify, deezer),
)
for i in range(n_files):
cur.execute(
"INSERT INTO tracks (id, album_id, artist_id, title, track_number, duration) "
"VALUES (?, 'alb1', 'art1', ?, ?, ?)",
(f"t{i}", f"Song {i+1}", i + 1, 180_000 + i * 10_000),
)
conn.commit()
conn.close()
return "alb1"
def test_resolve_and_store_picks_best_fit_and_persists(tmp_path):
db = MusicDatabase(str(tmp_path / "m.db"))
album_id = _seed(db, spotify="sp_deluxe", deezer="dz_std") # 11 files
table = {("spotify", "sp_deluxe"): DLX, ("deezer", "dz_std"): STD}
out = resolve_and_store_canonical_for_album(
db, album_id,
fetch_tracklist=lambda s, a: table.get((s, a)),
source_priority=["spotify", "deezer"],
mode="best_fit",
)
# best_fit: Deezer's standard matches the 11 files better than Spotify's deluxe.
assert out["source"] == "deezer" and out["album_id"] == "dz_std"
# ...and it was persisted.
stored = db.get_album_canonical(album_id)
assert stored["source"] == "deezer" and stored["album_id"] == "dz_std"
def test_default_mode_prefers_active_source(tmp_path):
# Same setup, but default (active_preferred) mode: primary = spotify, whose
# deluxe still clears the floor -> pinned, even though deezer fits better.
db = MusicDatabase(str(tmp_path / "m.db"))
album_id = _seed(db, spotify="sp_deluxe", deezer="dz_std")
table = {("spotify", "sp_deluxe"): DLX, ("deezer", "dz_std"): STD}
out = resolve_and_store_canonical_for_album(
db, album_id,
fetch_tracklist=lambda s, a: table.get((s, a)),
source_priority=["spotify", "deezer"], # default mode
)
assert out["source"] == "spotify" # active source preferred
def test_result_includes_artist_and_album_context(tmp_path):
db = MusicDatabase(str(tmp_path / "m.db"))
conn = db._get_connection()
cur = conn.cursor()
cur.execute("INSERT INTO artists (id, name, thumb_url) VALUES ('art1', 'Imagine Dragons', 'http://artist.jpg')")
cur.execute(
"INSERT INTO albums (id, title, artist_id, thumb_url, spotify_album_id) "
"VALUES ('alb1', 'Evolve', 'art1', 'http://album.jpg', 'sp1')"
)
for i in range(11):
cur.execute(
"INSERT INTO tracks (id, album_id, artist_id, title, track_number, duration) "
"VALUES (?, 'alb1', 'art1', ?, ?, ?)",
(f"t{i}", f"Song {i+1}", i + 1, 180_000 + i * 10_000),
)
conn.commit()
conn.close()
out = resolve_and_store_canonical_for_album(
db, "alb1", fetch_tracklist=lambda s, a: STD, source_priority=["spotify"],
)
assert out["artist_name"] == "Imagine Dragons"
assert out["album_thumb_url"] == "http://album.jpg"
assert out["artist_thumb_url"] == "http://artist.jpg"
# free context: db track count, linked sources, and both title lists
assert out["db_track_count"] == 11
assert out["linked_sources"] == {"spotify": "sp1"}
assert out["file_track_titles"][0] == "Song 1" and len(out["file_track_titles"]) == 11
assert "Song 1" in out["release_track_titles"]
def test_resolve_returns_none_when_album_has_no_source_ids(tmp_path):
db = MusicDatabase(str(tmp_path / "m.db"))
album_id = _seed(db, spotify=None, deezer=None)
out = resolve_and_store_canonical_for_album(
db, album_id, fetch_tracklist=lambda s, a: STD, source_priority=["spotify"],
)
assert out is None
assert db.get_album_canonical(album_id) is None
def test_resolve_returns_none_for_missing_album(tmp_path):
db = MusicDatabase(str(tmp_path / "m.db"))
out = resolve_and_store_canonical_for_album(
db, "does-not-exist", fetch_tracklist=lambda s, a: STD, source_priority=["spotify"],
)
assert out is None
# ── default_fetch_tracklist normalization (no DB / no live API) ────────────
def test_default_fetcher_normalizes_dict_items(monkeypatch):
import core.metadata_service as ms
monkeypatch.setattr(
ms, "get_album_tracks_for_source",
lambda s, a: [{"name": "A", "track_number": 1, "duration_ms": 200000},
{"title": "B", "track_number": 2, "duration": 210}], # seconds
raising=False,
)
out = default_fetch_tracklist("spotify", "x")
assert out[0] == {"title": "A", "track_number": 1, "duration_ms": 200000}
assert out[1] == {"title": "B", "track_number": 2, "duration_ms": 210_000} # sec->ms
def test_default_fetcher_handles_failure(monkeypatch):
import core.metadata_service as ms
monkeypatch.setattr(
ms, "get_album_tracks_for_source",
lambda s, a: (_ for _ in ()).throw(RuntimeError("boom")), raising=False,
)
assert default_fetch_tracklist("spotify", "x") is None

195
tests/test_canonical_resolver.py Normal file
View file

@ -0,0 +1,195 @@
"""Tests for resolve_canonical_for_album (#765 Stage 2 — injectable core)."""
from __future__ import annotations
from core.metadata.canonical_resolver import resolve_canonical_for_album
STD = [{"duration_ms": 180_000 + i * 10_000, "title": f"Song {i+1}"} for i in range(11)]
DLX = STD + [{"duration_ms": 300_000 + i * 10_000, "title": f"Bonus {i+1}"} for i in range(6)]
PRIORITY = ["spotify", "itunes", "deezer", "musicbrainz"]
def _fetcher(table):
"""fetch_tracklist backed by a {(source, album_id): tracks} table."""
def fetch(source, album_id):
return table.get((source, album_id))
return fetch
def test_best_fit_mode_picks_best_regardless_of_priority():
files = list(STD) # user owns the 11-track standard
table = {
("spotify", "sp_deluxe"): DLX, # spotify (primary) linked to deluxe (17)
("musicbrainz", "mb_std"): STD, # musicbrainz has standard (11)
}
out = resolve_canonical_for_album(
album_source_ids={"spotify": "sp_deluxe", "musicbrainz": "mb_std"},
file_tracks=files,
fetch_tracklist=_fetcher(table),
source_priority=PRIORITY,
mode="best_fit",
)
# best_fit: standard matches the files, deluxe doesn't — fit beats priority.
assert out["source"] == "musicbrainz" and out["album_id"] == "mb_std"
assert out["score"] > 0.9
# ── source-selection modes ────────────────────────────────────────────────
def test_active_preferred_uses_primary_when_it_fits():
files = list(STD)
table = {("spotify", "sp1"): STD, ("musicbrainz", "mb1"): STD} # both fit
out = resolve_canonical_for_album(
album_source_ids={"spotify": "sp1", "musicbrainz": "mb1"},
file_tracks=files, fetch_tracklist=_fetcher(table),
source_priority=PRIORITY, # primary = spotify
) # default mode = active_preferred
assert out["source"] == "spotify"
def test_active_preferred_falls_back_when_primary_clearly_misfits():
files = list(STD) # 11 tracks
table = {
("spotify", "sp_bad"): [{"duration_ms": 60_000, "title": "X"}] * 3, # 3-track, <floor
("musicbrainz", "mb_std"): STD,
}
out = resolve_canonical_for_album(
album_source_ids={"spotify": "sp_bad", "musicbrainz": "mb_std"},
file_tracks=files, fetch_tracklist=_fetcher(table),
source_priority=PRIORITY, mode="active_preferred",
)
# primary spotify scores below floor -> fall back to the fitting source.
assert out["source"] == "musicbrainz"
def test_active_preferred_keeps_primary_even_if_another_fits_better():
files = list(STD)
# primary spotify is a deluxe (decent fit, above floor); musicbrainz is exact.
table = {("spotify", "sp_dlx"): DLX, ("musicbrainz", "mb_std"): STD}
out = resolve_canonical_for_album(
album_source_ids={"spotify": "sp_dlx", "musicbrainz": "mb_std"},
file_tracks=files, fetch_tracklist=_fetcher(table),
source_priority=PRIORITY, mode="active_preferred",
)
# active_preferred respects the active source as long as it clears the floor,
# even though musicbrainz would fit better (use best_fit for that).
assert out["source"] == "spotify"
def test_active_only_pins_primary_and_never_falls_back():
files = list(STD)
# primary spotify is below floor; a perfect musicbrainz exists but is ignored.
table = {
("spotify", "sp_bad"): [{"duration_ms": 60_000, "title": "X"}] * 3,
("musicbrainz", "mb_std"): STD,
}
out = resolve_canonical_for_album(
album_source_ids={"spotify": "sp_bad", "musicbrainz": "mb_std"},
file_tracks=files, fetch_tracklist=_fetcher(table),
source_priority=PRIORITY, mode="active_only",
)
assert out is None # primary didn't fit, and active_only won't consider others
def test_result_includes_breakdown_and_candidate_comparison():
files = list(STD)
table = {("spotify", "sp1"): DLX, ("deezer", "dz1"): STD}
out = resolve_canonical_for_album(
album_source_ids={"spotify": "sp1", "deezer": "dz1"},
file_tracks=files, fetch_tracklist=_fetcher(table),
source_priority=["spotify", "deezer", "itunes", "musicbrainz"],
mode="best_fit",
)
assert out["source"] == "deezer"
assert out["file_track_count"] == 11
assert out["release_track_count"] == 11
assert out["count_fit"] == 1.0 and out["duration_fit"] == 1.0 and out["title_fit"] == 1.0
by_src = {c["source"]: c for c in out["candidates"]}
assert by_src["deezer"]["track_count"] == 11 and by_src["deezer"]["score"] > 0.9
assert by_src["spotify"]["track_count"] == 17 and by_src["spotify"]["score"] < 0.8
def test_active_only_pins_primary_when_it_fits():
files = list(STD)
table = {("spotify", "sp1"): STD, ("musicbrainz", "mb1"): STD}
out = resolve_canonical_for_album(
album_source_ids={"spotify": "sp1", "musicbrainz": "mb1"},
file_tracks=files, fetch_tracklist=_fetcher(table),
source_priority=PRIORITY, mode="active_only",
)
assert out["source"] == "spotify"
def test_priority_breaks_tie_between_equal_fits():
files = list(STD)
table = {("spotify", "a"): STD, ("itunes", "b"): STD} # identical fit
out = resolve_canonical_for_album(
album_source_ids={"itunes": "b", "spotify": "a"},
file_tracks=files,
fetch_tracklist=_fetcher(table),
source_priority=PRIORITY, # spotify before itunes
)
assert out["source"] == "spotify"
def test_skips_sources_without_ids_or_failed_fetch():
files = list(STD)
def fetch(source, album_id):
if source == "spotify":
raise RuntimeError("API down")
if source == "deezer":
return STD
return None
out = resolve_canonical_for_album(
album_source_ids={"spotify": "x", "deezer": "y"}, # no itunes id
file_tracks=files,
fetch_tracklist=fetch,
source_priority=PRIORITY,
)
assert out["source"] == "deezer"
def test_none_when_no_candidates():
out = resolve_canonical_for_album(
album_source_ids={},
file_tracks=list(STD),
fetch_tracklist=_fetcher({}),
source_priority=PRIORITY,
)
assert out is None
def test_none_when_no_files():
out = resolve_canonical_for_album(
album_source_ids={"spotify": "a"},
file_tracks=[],
fetch_tracklist=_fetcher({("spotify", "a"): STD}),
source_priority=PRIORITY,
)
assert out is None
def test_none_when_below_floor():
files = list(STD) # 11 tracks
# Only candidate is a wildly-wrong 3-track release.
table = {("spotify", "a"): [{"duration_ms": 60_000, "title": "X"}] * 3}
out = resolve_canonical_for_album(
album_source_ids={"spotify": "a"},
file_tracks=files,
fetch_tracklist=_fetcher(table),
source_priority=PRIORITY,
)
assert out is None
def test_score_is_rounded():
out = resolve_canonical_for_album(
album_source_ids={"spotify": "a"},
file_tracks=list(STD),
fetch_tracklist=_fetcher({("spotify", "a"): STD}),
source_priority=PRIORITY,
)
assert out["score"] == round(out["score"], 4)

143
tests/test_canonical_version.py Normal file
View file

@ -0,0 +1,143 @@
"""Extreme battery for canonical-album-version scoring (#765 / #767-Bug2).
The scorer must: pick the right EDITION by best-fit to the files on disk
(standard when you have the standard, deluxe when you have the deluxe), break
ties deterministically toward the higher-priority candidate (so every tool
agrees), degrade gracefully when durations/titles are missing, and never pin a
low-confidence guess.
"""
from __future__ import annotations
from core.metadata.canonical_version import (
pick_canonical_release,
score_release_against_files,
)
# Helpers — build track lists ----------------------------------------------
def _tracks(n, base_ms=180_000, step_ms=10_000, titles=None):
"""n tracks with distinct, deterministic durations + optional titles."""
out = []
for i in range(n):
t = {"duration_ms": base_ms + i * step_ms, "track_number": i + 1}
if titles:
t["title"] = titles[i]
out.append(t)
return out
STANDARD_TITLES = [f"Song {i+1}" for i in range(11)]
DELUXE_TITLES = STANDARD_TITLES + [f"Bonus {i+1}" for i in range(6)]
# ── edition discrimination ────────────────────────────────────────────────
def test_eleven_files_prefer_standard_over_deluxe():
files = _tracks(11, titles=STANDARD_TITLES)
standard = _tracks(11, titles=STANDARD_TITLES)
deluxe = _tracks(17, titles=DELUXE_TITLES)
s_std = score_release_against_files(files, standard)
s_dlx = score_release_against_files(files, deluxe)
assert s_std > s_dlx
best, score = pick_canonical_release(
files,
[{"source": "standard", "tracks": standard}, {"source": "deluxe", "tracks": deluxe}],
)
assert best["source"] == "standard" and score > 0.9
def test_seventeen_files_prefer_deluxe():
files = _tracks(17, titles=DELUXE_TITLES)
standard = _tracks(11, titles=STANDARD_TITLES)
deluxe = _tracks(17, titles=DELUXE_TITLES)
best, _ = pick_canonical_release(
files,
# deluxe deliberately listed SECOND to prove count/fit beats order here
[{"source": "standard", "tracks": standard}, {"source": "deluxe", "tracks": deluxe}],
)
assert best["source"] == "deluxe"
def test_exact_count_and_durations_scores_near_one():
files = _tracks(11, titles=STANDARD_TITLES)
assert score_release_against_files(files, _tracks(11, titles=STANDARD_TITLES)) > 0.99
# ── deterministic tiebreak (the #765 resolution) ──────────────────────────
def test_identical_releases_break_tie_to_first_candidate():
# Same album from two sources (same files match both equally) — must pick
# the FIRST (higher-priority) deterministically so both tools agree.
files = _tracks(11, titles=STANDARD_TITLES)
a = {"source": "spotify", "tracks": _tracks(11, titles=STANDARD_TITLES)}
b = {"source": "musicbrainz", "tracks": _tracks(11, titles=STANDARD_TITLES)}
best, _ = pick_canonical_release(files, [a, b])
assert best["source"] == "spotify"
# ...and stable when the order flips (priority is the caller's order).
best2, _ = pick_canonical_release(files, [b, a])
assert best2["source"] == "musicbrainz"
# ── duration disambiguation when counts tie ───────────────────────────────
def test_duration_breaks_tie_when_counts_equal():
# Two 11-track candidates; the files' durations match candidate A's lengths,
# not B's (e.g. album cuts vs radio edits). A must win on duration fit.
files = _tracks(11, base_ms=200_000, step_ms=5_000)
cand_a = {"source": "album", "tracks": _tracks(11, base_ms=200_000, step_ms=5_000)}
cand_b = {"source": "edits", "tracks": _tracks(11, base_ms=140_000, step_ms=5_000)}
best, _ = pick_canonical_release(files, [cand_b, cand_a]) # B listed first
assert best["source"] == "album" # duration fit overrides order
# ── graceful degradation ──────────────────────────────────────────────────
def test_no_durations_falls_back_to_count_and_title():
files = [{"title": t} for t in STANDARD_TITLES] # no durations
standard = [{"title": t} for t in STANDARD_TITLES]
deluxe = [{"title": t} for t in DELUXE_TITLES]
best, score = pick_canonical_release(
files,
[{"source": "standard", "tracks": standard}, {"source": "deluxe", "tracks": deluxe}],
)
assert best["source"] == "standard" and score > 0.5
def test_only_counts_available_still_scores():
files = [{} for _ in range(11)]
assert score_release_against_files(files, [{} for _ in range(11)]) > 0.99
assert score_release_against_files(files, [{} for _ in range(17)]) < 0.8
def test_fuzzy_titles_still_match():
files = _tracks(3, titles=["Believer", "Whatever It Takes", "Thunder"])
rel = _tracks(3, titles=["Believer (Remastered)", "Whatever It Takes", "Thunder!"])
assert score_release_against_files(files, rel) > 0.9
# ── confidence floor / guards ─────────────────────────────────────────────
def test_below_floor_returns_none():
files = _tracks(11, titles=STANDARD_TITLES)
# A wildly wrong candidate (3 unrelated tracks) must not be pinned.
bad = {"source": "wrong", "tracks": _tracks(3, base_ms=60_000, titles=["X", "Y", "Z"])}
best, score = pick_canonical_release(files, [bad])
assert best is None
assert score < 0.5
def test_empty_inputs_are_safe():
assert score_release_against_files([], _tracks(11)) == 0.0
assert score_release_against_files(_tracks(11), []) == 0.0
best, score = pick_canonical_release(_tracks(11), [])
assert best is None and score == 0.0
def test_min_score_is_tunable():
files = _tracks(11, titles=STANDARD_TITLES)
near = {"source": "near", "tracks": _tracks(10, titles=STANDARD_TITLES[:10])}
# default floor accepts a 10/11 fit, a strict floor rejects it
assert pick_canonical_release(files, [near])[0] is not None
assert pick_canonical_release(files, [near], min_score=0.99)[0] is None

160
tests/test_canonical_version_job.py Normal file
View file

@ -0,0 +1,160 @@
"""Backfill job: Resolve Canonical Album Versions (#765 Stage 2 trigger)."""
from __future__ import annotations
import types
import core.repair_jobs.canonical_version_resolve as cvr
from core.repair_jobs import get_all_jobs
from core.repair_jobs.canonical_version_resolve import (
CanonicalVersionResolveJob,
_describe_pin,
)
from database.music_database import MusicDatabase
def _ctx(db, findings):
return types.SimpleNamespace(
db=db,
config_manager=None, # -> active_server None -> all albums
check_stop=lambda: False,
wait_if_paused=lambda: False,
report_progress=None,
update_progress=None,
create_finding=lambda **kw: (findings.append(kw) or True),
)
def _seed_two_albums(db):
conn = db._get_connection()
cur = conn.cursor()
cur.execute("INSERT INTO artists (id, name) VALUES ('art1', 'A')")
cur.execute("INSERT INTO albums (id, title, artist_id) VALUES ('alb1', 'Album One', 'art1')")
cur.execute("INSERT INTO albums (id, title, artist_id) VALUES ('alb2', 'Album Two', 'art1')")
conn.commit()
conn.close()
def _fake_resolver(monkeypatch):
def fake(db, album_id, *, min_score=0.5, store=True, mode="active_preferred"):
res = {"source": "spotify", "album_id": f"sp_{album_id}", "score": 0.9}
if store:
db.set_album_canonical(album_id, res["source"], res["album_id"], res["score"])
return res
monkeypatch.setattr(cvr, "resolve_and_store_canonical_for_album", fake)
def test_job_is_registered():
jobs = get_all_jobs() # {job_id: cls}
assert "canonical_version_resolve" in jobs
assert jobs["canonical_version_resolve"] is CanonicalVersionResolveJob
def test_job_is_opt_in_and_dry_run_by_default():
assert CanonicalVersionResolveJob.default_enabled is False
assert CanonicalVersionResolveJob.default_settings["dry_run"] is True
def test_source_selection_defaults_to_active_preferred():
assert CanonicalVersionResolveJob.default_settings["source_selection"] == "active_preferred"
def test_source_selection_exposes_dropdown_options():
# The UI renders a <select> for keys listed in setting_options.
opts = CanonicalVersionResolveJob.setting_options.get("source_selection")
assert opts == ["active_preferred", "active_only", "best_fit"]
# default must be one of the offered options
assert CanonicalVersionResolveJob.default_settings["source_selection"] in opts
def test_describe_pin_is_judgeable():
desc = _describe_pin({
"source": "deezer", "album_id": "665666731", "score": 1.0,
"file_track_count": 11, "release_track_count": 11,
"count_fit": 1.0, "duration_fit": 1.0, "title_fit": 1.0,
"candidates": [
{"source": "deezer", "album_id": "665666731", "track_count": 11, "score": 1.0},
{"source": "spotify", "album_id": "sp", "track_count": 17, "score": 0.65},
],
})
assert "deezer" in desc and "665666731" in desc
assert "11 files vs 11 tracks" in desc # your library vs the release
assert "durations 100%" in desc and "titles 100%" in desc # the WHY
assert "Beat:" in desc and "spotify 65% (17 tk)" in desc # what it beat
def test_describe_pin_includes_year_linked_and_tracklist():
desc = _describe_pin({
"source": "deezer", "album_id": "dz1", "score": 1.0,
"artist_name": "Lenka", "album_title": "Souls of Serenity", "year": 2017,
"file_track_count": 3, "release_track_count": 3,
"count_fit": 1.0, "duration_fit": 1.0, "title_fit": 1.0,
"linked_sources": {"spotify": "sp1", "deezer": "dz1"},
"release_track_titles": ["The Show", "Trouble Is a Friend", "Everything at Once"],
"candidates": [{"source": "deezer", "album_id": "dz1", "track_count": 3, "score": 1.0}],
})
assert "Lenka — Souls of Serenity (2017)" in desc
assert "Currently linked: spotify=sp1, deezer=dz1 → pinning deezer" in desc
assert "Release tracks: 1. The Show; 2. Trouble Is a Friend; 3. Everything at Once" in desc
def test_describe_pin_single_source():
desc = _describe_pin({
"source": "spotify", "album_id": "x", "score": 0.9,
"file_track_count": 10, "release_track_count": 10,
"count_fit": 1.0, "duration_fit": None, "title_fit": 0.9,
"candidates": [{"source": "spotify", "album_id": "x", "track_count": 10, "score": 0.9}],
})
assert "Only this source" in desc
assert "durations n/a" in desc # missing signal shown honestly
def test_live_resolves_and_stores(tmp_path, monkeypatch):
db = MusicDatabase(str(tmp_path / "m.db"))
_seed_two_albums(db)
_fake_resolver(monkeypatch)
findings = []
ctx = _ctx(db, findings)
job = CanonicalVersionResolveJob()
# force live mode
monkeypatch.setattr(job, "_get_settings", lambda c: {"dry_run": False, "min_score": 0.5})
result = job.scan(ctx)
assert result.auto_fixed == 2
assert db.get_album_canonical("alb1")["source"] == "spotify"
assert db.get_album_canonical("alb2")["album_id"] == "sp_alb2"
assert findings == [] # live mode writes, doesn't create findings
def test_dry_run_creates_findings_without_storing(tmp_path, monkeypatch):
db = MusicDatabase(str(tmp_path / "m.db"))
_seed_two_albums(db)
_fake_resolver(monkeypatch)
findings = []
ctx = _ctx(db, findings)
job = CanonicalVersionResolveJob()
monkeypatch.setattr(job, "_get_settings", lambda c: {"dry_run": True, "min_score": 0.5})
result = job.scan(ctx)
assert result.findings_created == 2
assert len(findings) == 2
# dry run must NOT persist
assert db.get_album_canonical("alb1") is None
def test_skips_already_pinned_albums(tmp_path, monkeypatch):
db = MusicDatabase(str(tmp_path / "m.db"))
_seed_two_albums(db)
db.set_album_canonical("alb1", "deezer", "dz_pinned", 0.8) # alb1 already pinned
_fake_resolver(monkeypatch)
ctx = _ctx(db, [])
job = CanonicalVersionResolveJob()
monkeypatch.setattr(job, "_get_settings", lambda c: {"dry_run": False, "min_score": 0.5})
result = job.scan(ctx)
assert result.skipped == 1 # alb1 skipped
assert result.auto_fixed == 1 # only alb2 resolved
assert db.get_album_canonical("alb1")["album_id"] == "dz_pinned" # untouched

75
tests/test_reorganize_canonical_source.py Normal file
View file

@ -0,0 +1,75 @@
"""_resolve_source honors a pinned canonical release (#765 Stage 3, read side).
Gated + side-effect-free: only changes behavior for albums that already carry a
canonical_source/canonical_album_id, and an explicit user source pick
(strict_source) still wins. No canonical -> byte-identical to before.
"""
from __future__ import annotations
import core.library_reorganize as lr
def _patch_fetch(monkeypatch, tracklists):
"""tracklists: {(source, album_id): items_or_None}. Patches the album +
tracklist fetchers and the normaliser (pass-through)."""
def get_album(source, aid):
return {"name": f"{source}:{aid}"} if tracklists.get((source, aid)) else None
def get_tracks(source, aid):
return tracklists.get((source, aid))
monkeypatch.setattr(lr, "get_album_for_source", get_album)
monkeypatch.setattr(lr, "get_album_tracks_for_source", get_tracks)
monkeypatch.setattr(lr, "_normalize_album_tracks", lambda items: items or [])
monkeypatch.setattr(lr, "get_source_priority", lambda primary: ["spotify", "itunes", "deezer"])
def test_canonical_source_preferred_over_priority(monkeypatch):
# Album has spotify (priority winner) AND a pinned canonical = deezer.
_patch_fetch(monkeypatch, {
("spotify", "sp1"): [{"name": "x"}],
("deezer", "dz1"): [{"name": "y"}],
})
album_data = {
"spotify_album_id": "sp1", "deezer_id": "dz1",
"canonical_source": "deezer", "canonical_album_id": "dz1",
}
source, api_album, items = lr._resolve_source(album_data, "spotify")
assert source == "deezer" # canonical beats the priority walk
def test_canonical_fetch_failure_falls_back_to_priority(monkeypatch):
# Canonical points at musicbrainz but that fetch yields nothing -> fall back.
_patch_fetch(monkeypatch, {
("spotify", "sp1"): [{"name": "x"}],
# no entry for ('musicbrainz', 'mb1') -> get_tracks returns None
})
album_data = {
"spotify_album_id": "sp1",
"canonical_source": "musicbrainz", "canonical_album_id": "mb1",
}
source, _, _ = lr._resolve_source(album_data, "spotify")
assert source == "spotify" # fell back to priority
def test_strict_source_ignores_canonical(monkeypatch):
# User explicitly picked spotify in the modal — their choice wins over canonical.
_patch_fetch(monkeypatch, {
("spotify", "sp1"): [{"name": "x"}],
("deezer", "dz1"): [{"name": "y"}],
})
album_data = {
"spotify_album_id": "sp1", "deezer_id": "dz1",
"canonical_source": "deezer", "canonical_album_id": "dz1",
}
source, _, _ = lr._resolve_source(album_data, "spotify", strict_source=True)
assert source == "spotify"
def test_no_canonical_unchanged(monkeypatch):
# No canonical set -> identical to legacy priority resolution.
_patch_fetch(monkeypatch, {("spotify", "sp1"): [{"name": "x"}]})
album_data = {"spotify_album_id": "sp1"}
source, _, _ = lr._resolve_source(album_data, "spotify")
assert source == "spotify"

52
tests/test_track_repair_canonical.py Normal file
View file

@ -0,0 +1,52 @@
"""Track Number Repair canonical lookup (#765 Stage 4, read side)."""
from __future__ import annotations
import types
from core.repair_jobs.track_number_repair import _lookup_canonical_from_db
from database.music_database import MusicDatabase
def _ctx(db):
return types.SimpleNamespace(db=db)
def _seed(db, *, with_canonical: bool, file_path: str = "/music/Evolve/01 - Believer.flac"):
conn = db._get_connection()
cur = conn.cursor()
cur.execute("INSERT INTO artists (id, name) VALUES ('art1', 'Imagine Dragons')")
cur.execute("INSERT INTO albums (id, title, artist_id) VALUES ('alb1', 'Evolve', 'art1')")
cur.execute(
"INSERT INTO tracks (id, album_id, artist_id, title, track_number, duration, file_path) "
"VALUES ('t1', 'alb1', 'art1', 'Believer', 1, 204000, ?)",
(file_path,),
)
conn.commit()
conn.close()
if with_canonical:
db.set_album_canonical("alb1", "spotify", "sp_evolve", 0.96)
def test_returns_canonical_when_pinned(tmp_path):
db = MusicDatabase(str(tmp_path / "m.db"))
fp = "/music/Evolve/01 - Believer.flac"
_seed(db, with_canonical=True, file_path=fp)
assert _lookup_canonical_from_db([(fp, "01 - Believer.flac", 1)], _ctx(db)) == ("spotify", "sp_evolve")
def test_none_when_unresolved(tmp_path):
db = MusicDatabase(str(tmp_path / "m.db"))
fp = "/music/Evolve/01 - Believer.flac"
_seed(db, with_canonical=False, file_path=fp)
assert _lookup_canonical_from_db([(fp, "01 - Believer.flac", 1)], _ctx(db)) is None
def test_none_when_file_not_tracked(tmp_path):
db = MusicDatabase(str(tmp_path / "m.db"))
_seed(db, with_canonical=True)
assert _lookup_canonical_from_db([("/some/other/path.flac", "x.flac", 1)], _ctx(db)) is None
def test_none_when_no_db():
assert _lookup_canonical_from_db([("/p.flac", "p.flac", 1)], types.SimpleNamespace(db=None)) is None

12
webui/static/enrichment.js
View file

@ -1705,6 +1705,18 @@ async function loadRepairJobs() {
return `<div class="repair-setting-section">${val}</div>`;
}
const label = _prettifyRepairSettingKey(key);
// Dropdown when the job declares allowed values for this key.
const opts = job.setting_options && job.setting_options[key];
if (Array.isArray(opts) && opts.length) {
const optionsHtml = opts.map(o =>
`<option value="${o}"${o === val ? ' selected' : ''}>${_prettifyRepairSettingKey(String(o))}</option>`
).join('');
return `<div class="repair-setting-row">
<label>${label}</label>
<select class="repair-setting-input"
data-job="${job.job_id}" data-key="${key}">${optionsHtml}</select>
</div>`;
}
const inputType = typeof val === 'boolean' ? 'checkbox' :
typeof val === 'number' ? 'number' : 'text';
const inputVal = inputType === 'checkbox' ?