From 08fb21fb1376b288e9d7de7c1f99905f2307151f Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Thu, 18 Jun 2026 15:25:37 -0700 Subject: [PATCH] =?UTF-8?q?#889=20Phase=202:=20import=20seam=20=E2=80=94?= =?UTF-8?q?=20hint=20short-circuits=20identification,=20replaces=20on=20su?= =?UTF-8?q?ccess?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a staged single-file candidate carries a re-identify hint, the worker builds the identification straight from the user-chosen release (album_id/source) and skips the guessing tiers — so the ambiguity that mis-filed the track is gone. No hint → byte-identical to before (the lookup returns (None, None), fail-safe on any DB error). A hinted import auto-processes (explicit user choice), still gated on the global auto_process pref. After the re-import lands, _finalize_rematch_hint consumes the hint and (if replace was chosen) deletes the old row + file via delete_replaced_track — deferred to success so a failed import never loses the original. Safe by construction: unlink only when no surviving row references the file, and the modal never offers the track's current release so old path != new path. All hint logic lives in auto_import_worker.py + the pure rematch_hints helpers — pipeline.py / side_effects.py untouched. 18 tests; full auto-import suite green. --- core/auto_import_worker.py | 75 +++++++++- core/imports/rematch_hints.py | 61 ++++++++ tests/imports/test_rematch_hints_seam.py | 168 +++++++++++++++++++++++ 3 files changed, 301 insertions(+), 3 deletions(-) create mode 100644 tests/imports/test_rematch_hints_seam.py diff --git a/core/auto_import_worker.py b/core/auto_import_worker.py index 8bfbea34..c3f85603 100644 --- a/core/auto_import_worker.py +++ b/core/auto_import_worker.py @@ -660,8 +660,13 @@ class AutoImportWorker: auto_process = self._config_manager.get('auto_import.auto_process', True) try: - # Phase 3: Identify - identification = self._identify_folder(candidate) + # Phase 3: Identify. + # Re-identify (#889): if the user designated this exact file's release in + # the Re-identify modal, a hint short-circuits the guessing — we match + # straight against the chosen album. No hint → byte-identical to before. + rematch_hint, identification = self._resolve_rematch_hint(candidate) + if identification is None: + identification = self._identify_folder(candidate) if not identification: self._record_result(candidate, 'needs_identification', 0.0, error_message='Could not identify album from tags, folder name, or fingerprint') @@ -690,7 +695,10 @@ class AutoImportWorker: high_conf_matches = [m for m in match_result.get('matches', []) if m['confidence'] >= 0.8] has_strong_individual_matches = len(high_conf_matches) > 0 - if (confidence >= threshold or has_strong_individual_matches) and auto_process: + # A re-identify is an explicit user choice — let it auto-process like a + # strong match (still gated on the global auto_process preference). + if (confidence >= threshold or has_strong_individual_matches + or rematch_hint is not None) and auto_process: # Phase 5: Auto-process — insert an in-progress row # so the UI sees the import the moment it starts, # then update it with the final status when done. @@ -709,6 +717,11 @@ class AutoImportWorker: confidence = max(confidence, effective_conf) if success: self._bump_stat('auto_processed') + # Re-identify (#889): only NOW that the new home exists do we + # consume the hint and (if replace was chosen) delete the old + # row + file — so a failed import never loses the original. + if rematch_hint is not None: + self._finalize_rematch_hint(rematch_hint) else: self._bump_stat('failed') @@ -1003,6 +1016,62 @@ class AutoImportWorker: except Exception: return False + # ── Re-identify hints (#889) ── + + def _resolve_rematch_hint(self, candidate: 'FolderCandidate'): + """If this staged file carries a user-designated re-identify hint, return + ``(hint, identification)`` so matching skips the guessing tiers; otherwise + ``(None, None)`` and the caller falls back to normal identification. + + Fail-safe: ANY error (no table, DB hiccup) returns ``(None, None)`` so a + re-identify problem can never break ordinary auto-import. Only single-file + candidates are eligible — a re-identify always stages exactly one track.""" + try: + files = candidate.audio_files or [] + if len(files) != 1: + return None, None + from core.imports.rematch_hints import ( + build_identification_from_hint, + find_hint_for_file, + quick_file_signature, + ) + file_path = files[0] + sig = quick_file_signature(file_path) + conn = self.database._get_connection() + try: + cursor = conn.cursor() + hint = find_hint_for_file(cursor, file_path, sig) + finally: + conn.close() + if hint is None: + return None, None + logger.info("[Auto-Import] Re-identify hint for %s → %s '%s' (%s)", + candidate.name, hint.album_type or 'release', + hint.album_name or '?', hint.source) + return hint, build_identification_from_hint(hint) + except Exception as e: + logger.debug("[Auto-Import] rematch-hint lookup skipped: %s", e) + return None, None + + def _finalize_rematch_hint(self, hint) -> None: + """Post-success: delete the replaced library row + file (if the user chose + replace) and consume the hint so it's single-use. Best-effort — a cleanup + failure is logged, never raised, since the re-import already succeeded.""" + try: + from core.imports.rematch_hints import consume_hint, delete_replaced_track + conn = self.database._get_connection() + try: + cursor = conn.cursor() + removed = delete_replaced_track(cursor, hint.replace_track_id) + consume_hint(cursor, hint.id) + conn.commit() + finally: + conn.close() + if removed: + logger.info("[Auto-Import] Re-identify replaced old track — removed %s", removed) + except Exception as e: + logger.warning("[Auto-Import] rematch-hint finalize failed (import still OK): %s", e) + # ── Identification ── def _identify_folder(self, candidate: FolderCandidate) -> Optional[Dict]: diff --git a/core/imports/rematch_hints.py b/core/imports/rematch_hints.py index 6804e788..2a75e66a 100644 --- a/core/imports/rematch_hints.py +++ b/core/imports/rematch_hints.py @@ -193,6 +193,65 @@ def list_pending_hints(cursor: Any) -> list: return [_row_to_hint(r) for r in cursor.fetchall()] +def build_identification_from_hint(hint: RematchHint) -> dict: + """Turn a hint into the ``identification`` dict the auto-import matcher expects, + so a re-identify SKIPS the guessing tiers entirely and matches straight against + the user-chosen release. Mirrors the shape `_identify_folder` returns (album_id + / source / track_number drive the album fetch + file→track match).""" + return { + "album_id": hint.album_id or None, + "album_name": hint.album_name or hint.track_title or "", + "artist_name": hint.artist_name or "", + "artist_id": hint.artist_id or "", + "track_name": hint.track_title or "", + "track_id": hint.track_id or "", + "image_url": "", + "release_date": "", + "track_number": hint.track_number or 1, + "total_tracks": 1, + "source": hint.source, + "method": "rematch_hint", + "identification_confidence": 1.0, + "is_single": True, + "album_type": hint.album_type, + } + + +def delete_replaced_track(cursor: Any, replace_track_id: Any, *, unlink=os.remove) -> Optional[str]: + """Remove the OLD library row a re-identify replaces, and its file. + + Called only AFTER the re-import has landed the track at its new home, so the + original is never lost on failure. Safe by construction: the file is unlinked + only if it still exists and **no other track row references it** (guards against + yanking a file a different row legitimately points to). Returns the path it + removed, or ``None`` if there was nothing to do. ``unlink`` is injectable for + tests. Assumes the new home differs from the old — the Re-identify modal never + offers the track's current release as a target, so this holds.""" + if not replace_track_id: + return None + cursor.execute("SELECT file_path FROM tracks WHERE id = ?", (replace_track_id,)) + row = cursor.fetchone() + if row is None: + return None + old_path = (row["file_path"] if not isinstance(row, (tuple, list)) else row[0]) or "" + + cursor.execute("DELETE FROM tracks WHERE id = ?", (replace_track_id,)) + + if not old_path: + return None + # Only unlink if no surviving row still points at this file. + cursor.execute("SELECT 1 FROM tracks WHERE file_path = ? LIMIT 1", (old_path,)) + if cursor.fetchone() is not None: + return None + try: + if os.path.exists(old_path): + unlink(old_path) + return old_path + except OSError: + pass + return None + + def quick_file_signature(path: str, *, chunk: int = 65536) -> Optional[str]: """A cheap, rename-proof content fingerprint: size + first/last chunk, hashed. @@ -222,5 +281,7 @@ __all__ = [ "find_hint_for_file", "consume_hint", "list_pending_hints", + "build_identification_from_hint", + "delete_replaced_track", "quick_file_signature", ] diff --git a/tests/imports/test_rematch_hints_seam.py b/tests/imports/test_rematch_hints_seam.py new file mode 100644 index 00000000..58d479fd --- /dev/null +++ b/tests/imports/test_rematch_hints_seam.py @@ -0,0 +1,168 @@ +"""#889 Phase 2: the import seam — a hint short-circuits identification, and the +old library row is replaced only after the re-import succeeds. + +Two layers: + * pure helpers (build_identification_from_hint, delete_replaced_track) — exact + mapping + safe replacement against an in-memory DB, injectable unlink. + * the worker seam (_resolve_rematch_hint / _finalize_rematch_hint) — proves the + NO-HINT path is untouched, the hint path returns a ready identification, the + lookup is fail-safe, and finalize consumes + replaces. +""" + +from __future__ import annotations + +import sqlite3 +import types + +import pytest + +from core.auto_import_worker import AutoImportWorker, FolderCandidate +from core.imports.rematch_hints import ( + RematchHint, + build_identification_from_hint, + consume_hint, + create_hint, + delete_replaced_track, + find_hint_for_file, +) + +_SCHEMA = """ +CREATE TABLE rematch_hints ( + id INTEGER PRIMARY KEY AUTOINCREMENT, staged_path TEXT NOT NULL, content_hash TEXT, + source TEXT NOT NULL, isrc TEXT, track_id TEXT, album_id TEXT, artist_id TEXT, + track_title TEXT, album_name TEXT, artist_name TEXT, album_type TEXT, + track_number INTEGER, disc_number INTEGER, replace_track_id INTEGER, + exempt_dedup INTEGER NOT NULL DEFAULT 1, status TEXT NOT NULL DEFAULT 'pending', + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, consumed_at TIMESTAMP +); +CREATE TABLE tracks ( + id INTEGER PRIMARY KEY, album_id INTEGER, artist_id INTEGER, title TEXT, + track_number INTEGER, file_path TEXT +); +""" + + +@pytest.fixture +def conn(): + c = sqlite3.connect(":memory:") + c.row_factory = sqlite3.Row + c.executescript(_SCHEMA) + yield c + c.close() + + +def _hint(**kw): + base = dict(staged_path="/staging/Song.flac", source="spotify", album_id="alb_album1", + artist_id="art_1", track_id="trk_1", track_title="Song", album_name="Album1", + artist_name="Artist", album_type="album", track_number=5, disc_number=1) + base.update(kw) + return RematchHint(**base) + + +# ── pure: identification mapping ────────────────────────────────────────────── +def test_build_identification_maps_hint_fields(): + ident = build_identification_from_hint(_hint()) + assert ident["album_id"] == "alb_album1" + assert ident["source"] == "spotify" + assert ident["album_name"] == "Album1" + assert ident["artist_id"] == "art_1" + assert ident["track_number"] == 5 + assert ident["method"] == "rematch_hint" + assert ident["identification_confidence"] == 1.0 + assert ident["is_single"] is True + + +# ── pure: safe replacement ──────────────────────────────────────────────────── +def test_delete_replaced_track_removes_row_and_file(conn): + cur = conn.cursor() + cur.execute("INSERT INTO tracks (id, file_path) VALUES (7, '/lib/EP1/05 - Song.flac')") + removed = [] + out = delete_replaced_track(cur, 7, unlink=lambda p: removed.append(p)) + assert out == "/lib/EP1/05 - Song.flac" + assert removed == ["/lib/EP1/05 - Song.flac"] # file removed (we faked existence below) + cur.execute("SELECT 1 FROM tracks WHERE id = 7") + assert cur.fetchone() is None # row gone + + +def test_delete_replaced_track_keeps_file_if_another_row_points_at_it(conn): + cur = conn.cursor() + cur.execute("INSERT INTO tracks (id, file_path) VALUES (7, '/lib/shared.flac')") + cur.execute("INSERT INTO tracks (id, file_path) VALUES (8, '/lib/shared.flac')") + removed = [] + out = delete_replaced_track(cur, 7, unlink=lambda p: removed.append(p)) + assert out is None and removed == [] # row 8 still references it → no unlink + cur.execute("SELECT 1 FROM tracks WHERE id = 7") + assert cur.fetchone() is None # but row 7 still deleted + + +def test_delete_replaced_track_noops_on_missing_id(conn): + cur = conn.cursor() + assert delete_replaced_track(cur, None) is None + assert delete_replaced_track(cur, 999) is None # no such row + + +# patch os.path.exists so the unlink branch is reachable without real files +@pytest.fixture(autouse=True) +def _exists(monkeypatch): + monkeypatch.setattr("core.imports.rematch_hints.os.path.exists", lambda p: True) + + +# ── worker seam ─────────────────────────────────────────────────────────────── +def _worker(conn): + # Production hands out a FRESH connection per call (the worker closes it); + # here we share one in-memory DB, so proxy close() to a no-op. + w = AutoImportWorker.__new__(AutoImportWorker) + proxy = types.SimpleNamespace(cursor=conn.cursor, commit=conn.commit, close=lambda: None) + w.database = types.SimpleNamespace(_get_connection=lambda: proxy) + return w + + +def test_resolve_returns_none_when_no_hint(conn): + w = _worker(conn) + cand = FolderCandidate(path="/staging", name="Song", audio_files=["/staging/Song.flac"]) + assert w._resolve_rematch_hint(cand) == (None, None) # untouched → normal identify + + +def test_resolve_returns_identification_when_hinted(conn, monkeypatch): + # don't hash a real file + monkeypatch.setattr("core.imports.rematch_hints.quick_file_signature", lambda p: None) + create_hint(conn.cursor(), _hint(staged_path="/staging/Song.flac")) + conn.commit() + w = _worker(conn) + cand = FolderCandidate(path="/staging", name="Song", audio_files=["/staging/Song.flac"]) + hint, ident = w._resolve_rematch_hint(cand) + assert hint is not None and hint.album_id == "alb_album1" + assert ident["album_id"] == "alb_album1" and ident["method"] == "rematch_hint" + + +def test_resolve_ignores_multi_file_candidates(conn): + create_hint(conn.cursor(), _hint(staged_path="/staging/Song.flac")) + conn.commit() + w = _worker(conn) + cand = FolderCandidate(path="/staging", name="Album", + audio_files=["/staging/Song.flac", "/staging/Other.flac"]) + assert w._resolve_rematch_hint(cand) == (None, None) # re-identify is single-track only + + +def test_resolve_is_failsafe_on_db_error(): + w = AutoImportWorker.__new__(AutoImportWorker) + def _boom(): + raise RuntimeError("db down") + w.database = types.SimpleNamespace(_get_connection=_boom) + cand = FolderCandidate(path="/staging", name="Song", audio_files=["/staging/Song.flac"]) + assert w._resolve_rematch_hint(cand) == (None, None) # error never breaks auto-import + + +def test_finalize_consumes_and_replaces(conn, monkeypatch): + monkeypatch.setattr("core.imports.rematch_hints.quick_file_signature", lambda p: None) + cur = conn.cursor() + cur.execute("INSERT INTO tracks (id, file_path) VALUES (42, '/lib/EP1/05 - Song.flac')") + hid = create_hint(cur, _hint(replace_track_id=42)) + conn.commit() + w = _worker(conn) + hint = find_hint_for_file(conn.cursor(), "/staging/Song.flac") + w._finalize_rematch_hint(hint) + # old row deleted, hint consumed + cur.execute("SELECT 1 FROM tracks WHERE id = 42") + assert cur.fetchone() is None + assert find_hint_for_file(conn.cursor(), "/staging/Song.flac") is None # consumed