soulsync/core/downloads/history_match.py
BoulderBadgeDad 3a92571c71 downloads: stop AcoustID scan duplicating history rows / leaving verified tracks 'unverified' (#934)
the AcoustID scanner matched library_history rows by EXACT file_path, but that path is
frozen at import time while the file moves afterward (media-server import / reorganize) —
so tracks.file_path (what the scan reads) no longer equals it. two failures resulted, both
introduced in 37ea6604: verified status never reached the history row (verified tracks kept
showing 'unverified'), and a fresh acoustid_scan row was INSERTed every run (5551 rows for
3675 songs).

- new pure, tested matcher (core/downloads/history_match.py): exact path → filename guarded
  by title; prefers a real download row over a synthetic scan row.
- _persist_status now HEALS the matched row's path + status (so future scans match cleanly),
  DELETES synthetic acoustid_scan duplicates by exact path (collision-free, never a real row),
  and inserts only when the file genuinely has no row.
- a full AcoustID job now self-cleans existing duplicates — no destructive bulk migration.

8 matcher + 4 real-DB heal/dedup/insert tests; existing scanner tests updated to the new
seam (heal vs insert). 1076 acoustid/verification/download tests green.
2026-06-27 09:52:39 -07:00

62 lines
3.3 KiB
Python

"""Match a file back to its download-history row when its path has drifted (#934).
``library_history.file_path`` is frozen at import time, but the file moves afterward
(media-server import, library reorganize) and ``tracks.file_path`` — what the AcoustID
scanner reads — no longer equals it. Matching on the exact path alone then fails twice:
the verification status never reaches the history row (verified tracks read "unverified"),
and a fresh ``acoustid_scan`` row gets inserted every run (thousands of duplicates).
This module picks the canonical history row by exact path first, then by FILENAME guarded
by a title check — so a shared filename ("01 - Intro.flac") can never heal the wrong song.
Pure (no DB) so the matching rules are unit-testable; the caller does the SQL.
"""
from __future__ import annotations
import os
from typing import Iterable, Optional, Sequence, Tuple
def _norm_title(value) -> str:
"""Alphanumeric-only lowercase form, so "Song (Remaster)" vs "song remaster"
style drift between the download tag and the media-server tag still agrees."""
return ''.join(ch for ch in str(value or '').lower() if ch.isalnum())
def like_filename_filter(basename: str) -> str:
"""A ``LIKE ... ESCAPE '\\'`` pattern that coarsely matches rows whose path ends
in ``basename``. Escapes the LIKE metacharacters (``%`` ``_`` ``\\``) — filenames
routinely contain underscores. Callers MUST still confirm with an exact basename
compare (``pick_history_row`` does), since ``'%name'`` also matches ``'xname'``."""
esc = basename.replace('\\', '\\\\').replace('%', '\\%').replace('_', '\\_')
return '%' + esc
def pick_history_row(candidates: Sequence[Tuple], *, current_paths: Iterable[str],
basename: str, title: str) -> Optional[int]:
"""Return the id of the history row to update for this file, or None.
``candidates``: ``(id, file_path, title, download_source)`` rows the DB pre-filtered
(exact path or filename LIKE). A row matches when its path equals the current path OR
its filename matches AND its title agrees — the title guard prevents a shared filename
("01 - Intro.flac") from healing a different song's row. Among matches a REAL download
row is preferred over a synthetic ``acoustid_scan`` row, so the scanner heals the
genuine record and the caller can delete the synthetic duplicate. None when nothing
matches safely (caller then inserts a fresh row — the "file SoulSync never downloaded"
intent)."""
paths = {p for p in current_paths if p}
want = _norm_title(title)
matches: list = [] # (id, is_exact, is_real)
for cid, cpath, ctitle, csource in candidates:
is_real = csource != 'acoustid_scan'
if cpath and cpath in paths:
matches.append((cid, True, is_real))
elif (basename and cpath and os.path.basename(cpath) == basename
and (not want or not _norm_title(ctitle) or _norm_title(ctitle) == want)):
matches.append((cid, False, is_real))
if not matches:
return None
# Prefer a REAL download row over a synthetic acoustid_scan row; within that, prefer an
# exact-path match over a filename match. Stable, so ties keep DB order (first/oldest id).
matches.sort(key=lambda m: (m[2], m[1]), reverse=True)
return matches[0][0]