music-mirror/music_mirror.py

"""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.

Finished albums are levelled with rsgain, which writes ReplayGain tags into the
mirror. Rockbox applies the offset those tags carry but has no loudness
analysis of its own, so without them every album plays at whatever level it was
mastered to.
"""

import argparse
import concurrent.futures
import fcntl
import functools
import hashlib
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")

# What a copied cover is called in the mirror. Rockbox looks for album art on
# the filesystem -- cover.jpg, folder.jpg and friends beside the track -- and
# its search never touches the picture embedded in the tag, so a mirror that
# only embeds art shows none of it on the device.
MIRROR_COVER = "cover.jpg"

# Files the mirror is allowed to contain, and therefore allowed to delete.
MIRROR_SUFFIX = ".mp3"

# Measures loudness and writes the ReplayGain tags. Not a hard requirement: a
# pass without it still produces a correct mirror, only one the player cannot
# level, so a missing binary is a warning rather than a failure.
REPLAYGAIN_TOOL = "rsgain"

# Looked for in a file's ID3v2 tag to tell a levelled track from an unlevelled
# one. Album gain rather than track gain because the album value is the one
# this writes for; a file carrying only track gain came from somewhere else and
# should be rescanned.
REPLAYGAIN_TAG = b"replaygain_album_gain"

# Enough of a TXXX frame body to hold the encoding byte and the description.
# The value after it says what the gain is, which is not the question here.
TXXX_DESCRIPTION_BYTES = 128

# Filesystems disagree about mtime precision; SMB in particular rounds.
MTIME_TOLERANCE_SECONDS = 2

# What FAT32 refuses in a filename, plus the control characters. The mirror is
# copied onto a FAT32 device, and one of these in a path means a track that
# never arrives -- "Kick Out the Epic Motherf**ker" is a real example.
FAT32_RESERVED = re.compile(r'[<>:"/\\|?*\x00-\x1f]')

# Rockbox's MAX_PATH, from firmware/include/fs_defines.h. It bounds the whole
# path as the device sees it, so the budget for a mirror-relative path is this
# less whatever directory the mirror is copied into.
MAX_PATH = 260
DEVICE_PREFIX = "/Music"

# A component cut below this is no longer recognisable, and a path that cannot
# be brought under the limit without going there is better reported than
# mangled.
MIN_COMPONENT = 12

# The mirror exists to be read back by something else -- an SMB share, another
# account on the box -- so everything written into it has to be group-readable.
# Neither writer manages that unaided: tempfile.mkstemp forces 0600 whatever the
# umask, and shutil.copy2 carries the source file's mode across from a library
# that may be tighter still. Directories need the execute bit too, or the group
# cannot enter them to reach the readable files inside.
GROUP_READ = 0o040

# Cleared from the umask so directories this run creates can be listed and
# entered. Owner as well as group: a umask carrying 0400 -- which is unusual but
# not ours to assume away -- otherwise produces a mirror tree that not even the
# process that built it can read back.
DIRECTORY_ACCESS = 0o550


@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 fat32_safe(component):
    """Return a single path component FAT32 will accept.

    The mirror exists to be copied onto a FAT32 device, and a name that device
    will not take is a track that silently does not arrive. Cheaper to produce
    an acceptable name here than to discover the problem partway through
    copying fifty thousand files.

    Handled: the reserved characters, control characters, and the trailing dots
    and spaces that FAT quietly eats -- a name ending in one round-trips as a
    different name, which is worse than being rejected outright.
    """
    cleaned = FAT32_RESERVED.sub("_", component).rstrip(". ")
    # Stripping can empty a component outright: a directory named "..." is
    # legal on ext4 and nothing at all on FAT.
    return cleaned or "_"


def device_prefix_length(prefix):
    """Return the on-device prefix as it will actually appear, with slashes.

    "/Music" costs seven characters -- the leading slash, the name, and the
    separator before the mirror's own path -- while an empty prefix costs one.
    Approximating that loses a character at the root, which is precisely where
    the longest paths are.
    """
    cleaned = prefix.strip("/")
    return f"/{cleaned}/" if cleaned else "/"


