8e218303be
Picking the release a track is ALREADY in deleted the file: the re-import lands at the same path, record_soulsync_library_entry skips the insert (row exists), so no new row is created — then delete_replaced_track removed that very row AND unlinked the file (the freshly-imported one). This was the 'assumption' I documented but never enforced. Bulletproof guard: the pipeline writes its landing path into context['_final_processed_path']; _process_matches captures it onto the candidate, and _finalize_rematch_hint passes those new_paths to delete_replaced_track. If the old file canonically equals where the import landed, it's a NO-OP — the row and file are kept (that file IS the re-imported track). Canonical compare folds symlinks/case/sep. So same-release re-identify is now a harmless re-tag-in-place; only a genuine re-home (different path) deletes the old. 114 auto-import + rematch tests green.
334 lines
12 KiB
Python
334 lines
12 KiB
Python
"""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, Callable, 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 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 reflects the CHOSEN release, but force_album_match makes the
|
|
# matcher FETCH that release (even for a lone staged file) instead of taking
|
|
# the singles fast-path — so the re-imported track gets the real album
|
|
# metadata: year, the correct in-album track number, and the album art.
|
|
"is_single": (str(hint.album_type or "").lower() == "single"),
|
|
"force_album_match": True,
|
|
"album_type": hint.album_type,
|
|
}
|
|
|
|
|
|
def _canonical(path: Optional[str]) -> str:
|
|
"""Canonical form of a path for same-file comparison (symlinks + case + sep)."""
|
|
if not path:
|
|
return ""
|
|
try:
|
|
return os.path.normcase(os.path.realpath(path))
|
|
except OSError:
|
|
return os.path.normcase(os.path.normpath(path))
|
|
|
|
|
|
def delete_replaced_track(
|
|
cursor: Any,
|
|
replace_track_id: Any,
|
|
*,
|
|
unlink=os.remove,
|
|
resolve_fn: Optional[Callable[[str], Optional[str]]] = None,
|
|
new_paths: Optional[list] = None,
|
|
) -> 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:
|
|
|
|
* **Same-home guard (CRITICAL):** if the re-import landed at the SAME file as the
|
|
old one (``new_paths`` — the paths the import actually wrote), this is a no-op:
|
|
we DON'T delete the row or the file, because that file IS the re-imported track.
|
|
This is what stops "re-identify to the release it's already in" from deleting
|
|
the file (the import reuses the same row, so deleting it would orphan the file).
|
|
* 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. ``resolve_fn`` maps the STORED DB path to the file's actual
|
|
on-disk location (the stored path may be a Docker/media-server view this process
|
|
can't read literally — without it we'd delete the row but orphan the file)."""
|
|
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 ""
|
|
if not old_path:
|
|
cursor.execute("DELETE FROM tracks WHERE id = ?", (replace_track_id,))
|
|
return None
|
|
|
|
# Resolve the old stored path to its real on-disk location up front.
|
|
real_path = old_path
|
|
if resolve_fn is not None:
|
|
try:
|
|
real_path = resolve_fn(old_path) or old_path
|
|
except Exception:
|
|
real_path = old_path
|
|
|
|
# Same-home guard: if the re-import wrote to this very file, do NOTHING — the row
|
|
# is the re-imported track's row and the file is its file. Deleting either would
|
|
# be data loss (the "picked the same release" bug).
|
|
if new_paths:
|
|
landed = {_canonical(p) for p in new_paths if p}
|
|
if _canonical(real_path) in landed or _canonical(old_path) in landed:
|
|
return None
|
|
|
|
cursor.execute("DELETE FROM tracks WHERE id = ?", (replace_track_id,))
|
|
# Only unlink if no surviving row still points at this file (rows store the
|
|
# stored path, so compare against the stored path, not the resolved one).
|
|
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(real_path): # real_path resolved above
|
|
unlink(real_path)
|
|
return real_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.
|
|
|
|
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",
|
|
"build_identification_from_hint",
|
|
"delete_replaced_track",
|
|
"quick_file_signature",
|
|
]
|