feat: mirror a lossless library to MP3 for iPod sync
Apple's Music app cannot read FLAC, so getting a lossless library onto an iPod requires a converted copy somewhere. This keeps that copy beside the library on a NAS rather than on a laptop, and keeps it current unattended. Walks a source tree and reproduces it path for path as MP3: tags and cover art carried across, already-MP3 sources copied rather than re-encoded, and mirror files whose source has gone deleted along with any directories they emptied. The source library is never written to. Freshness is tracked by modification time -- an encoded file is stamped with its source's mtime, so a file is stale exactly when the two differ. That keeps runs idempotent without a database that could fall out of step with whatever owns the library, which here is Lidarr. Encodes go to a temporary file and are renamed into place, so an interrupted run cannot leave a truncated MP3 that the next run mistakes for finished work. A lock file in the mirror root prevents overlapping passes. Ships as a Python package with a console script, a container image with a TrueNAS Scale compose file, and a Nix flake providing the package, an overlay and a dev shell. The test suite runs real ffmpeg encodes rather than mocks -- the failures worth catching are in what ffmpeg does with tags, cover art and container formats. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Emma Thorpe
co-authored by
Claude Opus 5
Claude Opus 5
parent
38f545ac32
commit
177761bb75
@@ -0,0 +1,228 @@
|
||||
import os
|
||||
import subprocess
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
import music_mirror
|
||||
|
||||
|
||||
def run(source, mirror, *extra):
|
||||
return music_mirror.main(["--source", str(source), "--mirror", str(mirror), *extra])
|
||||
|
||||
|
||||
def test_parse_quality_accepts_vbr_and_cbr():
|
||||
assert music_mirror.parse_quality("V0") == ["-q:a", "0"]
|
||||
assert music_mirror.parse_quality("v2") == ["-q:a", "2"]
|
||||
assert music_mirror.parse_quality("256") == ["-b:a", "256k"]
|
||||
|
||||
|
||||
def test_parse_quality_rejects_nonsense():
|
||||
with pytest.raises(ValueError):
|
||||
music_mirror.parse_quality("best")
|
||||
|
||||
|
||||
def test_parse_interval_units():
|
||||
assert music_mirror.parse_interval("90") == 90
|
||||
assert music_mirror.parse_interval("30m") == 1800
|
||||
assert music_mirror.parse_interval("6h") == 21600
|
||||
assert music_mirror.parse_interval("1d") == 86400
|
||||
|
||||
|
||||
def test_encodes_and_preserves_layout_and_tags(tmp_path, make_flac, probe_tag):
|
||||
source = tmp_path / "src"
|
||||
mirror = tmp_path / "dst"
|
||||
make_flac(source / "Test Artist" / "Test Album" / "03 Song.flac")
|
||||
|
||||
assert run(source, mirror) == 0
|
||||
|
||||
output = mirror / "Test Artist" / "Test Album" / "03 Song.mp3"
|
||||
assert output.is_file()
|
||||
assert probe_tag(output, "title") == "Test Title"
|
||||
assert probe_tag(output, "album") == "Test Album"
|
||||
|
||||
|
||||
def test_output_is_mp3(tmp_path, make_flac):
|
||||
source = tmp_path / "src"
|
||||
mirror = tmp_path / "dst"
|
||||
make_flac(source / "a.flac")
|
||||
|
||||
run(source, mirror)
|
||||
|
||||
codec = subprocess.run(
|
||||
[
|
||||
"ffprobe",
|
||||
"-v",
|
||||
"error",
|
||||
"-select_streams",
|
||||
"a:0",
|
||||
"-show_entries",
|
||||
"stream=codec_name",
|
||||
"-of",
|
||||
"csv=p=0",
|
||||
str(mirror / "a.mp3"),
|
||||
],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
).stdout.strip()
|
||||
assert codec == "mp3"
|
||||
|
||||
|
||||
def test_second_pass_skips_unchanged_files(tmp_path, make_flac):
|
||||
source = tmp_path / "src"
|
||||
mirror = tmp_path / "dst"
|
||||
make_flac(source / "a.flac")
|
||||
|
||||
run(source, mirror)
|
||||
first = (mirror / "a.mp3").stat().st_mtime_ns
|
||||
|
||||
run(source, mirror)
|
||||
assert (mirror / "a.mp3").stat().st_mtime_ns == first
|
||||
|
||||
|
||||
def test_changed_source_is_re_encoded(tmp_path, make_flac):
|
||||
source = tmp_path / "src"
|
||||
mirror = tmp_path / "dst"
|
||||
track = make_flac(source / "a.flac")
|
||||
|
||||
run(source, mirror)
|
||||
before = (mirror / "a.mp3").stat().st_mtime
|
||||
|
||||
# A replaced file with a newer mtime is what a Lidarr quality upgrade
|
||||
# looks like on disk.
|
||||
later = time.time() + 120
|
||||
os.utime(track, (later, later))
|
||||
run(source, mirror)
|
||||
|
||||
assert (mirror / "a.mp3").stat().st_mtime > before
|
||||
|
||||
|
||||
def test_deleted_source_is_pruned(tmp_path, make_flac):
|
||||
source = tmp_path / "src"
|
||||
mirror = tmp_path / "dst"
|
||||
make_flac(source / "Album" / "a.flac")
|
||||
make_flac(source / "Album" / "b.flac")
|
||||
|
||||
run(source, mirror)
|
||||
(source / "Album" / "b.flac").unlink()
|
||||
run(source, mirror)
|
||||
|
||||
assert (mirror / "Album" / "a.mp3").is_file()
|
||||
assert not (mirror / "Album" / "b.mp3").exists()
|
||||
|
||||
|
||||
def test_emptied_directory_is_removed(tmp_path, make_flac):
|
||||
source = tmp_path / "src"
|
||||
mirror = tmp_path / "dst"
|
||||
make_flac(source / "Gone" / "a.flac")
|
||||
|
||||
run(source, mirror)
|
||||
(source / "Gone" / "a.flac").unlink()
|
||||
run(source, mirror)
|
||||
|
||||
assert not (mirror / "Gone").exists()
|
||||
|
||||
|
||||
def test_no_prune_keeps_orphans(tmp_path, make_flac):
|
||||
source = tmp_path / "src"
|
||||
mirror = tmp_path / "dst"
|
||||
make_flac(source / "a.flac")
|
||||
|
||||
run(source, mirror)
|
||||
(source / "a.flac").unlink()
|
||||
run(source, mirror, "--no-prune")
|
||||
|
||||
assert (mirror / "a.mp3").is_file()
|
||||
|
||||
|
||||
def test_existing_mp3_is_copied_not_re_encoded(tmp_path, make_flac):
|
||||
source = tmp_path / "src"
|
||||
mirror = tmp_path / "dst"
|
||||
flac = make_flac(source / "a.flac")
|
||||
subprocess.run(
|
||||
["ffmpeg", "-loglevel", "error", "-y", "-i", str(flac), str(source / "b.mp3")],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
)
|
||||
flac.unlink()
|
||||
|
||||
run(source, mirror)
|
||||
|
||||
assert (mirror / "b.mp3").read_bytes() == (source / "b.mp3").read_bytes()
|
||||
|
||||
|
||||
def test_dry_run_writes_nothing(tmp_path, make_flac):
|
||||
source = tmp_path / "src"
|
||||
mirror = tmp_path / "dst"
|
||||
make_flac(source / "a.flac")
|
||||
|
||||
assert run(source, mirror, "--dry-run") == 0
|
||||
assert not (mirror / "a.mp3").exists()
|
||||
|
||||
|
||||
def test_subdir_limits_the_pass_and_does_not_prune(tmp_path, make_flac):
|
||||
source = tmp_path / "src"
|
||||
mirror = tmp_path / "dst"
|
||||
make_flac(source / "One" / "a.flac")
|
||||
make_flac(source / "Two" / "b.flac")
|
||||
|
||||
assert run(source, mirror, "--subdir", "One") == 0
|
||||
|
||||
assert (mirror / "One" / "a.mp3").is_file()
|
||||
# Outside the requested directory: neither encoded nor treated as an orphan.
|
||||
assert not (mirror / "Two").exists()
|
||||
|
||||
|
||||
def test_external_cover_is_embedded(tmp_path, make_flac):
|
||||
source = tmp_path / "src"
|
||||
mirror = tmp_path / "dst"
|
||||
make_flac(source / "Album" / "a.flac")
|
||||
subprocess.run(
|
||||
[
|
||||
"ffmpeg",
|
||||
"-loglevel",
|
||||
"error",
|
||||
"-y",
|
||||
"-f",
|
||||
"lavfi",
|
||||
"-i",
|
||||
"color=c=red:s=64x64:d=1",
|
||||
"-frames:v",
|
||||
"1",
|
||||
str(source / "Album" / "cover.jpg"),
|
||||
],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
)
|
||||
|
||||
run(source, mirror)
|
||||
|
||||
streams = subprocess.run(
|
||||
[
|
||||
"ffprobe",
|
||||
"-v",
|
||||
"error",
|
||||
"-select_streams",
|
||||
"v",
|
||||
"-show_entries",
|
||||
"stream=codec_name",
|
||||
"-of",
|
||||
"csv=p=0",
|
||||
str(mirror / "Album" / "a.mp3"),
|
||||
],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
).stdout.strip()
|
||||
assert streams # a picture stream is present
|
||||
|
||||
|
||||
def test_mirror_inside_source_is_refused(tmp_path, make_flac):
|
||||
source = tmp_path / "src"
|
||||
make_flac(source / "a.flac")
|
||||
assert run(source, source / "mp3") == 2
|
||||
|
||||
|
||||
def test_missing_source_is_refused(tmp_path):
|
||||
assert run(tmp_path / "nope", tmp_path / "dst") == 2
|
||||
Reference in New Issue
Block a user