andrew
543f10e662
run-playlist.sh and import-track.sh had the same beets->#EXTM3U block, but import-track.sh wrote the file non-atomically. Move it to pipeline/lib/m3u.sh as regen_m3u <name> <out>, sourced by both. The helper always writes atomically (temp + rename), so a reader like Navidrome never sees a partial file and a stray wrong-owner leftover .m3u8 is replaced without a chown. Returns 0 always (empty result is a warning), so it's safe as a standalone call under set -e. Verified: tests/test_m3u.py (atomic write with tracks, no file when empty; 44 tests green) plus a live regen against the real library (techno -> 229 tracks, no temp leftover). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
39 lines
1.4 KiB
Python
39 lines
1.4 KiB
Python
"""regen_m3u (pipeline/lib/m3u.sh): writes #EXTM3U + track paths atomically,
|
|
and leaves the file untouched when the playlist has no tracks. `beet` is stubbed
|
|
so no real library is needed."""
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
REPO = Path(__file__).resolve().parent.parent
|
|
M3U_SH = REPO / "pipeline" / "lib" / "m3u.sh"
|
|
|
|
|
|
def _run(work: Path, playlist: str, out: Path) -> subprocess.CompletedProcess:
|
|
script = r"""
|
|
set -euo pipefail
|
|
export LOG="{work}/log.txt"; : > "$LOG"
|
|
log() {{ echo "$*" >> "$LOG"; }}
|
|
# stub beet: two paths only for grouping:techno, nothing otherwise
|
|
beet() {{ case "$*" in *grouping:techno*) printf '/music/a.flac\n/music/b.flac\n';; esac; }}
|
|
source "{m3u}"
|
|
regen_m3u "{playlist}" "{out}"
|
|
""".format(work=work, m3u=M3U_SH, playlist=playlist, out=out)
|
|
return subprocess.run(["bash", "-c", script], capture_output=True, text=True)
|
|
|
|
|
|
def test_writes_m3u_with_tracks(tmp_path):
|
|
out = tmp_path / "sub" / "techno.m3u8"
|
|
proc = _run(tmp_path, "techno", out)
|
|
assert proc.returncode == 0, proc.stderr
|
|
assert out.read_text() == "#EXTM3U\n/music/a.flac\n/music/b.flac\n"
|
|
# atomic write leaves no temp file behind
|
|
assert not list(out.parent.glob("*.tmp.*"))
|
|
|
|
|
|
def test_no_tracks_leaves_no_file(tmp_path):
|
|
out = tmp_path / "empty.m3u8"
|
|
proc = _run(tmp_path, "empty", out)
|
|
assert proc.returncode == 0, proc.stderr
|
|
assert not out.exists()
|
|
assert "no tracks" in (tmp_path / "log.txt").read_text()
|