def shorten_component(component, budget):
    """Return a component of at most `budget` characters, cut from the middle.

    From the middle, not the end, because of how these names are built. Lidarr
    writes "Artist - Album - 07 - Flamethrower.mp3" inside a directory already
    named for that artist and album, so the informative part -- the track
    number and title -- is at the very end. Cutting from the end discards it
    and leaves every track on the record with the same name.

    The four hex digits are of the original component. Two names sharing both a
    head and a tail would otherwise produce the same string, and a silent
    collision between two tracks is worse than an ugly filename.
    """
    stem, dot, extension = component.rpartition(".")
    if not dot or len(extension) > 4:
        stem, extension = component, ""
    else:
        extension = dot + extension

    digest = hashlib.blake2s(component.encode("utf-8"), digest_size=2).hexdigest()
    marker = f"~{digest}~"
    room = max(2, budget - len(extension) - len(marker))
    if room >= len(stem):
        return stem + extension

    # Two thirds to the tail: the head is usually a restatement of the
    # directory it sits in, and the tail is what tells two tracks apart.
    keep_end = min(len(stem), room * 2 // 3)
    keep_start = max(1, room - keep_end)
    return stem[:keep_start].rstrip(". ") + marker + stem[len(stem) - keep_end :] + extension


def fit_path(relative, budget):
    """Return a relative path within `budget` characters, or the best available.

    Shortened from the deepest component outward. The filename carries the least
    navigational value and the artist directory the most, so the track name is
    sacrificed before the album and the album before the artist.
    """
    parts = list(relative.parts)
    for index in reversed(range(len(parts))):
        overage = len(str(Path(*parts))) - budget
        if overage <= 0:
            break
        allowed = max(MIN_COMPONENT, len(parts[index]) - overage)
        if allowed < len(parts[index]):
            parts[index] = shorten_component(parts[index], allowed)
    fitted = Path(*parts)
    if len(str(fitted)) > budget:
        logger.warning(
            "%s is still %d characters over the limit after shortening; it is too"
            " deeply nested to fit",
            relative,
            len(str(fitted)) - budget,
        )
    return fitted


def mirror_path_for(source, source_root, mirror_root, safe=False, budget=0):
    """Return the mirror path corresponding to a source file."""
    relative = source.relative_to(source_root).with_suffix(MIRROR_SUFFIX)
    if safe:
        relative = Path(*(fat32_safe(part) for part in relative.parts))
        if budget > 0 and len(str(relative)) > budget:
            relative = fit_path(relative, budget)
    return mirror_root / relative


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


@functools.lru_cache(maxsize=4096)
def mirror_cover(source_directory, mirror_directory):
    """Put a copy of the album's cover beside its tracks in the mirror.

    Cached per directory: an album's tracks all ask for the same file, and the
    answer cannot change within a pass.
    """
    cover = find_cover(source_directory)
    if cover is None or cover.suffix.lower() not in (".jpg", ".jpeg"):
        # Only JPEG is copied. Rockbox will read a BMP too, but converting a
        # PNG is ffmpeg work for a file nothing else in the pass needs.
        return None
    destination = mirror_directory / MIRROR_COVER
    try:
        if destination.is_file() and destination.stat().st_size == cover.stat().st_size:
            return destination
        shutil.copy2(cover, destination)
        make_group_readable(destination)
    except OSError as error:
        logger.warning("could not copy cover for %s: %s", source_directory, error)
        return None
    return destination


def make_group_readable(path):
    """Add the group-read bit to a mirror file, leaving the rest of the mode alone."""
    mode = path.stat().st_mode
    if not mode & GROUP_READ:
        path.chmod(mode | GROUP_READ)


@functools.lru_cache(maxsize=4096)
def find_cover(directory):
    """Return an external cover image for a directory, if one is present.

    Cached because an album's tracks all ask the same question, and the answer
    costs one stat per candidate name.
    """
    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)
    mirror_cover(source.parent, mirror.parent)
    # Probing costs an ffprobe process per file, so only ask when the answer
    # can change the command. With no cover file beside the track, `-map
    # 0:v:0?` carries embedded art if there is any and shrugs if there is not.
    cover = find_cover(source.parent)
    if cover is not None and has_embedded_picture(source):
        cover = None

    # 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))
        # Before the rename, so the file is never visible in the mirror without
        # the bit.
        make_group_readable(temporary)
        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, atomically."""
    if dry_run:
        logger.info("would copy %s", source)
        return Result("copied", mirror)

    mirror.parent.mkdir(parents=True, exist_ok=True)
    mirror_cover(source.parent, mirror.parent)

    # Through a temporary file and a rename, for the same reason encodes go
    # that way, and a sharper one: copy2 reproduces the source's mtime as well
    # as its bytes, so a copy cut short by a full disk or a killed container
    # would leave a truncated MP3 that every later pass reads as current.
    handle, temporary = tempfile.mkstemp(dir=mirror.parent, suffix=".mp3.part")
    os.close(handle)
    temporary = Path(temporary)

    try:
        shutil.copy2(source, temporary)
        # copy2 brings the source's mode with it, and the source library is not
        # ours to have permissions opinions about.
        make_group_readable(temporary)
        os.replace(temporary, mirror)
    except OSError as error:
        return Result("failed", source, str(error))
    finally:
        temporary.unlink(missing_ok=True)

    logger.info("copied %s", source)
    return Result("copied", mirror)


def adopt_existing(source, mirror, candidates, dry_run=False):
    """Move an already-encoded file to its new name. Returns whether it moved.

    Turning on FAT32-safe naming changes the path of every track whose name
    held a reserved character. Without this the run would encode them all again
    and then prune the originals -- hours of work to produce files that already
    exist, byte for byte, under the old name.

    Several candidates are tried because there is more than one previous
    naming: the original, and the sanitised-but-not-yet-shortened form left by
    an earlier version.
    """
    for previous in candidates:
        if previous == mirror or not previous.is_file() or not is_current(source, previous):
            continue
        if dry_run:
            logger.info("would rename %s -> %s", previous.name, mirror.name)
            return True
        mirror.parent.mkdir(parents=True, exist_ok=True)
        os.replace(previous, mirror)
        logger.info("renamed %s -> %s", previous.name, mirror.name)
        return True
    return False


def process(source, mirror, quality_args, dry_run, previous=None):
    """Bring one source file's mirror entry up to date."""
    # Counted separately from an encode, and reported in a dry run, because the
    # difference between moving a file and re-encoding it is the difference
    # between a minute and an afternoon.
    if previous is not None and not mirror.exists():
        if adopt_existing(source, mirror, previous, dry_run):
            return Result("renamed", mirror)
    if is_current(source, mirror):
        # A mirror written before this bit was set has a correct mtime, so
        # nothing else in the pass would ever revisit it. Top it up here
        # instead: one stat per file, and no chmod at all once it is right.
        if not dry_run:
            try:
                make_group_readable(mirror)
            except OSError as error:
                return Result("failed", mirror, str(error))
        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, safe=False, budget=0):
    """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 = {}
    # Keyed case-insensitively when the target is FAT32, because two names
    # differing only in case are two files here and one file there. Detecting
    # that now beats discovering it as a silent overwrite during the copy.
    seen = {}
    for source in find_sources(scan_root):
        mirror = mirror_path_for(source, source_root, mirror_root, safe, budget)
        key = str(mirror).casefold() if safe else str(mirror)
        rival_path = seen.get(key)
        rival = chosen.get(rival_path) if rival_path else None
        if rival is None:
            seen[key] = mirror
            chosen[mirror] = source
            continue
        mirror = rival_path
        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.

    Returns the number of files removed and the directories they came out of.
    Losing a track changes an album's loudness, so those directories need
    levelling again even though nothing was written into them.
    """
    removed = 0
    emptied = set()

    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)
        emptied.add(mirror.parent)

    if not dry_run:
        # A cover copied for an album whose tracks have all gone is an orphan
        # too, and while it sits there the directory never looks empty.
        for cover in sorted(mirror_root.rglob(MIRROR_COVER)):
            if not any(cover.parent.glob(f"*{MIRROR_SUFFIX}")):
                logger.info("removing orphan %s", cover)
                cover.unlink(missing_ok=True)

        # 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, emptied


