feat: rebuild the moods from real tags, add exclusions, fix playlist ownership #12

Merged
lyrathorpe merged 3 commits from feat/moods-from-real-tags into main 2026-08-24 18:48:44 +01:00
3 changed files with 104 additions and 3 deletions
Showing only changes of commit a7d16ca0b2 - Show all commits
+7
View File
@@ -261,6 +261,13 @@ in step by hand. Override it if that guess is wrong.
Entries are written **relative to the playlist file**, so one playlist works
from the NAS, from a Mac over SMB, and from Linux, without rewriting.
Each playlist is given the **owner and group of the mirror** it is written
into. The image runs as root by default so that a bind mount of any ownership
stays writable, and the cost of that is output owned by root — which the account
serving the share cannot read, group bit or no group bit, because the group is
also root. Copying the mirror's own ownership avoids having to be told what it
should be, and does nothing when the two already agree.
A track is only listed once its mirror file has been confirmed to exist. Lidarr
holding the FLAC says nothing about whether the MP3 has been encoded yet. If a
large number are missing, the run says so — that is what a wrong `--library-root`
+44 -3
View File
@@ -1534,7 +1534,7 @@ def build_vibe_playlists(store, vibes, mirror_root, library_root, limit, now):
continue
entries.append({**dict(row), "mirror": mirror})
write_playlist(directory / f"{vibe['name']}.m3u", entries)
write_playlist(directory / f"{vibe['name']}.m3u", entries, Path(mirror_root))
total += len(entries)
logger.info("playlist %-20s %4d tracks -- by tag", vibe["name"], len(entries))
@@ -1572,7 +1572,40 @@ def mirror_path_for(source, library_root, mirror_root):
return (Path(mirror_root) / relative).with_suffix(MIRROR_SUFFIX)
def write_playlist(path, entries):
def set_ownership(path, uid, gid):
"""Give a path an owner and group. Returns whether anything changed."""
try:
current = path.stat()
if (current.st_uid, current.st_gid) == (uid, gid):
return False
os.chown(path, uid, gid)
except OSError:
# Not permitted unless running as root, which is the case where the
# ownership is already whatever the caller runs as.
return False
return True
def match_ownership(path, reference):
"""Give a path the owner and group of the tree it is joining.
The image runs as root by default, so that a bind mount of any ownership
stays writable. The cost is that everything it writes comes out root-owned,
and a root-owned playlist inside a mirror owned by the apps account is
unreadable to the thing that serves it -- the group bit does not help when
the group is root.
Copying the mirror's own ownership avoids having to be told what it should
be, and is a no-op when the two already agree.
"""
try:
wanted = reference.stat()
except OSError:
return False
return set_ownership(path, wanted.st_uid, wanted.st_gid)
def write_playlist(path, entries, reference=None):
"""Write one extended M3U, atomically.
Paths are relative to the playlist file, so the same playlist works from the
@@ -1584,7 +1617,11 @@ def write_playlist(path, entries):
lines.append(f"#EXTINF:{seconds},{entry['artist']} - {entry['title']}")
lines.append(os.path.relpath(entry["mirror"], path.parent))
fresh = not path.parent.exists()
path.parent.mkdir(parents=True, exist_ok=True)
if fresh and reference is not None:
match_ownership(path.parent, reference)
handle, temporary = tempfile.mkstemp(dir=path.parent, suffix=".m3u.part")
os.close(handle)
temporary = Path(temporary)
@@ -1595,6 +1632,10 @@ def write_playlist(path, entries):
mode = temporary.stat().st_mode
if not mode & GROUP_READ:
temporary.chmod(mode | GROUP_READ)
# Before the rename, so the playlist is never briefly visible owned by
# the wrong account.
if reference is not None:
match_ownership(temporary, reference)
os.replace(temporary, path)
finally:
temporary.unlink(missing_ok=True)
@@ -1625,7 +1666,7 @@ def build_playlists(store, mirror_root, library_root, limit, now):
missing += 1
continue
entries.append({**dict(row), "mirror": mirror})
write_playlist(directory / f"{name}.m3u", entries)
write_playlist(directory / f"{name}.m3u", entries, Path(mirror_root))
total += len(entries)
logger.info("playlist %-20s %4d tracks -- %s", name, len(entries), description)
+53
View File
@@ -1,4 +1,5 @@
import json
import os
import stat
import urllib.error
from pathlib import Path
@@ -1289,3 +1290,55 @@ def test_a_vibe_cannot_both_select_and_exclude_a_tag(tmp_path):
with pytest.raises(ValueError, match="selects on and excludes"):
music_curator.load_vibes(str(path))
def test_matching_ownership_is_a_no_op_when_it_already_agrees(tmp_path):
target = tmp_path / "file"
target.write_text("x")
assert music_curator.match_ownership(target, tmp_path) is False
def test_ownership_failure_is_tolerated(tmp_path):
"""Not permitted unless running as root -- which is exactly the case where
the ownership is already whatever the caller runs as."""
target = tmp_path / "file"
target.write_text("x")
# uid 0 from a non-root test process: refused, and must not raise.
assert music_curator.set_ownership(target, 0, 0) is False
def test_a_playlist_is_chowned_to_match_the_mirror(tmp_path, monkeypatch):
"""The image runs as root, so its output is root-owned, and a root-owned
playlist in an apps-owned mirror is unreadable to whatever serves it."""
api, source, mirror = playlist_library(tmp_path)
store = store_at(tmp_path)
music_curator.index_library(
music_curator.Lidarr("http://lidarr", "key", transport=api), store
)
music_curator.match_library(store)
attempted = []
real_stat = music_curator.Path.stat
def pretend_mirror_is_owned_by_568(self, *args, **kwargs):
info = real_stat(self, *args, **kwargs)
if self == mirror:
return os.stat_result(
(info.st_mode, info.st_ino, info.st_dev, info.st_nlink, 568, 568,
info.st_size, int(info.st_atime), int(info.st_mtime), int(info.st_ctime))
)
return info
monkeypatch.setattr(music_curator.Path, "stat", pretend_mirror_is_owned_by_568)
monkeypatch.setattr(
music_curator.os, "chown", lambda p, u, g: attempted.append((str(p), u, g))
)
music_curator.build_playlists(store, mirror, str(source), 100, NOW)
assert attempted, "no ownership was applied"
assert all(tuple(owner) == (568, 568) for _, *owner in attempted)
# The temporary file, before the rename, never the finished playlist.
assert all(path.endswith(".part") or path.endswith("_playlists") for path, *_ in attempted)