diff --git a/core/imports/rematch_hints.py b/core/imports/rematch_hints.py new file mode 100644 index 00000000..6804e788 --- /dev/null +++ b/core/imports/rematch_hints.py @@ -0,0 +1,226 @@ +"""Re-identify hints (#889) — a single-use, user-designated answer to "which +release does this already-imported track belong to". + +Flow: the user clicks *Re-identify* on a library track, searches a source, and +picks the exact release (single / EP / album) it should live under. We write a +**hint** here and stage the file for auto-import. The import flow then reads the +hint at the very TOP of matching — before any fuzzy tier — builds the match from +these exact IDs, and consumes the row. So the original ambiguity that mis-filed +the track (which release?) is gone: the user already answered it. + +Two safety properties live in the hint, not the import code: + +- ``replace_track_id`` — the library row to delete AFTER the re-import lands (so a + re-identify *replaces* rather than *duplicates*). Cleanup is deferred to success + so a failed import can never lose the file. +- ``exempt_dedup`` — always set: a re-identify is an explicit user action and must + not be silently dropped by the quality dedup-skip (which would otherwise see the + incoming file as a duplicate of the very row we're replacing). + +This module is pure DB mechanics over an injected ``cursor`` (sqlite3-style, +``?`` params) — no connection management, no app state — so the create / find / +consume seam is unit-tested against an in-memory DB with no live metadata client. +The binding is keyed on the staged path, with ``content_hash`` as a rename-proof +fallback in case the staging watcher normalizes the filename on ingest. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import Any, Optional + +# Columns in INSERT/SELECT order — single source of truth so the dataclass, the +# write, and the read can't drift apart. +_FIELDS = ( + "staged_path", + "content_hash", + "source", + "isrc", + "track_id", + "album_id", + "artist_id", + "track_title", + "album_name", + "artist_name", + "album_type", + "track_number", + "disc_number", + "replace_track_id", + "exempt_dedup", +) + + +@dataclass +class RematchHint: + """One user-designated re-identify answer. ``id``/``status`` are set by the DB.""" + staged_path: str + source: str + content_hash: Optional[str] = None + isrc: Optional[str] = None + track_id: Optional[str] = None + album_id: Optional[str] = None + artist_id: Optional[str] = None + track_title: Optional[str] = None + album_name: Optional[str] = None + artist_name: Optional[str] = None + album_type: Optional[str] = None + track_number: Optional[int] = None + disc_number: Optional[int] = None + replace_track_id: Optional[int] = None + exempt_dedup: bool = True + id: Optional[int] = None + status: str = "pending" + + def _values(self) -> tuple: + return ( + self.staged_path, + self.content_hash, + self.source, + self.isrc, + self.track_id, + self.album_id, + self.artist_id, + self.track_title, + self.album_name, + self.artist_name, + self.album_type, + self.track_number, + self.disc_number, + self.replace_track_id, + 1 if self.exempt_dedup else 0, + ) + + +def _row_to_hint(row: Any) -> RematchHint: + """Map a sqlite3.Row (or any mapping/sequence-by-name) to a RematchHint.""" + def g(key, default=None): + try: + return row[key] + except (KeyError, IndexError, TypeError): + return default + return RematchHint( + id=g("id"), + staged_path=g("staged_path") or "", + content_hash=g("content_hash"), + source=g("source") or "", + isrc=g("isrc"), + track_id=g("track_id"), + album_id=g("album_id"), + artist_id=g("artist_id"), + track_title=g("track_title"), + album_name=g("album_name"), + artist_name=g("artist_name"), + album_type=g("album_type"), + track_number=g("track_number"), + disc_number=g("disc_number"), + replace_track_id=g("replace_track_id"), + exempt_dedup=bool(g("exempt_dedup", 1)), + status=g("status") or "pending", + ) + + +def create_hint(cursor: Any, hint: RematchHint) -> int: + """Insert a pending hint; return its new id. Caller owns commit.""" + placeholders = ", ".join("?" for _ in _FIELDS) + cursor.execute( + f"INSERT INTO rematch_hints ({', '.join(_FIELDS)}) VALUES ({placeholders})", + hint._values(), + ) + new_id = cursor.lastrowid + hint.id = new_id + return new_id + + +def find_hint_for_file( + cursor: Any, + staged_path: str, + content_hash: Optional[str] = None, +) -> Optional[RematchHint]: + """Return the newest PENDING hint for a staged file, or ``None``. + + Matched by exact ``staged_path`` first; if that misses and a ``content_hash`` + is given, fall back to it (covers a staging watcher that renamed the file on + ingest). Only ``status='pending'`` rows are returned, so a consumed hint is + never reused.""" + if staged_path: + cursor.execute( + "SELECT * FROM rematch_hints WHERE staged_path = ? AND status = 'pending' " + "ORDER BY id DESC LIMIT 1", + (staged_path,), + ) + row = cursor.fetchone() + if row is not None: + return _row_to_hint(row) + # Try by basename too — the watcher may move the file into a different dir. + base = os.path.basename(staged_path) + if base and base != staged_path: + cursor.execute( + "SELECT * FROM rematch_hints WHERE staged_path LIKE ? AND status = 'pending' " + "ORDER BY id DESC LIMIT 1", + ("%/" + base,), + ) + row = cursor.fetchone() + if row is not None: + return _row_to_hint(row) + + if content_hash: + cursor.execute( + "SELECT * FROM rematch_hints WHERE content_hash = ? AND status = 'pending' " + "ORDER BY id DESC LIMIT 1", + (content_hash,), + ) + row = cursor.fetchone() + if row is not None: + return _row_to_hint(row) + + return None + + +def consume_hint(cursor: Any, hint_id: int) -> None: + """Mark a hint consumed (single-use). Caller owns commit.""" + cursor.execute( + "UPDATE rematch_hints SET status = 'consumed', consumed_at = CURRENT_TIMESTAMP " + "WHERE id = ?", + (hint_id,), + ) + + +def list_pending_hints(cursor: Any) -> list: + """All pending hints (newest first) — for a 'pending re-identify' view and + orphan recovery when a staged file never imports.""" + cursor.execute("SELECT * FROM rematch_hints WHERE status = 'pending' ORDER BY id DESC") + return [_row_to_hint(r) for r in cursor.fetchall()] + + +def quick_file_signature(path: str, *, chunk: int = 65536) -> Optional[str]: + """A cheap, rename-proof content fingerprint: size + first/last chunk, hashed. + + Audio files are large, so a full hash is wasteful when we only need to re-bind + a hint to *this* file after a possible rename. Size + head + tail is plenty to + distinguish staged files in practice. Returns ``None`` if the file can't be + read (caller falls back to path-only binding).""" + import hashlib + + try: + size = os.path.getsize(path) + h = hashlib.sha256() + h.update(str(size).encode()) + with open(path, "rb") as f: + h.update(f.read(chunk)) + if size > chunk: + f.seek(max(0, size - chunk)) + h.update(f.read(chunk)) + return h.hexdigest() + except OSError: + return None + + +__all__ = [ + "RematchHint", + "create_hint", + "find_hint_for_file", + "consume_hint", + "list_pending_hints", + "quick_file_signature", +] diff --git a/database/music_database.py b/database/music_database.py index 7f50d875..8691e8d1 100644 --- a/database/music_database.py +++ b/database/music_database.py @@ -750,6 +750,41 @@ class MusicDatabase: cursor.execute("CREATE INDEX IF NOT EXISTS idx_aih_status ON auto_import_history (status)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_aih_folder_hash ON auto_import_history (folder_hash)") + # Re-identify hints (#889) — a user-designated, single-use answer to "which + # release does this track belong to". Written when the user picks a release in + # the Re-identify modal and the file is staged for auto-import; the import flow + # reads the hint at the TOP of matching (keyed by staged path, content_hash as a + # rename-proof fallback), expedites the match to these exact IDs, then consumes + # the row. `replace_track_id` (when set) is the library row to delete AFTER the + # re-import lands; `exempt_dedup` is always 1 because a re-identify is an explicit + # user action that must not be silently dropped by the quality dedup-skip. + cursor.execute(""" + CREATE TABLE IF NOT EXISTS 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 + ) + """) + cursor.execute("CREATE INDEX IF NOT EXISTS idx_rmh_staged_path ON rematch_hints (staged_path)") + cursor.execute("CREATE INDEX IF NOT EXISTS idx_rmh_content_hash ON rematch_hints (content_hash)") + cursor.execute("CREATE INDEX IF NOT EXISTS idx_rmh_status ON rematch_hints (status)") + # Sync history table — tracks the last 100 sync operations with cached context for re-trigger cursor.execute(""" CREATE TABLE IF NOT EXISTS sync_history ( diff --git a/tests/imports/test_rematch_hints.py b/tests/imports/test_rematch_hints.py new file mode 100644 index 00000000..6e1d08de --- /dev/null +++ b/tests/imports/test_rematch_hints.py @@ -0,0 +1,155 @@ +"""#889 Phase 1: the re-identify hint store — create / find / consume. + +The hint is the single-use, user-designated "which release" answer the import +flow reads at the top of matching. These lock down: a hint round-trips, it's +found by staged path, found by content_hash when the path missed (rename-proof), +found by basename when the dir changed, consumed exactly once, and that a +consumed hint is never handed back. +""" + +from __future__ import annotations + +import sqlite3 + +import pytest + +from core.imports.rematch_hints import ( + RematchHint, + consume_hint, + create_hint, + find_hint_for_file, + list_pending_hints, + quick_file_signature, +) + +# The slice of the real schema this module touches (kept in sync with +# database/music_database.py's rematch_hints CREATE). +_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 +) +""" + + +@pytest.fixture +def cur(): + conn = sqlite3.connect(":memory:") + conn.row_factory = sqlite3.Row + conn.executescript(_SCHEMA) + yield conn.cursor() + conn.close() + + +def _hint(**kw): + base = dict( + staged_path="/staging/Song.flac", + source="spotify", + isrc="USABC1234567", + track_id="trk_1", + album_id="alb_album1", + artist_id="art_1", + track_title="Song", + album_name="Album1", + artist_name="Artist", + album_type="album", + track_number=5, + disc_number=1, + replace_track_id=42, + ) + base.update(kw) + return RematchHint(**base) + + +def test_create_and_find_by_path_roundtrips(cur): + new_id = create_hint(cur, _hint()) + assert new_id > 0 + got = find_hint_for_file(cur, "/staging/Song.flac") + assert got is not None + assert got.id == new_id + assert got.album_id == "alb_album1" and got.album_type == "album" + assert got.isrc == "USABC1234567" + assert got.track_number == 5 and got.disc_number == 1 + assert got.replace_track_id == 42 + assert got.exempt_dedup is True # always set for a user-designated re-identify + assert got.status == "pending" + + +def test_find_by_content_hash_when_path_missed(cur): + create_hint(cur, _hint(content_hash="deadbeef")) + # Watcher renamed/moved the file → path lookup misses, hash rescues it. + assert find_hint_for_file(cur, "/totally/different.flac") is None + got = find_hint_for_file(cur, "/totally/different.flac", content_hash="deadbeef") + assert got is not None and got.album_name == "Album1" + + +def test_find_by_basename_when_dir_changed(cur): + create_hint(cur, _hint(staged_path="/staging/in/Song.flac")) + # Same filename, different directory (watcher moved it deeper). + got = find_hint_for_file(cur, "/staging/processing/Song.flac") + assert got is not None and got.track_id == "trk_1" + + +def test_consume_is_single_use(cur): + new_id = create_hint(cur, _hint()) + assert find_hint_for_file(cur, "/staging/Song.flac") is not None + consume_hint(cur, new_id) + # Consumed → never handed back, by path or by hash. + assert find_hint_for_file(cur, "/staging/Song.flac") is None + assert find_hint_for_file(cur, "/x", content_hash=None) is None + + +def test_list_pending_excludes_consumed(cur): + a = create_hint(cur, _hint(staged_path="/staging/A.flac")) + create_hint(cur, _hint(staged_path="/staging/B.flac")) + assert len(list_pending_hints(cur)) == 2 + consume_hint(cur, a) + pend = list_pending_hints(cur) + assert len(pend) == 1 and pend[0].staged_path == "/staging/B.flac" + + +def test_newest_pending_wins_on_duplicate_path(cur): + create_hint(cur, _hint(album_id="alb_old")) + create_hint(cur, _hint(album_id="alb_new")) # user re-picked for the same file + got = find_hint_for_file(cur, "/staging/Song.flac") + assert got.album_id == "alb_new" + + +def test_exempt_dedup_false_roundtrips(cur): + create_hint(cur, _hint(staged_path="/staging/Keep.flac", exempt_dedup=False)) + got = find_hint_for_file(cur, "/staging/Keep.flac") + assert got.exempt_dedup is False + + +# ── content fingerprint ─────────────────────────────────────────────────────── +def test_quick_file_signature_stable_and_distinct(tmp_path): + a = tmp_path / "a.bin" + b = tmp_path / "b.bin" + a.write_bytes(b"hello world" * 1000) + b.write_bytes(b"goodbye moon" * 1000) + sig_a1 = quick_file_signature(str(a)) + sig_a2 = quick_file_signature(str(a)) + sig_b = quick_file_signature(str(b)) + assert sig_a1 and sig_a1 == sig_a2 # stable + assert sig_a1 != sig_b # distinct content → distinct sig + + +def test_quick_file_signature_missing_file_is_none(): + assert quick_file_signature("/no/such/file.flac") is None