def syncsafe(data):
    """Return the integer held in syncsafe bytes: seven bits of each."""
    value = 0
    for byte in data:
        value = (value << 7) | (byte & 0x7F)
    return value


def has_replaygain(path):
    """Return whether an MP3 already carries ReplayGain tags.

    Walks the ID3v2 frame headers and seeks over the bodies rather than reading
    the tag whole. Every file in this mirror has its cover art embedded, so the
    tag is routinely half a megabyte; reading all of it for every track on
    every pass would turn an idle pass into a full read of the library.
    """
    try:
        with open(path, "rb") as handle:
            header = handle.read(10)
            if len(header) < 10 or header[:3] != b"ID3" or header[3] not in (3, 4):
                return False
            remaining = syncsafe(header[6:10])

            # Unsynchronisation shifts every offset in the tag, and the two
            # versions describe an extended header differently. Nothing that
            # writes this mirror emits either, so reading the tag whole is a
            # cheaper answer than the code to walk one that does.
            if header[5] & 0xC0:
                return REPLAYGAIN_TAG in handle.read(remaining).lower()

            while remaining >= 10:
                frame = handle.read(10)
                remaining -= 10
                # Frame ids are upper-case letters and digits, so anything else
                # is the padding that follows the last frame.
                if len(frame) < 10 or not frame[:4].isalnum():
                    return False
                # 2.3 sizes count all eight bits per byte; 2.4 made them
                # syncsafe like the tag length above.
                length = (
                    int.from_bytes(frame[4:8], "big")
                    if header[3] == 3
                    else syncsafe(frame[4:8])
                )
                if length <= 0 or length > remaining:
                    return False
                if frame[:4] == b"TXXX":
                    body = handle.read(min(length, TXXX_DESCRIPTION_BYTES))
                    handle.seek(length - len(body), os.SEEK_CUR)
                    if REPLAYGAIN_TAG in body.lower():
                        return True
                else:
                    handle.seek(length, os.SEEK_CUR)
                remaining -= length
    except OSError:
        return False
    return False


