feat: FAT32-safe mirror, album art, and a Rockbox sync script #6
@@ -27,5 +27,7 @@ FROM runtime AS test
|
||||
|
||||
RUN pip install --no-cache-dir pytest
|
||||
COPY pytest.ini ./
|
||||
# Host-side tools; not in the runtime image, but the suite covers them.
|
||||
COPY tools ./tools
|
||||
COPY tests ./tests
|
||||
RUN python -m pytest
|
||||
|
||||
@@ -93,6 +93,7 @@ music-mirror --source /music --mirror /music-mp3 --subdir "Artist/Album"
|
||||
| `--jobs` | `MUSIC_MIRROR_JOBS` | CPU count | Concurrent encodes |
|
||||
| `--interval` | `MUSIC_MIRROR_INTERVAL` | unset | Repeat forever, e.g. `45m`, `6h`, `1d` |
|
||||
| `--subdir` | — | unset | Limit the pass to one directory; skips pruning |
|
||||
| `--fat32-safe` | `MUSIC_MIRROR_FAT32_SAFE` | off | Name files so a FAT32 device accepts them |
|
||||
| `--no-prune` | — | off | Keep mirror files whose source has gone |
|
||||
| `--dry-run` | — | off | Report what would change, write nothing |
|
||||
|
||||
@@ -137,6 +138,17 @@ New Lidarr imports are picked up on the next pass. With `MUSIC_MIRROR_INTERVAL`
|
||||
at `6h` that is the worst case; run `--subdir` by hand if you want an album
|
||||
immediately.
|
||||
|
||||
## Tools
|
||||
|
||||
Host-side scripts under `tools/`, not part of the container image.
|
||||
|
||||
`check_fat32.py` reports paths a FAT32 device will not accept — reserved
|
||||
characters, trailing dots and spaces, over-long components and paths, and names
|
||||
colliding case-insensitively. Run it against the mirror **before** an rsync:
|
||||
rsync reports the failures too, but scattered through fifty thousand files where
|
||||
they are easy to miss. Exits non-zero when it finds anything, so it can gate a
|
||||
script.
|
||||
|
||||
## Tests
|
||||
|
||||
```sh
|
||||
@@ -158,6 +170,41 @@ Nix machine:
|
||||
nix shell nixpkgs#python3Packages.pytest nixpkgs#ffmpeg -c pytest
|
||||
```
|
||||
|
||||
## FAT32 and Rockbox
|
||||
|
||||
`--fat32-safe` names mirror files so a FAT32 device will accept them. Off by
|
||||
default, because turning it on renames files and that should be a decision
|
||||
rather than a surprise.
|
||||
|
||||
What it handles, per path component:
|
||||
|
||||
| Problem | Treatment |
|
||||
| ------------------------------- | ------------------------------ |
|
||||
| `< > : " \ \| ? *` and control characters | replaced with `_` |
|
||||
| trailing dots and spaces | stripped — FAT eats them silently, so the name round-trips as a different name |
|
||||
| a component left empty | becomes `_` |
|
||||
| names differing only in case | detected and reported; one wins, as with any other collision |
|
||||
|
||||
`Dada Life - Kick Out the Epic Motherf**ker` is a real example from a real
|
||||
library. Without this it simply never arrives on the device.
|
||||
|
||||
**Turning it on does not re-encode anything.** Every track whose name held a
|
||||
reserved character changes path, and re-encoding those would be hours of work
|
||||
producing files that already exist byte for byte. The run moves them instead,
|
||||
and says so. Prune then finds nothing to remove because nothing was left
|
||||
behind.
|
||||
|
||||
### Album art
|
||||
|
||||
Rockbox looks for cover art **on the filesystem** — `cover.jpg`, `folder.jpg`
|
||||
and friends beside the track or in its parent — and that search never touches
|
||||
the picture embedded in the tag. So a JPEG cover found beside the source is now
|
||||
copied into the mirror as `cover.jpg`, in addition to being embedded. The iPod
|
||||
firmware reads the embedded one; Rockbox reads the file. Both are satisfied.
|
||||
|
||||
A cover left behind in a directory whose tracks have all gone is pruned, or the
|
||||
directory would never look empty and never be removed.
|
||||
|
||||
## Getting the result onto an iPod
|
||||
|
||||
The mirror is just a directory of MP3s, so any client will do:
|
||||
@@ -169,6 +216,30 @@ The mirror is just a directory of MP3s, so any client will do:
|
||||
- **Linux.** Rhythmbox links `libgpod` and handles iPod sync. An iPod Video
|
||||
(5th generation) predates the models whose database has to be signed, so no
|
||||
firmware-hash trickery is needed.
|
||||
- **Rockbox.** No database to write at all — it reads a plain directory tree
|
||||
and builds its own index from tags. Copy the mirror across with rsync:
|
||||
|
||||
```sh
|
||||
python3 tools/check_fat32.py /mnt/tank/media/music-mp3 # before, not during
|
||||
rsync -rtv --delete --modify-window=2 \
|
||||
/mnt/tank/media/music-mp3/ /media/IPOD/Music/
|
||||
```
|
||||
|
||||
`--modify-window=2` because FAT stores modification times to two-second
|
||||
resolution; without it rsync re-copies the entire library on every run. `-rt`
|
||||
rather than `-a` because owners, groups and permissions mean nothing on FAT
|
||||
and asking for them only produces errors.
|
||||
|
||||
Do not route this through Rhythmbox. Its `rb_ipod_helpers_is_ipod()` reads
|
||||
`access-protocols` from media-player-info first and returns true without
|
||||
looking at the filesystem at all, so an iPod in disk mode is identified by its
|
||||
USB id and managed as an iPod — writing a database Rockbox does not want.
|
||||
Deleting `iPod_Control` does not change this. Either use rsync, or untick
|
||||
Preferences → Plugins → Portable Players - iPod.
|
||||
|
||||
Faster still for the first bulk copy: take the card out of the iFlash adapter
|
||||
and use a card reader. Fifty thousand files over USB 2.0 through an iPod is a
|
||||
long evening, and it avoids Rockbox's USB stack entirely.
|
||||
|
||||
Neither client transcodes at sync time; they copy finished MP3s.
|
||||
|
||||
|
||||
+117
-9
@@ -71,12 +71,23 @@ SOURCE_PRIORITY = [
|
||||
# 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"
|
||||
|
||||
# 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]')
|
||||
|
||||
# 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
|
||||
@@ -126,9 +137,30 @@ def parse_interval(interval):
|
||||
return value * {"": 1, "s": 1, "m": 60, "h": 3600, "d": 86400}[match.group(2)]
|
||||
|
||||
|
||||
def mirror_path_for(source, source_root, mirror_root):
|
||||
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 mirror_path_for(source, source_root, mirror_root, safe=False):
|
||||
"""Return the mirror path corresponding to a source file."""
|
||||
return (mirror_root / source.relative_to(source_root)).with_suffix(MIRROR_SUFFIX)
|
||||
relative = source.relative_to(source_root).with_suffix(MIRROR_SUFFIX)
|
||||
if safe:
|
||||
relative = Path(*(fat32_safe(part) for part in relative.parts))
|
||||
return mirror_root / relative
|
||||
|
||||
|
||||
def is_current(source, mirror):
|
||||
@@ -138,6 +170,30 @@ def is_current(source, mirror):
|
||||
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
|
||||
@@ -228,6 +284,7 @@ def encode(source, mirror, quality_args, dry_run):
|
||||
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.
|
||||
@@ -275,6 +332,7 @@ def copy(source, mirror, dry_run):
|
||||
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
|
||||
@@ -299,8 +357,26 @@ def copy(source, mirror, dry_run):
|
||||
return Result("copied", mirror)
|
||||
|
||||
|
||||
def process(source, mirror, quality_args, dry_run):
|
||||
def adopt_existing(source, mirror, previous):
|
||||
"""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.
|
||||
"""
|
||||
if previous == mirror or not previous.is_file() or not is_current(source, previous):
|
||||
return False
|
||||
mirror.parent.mkdir(parents=True, exist_ok=True)
|
||||
os.replace(previous, mirror)
|
||||
logger.info("renamed %s -> %s", previous.name, mirror.name)
|
||||
return True
|
||||
|
||||
|
||||
def process(source, mirror, quality_args, dry_run, previous=None):
|
||||
"""Bring one source file's mirror entry up to date."""
|
||||
if previous is not None and not dry_run and not mirror.exists():
|
||||
adopt_existing(source, mirror, previous)
|
||||
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
|
||||
@@ -324,7 +400,7 @@ def find_sources(root):
|
||||
yield path
|
||||
|
||||
|
||||
def plan(scan_root, source_root, mirror_root):
|
||||
def plan(scan_root, source_root, mirror_root, safe=False):
|
||||
"""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
|
||||
@@ -335,12 +411,20 @@ def plan(scan_root, source_root, mirror_root):
|
||||
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)
|
||||
rival = chosen.get(mirror)
|
||||
mirror = mirror_path_for(source, source_root, mirror_root, safe)
|
||||
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
|
||||
@@ -374,6 +458,13 @@ def prune(mirror_root, expected, dry_run):
|
||||
mirror.unlink(missing_ok=True)
|
||||
|
||||
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()):
|
||||
@@ -382,7 +473,9 @@ def prune(mirror_root, expected, dry_run):
|
||||
return removed
|
||||
|
||||
|
||||
def run_once(scan_root, source_root, mirror_root, quality_args, jobs, dry_run, do_prune):
|
||||
def run_once(
|
||||
scan_root, source_root, mirror_root, quality_args, jobs, dry_run, do_prune, safe=False
|
||||
):
|
||||
"""Run a single pass. Returns the number of failures.
|
||||
|
||||
`scan_root` is what gets walked and `source_root` is what mirror paths are
|
||||
@@ -393,11 +486,18 @@ def run_once(scan_root, source_root, mirror_root, quality_args, jobs, dry_run, d
|
||||
counts = {"encoded": 0, "copied": 0, "skipped": 0, "failed": 0}
|
||||
failures = []
|
||||
|
||||
work = plan(scan_root, source_root, mirror_root)
|
||||
work = plan(scan_root, source_root, mirror_root, safe)
|
||||
|
||||
with concurrent.futures.ThreadPoolExecutor(max_workers=jobs) as pool:
|
||||
futures = [
|
||||
pool.submit(process, source, mirror, quality_args, dry_run)
|
||||
pool.submit(
|
||||
process,
|
||||
source,
|
||||
mirror,
|
||||
quality_args,
|
||||
dry_run,
|
||||
mirror_path_for(source, source_root, mirror_root) if safe else None,
|
||||
)
|
||||
for mirror, source in work.items()
|
||||
]
|
||||
for future in concurrent.futures.as_completed(futures):
|
||||
@@ -492,6 +592,13 @@ def build_parser():
|
||||
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(
|
||||
"--no-prune",
|
||||
action="store_true",
|
||||
@@ -575,6 +682,7 @@ def main(argv=None):
|
||||
args.jobs,
|
||||
args.dry_run,
|
||||
do_prune,
|
||||
args.fat32_safe,
|
||||
)
|
||||
if interval is None or stopping:
|
||||
return 1 if failures else 0
|
||||
|
||||
@@ -78,6 +78,23 @@ def make_flac():
|
||||
return factory
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def make_cover():
|
||||
"""Return a factory writing a small JPEG beside an album's tracks."""
|
||||
|
||||
def factory(path):
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
subprocess.run(
|
||||
["ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "error", "-y",
|
||||
"-f", "lavfi", "-i", "color=c=red:s=64x64:d=1", "-frames:v", "1", str(path)],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
)
|
||||
return path
|
||||
|
||||
return factory
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def probe_tag():
|
||||
"""Return a helper reading a single metadata tag from a file."""
|
||||
|
||||
@@ -0,0 +1,64 @@
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools"))
|
||||
|
||||
import check_fat32 # noqa: E402
|
||||
|
||||
|
||||
def problems(name):
|
||||
return check_fat32.problems_with(Path(name))
|
||||
|
||||
|
||||
def test_a_reserved_character_is_a_problem():
|
||||
assert any("reserved" in p for p in problems("Album/Motherf**ker.mp3"))
|
||||
|
||||
|
||||
def test_a_trailing_dot_or_space_is_a_problem():
|
||||
"""FAT eats them silently, so the name round-trips as a different name."""
|
||||
assert any("trailing" in p for p in problems("Trailing Dot./x.mp3"))
|
||||
assert any("trailing" in p for p in problems("Space /x.mp3"))
|
||||
|
||||
|
||||
def test_an_ordinary_name_is_fine():
|
||||
assert problems("Artist/Album/01 Fine Track.mp3") == []
|
||||
|
||||
|
||||
def test_accents_are_fine():
|
||||
assert problems("Mötley Crüe/Album/Track.mp3") == []
|
||||
|
||||
|
||||
def test_an_over_long_path_is_a_problem():
|
||||
deep = "/".join("d" * 40 for _ in range(10)) + "/track.mp3"
|
||||
assert any("path of" in p for p in problems(deep))
|
||||
|
||||
|
||||
def test_case_collisions_are_reported(tmp_path):
|
||||
album = tmp_path / "Album"
|
||||
album.mkdir()
|
||||
(album / "Song.mp3").write_bytes(b"x")
|
||||
(album / "SONG.mp3").write_bytes(b"x")
|
||||
|
||||
completed = subprocess.run(
|
||||
[sys.executable, str(Path(check_fat32.__file__)), str(tmp_path)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
|
||||
assert completed.returncode == 1
|
||||
assert "collides case-insensitively" in completed.stdout
|
||||
|
||||
|
||||
def test_a_clean_tree_exits_zero(tmp_path):
|
||||
(tmp_path / "Album").mkdir()
|
||||
(tmp_path / "Album" / "Fine.mp3").write_bytes(b"x")
|
||||
|
||||
completed = subprocess.run(
|
||||
[sys.executable, str(Path(check_fat32.__file__)), str(tmp_path)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
|
||||
assert completed.returncode == 0
|
||||
assert "0 problems" in completed.stderr
|
||||
@@ -443,3 +443,118 @@ def test_mirror_inside_source_is_refused(tmp_path, make_flac):
|
||||
|
||||
def test_missing_source_is_refused(tmp_path):
|
||||
assert run(tmp_path / "nope", tmp_path / "dst") == 2
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("name", "expected"),
|
||||
[
|
||||
("Kick Out the Epic Motherf**ker", "Kick Out the Epic Motherf__ker"),
|
||||
("Where Are You?", "Where Are You_"),
|
||||
("Song: Part 2", "Song_ Part 2"),
|
||||
('"Heroes"', "_Heroes_"),
|
||||
("trailing dot.", "trailing dot"),
|
||||
("trailing space ", "trailing space"),
|
||||
("...", "_"),
|
||||
("Mötley Crüe", "Mötley Crüe"),
|
||||
("perfectly ordinary", "perfectly ordinary"),
|
||||
],
|
||||
)
|
||||
def test_fat32_safe_names(name, expected):
|
||||
"""A name FAT32 will not take is a track that silently does not arrive."""
|
||||
assert music_mirror.fat32_safe(name) == expected
|
||||
|
||||
|
||||
def test_a_reserved_character_is_replaced_in_the_mirror_path(tmp_path, make_flac):
|
||||
source = tmp_path / "src"
|
||||
mirror = tmp_path / "dst"
|
||||
make_flac(source / "Dada Life" / "Album" / "Kick Out the Epic Motherf**ker.flac")
|
||||
|
||||
run(source, mirror, "--fat32-safe")
|
||||
|
||||
assert (mirror / "Dada Life" / "Album" / "Kick Out the Epic Motherf__ker.mp3").is_file()
|
||||
|
||||
|
||||
def test_without_the_flag_the_name_is_left_alone(tmp_path, make_flac):
|
||||
source = tmp_path / "src"
|
||||
mirror = tmp_path / "dst"
|
||||
make_flac(source / "Album" / "Where Are You?.flac")
|
||||
|
||||
run(source, mirror)
|
||||
|
||||
assert (mirror / "Album" / "Where Are You?.mp3").is_file()
|
||||
|
||||
|
||||
def test_an_existing_mirror_is_renamed_not_re_encoded(tmp_path, make_flac):
|
||||
"""Turning the flag on changes the path of every track holding a reserved
|
||||
character. Re-encoding those would be hours of work to produce files that
|
||||
already exist byte for byte."""
|
||||
source = tmp_path / "src"
|
||||
mirror = tmp_path / "dst"
|
||||
make_flac(source / "Album" / "Where Are You?.flac")
|
||||
|
||||
run(source, mirror)
|
||||
before = mirror / "Album" / "Where Are You?.mp3"
|
||||
contents = before.read_bytes()
|
||||
stamp = before.stat().st_mtime_ns
|
||||
|
||||
run(source, mirror, "--fat32-safe")
|
||||
|
||||
after = mirror / "Album" / "Where Are You_.mp3"
|
||||
assert after.is_file()
|
||||
assert not before.exists()
|
||||
assert after.read_bytes() == contents, "it was re-encoded rather than moved"
|
||||
assert after.stat().st_mtime_ns == stamp
|
||||
|
||||
|
||||
def test_names_colliding_only_by_case_are_caught(tmp_path, make_flac):
|
||||
"""Two files here, one file on FAT32. Better found now than as a silent
|
||||
overwrite partway through the copy."""
|
||||
source = tmp_path / "src"
|
||||
mirror = tmp_path / "dst"
|
||||
make_flac(source / "Album" / "Song.flac")
|
||||
make_flac(source / "Album" / "SONG.flac")
|
||||
|
||||
run(source, mirror, "--fat32-safe")
|
||||
|
||||
written = sorted(p.name for p in (mirror / "Album").glob("*.mp3"))
|
||||
assert len(written) == 1, written
|
||||
|
||||
|
||||
def test_the_album_cover_is_copied_beside_the_tracks(tmp_path, make_flac, make_cover):
|
||||
"""Rockbox looks for art on the filesystem; its search never touches the
|
||||
picture embedded in the tag."""
|
||||
source = tmp_path / "src"
|
||||
mirror = tmp_path / "dst"
|
||||
make_flac(source / "Album" / "a.flac")
|
||||
make_cover(source / "Album" / "cover.jpg")
|
||||
|
||||
run(source, mirror)
|
||||
|
||||
assert (mirror / "Album" / "cover.jpg").is_file()
|
||||
|
||||
|
||||
def test_a_copied_cover_is_group_readable(tmp_path, make_flac, make_cover, tight_umask):
|
||||
source = tmp_path / "src"
|
||||
mirror = tmp_path / "dst"
|
||||
make_flac(source / "Album" / "a.flac")
|
||||
make_cover(source / "Album" / "cover.jpg")
|
||||
|
||||
run(source, mirror)
|
||||
|
||||
assert (mirror / "Album" / "cover.jpg").stat().st_mode & stat.S_IRGRP
|
||||
|
||||
|
||||
def test_a_cover_left_without_tracks_is_pruned(tmp_path, make_flac, make_cover):
|
||||
"""Otherwise the directory never looks empty and never goes."""
|
||||
source = tmp_path / "src"
|
||||
mirror = tmp_path / "dst"
|
||||
make_flac(source / "Gone" / "a.flac")
|
||||
make_cover(source / "Gone" / "cover.jpg")
|
||||
|
||||
run(source, mirror)
|
||||
assert (mirror / "Gone" / "cover.jpg").is_file()
|
||||
|
||||
shutil.rmtree(source / "Gone")
|
||||
run(source, mirror)
|
||||
|
||||
assert not (mirror / "Gone").exists()
|
||||
|
||||
Executable
+93
@@ -0,0 +1,93 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Report paths a FAT32 device will not accept, before copying rather than during.
|
||||
|
||||
Run this against the mirror before an rsync to a Rockbox iPod. rsync will
|
||||
report the failures too, but scattered through a run of fifty thousand files,
|
||||
where they are easy to lose.
|
||||
|
||||
Checks the four ways a name fails on FAT32: reserved characters, trailing dots
|
||||
or spaces that FAT silently eats, components longer than 255 characters, and
|
||||
names that differ only in case -- two files here, one file there, and the
|
||||
second silently overwrites the first.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import unicodedata
|
||||
from collections import defaultdict
|
||||
from pathlib import Path
|
||||
|
||||
RESERVED = re.compile(r'[<>:"\\|?*\x00-\x1f]')
|
||||
COMPONENT_LIMIT = 255
|
||||
# Rockbox builds paths into a fixed buffer; long trees fail on the device even
|
||||
# when every individual component is legal.
|
||||
PATH_LIMIT = 260
|
||||
|
||||
|
||||
def problems_with(relative):
|
||||
"""Return every reason this relative path is unfit for FAT32."""
|
||||
found = []
|
||||
for part in relative.parts:
|
||||
if RESERVED.search(part):
|
||||
found.append(f"reserved character in {part!r}")
|
||||
if part != part.rstrip(". "):
|
||||
found.append(f"trailing dot or space in {part!r}")
|
||||
if len(part) > COMPONENT_LIMIT:
|
||||
found.append(f"component of {len(part)} characters")
|
||||
if len(str(relative)) > PATH_LIMIT:
|
||||
found.append(f"path of {len(str(relative))} characters")
|
||||
return found
|
||||
|
||||
|
||||
def walk(root):
|
||||
"""Yield every file below a root, as a path relative to it."""
|
||||
for base, _, names in os.walk(root):
|
||||
for name in names:
|
||||
yield Path(os.path.join(base, name)).relative_to(root)
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("root", help="directory to check, e.g. the mirror")
|
||||
parser.add_argument("--limit", type=int, default=0, help="show at most this many")
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
root = Path(args.root)
|
||||
if not root.is_dir():
|
||||
print(f"{root} is not a directory", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
faults = []
|
||||
by_case = defaultdict(list)
|
||||
total = 0
|
||||
for relative in walk(root):
|
||||
total += 1
|
||||
# NFC first: the same name written by two systems is otherwise two
|
||||
# different strings, and the collision check would miss it.
|
||||
key = unicodedata.normalize("NFC", str(relative)).casefold()
|
||||
by_case[key].append(relative)
|
||||
for problem in problems_with(relative):
|
||||
faults.append((relative, problem))
|
||||
|
||||
for relative, group in sorted(by_case.items()):
|
||||
if len(group) > 1:
|
||||
names = ", ".join(str(path) for path in sorted(group))
|
||||
faults.append((group[0], f"collides case-insensitively with: {names}"))
|
||||
|
||||
for relative, problem in faults[: args.limit or None]:
|
||||
print(f"{relative}\t{problem}")
|
||||
|
||||
print(f"\n{len(faults)} problems across {total} files", file=sys.stderr)
|
||||
if faults:
|
||||
print(
|
||||
"Run music-mirror with --fat32-safe to have the mirror named"
|
||||
" acceptably in the first place.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 1 if faults else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user