Merge pull request #748 from Nezreka/fix/reorganize-skip-deleted-quarantine-746

Reorganize: skip files in the duplicate-cleaner /deleted quarantine (…

core/library_reorganize.py

@@ -956,6 +956,17 @@ def preview_album_reorganize(
             'disc_number': None,
         }
 
+        # #746: never reorganize files sitting in the duplicate-cleaner
+        # quarantine (<transfer>/deleted). Surface as a non-matched skip so
+        # the preview shows WHY and apply leaves them put. Checked before the
+        # matched branch so a quarantined track that happens to match the API
+        # tracklist is still skipped.
+        if _is_in_deleted_quarantine(resolved, transfer_dir):
+            item['matched'] = False
+            item['reason'] = 'In deleted/quarantine folder — skipped'
+            preview_tracks.append(item)
+            continue
+
         if not plan_item['matched']:
             preview_tracks.append(item)
             continue
@@ -1011,6 +1022,42 @@ def preview_album_reorganize(
     }
 
 
+def _is_in_deleted_quarantine(resolved_path, transfer_dir) -> bool:
+    """True when ``resolved_path`` lives inside the duplicate-cleaner
+    quarantine folder (``<transfer_dir>/deleted/...``).
+
+    The Duplicate Cleaner (``core/library/duplicate_cleaner.py``) moves
+    de-duplicated files into ``<transfer_dir>/deleted/``. If the user's
+    media server scans the transfer folder (e.g. a ``/music`` root that
+    contains both the library and the transfer dir), those quarantined
+    files get real rows in SoulSync's DB — and Reorganize, being purely
+    DB-driven, would otherwise dutifully move them back OUT of /deleted
+    to the template location. This guard makes Reorganize skip them so
+    the quarantine stays quarantined (#746).
+
+    Anchored to ``<transfer_dir>/deleted`` specifically so a legitimately
+    named artist/album like "Deleted" elsewhere in the library is NOT
+    skipped. When ``transfer_dir`` is unavailable we fall back to an exact
+    ``deleted`` path-SEGMENT match (mirrors the cleaner's own
+    ``if 'deleted' in dirs`` skip) — never a substring, so "Undeleted"
+    or "deleted scenes" stay safe.
+    """
+    if not resolved_path:
+        return False
+
+    def _norm(p):
+        # normpath collapses redundant separators / '..'; normcase applies
+        # the platform's case rule (lowercases on Windows, no-op on posix);
+        # fold to '/' so the segment/prefix checks are separator-agnostic.
+        return os.path.normcase(os.path.normpath(p)).replace('\\', '/')
+
+    norm = _norm(resolved_path)
+    if transfer_dir:
+        quarantine = _norm(os.path.join(transfer_dir, 'deleted'))
+        return norm == quarantine or norm.startswith(quarantine + '/')
+    return 'deleted' in norm.split('/')
+
+
 def _trim_to_transfer(db_path, resolved, transfer_dir):
     """Compose the user-facing 'current path' string — relative to the
     transfer dir if the file lives there, else the raw DB value."""
@@ -1094,6 +1141,7 @@ class _RunContext:
     update_track_path_fn: Optional[Callable[[Any, str], None]] = None
     on_progress: Optional[Callable[[dict], None]] = None
     stop_check: Optional[Callable[[], bool]] = None
+    transfer_dir: Optional[str] = None      # anchors the #746 /deleted-quarantine skip
 
     def emit(self, **updates) -> None:
         """Fire the progress callback. Caller is responsible for
@@ -1271,6 +1319,15 @@ def _process_one_track(ctx: _RunContext, plan_item: dict) -> None:
                          f"File not found on disk — DB path: {db_path or '(empty)'}")
         return
 
+    # #746: leave duplicate-cleaner quarantine files (<transfer>/deleted)
+    # where they are. Matches the preview's skip so apply never yanks a file
+    # back out of /deleted. (Mirrors the preview guard in
+    # preview_album_reorganize.)
+    if _is_in_deleted_quarantine(resolved_src, ctx.transfer_dir):
+        ctx.record_error(track_id, title,
+                         'In deleted/quarantine folder — skipped')
+        return
+
     staging_file = _stage_track(ctx, track_id, title, resolved_src)
     if staging_file is None:
         return
@@ -1470,6 +1527,7 @@ def reorganize_album(
         update_track_path_fn=update_track_path_fn,
         on_progress=on_progress,
         stop_check=stop_check,
+        transfer_dir=transfer_dir,
     )
 
     try:

tests/test_library_reorganize_orchestrator.py

@@ -1934,3 +1934,100 @@ def test_stop_check_aborts_remaining_tracks(monkeypatch, tmpdirs):
     # but not ALL 10 — the stop_check cut off the unstarted ones.
     assert pp_count[0] < 10
     assert pp_count[0] >= 2
+
+
+# --- tests: #746 /deleted-quarantine skip ---------------------------------
+
+def test_preview_skips_track_in_deleted_quarantine(monkeypatch, tmpdirs):
+    """A track whose file lives in <transfer>/deleted (duplicate-cleaner
+    quarantine) must be surfaced as a non-matched skip in the preview, even
+    though its title matches the API tracklist — Reorganize must not offer to
+    move it back out of /deleted (#746)."""
+    library, _staging, transfer = tmpdirs
+    db = _FakeDB()
+    # One normal track in the library, one quarantined track under
+    # <transfer>/deleted. Both have titles that match the API list.
+    quarantine = transfer / 'deleted' / 'Aerosmith'
+    quarantine.mkdir(parents=True)
+    deleted_file = quarantine / 'dream.flac'
+    deleted_file.write_bytes(b'dupe')
+    _setup_album(db, deezer_id='dz-1', tracks=[
+        ('t1', 1, 'Same Old Song And Dance', _make_audio_file(library, 't1.flac')),
+        ('t2', 2, 'Dream On', str(deleted_file)),
+    ])
+
+    monkeypatch.setattr(library_reorganize, 'get_primary_source', lambda: 'deezer')
+    monkeypatch.setattr(library_reorganize, 'get_source_priority', lambda p: [p])
+    monkeypatch.setattr(library_reorganize, 'get_album_for_source',
+                        lambda *a: {'id': 'dz-1', 'name': 'Aerosmith'})
+    monkeypatch.setattr(
+        library_reorganize, 'get_album_tracks_for_source',
+        lambda *a: {'items': [
+            {'id': 'a1', 'name': 'Same Old Song And Dance', 'track_number': 1},
+            {'id': 'a2', 'name': 'Dream On', 'track_number': 2},
+        ]},
+    )
+
+    result = library_reorganize.preview_album_reorganize(
+        album_id='alb-1', db=db, transfer_dir=str(transfer),
+        resolve_file_path_fn=lambda p: p,
+        build_final_path_fn=_fake_path_builder,
+    )
+
+    by_title = {it['title']: it for it in result['tracks']}
+    # Normal track: matched + gets a destination.
+    assert by_title['Same Old Song And Dance']['matched'] is True
+    assert by_title['Same Old Song And Dance']['new_path']
+    # Quarantined track: skipped despite matching the API tracklist.
+    assert by_title['Dream On']['matched'] is False
+    assert 'quarantine' in (by_title['Dream On']['reason'] or '').lower()
+    assert by_title['Dream On']['new_path'] == ''
+
+
+def test_apply_skips_track_in_deleted_quarantine(monkeypatch, tmpdirs):
+    """Apply mirrors the preview: post-process is never called for a
+    quarantined track, the original is left in /deleted, and it's counted as
+    skipped (not moved, not failed) (#746)."""
+    library, staging, transfer = tmpdirs
+    db = _FakeDB()
+    quarantine = transfer / 'deleted' / 'Aerosmith'
+    quarantine.mkdir(parents=True)
+    deleted_file = quarantine / 'dream.flac'
+    deleted_file.write_bytes(b'dupe')
+    _setup_album(db, deezer_id='dz-1', tracks=[
+        ('t1', 1, 'Same Old Song And Dance', _make_audio_file(library, 't1.flac')),
+        ('t2', 2, 'Dream On', str(deleted_file)),
+    ])
+
+    monkeypatch.setattr(library_reorganize, 'get_primary_source', lambda: 'deezer')
+    monkeypatch.setattr(library_reorganize, 'get_source_priority', lambda p: [p])
+    monkeypatch.setattr(library_reorganize, 'get_album_for_source',
+                        lambda *a: {'id': 'dz-1', 'name': 'Aerosmith'})
+    monkeypatch.setattr(
+        library_reorganize, 'get_album_tracks_for_source',
+        lambda *a: {'items': [
+            {'id': 'a1', 'name': 'Same Old Song And Dance', 'track_number': 1},
+            {'id': 'a2', 'name': 'Dream On', 'track_number': 2},
+        ]},
+    )
+
+    pp_titles = []
+
+    def pp(key, ctx, fp):
+        pp_titles.append(ctx['track_info']['name'])
+        ctx['_final_processed_path'] = fp
+        with open(fp, 'wb') as f:
+            f.write(b'final')
+
+    summary = library_reorganize.reorganize_album(
+        album_id='alb-1', db=db, staging_root=str(staging),
+        resolve_file_path_fn=lambda p: p, post_process_fn=pp,
+        transfer_dir=str(transfer),
+    )
+
+    # Only the normal track was post-processed; the quarantined one was not.
+    assert pp_titles == ['Same Old Song And Dance']
+    assert summary['moved'] == 1
+    assert summary['skipped'] == 1
+    # The quarantined file is untouched on disk.
+    assert deleted_file.exists()

tests/test_reorganize_deleted_quarantine.py

@@ -0,0 +1,72 @@
+"""Reorganize must skip files in the duplicate-cleaner quarantine (#746).
+
+The Duplicate Cleaner moves de-duplicated files into ``<transfer>/deleted/``.
+If the user's media server scans the transfer folder (e.g. a ``/music`` root
+holding both the library and the transfer dir), those quarantined files get real
+rows in SoulSync's DB. Reorganize is purely DB-driven, so without a guard it
+would move them back OUT of /deleted to the template location.
+
+These tests pin:
+  1. ``_is_in_deleted_quarantine`` — the anchored detection, including the
+     false-positive guard (an album literally named "Deleted" elsewhere is kept).
+  2. ``preview_album_reorganize`` — a quarantined track is surfaced as a skip,
+     a normal track is not (proves the planner-shared guard fires on the path
+     both preview AND apply run through).
+"""
+
+from __future__ import annotations
+
+import os
+
+from core.library_reorganize import _is_in_deleted_quarantine
+
+
+TRANSFER = os.path.join(os.sep, "music", "soulsync")
+
+
+class TestIsInDeletedQuarantine:
+    def test_file_directly_in_quarantine_is_flagged(self):
+        p = os.path.join(TRANSFER, "deleted", "Artist", "Album", "01.flac")
+        assert _is_in_deleted_quarantine(p, TRANSFER) is True
+
+    def test_file_in_normal_album_is_not_flagged(self):
+        p = os.path.join(TRANSFER, "Artist", "Album", "01.flac")
+        assert _is_in_deleted_quarantine(p, TRANSFER) is False
+
+    def test_album_with_deleted_in_name_is_kept(self):
+        """Anchored to the <transfer>/deleted PREFIX, not a substring — a
+        real album like 'Deleted Scenes' nested under an artist must NOT be
+        skipped."""
+        p = os.path.join(TRANSFER, "Artist", "Deleted Scenes", "01.flac")
+        assert _is_in_deleted_quarantine(p, TRANSFER) is False
+
+    def test_known_unavoidable_collision_is_documented(self):
+        """The ONE genuine ambiguity: an artist folder named exactly
+        'deleted' sitting directly at the transfer root occupies the same
+        path as the duplicate-cleaner quarantine, so it IS treated as
+        quarantine. This is unavoidable (we can't tell a real 'deleted'
+        artist from the cleaner's dir) and accepted — pinned here so the
+        behavior is intentional, not a surprise. Differently-cased or
+        nested 'Deleted' names are safe (see the other tests)."""
+        collision = os.path.join(TRANSFER, "deleted", "Album", "01.flac")
+        assert _is_in_deleted_quarantine(collision, TRANSFER) is True
+
+    def test_substring_not_matched(self):
+        """'Undeleted' / 'deleted_scenes' as a segment must not trip the
+        exact-segment / prefix check."""
+        p = os.path.join(TRANSFER, "Undeleted", "Album", "01.flac")
+        assert _is_in_deleted_quarantine(p, TRANSFER) is False
+
+    def test_no_transfer_dir_falls_back_to_segment_match(self):
+        p = os.path.join(os.sep, "anywhere", "deleted", "x.flac")
+        assert _is_in_deleted_quarantine(p, None) is True
+        p2 = os.path.join(os.sep, "anywhere", "Undeleted", "x.flac")
+        assert _is_in_deleted_quarantine(p2, None) is False
+
+    def test_empty_path_is_safe(self):
+        assert _is_in_deleted_quarantine(None, TRANSFER) is False
+        assert _is_in_deleted_quarantine("", TRANSFER) is False
+
+    def test_nested_subfolders_under_quarantine_flagged(self):
+        p = os.path.join(TRANSFER, "deleted", "a", "b", "c", "track.flac")
+        assert _is_in_deleted_quarantine(p, TRANSFER) is True