def replaygain_albums(expected, written):
    """Return the album directories needing a scan, each with its tracks.

    A directory is scanned when this pass changed what is in it, because album
    gain is a property of the whole album: one track added, replaced or removed
    makes the value stored on every one of its siblings wrong. It is also
    scanned when a track in it has never been levelled, which is what backfills
    a mirror built before any of this existed.
    """
    albums = {}
    for mirror in expected:
        albums.setdefault(mirror.parent, []).append(mirror)

    needed = {}
    for directory, tracks in sorted(albums.items()):
        # A dry run reaches here before anything has been encoded, so the
        # tracks a changed album is going to hold do not exist yet.
        present = sorted(track for track in tracks if track.is_file())
        if directory in written:
            needed[directory] = present
        elif present and not all(map(has_replaygain, present)):
            needed[directory] = present
    return needed


def replaygain_command(tracks):
    """Return the rsgain command that levels one album directory."""
    return [
        REPLAYGAIN_TOOL,
        "custom",
        # Album mode writes the per-track tags as well as the album ones, so
        # the device is left to choose between them -- Rockbox can apply track
        # gain when shuffling and album gain otherwise, and only if both are
        # present.
        "--album",
        "--tagmode=i",
        # The mirror is ID3v2.3 for the iPod firmware's sake. rsgain would
        # otherwise keep whatever version it found, and "whatever it found" is
        # not a guarantee.
        "--id3v2-version=3",
        # Staleness here is an mtime comparison and tagging rewrites the file.
        # Without this every levelled track would look newer than its source
        # and the next pass would re-encode the entire library, forever.
        "--preserve-mtimes",
        "--quiet",
        *[str(track) for track in tracks],
    ]


def scan_album(directory, tracks):
    """Write ReplayGain tags across one album. Returns whether it worked."""
    completed = subprocess.run(replaygain_command(tracks), capture_output=True, text=True)
    if completed.returncode != 0:
        lines = completed.stderr.strip().splitlines()
        logger.warning("could not level %s: %s", directory, lines[-1] if lines else "rsgain failed")
        return False
    logger.info("levelled %s", directory)
    return True


def replaygain(expected, written, jobs, dry_run):
    """Write ReplayGain tags into the albums that need them. Returns how many.

    A failure here is reported and then left alone. The mirror is still correct
    audio in the right place; it just plays at the level it was mastered to,
    which is what every pass before this one produced.
    """
    albums = {
        directory: tracks
        for directory, tracks in replaygain_albums(expected, written).items()
        if tracks or dry_run
    }
    if not albums:
        return 0

    if dry_run:
        logger.info("would level %d album%s", len(albums), "" if len(albums) == 1 else "s")
        return len(albums)

    if shutil.which(REPLAYGAIN_TOOL) is None:
        logger.warning(
            "%s is not on PATH; %d albums are left without ReplayGain tags",
            REPLAYGAIN_TOOL,
            len(albums),
        )
        return 0

    levelled = 0
    with concurrent.futures.ThreadPoolExecutor(max_workers=jobs) as pool:
        futures = [
            pool.submit(scan_album, directory, tracks) for directory, tracks in albums.items()
        ]
        for future in concurrent.futures.as_completed(futures):
            if future.result():
                levelled += 1
    return levelled


