Emma ThorpeandClaude Opus 5
9a3b4f9955
Build and publish container / build (pull_request) Successful in 1m43s
A ReplayGain album gain belongs to the whole record, so a track arriving or leaving rewrites the tags on every one of its siblings. rsgain fits the new values into the padding its previous write left behind, which changes neither the file's size nor its mtime: before: 277757 bytes, mtime 1577880000, album gain 3.75 dB after: 277757 bytes, mtime 1577880000, album gain 6.25 dB Those are the two things rsync's quick check compares, so the siblings are invisible to it and the device keeps the old gains indefinitely. The track that arrived or left is always visible. So sync-to-ipod.sh now runs a second pass over the albums the first one touched, with --ignore-times to defeat the same quick check. touched_albums.py derives the list from rsync's own report of what it moved, which costs no extra traversal of either tree, and leaves out the tracks the first pass has already copied so a whole-album quality upgrade is not sent twice. Albums whose file set has not changed are left alone. A dry run reports how many further tracks are involved. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
129 lines
4.0 KiB
Python
129 lines
4.0 KiB
Python
"""A ReplayGain re-level rewrites a track's tags without changing its size or
|
|
its mtime, which is precisely the pair rsync's quick check compares. These
|
|
cover the list that is fed back to rsync to copy those tracks anyway."""
|
|
|
|
import io
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools"))
|
|
|
|
import touched_albums # noqa: E402
|
|
|
|
|
|
def album(root, name, *tracks):
|
|
directory = root / name
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
for track in tracks:
|
|
(directory / track).write_bytes(b"x")
|
|
return directory
|
|
|
|
|
|
def listing(mirror, *lines):
|
|
return touched_albums.remaining(mirror, *touched_albums.touched(list(lines)))
|
|
|
|
|
|
def test_the_siblings_of_a_new_track_are_listed(tmp_path):
|
|
album(tmp_path, "Artist/Album", "01.mp3", "02.mp3", "03.mp3")
|
|
|
|
assert listing(tmp_path, "4096 Artist/Album/03.mp3") == [
|
|
"Artist/Album/01.mp3",
|
|
"Artist/Album/02.mp3",
|
|
]
|
|
|
|
|
|
def test_a_track_rsync_just_copied_is_not_copied_twice(tmp_path):
|
|
"""A quality upgrade replaces every track on the record. Listing them again
|
|
would send the album across twice."""
|
|
album(tmp_path, "Artist/Album", "01.mp3", "02.mp3")
|
|
|
|
assert listing(tmp_path, "4096 Artist/Album/01.mp3", "4096 Artist/Album/02.mp3") == []
|
|
|
|
|
|
def test_a_deletion_relevels_what_is_left(tmp_path):
|
|
album(tmp_path, "Artist/Album", "01.mp3", "02.mp3")
|
|
|
|
assert listing(tmp_path, "deleting Artist/Album/03.mp3") == [
|
|
"Artist/Album/01.mp3",
|
|
"Artist/Album/02.mp3",
|
|
]
|
|
|
|
|
|
def test_an_album_deleted_outright_lists_nothing(tmp_path):
|
|
assert listing(tmp_path, "deleting Artist/Gone/01.mp3") == []
|
|
|
|
|
|
def test_untouched_albums_are_left_alone(tmp_path):
|
|
album(tmp_path, "Artist/Changed", "01.mp3", "02.mp3")
|
|
album(tmp_path, "Artist/Quiet", "01.mp3", "02.mp3")
|
|
|
|
assert listing(tmp_path, "4096 Artist/Changed/01.mp3") == ["Artist/Changed/02.mp3"]
|
|
|
|
|
|
def test_a_replaced_cover_is_not_an_album_change(tmp_path):
|
|
"""Only a track can change an album's gains, and covers are replaced often
|
|
enough that treating one as a re-level would copy records for nothing."""
|
|
album(tmp_path, "Artist/Album", "01.mp3", "02.mp3")
|
|
|
|
assert listing(tmp_path, "17408 Artist/Album/cover.jpg") == []
|
|
|
|
|
|
def test_directories_are_not_mistaken_for_tracks(tmp_path):
|
|
album(tmp_path, "Artist/Album", "01.mp3")
|
|
|
|
assert listing(tmp_path, "4096 Artist/Album/", "deleting Artist/Old/") == []
|
|
|
|
|
|
def test_rsync_talking_to_the_operator_is_not_a_path(tmp_path):
|
|
album(tmp_path, "Artist/Album", "01.mp3", "02.mp3")
|
|
|
|
assert (
|
|
listing(
|
|
tmp_path,
|
|
"sending incremental file list",
|
|
"",
|
|
"sent 1,234 bytes received 56 bytes 2,580.00 bytes/sec",
|
|
"total size is 7,890 speedup is 6.12",
|
|
)
|
|
== []
|
|
)
|
|
|
|
|
|
def test_a_track_at_the_mirror_root_does_not_pull_in_the_whole_tree(tmp_path):
|
|
"""Nothing writes a mirror this way, but the directory of a root-level file
|
|
is the root, and recursing from there would be the whole library."""
|
|
(tmp_path / "loose.mp3").write_bytes(b"x")
|
|
(tmp_path / "other.mp3").write_bytes(b"x")
|
|
album(tmp_path, "Artist/Album", "01.mp3")
|
|
|
|
assert listing(tmp_path, "4096 loose.mp3") == ["other.mp3"]
|
|
|
|
|
|
def test_the_paths_are_written_one_per_line(tmp_path):
|
|
"""They are fed straight back to rsync as --files-from."""
|
|
album(tmp_path, "Artist/Album", "01.mp3", "02.mp3")
|
|
out = io.StringIO()
|
|
|
|
touched_albums.main(
|
|
["--mirror", str(tmp_path)],
|
|
stream=["4096 Artist/Album/01.mp3"],
|
|
out=out,
|
|
)
|
|
|
|
assert out.getvalue() == "Artist/Album/02.mp3\n"
|
|
|
|
|
|
def test_it_runs_as_a_script(tmp_path):
|
|
album(tmp_path, "Artist/Album", "01.mp3", "02.mp3")
|
|
|
|
completed = subprocess.run(
|
|
[sys.executable, touched_albums.__file__, "--mirror", str(tmp_path)],
|
|
input="4096 Artist/Album/01.mp3\n",
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
assert completed.returncode == 0
|
|
assert completed.stdout == "Artist/Album/02.mp3\n"
|