Downloads: reuse an album's existing folder so batches don't split it (#829)

Tacobell444: when tracks land in an album across multiple batches (a wishlist
run, the Album Completeness job, a missed track re-downloaded later), the folder
is rebuilt from API metadata each time — so when $albumtype or $year come back
blank/different on a later batch, the folder NAME changes and the album splits,
forcing a Reorganize.

Fix: build_final_path_for_track now checks whether the album already lives in a
single folder on disk and, if so, drops the new track there instead of a freshly
templated folder. Match (chosen): exact stored Spotify album id first, then a
STRICT >=0.85 name+artist match (vs the 0.7 used elsewhere) — a wrong match here
misplaces a file. New core/library/existing_album_folder.resolve_existing_album_folder
holds the logic; always-on with template fallback.

Safety rails: only returns a folder UNDER the transfer dir (never a read-only
library/NAS mount), only when the album lives in EXACTLY ONE folder (multiple =
disc subfolders, which DatabaseTrack can't disambiguate — those defer to the
template), and any failure falls through to the template path. Added
MusicDatabase.get_album_by_spotify_album_id for the id-first lookup.

Tests: single-folder reuse, no-match, below-threshold, multi-folder defer,
outside-transfer reject, id-first, missing transfer dir, no-files-on-disk.
8 tests; 1556 path/import/download tests pass (only the known soundcloud
failures remain).

core/imports/paths.py

@@ -587,6 +587,45 @@ def build_final_path_for_track(context, artist_context, album_info, file_ext, cr
         disc_label = _get_config_manager().get("file_organization.disc_label", "Disc")
 
         folder_path, filename_base = get_file_path_from_template(template_context, "album_path")
+
+        # #829: if this album already lives in a single folder on disk, drop the
+        # new track there instead of a freshly-templated folder — this is what
+        # keeps an album from splitting when $albumtype/$year drift between
+        # batches (wishlist, Album Completeness, a missed track later). Strict
+        # match + transfer-dir-only + single-folder-only inside the resolver;
+        # any miss falls through to the template path below. Best-effort.
+        reuse_folder = None
+        if filename_base:
+            try:
+                from core.library.existing_album_folder import resolve_existing_album_folder
+                from database.music_database import get_database
+                try:
+                    _active_server = _get_config_manager().get_active_media_server()
+                except Exception:
+                    _active_server = None
+                _spotify_album_id = (album_context.get("id")
+                                     if album_context and str(source).startswith("spotify") else None)
+                _expected_tracks = None
+                if album_context and album_context.get("total_tracks"):
+                    _expected_tracks = _coerce_int(album_context.get("total_tracks"), 0) or None
+                reuse_folder = resolve_existing_album_folder(
+                    db=get_database(),
+                    transfer_dir=transfer_dir,
+                    album_name=album_info.get("album_name"),
+                    album_artist=template_context.get("albumartist"),
+                    spotify_album_id=_spotify_album_id,
+                    active_server=_active_server,
+                    expected_track_count=_expected_tracks,
+                    config_manager=_get_config_manager(),
+                )
+            except Exception as _reuse_err:
+                logger.debug("[Existing Album Folder] lookup failed: %s", _reuse_err)
+                reuse_folder = None
+        if reuse_folder and filename_base:
+            final_path = os.path.join(reuse_folder, filename_base + file_ext)
+            _ensure_dir(reuse_folder, exist_ok=True)
+            return final_path, True
+
         if folder_path and filename_base:
             if total_discs > 1 and not user_controls_disc:
                 disc_folder = f"{disc_label} {disc_number}"

core/library/existing_album_folder.py

@@ -0,0 +1,129 @@
+"""Reuse an album's existing on-disk folder for new downloads (#829).
+
+When tracks are added to an album across multiple batches (a wishlist run, the
+Album Completeness job, a missed track re-downloaded later), the destination
+folder is normally rebuilt from API metadata each time. If ``$albumtype`` or
+``$year`` come back blank/different on a later batch, the folder *name* changes
+and the album splits across folders — forcing a Reorganize afterwards.
+
+This resolves the folder the album *already* lives in so the new track joins its
+existing files instead. Matching is deliberately conservative: the exact stored
+Spotify album id first (definitive), then a STRICT (>= 0.85) name+artist match —
+higher than the 0.7 used elsewhere, because a wrong match here misplaces a file.
+
+Safety rails:
+  * Only ever returns a folder UNDER the transfer dir (the managed download
+    tree) — never a read-only library/NAS mount the resolver happens to find.
+  * Only reuses when the album lives in EXACTLY ONE folder on disk. Multiple
+    folders means disc subfolders (DatabaseTrack carries no disc number, so we
+    can't safely pick the right one) — those defer to the template path.
+  * Any failure returns None — the caller falls back to the normal template.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any, Optional
+
+from core.library.path_resolver import resolve_library_file_path
+from utils.logging_config import get_logger
+
+logger = get_logger("library.existing_album_folder")
+
+# Strict — a wrong album match drops the file in the wrong folder.
+_STRICT_ALBUM_CONFIDENCE = 0.85
+
+
+def _is_under(child: str, parent: str) -> bool:
+    """True if ``child`` is the same as or inside ``parent`` (normalized)."""
+    try:
+        child_n = os.path.normcase(os.path.normpath(os.path.abspath(child)))
+        parent_n = os.path.normcase(os.path.normpath(os.path.abspath(parent)))
+        return child_n == parent_n or child_n.startswith(parent_n + os.sep)
+    except Exception:
+        return False
+
+
+def _find_album(db: Any, spotify_album_id: Optional[str], album_name: Optional[str],
+                album_artist: Optional[str], active_server: Optional[str],
+                expected_track_count: Optional[int]):
+    """Stored Spotify id first, then a strict name+artist match. None on no match."""
+    if spotify_album_id:
+        try:
+            album = db.get_album_by_spotify_album_id(spotify_album_id)
+            if album:
+                return album
+        except Exception as e:
+            logger.debug("album-by-spotify-id lookup failed: %s", e)
+    if album_name and album_artist:
+        try:
+            match, confidence = db.check_album_exists_with_editions(
+                title=album_name, artist=album_artist,
+                confidence_threshold=_STRICT_ALBUM_CONFIDENCE,
+                expected_track_count=expected_track_count,
+                server_source=active_server,
+            )
+            if match and confidence >= _STRICT_ALBUM_CONFIDENCE:
+                return match
+        except Exception as e:
+            logger.debug("strict album name+artist match failed: %s", e)
+    return None
+
+
+def resolve_existing_album_folder(
+    *,
+    db: Any,
+    transfer_dir: Optional[str],
+    album_name: Optional[str] = None,
+    album_artist: Optional[str] = None,
+    spotify_album_id: Optional[str] = None,
+    active_server: Optional[str] = None,
+    expected_track_count: Optional[int] = None,
+    config_manager: Any = None,
+    resolver=resolve_library_file_path,
+) -> Optional[str]:
+    """Return the on-disk folder an existing album lives in (so a new track joins
+    it) or None to fall back to the templated path. See module docstring."""
+    if not transfer_dir or not os.path.isdir(transfer_dir):
+        return None
+    if not db:
+        return None
+
+    album = _find_album(db, spotify_album_id, album_name, album_artist,
+                        active_server, expected_track_count)
+    if not album:
+        return None
+
+    try:
+        tracks = db.get_tracks_by_album(album.id)
+    except Exception as e:
+        logger.debug("get_tracks_by_album(%s) failed: %s", getattr(album, 'id', '?'), e)
+        return None
+
+    folders = set()
+    for t in tracks:
+        file_path = getattr(t, 'file_path', None)
+        if not file_path:
+            continue
+        try:
+            resolved = resolver(file_path, transfer_folder=transfer_dir,
+                                config_manager=config_manager)
+        except Exception:
+            resolved = None
+        if not resolved:
+            continue
+        folder = os.path.dirname(resolved)
+        if _is_under(folder, transfer_dir):
+            folders.add(os.path.normpath(folder))
+
+    # Single folder under the transfer dir → reuse it. Zero (nothing on disk yet)
+    # or many (disc subfolders) → let the template decide.
+    if len(folders) == 1:
+        reuse = next(iter(folders))
+        logger.info("[Existing Album Folder] Reusing '%s' for album '%s'",
+                    reuse, getattr(album, 'title', album_name))
+        return reuse
+    return None
+
+
+__all__ = ["resolve_existing_album_folder"]

database/music_database.py

@@ -6260,11 +6260,47 @@ class MusicDatabase:
                 ))
             
             return tracks
-            
+
         except Exception as e:
             logger.error(f"Error getting tracks for album {album_id}: {e}")
             return []
-    
+
+    def get_album_by_spotify_album_id(self, spotify_album_id: str) -> Optional[DatabaseAlbum]:
+        """Fetch a single album by its (enriched) Spotify album id, or None.
+
+        Used by the download path builder (#829) to reuse an album's existing
+        on-disk folder when re-downloading into the same album — matching the
+        exact stored Spotify id before falling back to fuzzy name+artist.
+        """
+        if not spotify_album_id:
+            return None
+        try:
+            conn = self._get_connection()
+            cursor = conn.cursor()
+            cursor.execute("""
+                SELECT albums.*, artists.name as artist_name
+                FROM albums
+                JOIN artists ON albums.artist_id = artists.id
+                WHERE albums.spotify_album_id = ?
+                LIMIT 1
+            """, (spotify_album_id,))
+            row = cursor.fetchone()
+            if not row:
+                return None
+            genres = json.loads(row['genres']) if row['genres'] else None
+            album = DatabaseAlbum(
+                id=row['id'], artist_id=row['artist_id'], title=row['title'],
+                year=row['year'], thumb_url=row['thumb_url'], genres=genres,
+                track_count=row['track_count'], duration=row['duration'],
+                created_at=datetime.fromisoformat(row['created_at']) if row['created_at'] else None,
+                updated_at=datetime.fromisoformat(row['updated_at']) if row['updated_at'] else None,
+            )
+            album.artist_name = row['artist_name']
+            return album
+        except Exception as e:
+            logger.error(f"Error getting album by spotify_album_id {spotify_album_id}: {e}")
+            return None
+
     def search_artists(self, query: str, limit: int = 50, server_source: str = None) -> List[DatabaseArtist]:
         """Search artists by name, optionally filtered by server source.
         Uses diacritic-insensitive matching so 'Tiesto' finds 'Tiësto'."""

tests/test_existing_album_folder.py

@@ -0,0 +1,114 @@
+"""Reuse an album's existing on-disk folder for new downloads (#829).
+
+Tacobell444: tracks added to an album across batches split into different folders
+when $albumtype/$year drift. The resolver finds the album's existing single
+folder (under the transfer dir) so the new track joins it. These pin the safety
+rails: strict match, transfer-dir-only, single-folder-only, id-first.
+"""
+
+from __future__ import annotations
+
+import os
+from types import SimpleNamespace
+
+from core.library.existing_album_folder import resolve_existing_album_folder
+
+
+class _FakeDb:
+    def __init__(self, album=None, album_conf=0.0, tracks=None, by_spotify=None):
+        self._album = album
+        self._album_conf = album_conf
+        self._tracks = tracks or []
+        self._by_spotify = by_spotify
+
+    def get_album_by_spotify_album_id(self, sid):
+        return self._by_spotify
+
+    def check_album_exists_with_editions(self, title, artist, confidence_threshold=0.8,
+                                         expected_track_count=None, server_source=None, **kw):
+        return (self._album, self._album_conf)
+
+    def get_tracks_by_album(self, album_id):
+        return self._tracks
+
+
+def _track(path):
+    return SimpleNamespace(file_path=path)
+
+
+def _album(id=1, title="Ocean Avenue"):
+    return SimpleNamespace(id=id, title=title)
+
+
+def _mkfile(folder, name):
+    folder.mkdir(parents=True, exist_ok=True)
+    f = folder / name
+    f.write_text("x")
+    return str(f)
+
+
+def test_reuses_single_folder_under_transfer(tmp_path):
+    album_dir = tmp_path / "Yellowcard - Ocean Avenue"
+    f1 = _mkfile(album_dir, "01 - Way Away.mp3")
+    f2 = _mkfile(album_dir, "02 - Breathing.mp3")
+    db = _FakeDb(album=_album(), album_conf=0.95, tracks=[_track(f1), _track(f2)])
+    out = resolve_existing_album_folder(
+        db=db, transfer_dir=str(tmp_path),
+        album_name="Ocean Avenue", album_artist="Yellowcard")
+    assert out == os.path.normpath(str(album_dir))
+
+
+def test_no_match_returns_none(tmp_path):
+    db = _FakeDb(album=None, album_conf=0.0)
+    assert resolve_existing_album_folder(
+        db=db, transfer_dir=str(tmp_path), album_name="X", album_artist="Y") is None
+
+
+def test_below_strict_threshold_returns_none(tmp_path):
+    f = _mkfile(tmp_path / "A", "1.mp3")
+    # 0.80 < the resolver's 0.85 strict gate -> not reused.
+    db = _FakeDb(album=_album(), album_conf=0.80, tracks=[_track(f)])
+    assert resolve_existing_album_folder(
+        db=db, transfer_dir=str(tmp_path), album_name="A", album_artist="B") is None
+
+
+def test_multi_folder_defers_to_template(tmp_path):
+    f1 = _mkfile(tmp_path / "Album" / "Disc 01", "1.mp3")
+    f2 = _mkfile(tmp_path / "Album" / "Disc 02", "1.mp3")
+    db = _FakeDb(album=_album(), album_conf=0.95, tracks=[_track(f1), _track(f2)])
+    assert resolve_existing_album_folder(
+        db=db, transfer_dir=str(tmp_path), album_name="A", album_artist="B") is None
+
+
+def test_folder_outside_transfer_returns_none(tmp_path):
+    f = _mkfile(tmp_path / "outside", "1.mp3")
+    transfer = tmp_path / "transfer"
+    transfer.mkdir()
+    db = _FakeDb(album=_album(), album_conf=0.95, tracks=[_track(f)])
+    assert resolve_existing_album_folder(
+        db=db, transfer_dir=str(transfer), album_name="A", album_artist="B") is None
+
+
+def test_id_first_match_skips_name_lookup(tmp_path):
+    album_dir = tmp_path / "Album"
+    f = _mkfile(album_dir, "1.mp3")
+    # name match would FAIL (album=None); the stored spotify id hits.
+    db = _FakeDb(album=None, album_conf=0.0, tracks=[_track(f)], by_spotify=_album())
+    out = resolve_existing_album_folder(
+        db=db, transfer_dir=str(tmp_path), spotify_album_id="sp123",
+        album_name="X", album_artist="Y")
+    assert out == os.path.normpath(str(album_dir))
+
+
+def test_missing_transfer_dir_returns_none(tmp_path):
+    db = _FakeDb(album=_album(), album_conf=0.95, tracks=[])
+    assert resolve_existing_album_folder(
+        db=db, transfer_dir=str(tmp_path / "nope"), album_name="A", album_artist="B") is None
+
+
+def test_album_with_no_files_on_disk_returns_none(tmp_path):
+    # Album matched but its tracks have no resolvable file -> nothing to reuse.
+    db = _FakeDb(album=_album(), album_conf=0.95,
+                 tracks=[_track("/gone/1.mp3"), _track(None)])
+    assert resolve_existing_album_folder(
+        db=db, transfer_dir=str(tmp_path), album_name="A", album_artist="B") is None