lyrathorpe/music-mirror
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat: mirror a lossless library to MP3 for iPod sync (#1)
Build and publish container / build (push) Failing after 51s
Details

Browse Source 
## What

A path-for-path MP3 mirror of a lossless library. FLAC in, MP3 out, same relative layout, tags and cover art carried across. Already-MP3 sources are copied rather than re-encoded. Mirror files whose source has gone are deleted, along with any directory they emptied. The source library is never written to — mounted read-only in the compose file, and never opened for writing in code.

| Situation                        | Action                               |
| -------------------------------- | ------------------------------------ |
| No mirror file                   | encode                               |
| Source modified since the mirror | re-encode (a Lidarr quality upgrade) |
| Mirror up to date                | skip                                 |
| Source is already MP3            | copy verbatim                        |
| Source gone                      | delete, prune empty dirs             |

## Design notes

- **No database.** Freshness is mtime: an encode is stamped with its source's mtime, so a file is stale exactly when the two differ. Lidarr owns the library; a second tool with its own index would only fall out of step with it. This is also why it is not beets.
- **Atomic writes.** Encode to a temp file, rename into place. An interrupted run cannot leave a truncated MP3 that the next run treats as finished.
- **A lock file** in the mirror root stops two passes overlapping.
- **Refuses a mirror inside the source tree**, which would otherwise recurse.
- `--subdir` never prunes: a partial pass cannot distinguish an orphan from a file outside its scope.

## Shipping

- Python package with a `music-mirror` console script, no runtime dependencies beyond ffmpeg.
- `Dockerfile` plus `compose.yaml` as a TrueNAS Scale Custom App: source dataset read-only, mirror dataset writable, `MUSIC_MIRROR_INTERVAL=6h`.
- CI runs the tests **inside the image**, against the ffmpeg that ships, on every pull request, and on merge publishes multi-arch (amd64 + arm64) to this Gitea's registry using the `PACKAGES_SECRET` repository secret. Versioning follows the same conventional-commit scheme as `legacy-email-proxy`, including writing the released version back into `pyproject.toml` so the packaging metadata cannot drift behind the tag.

## Behaviour under Lidarr

| Lidarr does this                        | The mirror does this                                                             |
| --------------------------------------- | -------------------------------------------------------------------------------- |
| Replaces a file with a better rip        | Re-encodes in place; same path in, same path out, so no duplicate                 |
| Upgrades MP3 to FLAC                     | Both map to the same `.mp3` mirror path, so the old one is overwritten            |
| Renames a track, album or artist folder  | Old path pruned, new path encoded — correct, though it re-encodes rather than moving |
| Deletes an album or artist               | Every orphaned mirror file goes, and the directories they emptied with them        |

Three defects were found and fixed while writing those tests, each verified to
fail against the previous code:

- Pruning probed the source tree for a mirror file's original name, lowercase
  extensions only. A `.FLAC` source was never found, so its mirror file was
  deleted as an orphan and rebuilt on the next pass, for ever. Pruning now works
  from the set of paths the pass actually accounted for.
- Two sources could claim one mirror path — `01 Song.flac` beside a leftover
  `01 Song.mp3`. Both encoded to the same destination and each pass found the
  loser stale. The better format now wins, ties break on path.
- The source mtime was read after encoding rather than before, so a file still
  being written when the pass reached it could be stamped current while holding
  truncated audio.

## Why there is no flake

The deployment target is a container. A flake here would sit on no path between
the source and the NAS, and `nix flake check` would test against nixpkgs' ffmpeg
while the artefact ships Debian's — precisely the layer these tests exercise. It
was removed in favour of running the suite inside the image. The sibling
`legacy-email-proxy` keeps its flake because there the flake *is* the deployment
mechanism.

## Verification

- 21 tests, all real ffmpeg round-trips: layout, tag survival, MP3 output, skip-when-current, re-encode-on-change, orphan pruning, empty-dir removal, `--no-prune`, MP3 passthrough, `--dry-run`, `--subdir`, external cover art embedding, both refusal paths, and the Lidarr lifecycle cases below.
- The suite also runs in the multi-stage Docker `test` stage: 21 passed. The published `runtime` stage carries neither the tests nor pytest, verified by inspecting the image.
- Container built and run locally against a sample library: correct output path, Cyrillic tags intact.
- The release step was extracted from the workflow and run against a scratch repository: it commits and tags when the version changes, and skips the commit but still tags when `pyproject.toml` already carries it.
- **Not tested against the real library** — the first run there should be `--dry-run`.

## Follow-ups, not in this PR

- Lidarr imports are picked up on the next scheduled pass rather than instantly. A webhook trigger is the obvious next step if six hours feels slow.
- `PACKAGES_SECRET` must exist as a repository secret before the first merge, or the login step fails.

---------

Co-authored-by: Emma Thorpe <emma.thorpe@citrix.com>
Reviewed-on: #1
This commit was merged in pull request #1.
This commit is contained in:
lyrathorpe
2026-08-21 15:30:58 +01:00
co-authored by Emma Thorpe
parent 38f545ac32
commit d4ccff3b75
11 changed files with 1346 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
music_mirror.py
+509
View File
@@ -0,0 +1,509 @@
"""Maintain a lossy MP3 mirror of a lossless music library.
Walks a source library and reproduces it, path for path, as MP3 in a separate
directory tree: FLAC in, MP3 out, same relative layout, tags and cover art
carried across. Sources that are already MP3 are copied rather than re-encoded.
The mirror is derived state. It is only ever written to, never read as a
source of truth, so it can be deleted and rebuilt at any time. Nothing here
writes to the source library.
Staleness is tracked by modification time: an encoded file is given its
source's mtime, so a file is out of date exactly when the two differ. That
makes runs idempotent without a database to keep in step.
"""
import argparse
import concurrent.futures
import fcntl
import logging
import os
import re
import shutil
import signal
import subprocess
import sys
import tempfile
import time
from dataclasses import dataclass
from pathlib import Path
logger = logging.getLogger("music-mirror")
# Sources that are transcoded. Anything ffmpeg can decode works; this list
# decides what the walker picks up in the first place.
SOURCE_EXTENSIONS = {
".flac",
".wav",
".aif",
".aiff",
".ape",
".wv",
".m4a",
".alac",
".ogg",
".opus",
".wma",
}
# Already MP3: copied through. Re-encoding lossy audio to lossy audio costs
# quality for nothing.
COPY_EXTENSIONS = {".mp3"}
# Best first. Used only to settle which source wins when two of them want the
# same mirror path; see plan().
SOURCE_PRIORITY = [
".flac",
".wav",
".aif",
".aiff",
".ape",
".wv",
".alac",
".m4a",
".ogg",
".opus",
".wma",
".mp3",
]
# Looked for in the source directory when a file has no embedded picture.
COVER_NAMES = ("cover.jpg", "folder.jpg", "front.jpg", "cover.png", "folder.png")
# Files the mirror is allowed to contain, and therefore allowed to delete.
MIRROR_SUFFIX = ".mp3"
# Filesystems disagree about mtime precision; SMB in particular rounds.
MTIME_TOLERANCE_SECONDS = 2
@dataclass
class Result:
"""Outcome of processing one file."""
action: str # encoded | copied | skipped | failed
path: Path
error: str = ""
def parse_quality(quality):
"""Return the ffmpeg arguments for a quality setting.
Accepts LAME VBR levels (``V0``..``V9``) or a constant bitrate in kbps
(``256``). VBR is the better trade at a given average bitrate; CBR is for
when a fixed size matters more.
"""
text = str(quality).strip().lower()
if re.fullmatch(r"v[0-9]", text):
return ["-q:a", text[1:]]
if re.fullmatch(r"[0-9]{2,3}", text):
return ["-b:a", f"{text}k"]
raise ValueError(f"unrecognised quality {quality!r}: expected V0-V9 or a bitrate like 256")
def parse_interval(interval):
"""Return seconds for an interval such as ``30m``, ``6h`` or ``90``."""
text = str(interval).strip().lower()
match = re.fullmatch(r"([0-9]+)([smhd]?)", text)
if not match:
raise ValueError(f"unrecognised interval {interval!r}: expected e.g. 45m, 6h, 1d")
value = int(match.group(1))
return value * {"": 1, "s": 1, "m": 60, "h": 3600, "d": 86400}[match.group(2)]
def mirror_path_for(source, source_root, mirror_root):
"""Return the mirror path corresponding to a source file."""
return (mirror_root / source.relative_to(source_root)).with_suffix(MIRROR_SUFFIX)
def is_current(source, mirror):
"""Return whether the mirror file is up to date with its source."""
if not mirror.exists():
return False
return abs(source.stat().st_mtime - mirror.stat().st_mtime) <= MTIME_TOLERANCE_SECONDS
def find_cover(directory):
"""Return an external cover image for a directory, if one is present."""
for name in COVER_NAMES:
candidate = directory / name
if candidate.is_file():
return candidate
return None
def has_embedded_picture(source):
"""Return whether the source carries its own cover art."""
try:
probe = subprocess.run(
[
"ffprobe",
"-v",
"error",
"-select_streams",
"v",
"-show_entries",
"stream=index",
"-of",
"csv=p=0",
str(source),
],
capture_output=True,
text=True,
check=True,
)
except (subprocess.CalledProcessError, FileNotFoundError):
return False
return bool(probe.stdout.strip())
def build_command(source, destination, quality_args, cover):
"""Return the ffmpeg command that encodes one file."""
command = ["ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "error", "-y", "-i", str(source)]
if cover is not None:
command += ["-i", str(cover), "-map", "1:v:0"]
else:
# Optional: the source may have no picture stream at all.
command += ["-map", "0:v:0?"]
command += [
"-map",
"0:a:0",
"-map_metadata",
"0",
"-c:a",
"libmp3lame",
*quality_args,
"-c:v",
"copy",
"-disposition:v",
"attached_pic",
# ID3v2.3 is the widest-compatibility tag version, and what the iPod
# firmware is happiest with; the v1 tag costs 128 bytes.
"-id3v2_version",
"3",
"-write_id3v1",
"1",
# Stated rather than inferred: the destination is a temporary file
# whose suffix ffmpeg would not recognise.
"-f",
"mp3",
str(destination),
]
return command
def encode(source, mirror, quality_args, dry_run):
"""Encode one source file into the mirror, atomically."""
if dry_run:
logger.info("would encode %s", source)
return Result("encoded", mirror)
mirror.parent.mkdir(parents=True, exist_ok=True)
cover = None if has_embedded_picture(source) else find_cover(source.parent)
# Read the source's mtime before encoding, not after. If the file is still
# being written -- a Lidarr import landing mid-pass -- stamping the mirror
# with the later mtime would mark truncated output as current. Stamping the
# earlier one leaves the two mismatched, so the next pass re-encodes it.
stat = source.stat()
# Encode to a temporary file in the destination directory and rename it
# into place, so an interrupted run cannot leave a truncated MP3 that the
# next run would treat as complete.
handle, temporary = tempfile.mkstemp(dir=mirror.parent, suffix=".mp3.part")
os.close(handle)
temporary = Path(temporary)
try:
command = build_command(source, temporary, quality_args, cover)
completed = subprocess.run(command, capture_output=True, text=True)
if completed.returncode != 0:
lines = completed.stderr.strip().splitlines()
return Result("failed", source, lines[-1] if lines else "ffmpeg failed")
os.utime(temporary, (stat.st_atime, stat.st_mtime))
os.replace(temporary, mirror)
except Exception as error: # noqa: BLE001 - reported per file, run continues
return Result("failed", source, str(error))
finally:
temporary.unlink(missing_ok=True)
logger.info("encoded %s", source)
return Result("encoded", mirror)
def copy(source, mirror, dry_run):
"""Copy an already-MP3 source into the mirror."""
if dry_run:
logger.info("would copy %s", source)
return Result("copied", mirror)
mirror.parent.mkdir(parents=True, exist_ok=True)
try:
shutil.copy2(source, mirror)
except OSError as error:
return Result("failed", source, str(error))
logger.info("copied %s", source)
return Result("copied", mirror)
def process(source, mirror, quality_args, dry_run):
"""Bring one source file's mirror entry up to date."""
if is_current(source, mirror):
return Result("skipped", mirror)
if source.suffix.lower() in COPY_EXTENSIONS:
return copy(source, mirror, dry_run)
return encode(source, mirror, quality_args, dry_run)
def find_sources(root):
"""Yield every audio file under a root, in a stable order."""
extensions = SOURCE_EXTENSIONS | COPY_EXTENSIONS
for path in sorted(root.rglob("*")):
if path.is_file() and path.suffix.lower() in extensions:
yield path
def plan(scan_root, source_root, mirror_root):
"""Map each mirror path to the one source that should produce it.
Two sources can want the same mirror path -- `01 Song.flac` alongside a
leftover `01 Song.mp3`, which is what an interrupted Lidarr upgrade leaves
behind. Without a decision here both would encode to the same destination,
each pass would find the loser stale, and the mirror would be rewritten
forever. Preferring the highest-quality source, ties broken by path, makes
the outcome stable and predictable instead.
"""
chosen = {}
for source in find_sources(scan_root):
mirror = mirror_path_for(source, source_root, mirror_root)
rival = chosen.get(mirror)
if rival is None:
chosen[mirror] = source
continue
winner, loser = sorted((source, rival), key=source_rank)
logger.warning("%s and %s both map to %s; using %s", rival, source, mirror, winner)
chosen[mirror] = winner
return chosen
def source_rank(source):
"""Sort key preferring better source formats, then a stable path order."""
suffix = source.suffix.lower()
position = SOURCE_PRIORITY.index(suffix) if suffix in SOURCE_PRIORITY else len(SOURCE_PRIORITY)
return (position, str(source))
def prune(mirror_root, expected, dry_run):
"""Delete mirror files this pass did not account for, and empty dirs.
Driven by the set of paths the pass expects to exist rather than by
probing the source tree for names, which would disagree with it over
letter case and over any extension the walker does not collect.
"""
removed = 0
for mirror in sorted(mirror_root.rglob(f"*{MIRROR_SUFFIX}")):
if mirror in expected:
continue
removed += 1
if dry_run:
logger.info("would remove orphan %s", mirror)
continue
logger.info("removing orphan %s", mirror)
mirror.unlink(missing_ok=True)
if not dry_run:
# Deepest first, so a directory emptied by the loop above is caught.
for directory in sorted(mirror_root.rglob("*"), reverse=True):
if directory.is_dir() and not any(directory.iterdir()):
directory.rmdir()
return removed
def run_once(scan_root, source_root, mirror_root, quality_args, jobs, dry_run, do_prune):
"""Run a single pass. Returns the number of failures.
`scan_root` is what gets walked and `source_root` is what mirror paths are
computed against; they differ only for a partial pass over one directory.
"""
started = time.monotonic()
counts = {"encoded": 0, "copied": 0, "skipped": 0, "failed": 0}
failures = []
work = plan(scan_root, source_root, mirror_root)
with concurrent.futures.ThreadPoolExecutor(max_workers=jobs) as pool:
futures = [
pool.submit(process, source, mirror, quality_args, dry_run)
for mirror, source in work.items()
]
for future in concurrent.futures.as_completed(futures):
result = future.result()
counts[result.action] += 1
if result.action == "failed":
failures.append(result)
removed = prune(mirror_root, set(work), dry_run) if do_prune else 0
for failure in failures:
logger.error("failed: %s: %s", failure.path, failure.error)
logger.info(
"pass complete in %.1fs: %d encoded, %d copied, %d up to date, %d removed, %d failed",
time.monotonic() - started,
counts["encoded"],
counts["copied"],
counts["skipped"],
removed,
counts["failed"],
)
return counts["failed"]
def acquire_lock(mirror_root):
"""Take an exclusive lock so two passes cannot run over one mirror."""
mirror_root.mkdir(parents=True, exist_ok=True)
handle = open(mirror_root / ".music-mirror.lock", "w") # noqa: SIM115 - held for the process
try:
fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
handle.close()
return None
return handle
def build_parser():
"""Return the argument parser. Every option also reads an env var, so the
container can be configured without a command line."""
parser = argparse.ArgumentParser(
prog="music-mirror",
description="Maintain a lossy MP3 mirror of a lossless music library.",
)
parser.add_argument(
"--source",
default=os.getenv("MUSIC_MIRROR_SOURCE"),
help="root of the lossless library; never written to (env MUSIC_MIRROR_SOURCE)",
)
parser.add_argument(
"--mirror",
default=os.getenv("MUSIC_MIRROR_MIRROR"),
help="root of the MP3 mirror (env MUSIC_MIRROR_MIRROR)",
)
parser.add_argument(
"--quality",
default=os.getenv("MUSIC_MIRROR_QUALITY", "V0"),
help="LAME VBR level (V0-V9) or a CBR bitrate in kbps (env MUSIC_MIRROR_QUALITY)",
)
parser.add_argument(
"--jobs",
type=int,
default=int(os.getenv("MUSIC_MIRROR_JOBS", "0")) or (os.cpu_count() or 4),
help="concurrent encodes (env MUSIC_MIRROR_JOBS; default: CPU count)",
)
parser.add_argument(
"--interval",
default=os.getenv("MUSIC_MIRROR_INTERVAL"),
help="repeat forever, waiting this long between passes, e.g. 6h (env MUSIC_MIRROR_INTERVAL)",
)
parser.add_argument(
"--subdir",
default=None,
help="limit the pass to one directory below --source; skips pruning",
)
parser.add_argument(
"--no-prune",
action="store_true",
help="keep mirror files whose source has been deleted",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="report what would change without touching the mirror",
)
return parser
def main(argv=None):
"""Entry point. Returns a process exit code."""
logging.basicConfig(format="%(asctime)s %(levelname)s %(message)s", level=logging.INFO)
args = build_parser().parse_args(argv)
if not args.source or not args.mirror:
logger.error("both --source and --mirror are required")
return 2
source_root = Path(args.source).resolve()
mirror_root = Path(args.mirror).resolve()
if not source_root.is_dir():
logger.error("source %s is not a directory", source_root)
return 2
if mirror_root == source_root or mirror_root.is_relative_to(source_root):
logger.error("mirror %s must not sit inside the source library", mirror_root)
return 2
try:
quality_args = parse_quality(args.quality)
interval = parse_interval(args.interval) if args.interval else None
except ValueError as error:
logger.error("%s", error)
return 2
scan_root = source_root
do_prune = not args.no_prune
if args.subdir:
scan_root = (source_root / args.subdir).resolve()
if not scan_root.is_relative_to(source_root) or not scan_root.is_dir():
logger.error("--subdir %s is not a directory below the source", args.subdir)
return 2
# A partial pass cannot tell an orphan from a file outside its scope.
do_prune = False
lock = acquire_lock(mirror_root)
if lock is None:
logger.error("another pass is already running over %s", mirror_root)
return 3
stopping = False
def stop(signum, _frame):
nonlocal stopping
stopping = True
logger.info("signal %d received; finishing the current pass", signum)
signal.signal(signal.SIGTERM, stop)
signal.signal(signal.SIGINT, stop)
try:
while True:
failures = run_once(
scan_root,
source_root,
mirror_root,
quality_args,
args.jobs,
args.dry_run,
do_prune,
)
if interval is None or stopping:
return 1 if failures else 0
logger.info("sleeping %ds", interval)
for _ in range(interval):
if stopping:
return 1 if failures else 0
time.sleep(1)
finally:
lock.close()
def run():
"""Console-script entry point."""
sys.exit(main())
if __name__ == "__main__":
run()