9aaafaf341
Files inside an Organize-by-Playlist folder were stuck with the library filename (materializer hardcoded os.path.basename) — users wanted control over the naming, e.g. a playlist-order prefix so a DAP plays them in order. Add an opt-in FILENAME template "Playlist File Naming" (file_organization.templates .playlist_item), tokens $position/$artist/$album/$track/$title. It is a filename, not a path: validated to forbid "/" or "\" and to require $title, both in the Settings UI (blocks save with a reason) and in core/playlists/item_naming.py, which also fails safe at apply time — a bad/empty template falls back to the library filename, so it can never produce a broken name. Default empty = current behavior. Works for symlink AND copy modes (a symlink name is independent of its target). Applied in _rebuild_one_from_db (the live reconcile/rebuild path), which has the per-track metadata + playlist order; the pure FS materializer just gained an optional dest_names override and is otherwise untouched. $position is zero-padded to playlist width for correct sorting. Tests: pure validate/render (slash + missing-title rejected, fallback, sanitize, no-separator guarantee), FS-layer dest_names + collision disambiguation + back- compat, and end-to-end through the DB rebuild (07->01 rename + empty-template keeps library filename).
247 lines
11 KiB
Python
247 lines
11 KiB
Python
"""Build a playlist's materialized folder from a FINISHED organize-by-playlist
|
||
download batch.
|
||
|
||
The batch already carries the real on-disk location of every resolved track:
|
||
|
||
- **owned** tracks → ``analysis_results[*]['matched_file_path']`` (the library
|
||
matcher's result, captured during analysis), and
|
||
- **downloaded** tracks → ``download_tasks[tid]['final_file_path']`` (where the
|
||
import landed).
|
||
|
||
So this is pure stitching + filesystem work: **no re-matching, no source-ID
|
||
lookup, no mirrored-playlist resolution**. It works for any organize-by-playlist
|
||
download, mirrored or not, and for the all-owned case (where nothing downloads).
|
||
|
||
Each stored path is run through ``resolve_library_file_path`` — the same resolver
|
||
playback uses — so a DB path that's host-formatted (Docker) maps to the real file
|
||
the container can see; a freshly-downloaded path already exists and passes
|
||
through unchanged.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any, List, Optional
|
||
|
||
from core.imports.paths import docker_resolve_path
|
||
from core.playlists.materialize import (
|
||
RebuildSummary,
|
||
normalize_mode,
|
||
rebuild_playlist_folder,
|
||
)
|
||
from utils.logging_config import get_logger
|
||
|
||
logger = get_logger("playlists.materialize")
|
||
|
||
|
||
def collect_batch_real_paths(batch: dict, download_tasks: dict, *, config_manager) -> List[str]:
|
||
"""Real on-disk paths of every track in a finished batch that resolved —
|
||
owned (from analysis) + downloaded (from completed tasks) — resolved to this
|
||
process's filesystem and de-duplicated, in playlist order then download order."""
|
||
from core.library.path_resolver import resolve_library_file_path
|
||
|
||
out: List[str] = []
|
||
seen = set()
|
||
|
||
def _add(stored_path: Any) -> None:
|
||
if not stored_path:
|
||
return
|
||
real = resolve_library_file_path(str(stored_path), config_manager=config_manager)
|
||
if real and real not in seen:
|
||
seen.add(real)
|
||
out.append(real)
|
||
|
||
for res in (batch.get("analysis_results") or []):
|
||
if res.get("found"):
|
||
_add(res.get("matched_file_path"))
|
||
|
||
for tid in (batch.get("queue") or []):
|
||
task = (download_tasks or {}).get(tid) or {}
|
||
if task.get("status") == "completed":
|
||
_add(task.get("final_file_path"))
|
||
|
||
return out
|
||
|
||
|
||
def materialize_playlist_from_batch(batch: dict, download_tasks: dict, config_manager) -> Optional[RebuildSummary]:
|
||
"""(Re)build the playlist folder for a finished organize-by-playlist batch.
|
||
|
||
Returns the :class:`RebuildSummary`, or ``None`` when the batch isn't an
|
||
organize-by-playlist batch. Reads ``playlists.materialize_path`` /
|
||
``playlists.materialize_mode`` from config."""
|
||
if not batch or not batch.get("playlist_folder_mode"):
|
||
return None
|
||
name = batch.get("playlist_name") or "Unknown Playlist"
|
||
real_paths = collect_batch_real_paths(batch, download_tasks, config_manager=config_manager)
|
||
root = docker_resolve_path(config_manager.get("playlists.materialize_path", "./Playlists"))
|
||
mode = normalize_mode(config_manager.get("playlists.materialize_mode", "symlink"))
|
||
return rebuild_playlist_folder(root, name, real_paths, mode)
|
||
|
||
|
||
def reconcile_batch_playlists(db, batch: dict, download_tasks: dict, config_manager, *, profile_id: int = 1):
|
||
"""One post-batch step: rebuild every organize-by-playlist playlist this batch
|
||
TOUCHED, from CURRENT library ownership.
|
||
|
||
Touched playlists = the batch's own organize playlist + any playlist a completed
|
||
track belongs to (via the per-track ``source_info`` provenance — covers a
|
||
wishlist batch fulfilling a track that belongs to an organize playlist). Each is
|
||
rebuilt via ``_rebuild_one_from_db`` (``check_track_exists`` over its membership),
|
||
so it's robust to HOW a track imported — modal worker, slskd monitor, or the
|
||
verification worker, which don't all set the same task fields. It simply asks the
|
||
library what's owned, and prunes tracks that have left the playlist.
|
||
|
||
Returns ``[(playlist_name, RebuildSummary)]``. Callers wrap non-fatally."""
|
||
import json as _json
|
||
|
||
# Collect the (ref, source) of every organize playlist this batch touched.
|
||
# (ref, source, force). force=True for the batch's OWN playlist: the per-download
|
||
# "organize by playlist" toggle (batch playlist_folder_mode) is the user's
|
||
# intent, so it rebuilds regardless of the saved row preference. Provenance
|
||
# playlists (force=False) require the saved organize_by_playlist flag.
|
||
wanted, seen_ref = [], set()
|
||
|
||
def _want(ref, source, force):
|
||
ref = str(ref or "").strip()
|
||
if not ref:
|
||
return
|
||
key = (ref, source or "spotify")
|
||
if key not in seen_ref:
|
||
seen_ref.add(key)
|
||
wanted.append((ref, source or "spotify", force))
|
||
|
||
if batch.get("playlist_folder_mode"):
|
||
_want(batch.get("source_playlist_ref") or batch.get("playlist_id"),
|
||
batch.get("batch_source") or "spotify", True)
|
||
|
||
for tid in (batch.get("queue") or []):
|
||
task = (download_tasks or {}).get(tid) or {}
|
||
if task.get("status") != "completed":
|
||
continue
|
||
si = (task.get("track_info") or {}).get("source_info") or {}
|
||
if isinstance(si, str):
|
||
try:
|
||
si = _json.loads(si)
|
||
except Exception:
|
||
si = {}
|
||
if isinstance(si, dict) and si.get("playlist_id"):
|
||
_want(si["playlist_id"], si.get("source") or "spotify", False)
|
||
|
||
results, seen_id = [], set()
|
||
for ref, source, force in wanted:
|
||
try:
|
||
pl = db.resolve_mirrored_playlist(ref, profile_id, default_source=source)
|
||
except Exception as e:
|
||
logger.debug("[Playlist Folder] resolve failed for ref=%s source=%s: %s", ref, source, e)
|
||
pl = None
|
||
if not pl:
|
||
logger.info("[Playlist Folder] no mirrored playlist resolved for ref=%s source=%s "
|
||
"(force=%s) — folder not built", ref, source, force)
|
||
continue
|
||
if pl.get("id") in seen_id:
|
||
continue
|
||
if not force and not pl.get("organize_by_playlist"):
|
||
continue
|
||
seen_id.add(pl.get("id"))
|
||
try:
|
||
results.append(_rebuild_one_from_db(db, config_manager, pl))
|
||
except Exception as e:
|
||
logger.error("[Playlist Folder] rebuild failed for '%s': %s", pl.get("name"), e)
|
||
continue
|
||
return results
|
||
|
||
|
||
def _rebuild_one_from_db(db, config_manager, playlist: dict):
|
||
"""Rebuild ONE playlist's folder from its CURRENT membership × ownership —
|
||
re-matching each member via the app's own ``check_track_exists`` (by name, not
|
||
source IDs), resolving to disk, and rebuilding WITH prune. Because it's driven
|
||
by current membership, a track that has LEFT the playlist drops out of the set
|
||
and its symlink is pruned. Returns ``(playlist_name, RebuildSummary)``."""
|
||
import os as _os
|
||
|
||
from core.library.path_resolver import resolve_library_file_path
|
||
from core.playlists.item_naming import render_playlist_item_name
|
||
|
||
root = docker_resolve_path(config_manager.get("playlists.materialize_path", "./Playlists"))
|
||
mode = normalize_mode(config_manager.get("playlists.materialize_mode", "symlink"))
|
||
item_template = ((config_manager.get("file_organization.templates", {}) or {})
|
||
.get("playlist_item", "") or "").strip()
|
||
|
||
# Resolve owned tracks to real paths IN PLAYLIST ORDER, keeping the metadata
|
||
# so an optional custom filename template ($position/$artist/$title/...) can
|
||
# be applied. $position is the playlist index, which is exactly this order.
|
||
resolved: List[dict] = []
|
||
seen = set()
|
||
for t in (db.get_mirrored_playlist_tracks(playlist["id"]) or []):
|
||
title = (t.get("track_name") or "").strip()
|
||
artist = (t.get("artist_name") or "").strip()
|
||
if not title:
|
||
continue
|
||
try:
|
||
db_track, conf = db.check_track_exists(title, artist, confidence_threshold=0.7)
|
||
except Exception:
|
||
continue
|
||
if db_track is None or conf < 0.7:
|
||
continue
|
||
real = resolve_library_file_path(getattr(db_track, "file_path", None), config_manager=config_manager)
|
||
if real and real not in seen:
|
||
seen.add(real)
|
||
resolved.append({
|
||
"real": real,
|
||
"title": title,
|
||
"artist": artist,
|
||
"album": (t.get("album_name") or t.get("album") or "").strip(),
|
||
"track": getattr(db_track, "track_number", None),
|
||
})
|
||
|
||
real_paths: List[str] = [r["real"] for r in resolved]
|
||
|
||
dest_names = None
|
||
if item_template:
|
||
width = max(2, len(str(len(resolved)))) # zero-pad $position for correct sorting
|
||
dest_names = [
|
||
render_playlist_item_name(
|
||
item_template,
|
||
title=r["title"], artist=r["artist"], album=r["album"], track=r["track"],
|
||
position=f"{i:0{width}d}",
|
||
ext=_os.path.splitext(r["real"])[1],
|
||
fallback_name=_os.path.basename(r["real"]),
|
||
)
|
||
for i, r in enumerate(resolved, start=1)
|
||
]
|
||
|
||
name = playlist.get("name") or "Unnamed Playlist"
|
||
return name, rebuild_playlist_folder(root, name, real_paths, mode, dest_names=dest_names) # prune_stale=True
|
||
|
||
|
||
def rebuild_organized_playlists_from_db(db, config_manager, *, profile_id: int = 1):
|
||
"""Rebuild EVERY "organize by playlist" folder from current library ownership —
|
||
for the manual "Rebuild" button. Self-heals after a library reorganize moves
|
||
files or membership changes. Returns a list of ``(playlist_name, summary)``."""
|
||
return [
|
||
_rebuild_one_from_db(db, config_manager, pl)
|
||
for pl in (db.get_mirrored_playlists(profile_id) or [])
|
||
if pl.get("organize_by_playlist")
|
||
]
|
||
|
||
|
||
def rebuild_mirrored_playlist_if_organized(db, config_manager, playlist_id, *, profile_id: int = 1):
|
||
"""Mirror-update hook: after a playlist's membership is re-synced, rebuild its
|
||
folder (with prune) IF it's organize-by-playlist — so a track that just LEFT
|
||
the playlist has its symlink cleaned up the instant membership changes (the
|
||
mirror image of the post-download reconcile that handles additions). Returns
|
||
the summary, or ``None`` when the playlist isn't organized / can't be found."""
|
||
if playlist_id is None:
|
||
return None
|
||
pl = db.get_mirrored_playlist(playlist_id)
|
||
if not pl or not pl.get("organize_by_playlist"):
|
||
return None
|
||
_name, summary = _rebuild_one_from_db(db, config_manager, pl)
|
||
return summary
|
||
|
||
|
||
__all__ = [
|
||
"collect_batch_real_paths",
|
||
"materialize_playlist_from_batch",
|
||
"reconcile_batch_playlists",
|
||
"rebuild_organized_playlists_from_db",
|
||
"rebuild_mirrored_playlist_if_organized",
|
||
]
|