"""Filename sanitisation, path templates, and container extensions.
The two failure modes here are data loss (two different tracks landing
in the same file) and unreadable files (illegal characters, 255-byte
paths, control bytes). Sanitisation is conservative: characters that
are illegal on any common OS are replaced, unicode is normalised
(NFKC) but otherwise preserved, and names are length-capped *before*
the extension is attached.
"""
from __future__ import annotations
import re
import unicodedata
from pathlib import Path
from .models import Album, Playlist, Track
from .quality import quality_label
# Illegal on Windows/POSIX (POSIX: '/', NUL; Windows: the rest).
_ILLEGAL_RE = re.compile(r'[<>:"/\\|?*\x00-\x1f\x7f]')
# Characters that are legal but poison folder/file UX (trailing dots,
# device names on Windows are handled separately).
_DEVICE_NAMES = frozenset(
{
"CON",
"PRN",
"AUX",
"NUL",
"COM1",
"COM2",
"COM3",
"COM4",
"LPT1",
"LPT2",
"LPT3",
"LPT4",
"LPT5",
"LPT6",
"LPT7",
}
)
DEFAULT_MAX_NAME_LEN = 120
[docs]
def sanitize_name(name: str | None, max_length: int = DEFAULT_MAX_NAME_LEN) -> str:
"""Make an arbitrary title safe for use as a single path component."""
if not name or not str(name).strip():
return "untitled"
text = unicodedata.normalize("NFKC", str(name))
text = _ILLEGAL_RE.sub(" ", text)
text = re.sub(r"\s+", " ", text).strip()
text = text.strip(". ")
if not text:
return "untitled"
if text.upper() in _DEVICE_NAMES:
text = f"_{text}"
if max_length > 0 and len(text) > max_length:
text = text[:max_length].rstrip(" .")
return text or "untitled"
def _year_from_release(release_date: str | None) -> str:
if not release_date:
return ""
match = re.match(r"(\d{4})", str(release_date))
return match.group(1) if match else ""
def _template(text: str, mapping: dict[str, str]) -> str:
result = text
for key, value in mapping.items():
result = result.replace(key, value or "")
return re.sub(r"\s+", " ", result).strip(" -/")
[docs]
def quality_flag(album: Album) -> str:
"""The TMD-style ``[M]``/``[E]`` marker letters for a collection.
``M`` when the catalog advertises hi-res, ``E`` when explicit,
joined without a separator (``"ME"``). Empty when neither — the
``{flag}`` placeholder substitutes nothing in that case.
"""
parts: list[str] = []
if (album.audio_quality or "").upper() in ("HI_RES", "HI_RES_LOSSLESS"):
parts.append("M")
if album.explicit:
parts.append("E")
return "".join(parts)
def _collection_dir(
template: str,
artist: str,
title: str,
year: str = "",
flag: str = "",
collection_id: str = "",
quality: str = "",
collection_type: str = "",
) -> str:
return (
_template(
template,
{
"{artist}": artist,
"{album}": title,
"{title}": title,
"{year}": year,
"{flag}": f"[{flag}] " if flag else "",
"{id}": collection_id,
"{quality}": quality,
"{type}": collection_type,
},
)
or "untitled"
)
[docs]
def album_dir(album: Album, template: str) -> str:
artist = ""
if album.primary_artist and album.primary_artist.name:
artist = album.primary_artist.name
elif album.artists:
artist = ", ".join(a.name for a in album.artists if a.name)
title = album.title or f"album-{album.id}"
return _collection_dir(
template,
sanitize_name(artist),
sanitize_name(title),
_year_from_release(album.release_date),
quality_flag(album),
str(album.id or ""),
quality_label(album.audio_quality),
album.album_type or "",
)
[docs]
def playlist_dir(playlist: Playlist, template: str) -> str:
title = playlist.title or f"{playlist.kind}-{playlist.id}"
return _collection_dir(
template,
"",
sanitize_name(title),
collection_id=str(playlist.id or ""),
collection_type=playlist.kind,
)
[docs]
def track_stem(
track: Track,
track_format: str,
album: Album | None = None,
position: int | None = None,
quality: str | None = None,
) -> str:
"""The final filename without extension.
``quality`` is the *delivered* quality (the stream's
``delivered_quality``); when absent, ``{quality}`` falls back to
the catalog's ``track.audio_quality`` (reference-client parity:
the label reflects what is actually on disk, downgrades included).
"""
number = track.number_label if not position else f"{position:02d}"
title = track.title or f"track-{track.id}"
if track.version:
title = f"{title} ({track.version})"
explicit = " (Explicit)" if track.explicit else ""
artist = track.artist_name or (
(album.primary_artist.name if album and album.primary_artist else "")
)
artists = ", ".join(a.name for a in track.artists if a.name)
album_title = album.title if album is not None and album.title else ""
year = _year_from_release(album.release_date) if album is not None else ""
stem = _template(
track_format,
{
"{number}": number,
"{artist}": sanitize_name(artist),
"{artists}": sanitize_name(artists) if artists else "",
"{title}": sanitize_name(title),
"{explicit}": explicit,
"{album}": sanitize_name(album_title) if album_title else "",
"{year}": year,
"{quality}": quality_label(quality or track.audio_quality),
"{id}": str(track.id or ""),
},
)
return sanitize_name(stem)
[docs]
def cd_folder(album: Album | None, track: Track) -> str | None:
"""``CD2``-style subfolder for multi-disc albums (TMD parity).
Single-disc (or unknown) albums keep a flat layout: ``None``.
"""
if album is not None and album.number_of_volumes > 1 and track.volume_number:
return f"CD{track.volume_number}"
return None
[docs]
def track_extension(codec: str | None, stream_url: str = "") -> str:
"""Container extension from the *delivered* codec (not the request).
Tidal serves FLAC both in native FLAC and inside an MP4/DASH
container; lossless-in-MP4 keeps ``.m4a`` unless the host remuxes
(see :mod:`tidalpy.media`). Dolby/MQA renditions land in ``.mp4``.
"""
lowered = (codec or "").lower()
url = (stream_url or "").lower()
if ".flac" in url:
return ".flac"
if "flac" in lowered:
return ".flac"
if (
"ac4" in lowered
or "ac-4" in lowered
or "mha1" in lowered
or "mqa" in lowered
or "dolby" in lowered
):
return ".mp4"
return ".m4a"
[docs]
def unique_path(path: Path) -> Path:
"""``a.flac`` -> ``a (1).flac`` ... when the target already exists.
A host-side utility for layouts the library does not manage. The
pipeline itself uses deterministic in-run disambiguation (see
:mod:`tidalpy.pipeline`) instead, so re-runs map to the same
paths — a disk-scan like this one would not.
"""
if not path.exists():
return path
for i in range(1, 1000):
candidate = path.with_name(f"{path.stem} ({i}){path.suffix}")
if not candidate.exists():
return candidate
return path.with_name(f"{path.stem} ({i + 1}){path.suffix}")
[docs]
def track_path(
track: Track,
extension: str,
dest_dir: Path,
track_format: str,
album: Album | None = None,
position: int | None = None,
quality: str | None = None,
) -> Path:
"""Full destination path for one track inside ``dest_dir``.
``extension`` is the *final* container extension (decided by the
caller, which knows whether a FLAC-in-MP4 stream will be remuxed).
The path is *canonical* (no disambiguation suffixes); if two
different tracks would collide, the pipeline assigns the later one
a deterministic `` (index)`` suffix — the skip-existing/resume
logic relies on stable, reproducible paths. Multi-disc albums land
in ``CD{n}`` subfolders (TMD parity).
"""
stem = track_stem(
track, track_format, album=album, position=position, quality=quality
)
base = dest_dir
cd = cd_folder(album, track)
if cd:
base = base / cd
return base / f"{stem}{extension}"