def run_once(
    scan_root,
    source_root,
    mirror_root,
    quality_args,
    jobs,
    dry_run,
    do_prune,
    safe=False,
    budget=0,
    do_replaygain=True,
):
    """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()
    logger.info("pass starting with %d concurrent encoders", jobs)
    counts = {"encoded": 0, "copied": 0, "renamed": 0, "skipped": 0, "failed": 0}
    failures = []
    written = set()

    work = plan(scan_root, source_root, mirror_root, safe, budget)

    with concurrent.futures.ThreadPoolExecutor(max_workers=jobs) as pool:
        futures = [
            pool.submit(
                process,
                source,
                mirror,
                quality_args,
                dry_run,
                (
                    [
                        mirror_path_for(source, source_root, mirror_root),
                        mirror_path_for(source, source_root, mirror_root, True),
                    ]
                    if safe
                    else None
                ),
            )
            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)
            elif result.action != "skipped":
                written.add(result.path.parent)

    expected = set(work)
    if safe and dry_run:
        # Nothing was actually renamed, so the pre-sanitisation files are still
        # on disk. They are not orphans -- they are the files a real run would
        # move -- and reporting them for deletion would misrepresent the pass
        # twice over.
        for source in work.values():
            expected.add(mirror_path_for(source, source_root, mirror_root))
            expected.add(mirror_path_for(source, source_root, mirror_root, True))
    removed, emptied = prune(mirror_root, expected, dry_run) if do_prune else (0, set())

    # After pruning, so an album is not measured with a track in it that is
    # about to be deleted.
    levelled = replaygain(expected, written | emptied, jobs, dry_run) if do_replaygain 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 renamed, %d up to date,"
        " %d removed, %d levelled, %d failed",
        time.monotonic() - started,
        counts["encoded"],
        counts["copied"],
        counts["renamed"],
        counts["skipped"],
        removed,
        levelled,
        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 default_jobs():
    """Return the number of CPUs this process may actually use.

    os.cpu_count() reports the host's total, which in a container with a `cpus:`
    limit means starting several times more encoders than there is CPU to run
    them. libmp3lame is single-threaded, so one process per available CPU is the
    whole of the concurrency story.
    """
    try:
        quota, period = Path("/sys/fs/cgroup/cpu.max").read_text().split()
        if quota != "max":
            return max(1, round(int(quota) / int(period)))
    except (OSError, ValueError):
        pass
    if hasattr(os, "process_cpu_count"):  # 3.13+, respects CPU affinity
        return os.process_cpu_count() or 4
    return os.cpu_count() or 4


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 default_jobs(),
        help="concurrent encodes (env MUSIC_MIRROR_JOBS; default: available CPUs)",
    )
    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(
        "--fat32-safe",
        action="store_true",
        default=os.getenv("MUSIC_MIRROR_FAT32_SAFE", "").lower() in ("1", "true", "yes"),
        help="name mirror files so a FAT32 device will accept them"
        " (env MUSIC_MIRROR_FAT32_SAFE)",
    )
    parser.add_argument(
        "--max-path",
        type=int,
        default=int(os.getenv("MUSIC_MIRROR_MAX_PATH", str(MAX_PATH))),
        help=f"longest path the device will take, counted from its root; Rockbox's"
        f" MAX_PATH is {MAX_PATH} (env MUSIC_MIRROR_MAX_PATH)",
    )
    parser.add_argument(
        "--device-prefix",
        default=os.getenv("MUSIC_MIRROR_DEVICE_PREFIX", DEVICE_PREFIX),
        help="directory the mirror is copied into on the device, whose length comes"
        " out of the path budget (env MUSIC_MIRROR_DEVICE_PREFIX)",
    )
    parser.add_argument(
        "--no-prune",
        action="store_true",
        help="keep mirror files whose source has been deleted",
    )
    parser.add_argument(
        "--no-replaygain",
        action="store_true",
        default=os.getenv("MUSIC_MIRROR_REPLAYGAIN", "").lower() in ("0", "false", "no"),
        help="do not write ReplayGain tags; skips the rsgain pass over changed"
        " albums (env MUSIC_MIRROR_REPLAYGAIN=0)",
    )
    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)

    # Directories are created with 0o777 masked by the umask, so clear the bits
    # that matter from it once here rather than chmod'ing every directory the
    # walk creates. The `other` bits are left alone, since whether the mirror is
    # world-readable is a real policy question; owner and group access is not.
    # Files cannot be handled this way -- mkstemp and copy2 both set a mode
    # outright, ignoring the umask -- so they get an explicit chmod instead.
    inherited = os.umask(0o077)
    os.umask(inherited & ~DIRECTORY_ACCESS)

    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

    # The device's limit covers the whole path it will see, so what the mirror
    # may spend is that less the directory it gets copied into.
    budget = max(0, args.max_path - len(device_prefix_length(args.device_prefix)))
    if args.fat32_safe:
        logger.info(
            "paths are limited to %d characters, from --max-path %d less the %r prefix",
            budget,
            args.max_path,
            args.device_prefix,
        )

    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,
                args.fat32_safe,
                budget,
                not args.no_replaygain,
            )
            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()