diff --git a/config/settings.py b/config/settings.py index a522d4a7..ff232ad6 100644 --- a/config/settings.py +++ b/config/settings.py @@ -696,6 +696,17 @@ class ConfigManager: "enabled": False, "entry_base_path": "" }, + "playlists": { + # Where "Organize by playlist" materializes playlist folders. + # MUST be a separate root from the music library so the media + # server (and the maintenance jobs) never scan it — otherwise the + # same track would show up twice. Mapped separately for Docker. + "materialize_path": "./Playlists", + # "symlink" (relative links, ~zero disk) or "copy" (real + # duplicates for FAT/USB/DAPs that can't follow links). Symlink + # auto-falls back to copy when the filesystem can't link. + "materialize_mode": "symlink" + }, "youtube": { "cookies_browser": "", # "", "chrome", "firefox", "edge", "brave", "opera", "safari" "download_delay": 3, # seconds between sequential downloads diff --git a/core/playlists/materialize.py b/core/playlists/materialize.py new file mode 100644 index 00000000..f48a9001 --- /dev/null +++ b/core/playlists/materialize.py @@ -0,0 +1,214 @@ +"""Materialize a playlist as a folder of links into the real music library. + +A playlist folder is a **view**, not storage. Every entry points at the one real +file that already lives in the library (``Artist/Album/track.ext``); a track is +never stored twice no matter how many playlists it's in. Two modes: + +- ``symlink`` (default): a *relative* symlink to the real file — ~zero disk, and + relative so the tree stays valid if the parent folder is moved. +- ``copy``: a real duplicate of the file — for filesystems/players that can't + follow symlinks (FAT USB sticks, some DAPs) or when a self-contained, + portable folder is wanted. + +Symlinks silently fail or are unsupported on a lot of real setups (Windows +without the privilege, SMB/CIFS shares, FAT/exFAT). So when a symlink can't be +created we **fall back to a copy automatically** — the folder is always fully +populated, never left with dangling links. + +This module is pure filesystem mechanics: no DB, no app state. Given a list of +real file paths, a playlists root, a playlist name and a mode, it (re)builds the +folder to match. That makes it the single unit-tested source of truth for "how a +playlist folder looks on disk", and means the folder is a *derived view* that can +be rebuilt from scratch at any time. Filesystem ops are injectable so the +behaviour — including the symlink→copy fallback — is testable without depending +on the host filesystem's symlink support. +""" + +from __future__ import annotations + +import os +import shutil +from dataclasses import dataclass, field +from typing import Callable, List, Optional, Sequence + +from core.imports.paths import sanitize_filename + +MATERIALIZE_MODES = ("symlink", "copy") +DEFAULT_MODE = "symlink" + + +def normalize_mode(mode: Optional[str]) -> str: + """Coerce a config value to a valid mode (``symlink`` default).""" + m = (mode or "").strip().lower() + return m if m in MATERIALIZE_MODES else DEFAULT_MODE + + +@dataclass +class RebuildSummary: + """Outcome of rebuilding one playlist folder. ``copied`` may be non-zero even + in symlink mode when the fallback kicked in (flagged by ``fellback``).""" + playlist_dir: str = "" + linked: int = 0 + copied: int = 0 + unchanged: int = 0 + removed_stale: int = 0 + missing_source: int = 0 + failed: int = 0 + fellback: bool = False + mode_requested: str = DEFAULT_MODE + errors: List[str] = field(default_factory=list) + + +def playlist_dir_for(playlists_root: str, playlist_name: str) -> str: + """Absolute path of one playlist's folder, sanitized and guaranteed to stay + directly under ``playlists_root`` (defends against ``..`` / separators in a + playlist name).""" + root = os.path.abspath(playlists_root) + safe_name = sanitize_filename(playlist_name or "").strip() or "Unnamed Playlist" + candidate = os.path.abspath(os.path.join(root, safe_name)) + if os.path.dirname(candidate) != root: + safe_name = sanitize_filename(os.path.basename(candidate)) or "Unnamed Playlist" + candidate = os.path.join(root, safe_name) + return candidate + + +def _desired_entries(playlist_dir: str, real_paths: Sequence[str]) -> "list[tuple[str, str]]": + """Map each real file to a flat destination inside ``playlist_dir``, preserving + the source filename. On a basename collision between two *different* sources, + disambiguate with a numeric suffix rather than silently overwriting.""" + entries: list[tuple[str, str]] = [] + used: dict[str, str] = {} # dest basename -> source real path + for real in real_paths: + if not real: + continue + base = os.path.basename(real) + name = base + stem, ext = os.path.splitext(base) + counter = 1 + while name in used and used[name] != os.path.abspath(real): + counter += 1 + name = f"{stem} ({counter}){ext}" + used[name] = os.path.abspath(real) + entries.append((os.path.abspath(real), os.path.join(playlist_dir, name))) + return entries + + +def _symlink_is_current(dest: str, rel_target: str) -> bool: + try: + return os.path.islink(dest) and os.readlink(dest) == rel_target + except OSError: + return False + + +def _remove_entry(path: str) -> None: + """Remove an existing file/symlink at ``path`` (incl. a broken symlink).""" + if os.path.islink(path) or os.path.exists(path): + try: + os.remove(path) + except OSError: + pass + + +def materialize_one( + real_path: str, + dest_path: str, + mode: str = DEFAULT_MODE, + *, + symlink_fn: Callable[[str, str], None] = os.symlink, + copy_fn: Callable[[str, str], object] = shutil.copy2, +) -> str: + """Create one playlist entry at ``dest_path`` pointing at ``real_path``. + + Idempotent: a correct existing entry is left alone. In ``symlink`` mode a + relative link is used; if it can't be created (unsupported FS, no privilege) + it falls back to a copy so the entry is never left broken. Returns one of: + ``'linked'``, ``'copied'``, ``'unchanged'``, ``'fellback'`` (symlink + requested but copied), ``'missing'`` (source gone).""" + if not real_path or not os.path.exists(real_path): + return "missing" + + dest_dir = os.path.dirname(dest_path) + os.makedirs(dest_dir, exist_ok=True) + rel_target = os.path.relpath(os.path.abspath(real_path), start=dest_dir) + + if mode == "copy": + if os.path.isfile(dest_path) and not os.path.islink(dest_path): + return "unchanged" + _remove_entry(dest_path) + copy_fn(real_path, dest_path) + return "copied" + + # symlink mode + if _symlink_is_current(dest_path, rel_target): + return "unchanged" + _remove_entry(dest_path) + try: + symlink_fn(rel_target, dest_path) + return "linked" + except (OSError, NotImplementedError): + copy_fn(real_path, dest_path) + return "fellback" + + +def rebuild_playlist_folder( + playlists_root: str, + playlist_name: str, + real_paths: Sequence[str], + mode: str = DEFAULT_MODE, + *, + prune_stale: bool = True, + symlink_fn: Callable[[str, str], None] = os.symlink, + copy_fn: Callable[[str, str], object] = shutil.copy2, +) -> RebuildSummary: + """(Re)build ``playlists_root//`` so it contains exactly one + entry per real file in ``real_paths`` — adding missing entries, leaving correct + ones untouched, and (when ``prune_stale``) removing entries no longer present. + Idempotent and safe to re-run any time. Filesystem ops are injectable.""" + mode = normalize_mode(mode) + pdir = playlist_dir_for(playlists_root, playlist_name) + summary = RebuildSummary(playlist_dir=pdir, mode_requested=mode) + os.makedirs(pdir, exist_ok=True) + + entries = _desired_entries(pdir, real_paths) + keep = {dest for _real, dest in entries} + + for real, dest in entries: + try: + outcome = materialize_one(real, dest, mode, symlink_fn=symlink_fn, copy_fn=copy_fn) + except OSError as e: + summary.failed += 1 + summary.errors.append(f"{os.path.basename(dest)}: {e}") + continue + if outcome == "linked": + summary.linked += 1 + elif outcome == "copied": + summary.copied += 1 + elif outcome == "fellback": + summary.copied += 1 + summary.fellback = True + elif outcome == "unchanged": + summary.unchanged += 1 + elif outcome == "missing": + summary.missing_source += 1 + + if prune_stale and os.path.isdir(pdir): + for name in os.listdir(pdir): + full = os.path.join(pdir, name) + if full in keep: + continue + if os.path.islink(full) or os.path.isfile(full): + _remove_entry(full) + summary.removed_stale += 1 + + return summary + + +__all__ = [ + "MATERIALIZE_MODES", + "DEFAULT_MODE", + "normalize_mode", + "RebuildSummary", + "playlist_dir_for", + "materialize_one", + "rebuild_playlist_folder", +] diff --git a/tests/test_playlist_materialize.py b/tests/test_playlist_materialize.py new file mode 100644 index 00000000..daf397e8 --- /dev/null +++ b/tests/test_playlist_materialize.py @@ -0,0 +1,143 @@ +"""Playlist materialization seam — a playlist folder is a derived view of links +into the real library. Locks down: symlink vs copy, the auto-fallback when +symlinks aren't supported, idempotency, stale-link pruning, collision handling, +and that nothing is ever written outside the playlists root.""" + +from __future__ import annotations + +import os +from pathlib import Path + +from core.playlists.materialize import ( + DEFAULT_MODE, + materialize_one, + normalize_mode, + playlist_dir_for, + rebuild_playlist_folder, +) + + +def _library(tmp_path: Path) -> list[str]: + a = tmp_path / "Music" / "Daft Punk" / "Discovery" / "One More Time.mp3" + b = tmp_path / "Music" / "Queen" / "A Night at the Opera" / "Bohemian Rhapsody.mp3" + for f in (a, b): + f.parent.mkdir(parents=True, exist_ok=True) + f.write_bytes(b"audio") + return [str(a), str(b)] + + +def test_normalize_mode(): + assert normalize_mode("symlink") == "symlink" + assert normalize_mode("COPY") == "copy" + assert normalize_mode("") == DEFAULT_MODE + assert normalize_mode(None) == DEFAULT_MODE + assert normalize_mode("nonsense") == DEFAULT_MODE + + +def test_symlink_mode_creates_relative_links(tmp_path: Path): + real = _library(tmp_path) + s = rebuild_playlist_folder(str(tmp_path / "Playlists"), "Road Trip", real, mode="symlink") + assert s.linked == 2 and s.copied == 0 and not s.fellback + link = Path(s.playlist_dir) / "One More Time.mp3" + assert link.is_symlink() + assert not os.path.isabs(os.readlink(link)) # relative for portability + assert link.resolve() == Path(real[0]).resolve() + assert link.read_bytes() == b"audio" + + +def test_symlink_mode_idempotent(tmp_path: Path): + real = _library(tmp_path) + root = str(tmp_path / "Playlists") + rebuild_playlist_folder(root, "Mix", real, mode="symlink") + s2 = rebuild_playlist_folder(root, "Mix", real, mode="symlink") + assert s2.unchanged == 2 and s2.linked == 0 and s2.removed_stale == 0 + + +def test_copy_mode_duplicates_real_files(tmp_path: Path): + real = _library(tmp_path) + s = rebuild_playlist_folder(str(tmp_path / "Playlists"), "USB", real, mode="copy") + assert s.copied == 2 and s.linked == 0 + f = Path(s.playlist_dir) / "Bohemian Rhapsody.mp3" + assert f.is_file() and not f.is_symlink() and f.read_bytes() == b"audio" + + +def test_copy_mode_idempotent(tmp_path: Path): + real = _library(tmp_path) + root = str(tmp_path / "Playlists") + rebuild_playlist_folder(root, "Mix", real, mode="copy") + s2 = rebuild_playlist_folder(root, "Mix", real, mode="copy") + assert s2.unchanged == 2 and s2.copied == 0 + + +def test_falls_back_to_copy_when_symlinks_unsupported(tmp_path: Path): + real = _library(tmp_path) + + def _no_symlinks(target, link): + raise OSError("symlinks not supported here") + + s = rebuild_playlist_folder(str(tmp_path / "Playlists"), "Car", real, + mode="symlink", symlink_fn=_no_symlinks) + assert s.fellback is True and s.copied == 2 and s.linked == 0 + f = Path(s.playlist_dir) / "One More Time.mp3" + assert f.is_file() and not f.is_symlink() and f.read_bytes() == b"audio" + + +def test_rebuild_prunes_entries_no_longer_in_playlist(tmp_path: Path): + real = _library(tmp_path) + root = str(tmp_path / "Playlists") + rebuild_playlist_folder(root, "Mix", real, mode="symlink") # 2 entries + s = rebuild_playlist_folder(root, "Mix", real[:1], mode="symlink") # drop one + assert s.removed_stale == 1 + pdir = Path(s.playlist_dir) + assert (pdir / "One More Time.mp3").exists() + assert not (pdir / "Bohemian Rhapsody.mp3").exists() + + +def test_prune_stale_can_be_disabled(tmp_path: Path): + real = _library(tmp_path) + root = str(tmp_path / "Playlists") + rebuild_playlist_folder(root, "Mix", real, mode="symlink") + s = rebuild_playlist_folder(root, "Mix", real[:1], mode="symlink", prune_stale=False) + assert s.removed_stale == 0 + assert (Path(s.playlist_dir) / "Bohemian Rhapsody.mp3").exists() + + +def test_switching_mode_replaces_links_with_copies(tmp_path: Path): + real = _library(tmp_path) + root = str(tmp_path / "Playlists") + rebuild_playlist_folder(root, "Mix", real, mode="symlink") + s = rebuild_playlist_folder(root, "Mix", real, mode="copy") + f = Path(s.playlist_dir) / "One More Time.mp3" + assert f.is_file() and not f.is_symlink() and s.copied == 2 + + +def test_basename_collision_is_disambiguated_not_overwritten(tmp_path: Path): + p1 = tmp_path / "Music" / "A" / "AlbumA" / "01 - Intro.mp3" + p2 = tmp_path / "Music" / "B" / "AlbumB" / "01 - Intro.mp3" + for f, data in ((p1, b"one"), (p2, b"two")): + f.parent.mkdir(parents=True, exist_ok=True) + f.write_bytes(data) + s = rebuild_playlist_folder(str(tmp_path / "Playlists"), "Dup", [str(p1), str(p2)], mode="copy") + pdir = Path(s.playlist_dir) + assert (pdir / "01 - Intro.mp3").read_bytes() == b"one" + assert (pdir / "01 - Intro (2).mp3").read_bytes() == b"two" # second kept, not lost + assert s.copied == 2 + + +def test_missing_source_is_counted_not_fatal(tmp_path: Path): + real = _library(tmp_path) + s = rebuild_playlist_folder(str(tmp_path / "Playlists"), "Mix", + real + [str(tmp_path / "gone.mp3")], mode="symlink") + assert s.linked == 2 and s.missing_source == 1 + + +def test_playlist_name_cannot_escape_root(tmp_path: Path): + root = tmp_path / "Playlists" + nasty = playlist_dir_for(str(root), "../../etc/evil") + assert os.path.abspath(nasty).startswith(os.path.abspath(str(root)) + os.sep) + + +def test_materialize_one_missing_source(tmp_path: Path): + dest = tmp_path / "Playlists" / "X" / "x.mp3" + assert materialize_one(str(tmp_path / "nope.mp3"), str(dest), "symlink") == "missing" + assert not dest.exists()