Playlists: config (separate root + symlink/copy) + pure materializer seam
- settings: playlists.materialize_path (separate root, mapped apart from the music library so the media server never double-scans it) + materialize_mode (symlink|copy). - core/playlists/materialize.py: pure filesystem engine that (re)builds a playlist folder of relative symlinks (or copies) into the real library — idempotent, prunes stale entries, disambiguates filename collisions, never escapes the root, and auto-falls-back to copy when the FS can't symlink. No DB, no app state; ops injectable. 13 unit tests. Isolated + additive — nothing live calls this yet (stitcher/trigger/routing come next).
This commit is contained in:
BoulderBadgeDad
parent
37431ea82b
commit
3a6cb8cda5
3 changed files with 368 additions and 0 deletions
|
|
@ -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
|
||||
|
|
|
|||
214
core/playlists/materialize.py
Normal file
214
core/playlists/materialize.py
Normal file
|
|
@ -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/<playlist_name>/`` 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",
|
||||
]
|
||||
143
tests/test_playlist_materialize.py
Normal file
143
tests/test_playlist_materialize.py
Normal file
|
|
@ -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()
|
||||
Loading…
Reference in a new issue