From f5752e3dc056c62937b878677692e9df150ca10f Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Tue, 2 Jun 2026 11:47:42 -0700 Subject: [PATCH] =?UTF-8?q?Canonical=20album=20version=20=E2=80=94=20Stage?= =?UTF-8?q?=204:=20Track=20Number=20Repair=20prefers=20canonical=20(read)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _resolve_album_tracklist gains a Fallback -1: if the album has a pinned canonical (source, album_id), use it before the existing 6-level cascade — so Track Number Repair resolves the SAME release the Reorganizer does (Stage 3) and the two stop contradicting each other (#765, the Spotify-4 vs MusicBrainz-3 conflict). Gated + additive: the entire existing cascade is untouched for albums without a canonical, so this job's all-01-album rescue (which relies on the MusicBrainz/ AudioDB fallbacks for albums with no DB source ID) is fully preserved — that's the regression we explicitly refused to take in a reactive fix. New helper _lookup_canonical_from_db() mirrors _lookup_album_ids_from_db (file-path -> track -> album), returns None when no DB / no match / columns absent / unresolved. Tests: tests/test_track_repair_canonical.py (4) — returns canonical when pinned, None when unresolved / file untracked / no DB. Existing track_number_repair tests still pass (no regression). --- core/repair_jobs/track_number_repair.py | 56 +++++++++++++++++++++++++ tests/test_track_repair_canonical.py | 52 +++++++++++++++++++++++ 2 files changed, 108 insertions(+) create mode 100644 tests/test_track_repair_canonical.py diff --git a/core/repair_jobs/track_number_repair.py b/core/repair_jobs/track_number_repair.py index 8a569de6..a7c4292a 100644 --- a/core/repair_jobs/track_number_repair.py +++ b/core/repair_jobs/track_number_repair.py @@ -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. diff --git a/tests/test_track_repair_canonical.py b/tests/test_track_repair_canonical.py new file mode 100644 index 00000000..72879151 --- /dev/null +++ b/tests/test_track_repair_canonical.py @@ -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