Author SHA1 Message Date
Emma Thorpe 658c70a3dc chore: add a tool for finding Music tracks whose files have gone
Build and publish container / build (pull_request) Successful in 3m8s
Lidarr renames an artist or album folder, music-mirror prunes the old path and
encodes the new one, and every entry in Apple Music pointing at the old path
dies. Music offers no view of those, and the exclamation mark only appears once
a track is touched.

Reads the XML from Music's own Export Library rather than asking Music itself.
A broken track makes AppleScript's `location` raise instead of returning a
value, so a bulk query dies on the first one with error -1728 and a per-track
loop costs an Apple event apiece.

Checked against a single directory walk rather than a test per file. On a
fifty-thousand-track library that is around seven thousand directory reads
instead of fifty thousand stat calls, and over SMB each of those stats is a
network round trip -- which is the difference between seconds and minutes.

Two comparisons have to be loosened or most of the library reads as missing.
macOS stores filenames decomposed while the share composes them, so the umlauts
in Motley Crue are two different byte strings depending on which side wrote the
name; both are normalised to NFC. And the share is very likely
case-insensitive, so a file is not missing because someone capitalised it
differently.

The test stage now copies tools/ as well, since the suite covers this and the
runtime image deliberately does not carry it.
2026-08-24 19:36:06 +01:00
7 changed files with 367 additions and 506 deletions
+3
View File
@@ -28,5 +28,8 @@ FROM runtime AS test
RUN pip install --no-cache-dir pytest RUN pip install --no-cache-dir pytest
COPY pytest.ini ./ COPY pytest.ini ./
# The host-side tools are not part of the runtime image, but their tests are
# part of the suite, so they have to be present for it.
COPY tools ./tools
COPY tests ./tests COPY tests ./tests
RUN python -m pytest RUN python -m pytest
+41 -47
View File
@@ -7,7 +7,7 @@ which keeps an MP3 copy of a lossless library for an iPod. This one answers the
question that mirror cannot: which of it is worth carrying, and which of it has question that mirror cannot: which of it is worth carrying, and which of it has
not been played in years. not been played in years.
**This is stage four, read-only.** It ingests the scrobble history, indexes the library **This is stage three.** It ingests the scrobble history, indexes the library
from Lidarr, matches one to the other, and writes playlists into the mirror — from Lidarr, matches one to the other, and writes playlists into the mirror —
by listening history and by mood. by listening history and by mood.
Nothing is written back to Lidarr — every call there is a `GET`. See "Where this Nothing is written back to Lidarr — every call there is a `GET`. See "Where this
@@ -20,8 +20,7 @@ is going" below.
- Indexes every artist, album and track Lidarr knows about, with file paths and - Indexes every artist, album and track Lidarr knows about, with file paths and
the date each file landed. the date each file landed.
- Ties the two together and reports how well it managed. - Ties the two together and reports how well it managed.
- Writes M3U playlists into the mirror, from the listening history and by mood. - Writes M3U playlists into the mirror, from the listening history.
- Reports which albums have never been played. It does not act on that.
## Matching ## Matching
@@ -132,13 +131,7 @@ overstates the problem and would over-block the cull.
## Playlists ## Playlists
Written into `<mirror>/_playlists/` as extended M3U, rebuilt every pass. Written into `<mirror>/_playlists/` as extended M3U, rebuilt every pass. Six
The extension is **`.m3u8`**, not `.m3u`. Rockbox's `is_m3u8_name()` treats
every extension as UTF-8 *except* an explicit `.m3u`, which it decodes through
the user's configured codepage instead — so a plain `.m3u` mangles every
accented filename. No byte order mark is written: Rockbox does not need one at
this extension, and a BOM upsets players that do not expect it. Six
rules, capped at `--playlist-limit` tracks each: rules, capped at `--playlist-limit` tracks each:
| Playlist | Rule | | Playlist | Rule |
@@ -283,40 +276,6 @@ or `--mirror` looks like, since the paths then map to nothing at all.
`_playlists/` survives music-mirror's prune: it only deletes `*.mp3`, and its `_playlists/` survives music-mirror's prune: it only deletes `*.mp3`, and its
empty-directory sweep skips a directory holding M3Us. empty-directory sweep skips a directory holding M3Us.
## The cold report
Which albums have files, have never had a single track played in the whole
history, and have sat there long enough to have had the chance. Ranked by the
disk they occupy, because that is the point of the exercise.
**Album-level, not track-level.** A record with two played tracks is a record
that gets played; picking the other ten off it leaves gaps rather than
reclaiming anything worth having.
The age floor is measured from the newest file in the album, not from the
release date — what matters is how long it has been available to play, not how
old the record is. `--cold-after` sets it, defaulting to a year.
Artists whose *every* album is cold are counted separately. That is a different
proposition from one cold record by somebody otherwise played, and Lidarr can
only tag at artist level anyway.
### What stops it
The report refuses to produce anything at all when:
- the scrobble backfill is unfinished
- any artist failed to index, or any artist has no albums indexed
- the library has not been indexed, or there is no history to judge against
Each of those makes played music look unplayed, which is the single failure that
costs a library. They are checked rather than trusted, because the report they
gate is the one that ends in deletion.
**Nothing is written to Lidarr.** Every call there is still a `GET`. Unmonitoring
comes once the list has been looked at, because no flag protects against a list
that is wrong.
## How the ingest works ## How the ingest works
Two halves, both taking their bounds from the database rather than from a saved Two halves, both taking their bounds from the database rather than from a saved
@@ -376,7 +335,6 @@ music-curator --report-only # report on the store, fetch nothing
| `--playlist-limit` | `MUSIC_CURATOR_PLAYLIST_LIMIT` | `100` | Most tracks in any one playlist | | `--playlist-limit` | `MUSIC_CURATOR_PLAYLIST_LIMIT` | `100` | Most tracks in any one playlist |
| `--vibes` | `MUSIC_CURATOR_VIBES` | built-in | JSON file of mood definitions | | `--vibes` | `MUSIC_CURATOR_VIBES` | built-in | JSON file of mood definitions |
| `--tag-limit` | `MUSIC_CURATOR_TAG_LIMIT` | `0` | Cap artist tag lookups per pass | | `--tag-limit` | `MUSIC_CURATOR_TAG_LIMIT` | `0` | Cap artist tag lookups per pass |
| `--cold-after` | `MUSIC_CURATOR_COLD_AFTER` | `365` | Days a file must sit unplayed to count as cold |
| `--skip-index` | — | off | Match against the index already held | | `--skip-index` | — | off | Match against the index already held |
| `--report-only` | — | off | Report without fetching | | `--report-only` | — | off | Report without fetching |
@@ -419,6 +377,43 @@ stubbed. No network, no credentials, no rate limit. On a Nix machine:
nix shell nixpkgs#python3Packages.pytest -c pytest nix shell nixpkgs#python3Packages.pytest -c pytest
``` ```
## Tools
Host-side scripts under `tools/`. Not part of the container image; run them
wherever they are needed.
### `find_missing_tracks.py`
Reports tracks in an Apple Music library whose files are no longer on disk —
which happens whenever Lidarr renames an artist or album folder and
music-mirror prunes the old path.
```sh
# Music: File > Library > Export Library... then, with the share mounted:
python3 tools/find_missing_tracks.py Library.xml --root /Volumes/music-mp3
```
Reading the exported XML rather than asking Music itself is deliberate. A
broken track makes AppleScript's `location` raise instead of returning a value,
so a bulk query dies on the first one with `-1728` and a per-track loop costs an
Apple event apiece.
It checks against a **single directory walk**, not a test per file. On a
50,000-track library that is ~7,000 directory reads instead of 50,000 stat
calls, and over SMB every one of those stats is a network round trip.
Two comparisons that have to be loosened, or most of the library reads as
missing:
- **Unicode.** macOS stores filenames decomposed; the share composes them.
`Mötley Crüe` is two different byte strings depending on which side wrote it.
Both sides are normalised to NFC.
- **Case.** The share is very likely case-insensitive. A file is not missing
because someone capitalised it differently.
Pass `--root` if the library holds anything outside the mirror: the root it
otherwise derives is the common parent of every track, which can be `/`.
## Where this is going ## Where this is going
| Stage | Status | | Stage | Status |
@@ -427,8 +422,7 @@ nix shell nixpkgs#python3Packages.pytest -c pytest
| Lidarr index and the scrobble-to-track matcher | done | | Lidarr index and the scrobble-to-track matcher | done |
| M3U playlists from the listening history | done | | M3U playlists from the listening history | done |
| Genre and mood playlists from Last.fm tags | done | | Genre and mood playlists from Last.fm tags | done |
| Cold-music report | done | | Cold-music report, unmonitoring what is not played | last |
| Unmonitoring what is not played | last |
The cull will unmonitor cold albums in Lidarr and tag their artists. It will The cull will unmonitor cold albums in Lidarr and tag their artists. It will
never delete files: Lidarr's `AlbumResource` has no tags at all, so tagging never delete files: Lidarr's `AlbumResource` has no tags at all, so tagging
+18 -243
View File
@@ -69,17 +69,6 @@ BACKOFF_CEILING_SECONDS = 60.0
MIRROR_SUFFIX = ".mp3" MIRROR_SUFFIX = ".mp3"
PLAYLIST_DIRECTORY = "_playlists" PLAYLIST_DIRECTORY = "_playlists"
# .m3u8, not .m3u. Rockbox's is_m3u8_name() treats every extension as UTF-8
# except an explicit ".m3u", which it decodes through the user's configured
# codepage instead -- so a plain .m3u mangles every accented filename, and this
# library holds Mötley Crüe, Beyoncé and Sigur Rós.
PLAYLIST_SUFFIX = ".m3u8"
# Extensions this tool has written in the past. A playlist of the same name
# under one of these is a leftover of its own, and is cleaned up even though no
# record of writing it survives.
SUPERSEDED_SUFFIXES = (".m3u",)
# Tags are re-fetched this often. They move slowly, and the first pass over a # Tags are re-fetched this often. They move slowly, and the first pass over a
# library already costs one request per artist. # library already costs one request per artist.
TAG_REFRESH_SECONDS = 90 * 86400 TAG_REFRESH_SECONDS = 90 * 86400
@@ -96,17 +85,12 @@ SAFE_VIBE_NAME = re.compile(r"^[a-z0-9][a-z0-9-]*$")
# back by whatever serves it, and a playlist nobody can read is not a playlist. # back by whatever serves it, and a playlist nobody can read is not a playlist.
GROUP_READ = 0o040 GROUP_READ = 0o040
SCHEMA_VERSION = "3" SCHEMA_VERSION = "2"
# Versions this build upgrades in place. Everything added since version 1 is a # Versions this build upgrades in place. Everything added since version 1 is a
# new table or a new column, both of which are applied in place, so an existing # new table, and the schema script only ever creates what is missing, so running
# history is never re-downloaded -- a full backfill is thousands of requests. # it is the whole migration -- an existing history is not re-downloaded.
MIGRATABLE_FROM = {"1", "2"} MIGRATABLE_FROM = {"1"}
# Columns added to existing tables after the fact. CREATE TABLE IF NOT EXISTS
# will not add a column to a table that already exists, so these are applied
# separately and only when missing.
ADDED_COLUMNS = (("lidarr_track", "size", "INTEGER"),)
SCHEMA = """ SCHEMA = """
-- One row per scrobble. The primary key collapses two plays of the same track -- One row per scrobble. The primary key collapses two plays of the same track
@@ -176,8 +160,7 @@ CREATE TABLE IF NOT EXISTS lidarr_track (
has_file INTEGER NOT NULL DEFAULT 0, has_file INTEGER NOT NULL DEFAULT 0,
path TEXT, path TEXT,
added INTEGER, added INTEGER,
duration INTEGER, duration INTEGER
size INTEGER
); );
CREATE INDEX IF NOT EXISTS lidarr_track_recording ON lidarr_track (recording_mbid); CREATE INDEX IF NOT EXISTS lidarr_track_recording ON lidarr_track (recording_mbid);
CREATE INDEX IF NOT EXISTS lidarr_track_norm ON lidarr_track (norm_artist, norm_title); CREATE INDEX IF NOT EXISTS lidarr_track_norm ON lidarr_track (norm_artist, norm_title);
@@ -514,18 +497,8 @@ class Store:
# thousand name pairs into Python to compare them one at a time. # thousand name pairs into Python to compare them one at a time.
self.connection.create_function("normalise", 1, normalise, deterministic=True) self.connection.create_function("normalise", 1, normalise, deterministic=True)
self.connection.executescript(SCHEMA) self.connection.executescript(SCHEMA)
self._add_missing_columns()
self._check_version() self._check_version()
def _add_missing_columns(self):
"""Apply column additions to tables that predate them."""
for table, column, kind in ADDED_COLUMNS:
held = {row[1] for row in self.connection.execute(f"PRAGMA table_info({table})")}
if column not in held:
logger.info("adding %s.%s to the store", table, column)
with self.connection:
self.connection.execute(f"ALTER TABLE {table} ADD COLUMN {column} {kind}")
def _check_version(self): def _check_version(self):
held = self.get_state("schema_version") held = self.get_state("schema_version")
if held is None or held in MIGRATABLE_FROM: if held is None or held in MIGRATABLE_FROM:
@@ -607,8 +580,8 @@ class Store:
self.connection.executemany( self.connection.executemany(
"INSERT INTO lidarr_track" "INSERT INTO lidarr_track"
" (id, artist_id, album_id, recording_mbid, title, norm_artist, norm_title," " (id, artist_id, album_id, recording_mbid, title, norm_artist, norm_title,"
" has_file, path, added, duration, size)" " has_file, path, added, duration)"
" VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
tracks, tracks,
) )
@@ -1067,7 +1040,6 @@ def index_library(client, store):
handle.get("path"), handle.get("path"),
parse_added(handle.get("dateAdded")), parse_added(handle.get("dateAdded")),
track.get("duration"), track.get("duration"),
handle.get("size"),
) )
) )
@@ -1501,12 +1473,11 @@ def sync_tags(client, store, now, limit=0):
def build_vibe_playlists(store, vibes, mirror_root, library_root, limit, now): def build_vibe_playlists(store, vibes, mirror_root, library_root, limit, now):
"""Write one playlist per mood. Returns how many tracks were listed.""" """Write one playlist per mood. Returns how many tracks were listed."""
if not store.scalar("SELECT COUNT(*) FROM artist_tag"): if not store.scalar("SELECT COUNT(*) FROM artist_tag"):
return 0, [] return 0
directory = Path(mirror_root) / PLAYLIST_DIRECTORY directory = Path(mirror_root) / PLAYLIST_DIRECTORY
week = now // ROTATION_PERIOD_SECONDS week = now // ROTATION_PERIOD_SECONDS
total = 0 total = 0
produced = []
for vibe in vibes: for vibe in vibes:
tags = [str(tag).strip().casefold() for tag in vibe["tags"]] tags = [str(tag).strip().casefold() for tag in vibe["tags"]]
@@ -1563,14 +1534,11 @@ def build_vibe_playlists(store, vibes, mirror_root, library_root, limit, now):
continue continue
entries.append({**dict(row), "mirror": mirror}) entries.append({**dict(row), "mirror": mirror})
write_playlist( write_playlist(directory / f"{vibe['name']}.m3u", entries, Path(mirror_root))
directory / f"{vibe['name']}{PLAYLIST_SUFFIX}", entries, Path(mirror_root)
)
produced.append(f"{vibe['name']}{PLAYLIST_SUFFIX}")
total += len(entries) total += len(entries)
logger.info("playlist %-20s %4d tracks -- by tag", vibe["name"], len(entries)) logger.info("playlist %-20s %4d tracks -- by tag", vibe["name"], len(entries))
return total, produced return total
def library_root_of(store): def library_root_of(store):
@@ -1654,7 +1622,7 @@ def write_playlist(path, entries, reference=None):
if fresh and reference is not None: if fresh and reference is not None:
match_ownership(path.parent, reference) match_ownership(path.parent, reference)
handle, temporary = tempfile.mkstemp(dir=path.parent, suffix=".m3u8.part") handle, temporary = tempfile.mkstemp(dir=path.parent, suffix=".m3u.part")
os.close(handle) os.close(handle)
temporary = Path(temporary) temporary = Path(temporary)
try: try:
@@ -1673,35 +1641,6 @@ def write_playlist(path, entries, reference=None):
temporary.unlink(missing_ok=True) temporary.unlink(missing_ok=True)
def prune_playlists(directory, produced, store):
"""Delete playlists this build no longer produces.
Driven by a record of what was written last time rather than by "every M3U
that is not one of ours", so a playlist put there by hand is left alone. A
renamed mood otherwise leaves its old file on the device for ever.
That record cannot reach back before it existed, which is how the switch
from .m3u to .m3u8 left every playlist on the device twice: the old files
were written by a version that kept no record, so there was nothing to
prune them by. A file with the same name as one being written now, under a
superseded extension, is therefore removed as well -- narrow enough that
only this tool's own leavings match it.
"""
previous = set(json.loads(store.get_state("playlist_files") or "[]"))
stale_names = previous - set(produced)
for name in produced:
for superseded in SUPERSEDED_SUFFIXES:
stale_names.add(Path(name).with_suffix(superseded).name)
stale_names -= set(produced)
for name in sorted(stale_names):
stale = directory / name
if stale.is_file():
logger.info("removing playlist %s, no longer produced", name)
stale.unlink(missing_ok=True)
store.set_state("playlist_files", json.dumps(sorted(produced)))
def build_playlists(store, mirror_root, library_root, limit, now): def build_playlists(store, mirror_root, library_root, limit, now):
"""Write every playlist into the mirror. Returns how many tracks were listed. """Write every playlist into the mirror. Returns how many tracks were listed.
@@ -1719,7 +1658,6 @@ def build_playlists(store, mirror_root, library_root, limit, now):
total = 0 total = 0
missing = 0 missing = 0
produced = []
for name, description, sql in PLAYLISTS: for name, description, sql in PLAYLISTS:
entries = [] entries = []
for row in store.connection.execute(sql, parameters): for row in store.connection.execute(sql, parameters):
@@ -1728,8 +1666,7 @@ def build_playlists(store, mirror_root, library_root, limit, now):
missing += 1 missing += 1
continue continue
entries.append({**dict(row), "mirror": mirror}) entries.append({**dict(row), "mirror": mirror})
write_playlist(directory / f"{name}{PLAYLIST_SUFFIX}", entries, Path(mirror_root)) write_playlist(directory / f"{name}.m3u", entries, Path(mirror_root))
produced.append(f"{name}{PLAYLIST_SUFFIX}")
total += len(entries) total += len(entries)
logger.info("playlist %-20s %4d tracks -- %s", name, len(entries), description) logger.info("playlist %-20s %4d tracks -- %s", name, len(entries), description)
@@ -1741,7 +1678,7 @@ def build_playlists(store, mirror_root, library_root, limit, now):
" place, since the paths are then being mapped to nothing.", " place, since the paths are then being mapped to nothing.",
missing, missing,
) )
return total, produced return total
def coverage_report(store): def coverage_report(store):
@@ -1915,156 +1852,7 @@ def coverage_report(store):
logger.info(" unmatched %2d: %-60s %d plays", position, label[:60], row["plays"]) logger.info(" unmatched %2d: %-60s %d plays", position, label[:60], row["plays"])
# An album has to have sat unplayed for at least this long before it counts as def report(store, now):
# cold. Measured from the newest file in it, not from the release date: what
# matters is how long it has been available to play, not how old the record is.
COLD_AFTER_DAYS = 365
def human_bytes(count):
"""Return a size that can be read at a glance."""
size = float(count or 0)
for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
if size < 1024 or unit == "TiB":
return f"{size:.1f} {unit}"
size /= 1024
def cull_is_safe(store):
"""Return why a cull must not run, or None if it may.
Every one of these makes played music look unplayed, which is the single
failure that costs a library. They are checked rather than trusted because
the report they gate is the one that ends in deletion.
"""
if store.get_state("backfill_complete") != "yes":
return "the scrobble history is still being backfilled"
if int(store.get_state("index_skipped") or 0):
return "some artists could not be indexed, so their tracks are missing"
if int(store.get_state("index_albums_skipped") or 0):
return "some artists have no albums indexed"
if not store.scalar("SELECT COUNT(*) FROM lidarr_track"):
return "the library has not been indexed"
if not store.scalar("SELECT COUNT(*) FROM scrobble"):
return "there is no listening history to judge against"
return None
COLD_ALBUMS = """
WITH played AS (
SELECT DISTINCT sk.track_id AS track_id
FROM scrobble_key sk
WHERE sk.track_id IS NOT NULL
),
album AS (
SELECT t.album_id AS album_id,
COUNT(*) AS tracks,
SUM(COALESCE(t.size, 0)) AS bytes,
MAX(COALESCE(t.added, 0)) AS newest,
SUM(CASE WHEN p.track_id IS NULL THEN 0 ELSE 1 END) AS plays
FROM lidarr_track t
LEFT JOIN played p ON p.track_id = t.id
WHERE t.has_file = 1
GROUP BY t.album_id
)
SELECT al.id AS album_id,
al.title AS title,
ar.id AS artist_id,
ar.name AS artist,
album.tracks AS tracks,
album.bytes AS bytes,
album.newest AS newest
FROM album
JOIN lidarr_album al ON al.id = album.album_id
JOIN lidarr_artist ar ON ar.id = al.artist_id
WHERE album.plays = 0
AND album.newest > 0
AND album.newest < :cutoff
"""
def cold_albums(store, cutoff):
"""Return every album with files, none of them ever played, old enough to judge."""
return store.connection.execute(COLD_ALBUMS, {"cutoff": cutoff}).fetchall()
def cold_report(store, now, cold_after_days=COLD_AFTER_DAYS):
"""Report what has never been played. Writes nothing, anywhere.
Album-level rather than track-level: a record with two played tracks is a
record that gets played, and picking the other ten off it leaves gaps rather
than reclaiming anything worth having.
"""
refusal = cull_is_safe(store)
if refusal is not None:
logger.warning("no cold report: %s", refusal)
return []
cutoff = now - cold_after_days * 86400
rows = cold_albums(store, cutoff)
if not rows:
logger.info("--- cold albums --- none: everything with a file has been played")
return []
total_bytes = sum(row["bytes"] or 0 for row in rows)
total_tracks = sum(row["tracks"] or 0 for row in rows)
albums_with_files = store.scalar(
"SELECT COUNT(DISTINCT album_id) FROM lidarr_track WHERE has_file = 1"
)
logger.info("--- cold albums (never played, files older than %d days) ---", cold_after_days)
logger.info(
"%d of %d albums, %d tracks, %s",
len(rows),
albums_with_files,
total_tracks,
human_bytes(total_bytes),
)
# An artist every one of whose albums is cold is a different proposition
# from one cold record by someone otherwise played, and Lidarr can only tag
# at artist level anyway.
cold_by_artist = {}
for row in rows:
held = cold_by_artist.setdefault(row["artist_id"], {"name": row["artist"], "albums": 0,
"bytes": 0})
held["albums"] += 1
held["bytes"] += row["bytes"] or 0
owned = dict(
store.connection.execute(
"SELECT al.artist_id, COUNT(DISTINCT al.id) FROM lidarr_album al"
" JOIN lidarr_track t ON t.album_id = al.id AND t.has_file = 1"
" GROUP BY al.artist_id"
).fetchall()
)
entirely = [
held for artist_id, held in cold_by_artist.items()
if held["albums"] == owned.get(artist_id)
]
logger.info(
"%d artists are cold in their entirety (%s), %d have some cold records",
len(entirely),
human_bytes(sum(held["bytes"] for held in entirely)),
len(cold_by_artist) - len(entirely),
)
for position, held in enumerate(
sorted(cold_by_artist.values(), key=lambda held: -held["bytes"])[:20], start=1
):
logger.info(
" cold %2d: %-40s %2d albums, %s",
position,
held["name"][:40],
held["albums"],
human_bytes(held["bytes"]),
)
logger.info("nothing was changed in Lidarr: this report does not write")
return rows
def report(store, now, cold_after=COLD_AFTER_DAYS):
"""Log what the store holds. """Log what the store holds.
The MBID coverage line is the one to watch: scrobbles carrying a MusicBrainz The MBID coverage line is the one to watch: scrobbles carrying a MusicBrainz
@@ -2098,7 +1886,6 @@ def report(store, now, cold_after=COLD_AFTER_DAYS):
logger.info(" top artist %2d: %-40s %d", position, row["artist"][:40], row["plays"]) logger.info(" top artist %2d: %-40s %d", position, row["artist"][:40], row["plays"])
coverage_report(store) coverage_report(store)
cold_report(store, now, cold_after)
recent = store.connection.execute( recent = store.connection.execute(
"SELECT artist, track, COUNT(*) AS plays FROM scrobble WHERE uts >= ?" "SELECT artist, track, COUNT(*) AS plays FROM scrobble WHERE uts >= ?"
@@ -2134,14 +1921,9 @@ def run_once(
" written; set --library-root" " written; set --library-root"
) )
else: else:
_, from_history = build_playlists(store, mirror, root, playlist_limit, now) build_playlists(store, mirror, root, playlist_limit, now)
sync_tags(client, store, now, limit=tag_limit) sync_tags(client, store, now, limit=tag_limit)
_, from_tags = build_vibe_playlists( build_vibe_playlists(store, vibes, mirror, root, playlist_limit, now)
store, vibes, mirror, root, playlist_limit, now
)
prune_playlists(
Path(mirror) / PLAYLIST_DIRECTORY, from_history + from_tags, store
)
logger.info( logger.info(
"pass complete in %.1fs: %d scrobbles added, %d loved", "pass complete in %.1fs: %d scrobbles added, %d loved",
@@ -2245,13 +2027,6 @@ def build_parser():
default=int(os.getenv("MUSIC_CURATOR_TAG_LIMIT", "0")), default=int(os.getenv("MUSIC_CURATOR_TAG_LIMIT", "0")),
help="cap artist tag lookups per pass; 0 for no cap (env MUSIC_CURATOR_TAG_LIMIT)", help="cap artist tag lookups per pass; 0 for no cap (env MUSIC_CURATOR_TAG_LIMIT)",
) )
parser.add_argument(
"--cold-after",
type=int,
default=int(os.getenv("MUSIC_CURATOR_COLD_AFTER", str(COLD_AFTER_DAYS))),
help="days a file must sit unplayed before its album counts as cold"
" (env MUSIC_CURATOR_COLD_AFTER)",
)
parser.add_argument( parser.add_argument(
"--skip-index", "--skip-index",
action="store_true", action="store_true",
@@ -2290,7 +2065,7 @@ def main(argv=None, clock=time.time):
store = Store(database) store = Store(database)
if args.report_only: if args.report_only:
report(store, int(clock()), args.cold_after) report(store, int(clock()))
store.close() store.close()
lock.close() lock.close()
return 0 return 0
@@ -2334,7 +2109,7 @@ def main(argv=None, clock=time.time):
logger.error("%s", error) logger.error("%s", error)
if interval is None: if interval is None:
return 1 return 1
report(store, now, args.cold_after) report(store, now)
if interval is None or stopping: if interval is None or stopping:
return 0 return 0
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "music-curator" name = "music-curator"
version = "0.7.1" version = "0.6.0"
description = "Ingest a Last.fm listening history and curate a music library from it" description = "Ingest a Last.fm listening history and curate a music library from it"
readme = "README.md" readme = "README.md"
requires-python = ">=3.11" requires-python = ">=3.11"
+168
View File
@@ -0,0 +1,168 @@
import os
import plistlib
import sys
import unicodedata
import urllib.parse
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools"))
import find_missing_tracks # noqa: E402
def write_library(path, tracks):
"""Write a Library.xml in the shape Music exports."""
path.write_bytes(plistlib.dumps({"Tracks": {str(i): t for i, t in enumerate(tracks)}}))
return path
def track(root, name, filename=None):
"""A local track entry. Omit filename for a streaming entry with no file."""
entry = {"Name": name, "Artist": "An Artist", "Album": "An Album"}
if filename is not None:
entry["Location"] = "file://" + urllib.parse.quote(str(root / filename))
return entry
def missing_from(library, roots):
with open(library, "rb") as handle:
loaded = plistlib.load(handle)
present = find_missing_tracks.existing_under(roots)
gone = []
for entry in loaded["Tracks"].values():
path = find_missing_tracks.location_of(entry)
if path is not None and find_missing_tracks.key_for(path) not in present:
gone.append(entry["Name"])
return sorted(gone)
def test_a_deleted_file_is_reported(tmp_path):
(tmp_path / "kept.mp3").write_bytes(b"x")
library = write_library(
tmp_path / "Library.xml",
[track(tmp_path, "Kept", "kept.mp3"), track(tmp_path, "Gone", "gone.mp3")],
)
assert missing_from(library, [tmp_path]) == ["Gone"]
def test_a_decomposed_filename_is_not_reported_missing(tmp_path):
"""macOS stores filenames decomposed and most everything else composes them,
so the umlauts in Motley Crue are two different byte strings depending on
which side wrote the name."""
composed = unicodedata.normalize("NFC", "Mötley Crüe.mp3")
(tmp_path / composed).write_bytes(b"x")
library = write_library(
tmp_path / "Library.xml",
[track(tmp_path, "Umlauts", unicodedata.normalize("NFD", "Mötley Crüe.mp3"))],
)
assert missing_from(library, [tmp_path]) == []
def test_a_case_difference_is_not_reported_missing(tmp_path):
"""The share is very likely case-insensitive, and a file is not missing
because someone capitalised it differently."""
(tmp_path / "Hells Bells.mp3").write_bytes(b"x")
library = write_library(
tmp_path / "Library.xml", [track(tmp_path, "Bells", "hells bells.MP3")]
)
assert missing_from(library, [tmp_path]) == []
def test_a_track_with_no_file_is_not_a_missing_file(tmp_path):
"""A streaming entry has never had a file to lose."""
library = write_library(tmp_path / "Library.xml", [track(tmp_path, "Streamed", None)])
assert missing_from(library, [tmp_path]) == []
def test_the_walk_finds_files_nested_below_the_root(tmp_path):
nested = tmp_path / "Artist" / "Album"
nested.mkdir(parents=True)
(nested / "deep.mp3").write_bytes(b"x")
found = find_missing_tracks.existing_under([tmp_path])
assert find_missing_tracks.key_for(nested / "deep.mp3") in found
def test_the_root_is_derived_from_the_tracks(tmp_path):
paths = [tmp_path / "a" / "one.mp3", tmp_path / "b" / "two.mp3"]
assert find_missing_tracks.roots_of(paths, None) == [tmp_path]
def test_an_explicit_root_overrides_the_derived_one(tmp_path):
"""Worth using: tracks spread beyond the mirror can derive a common parent
of / and send the walk across the whole disk."""
paths = [tmp_path / "a" / "one.mp3"]
assert find_missing_tracks.roots_of(paths, ["/Volumes/music-mp3"]) == [
Path("/Volumes/music-mp3")
]
def test_the_whole_library_missing_is_called_out(tmp_path, capsys):
"""Almost always an unmounted share rather than an empty library."""
library = write_library(
tmp_path / "Library.xml", [track(tmp_path / "elsewhere", "Gone", "gone.mp3")]
)
find_missing_tracks.main([str(library), "--root", str(tmp_path)])
assert "not mounted" in capsys.readouterr().err
def test_a_healthy_library_says_nothing_alarming(tmp_path, capsys):
(tmp_path / "kept.mp3").write_bytes(b"x")
library = write_library(tmp_path / "Library.xml", [track(tmp_path, "Kept", "kept.mp3")])
find_missing_tracks.main([str(library), "--root", str(tmp_path)])
captured = capsys.readouterr()
assert "0 of 1 local tracks are missing" in captured.err
assert "not mounted" not in captured.err
assert captured.out == ""
@pytest.mark.parametrize(
("location", "expected"),
[
("file:///music/a%20b.mp3", Path("/music/a b.mp3")),
("file:///music/plain.mp3", Path("/music/plain.mp3")),
("https://example.invalid/stream", None),
("", None),
],
)
def test_locations_are_decoded(location, expected):
assert find_missing_tracks.location_of({"Location": location} if location else {}) == expected
def test_the_walk_costs_one_read_per_directory_not_one_per_file(tmp_path):
"""The whole point. Over SMB a per-file check is a round trip per track."""
for album in range(20):
directory = tmp_path / f"Album {album}"
directory.mkdir()
for index in range(25):
(directory / f"{index}.mp3").write_bytes(b"x")
reads = {"n": 0}
real_walk = os.walk
def counting_walk(*args, **kwargs):
for entry in real_walk(*args, **kwargs):
reads["n"] += 1
yield entry
find_missing_tracks.os.walk = counting_walk
try:
found = find_missing_tracks.existing_under([tmp_path])
finally:
find_missing_tracks.os.walk = real_walk
assert len(found) == 500
assert reads["n"] == 21 # the root and its twenty albums, not 500 files
+21 -215
View File
@@ -2,7 +2,6 @@ import json
import os import os
import stat import stat
import urllib.error import urllib.error
from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
import pytest import pytest
@@ -875,23 +874,23 @@ def test_playlists_are_written_into_the_mirror(tmp_path):
music_curator.build_playlists(store, mirror, str(source), 100, NOW) music_curator.build_playlists(store, mirror, str(source), 100, NOW)
written = sorted(p.name for p in (mirror / "_playlists").glob("*.m3u8")) written = sorted(p.name for p in (mirror / "_playlists").glob("*.m3u"))
assert written == [ assert written == [
"all-time.m3u8", "all-time.m3u",
"deep-cuts.m3u8", "deep-cuts.m3u",
"heavy-rotation.m3u8", "heavy-rotation.m3u",
"neglected.m3u8", "neglected.m3u",
"unheard-favourites.m3u8", "unheard-favourites.m3u",
"unheard.m3u8", "unheard.m3u",
] ]
played = (mirror / "_playlists" / "all-time.m3u8").read_text().splitlines() played = (mirror / "_playlists" / "all-time.m3u").read_text().splitlines()
assert played[0] == "#EXTM3U" assert played[0] == "#EXTM3U"
assert played[1].startswith("#EXTINF:") assert played[1].startswith("#EXTINF:")
assert "Played Band - Hit" in played[1] assert "Played Band - Hit" in played[1]
# Relative to the playlist file, so the same file works from any mount. # Relative to the playlist file, so the same file works from any mount.
assert played[2] == "../Played Band/Known/Hit.mp3" assert played[2] == "../Played Band/Known/Hit.mp3"
assert (mirror / "_playlists" / "all-time.m3u8").parent.joinpath(played[2]).resolve().is_file() assert (mirror / "_playlists" / "all-time.m3u").parent.joinpath(played[2]).resolve().is_file()
def test_an_unplayed_track_lands_in_the_unheard_playlist(tmp_path): def test_an_unplayed_track_lands_in_the_unheard_playlist(tmp_path):
@@ -905,7 +904,7 @@ def test_an_unplayed_track_lands_in_the_unheard_playlist(tmp_path):
music_curator.build_playlists(store, mirror, str(source), 100, NOW) music_curator.build_playlists(store, mirror, str(source), 100, NOW)
unheard = (mirror / "_playlists" / "unheard.m3u8").read_text() unheard = (mirror / "_playlists" / "unheard.m3u").read_text()
assert "Never Heard" in unheard assert "Never Heard" in unheard
assert "Album Track" in unheard assert "Album Track" in unheard
# The one thing that was played must not be in it. # The one thing that was played must not be in it.
@@ -925,11 +924,10 @@ def test_a_track_with_no_mirror_file_is_left_out(tmp_path):
) )
music_curator.match_library(store) music_curator.match_library(store)
total, produced = music_curator.build_playlists(store, mirror, str(source), 100, NOW) total = music_curator.build_playlists(store, mirror, str(source), 100, NOW)
assert total == 0 assert total == 0
assert produced, "the playlists are still written, they are simply empty" assert (mirror / "_playlists" / "all-time.m3u").read_text() == "#EXTM3U\n"
assert (mirror / "_playlists" / "all-time.m3u8").read_text() == "#EXTM3U\n"
def test_the_library_root_is_derived_from_the_artist_folders(tmp_path): def test_the_library_root_is_derived_from_the_artist_folders(tmp_path):
@@ -952,7 +950,7 @@ def test_the_playlist_limit_is_honoured(tmp_path):
music_curator.build_playlists(store, mirror, str(source), 1, NOW) music_curator.build_playlists(store, mirror, str(source), 1, NOW)
unheard = (mirror / "_playlists" / "unheard.m3u8").read_text().splitlines() unheard = (mirror / "_playlists" / "unheard.m3u").read_text().splitlines()
assert len([line for line in unheard if line.startswith("#EXTINF")]) == 1 assert len([line for line in unheard if line.startswith("#EXTINF")]) == 1
@@ -968,7 +966,7 @@ def test_the_rotation_moves_weekly_not_every_pass(tmp_path):
def unheard_at(when): def unheard_at(when):
music_curator.build_playlists(store, mirror, str(source), 100, when) music_curator.build_playlists(store, mirror, str(source), 100, when)
return (mirror / "_playlists" / "unheard.m3u8").read_text() return (mirror / "_playlists" / "unheard.m3u").read_text()
same_week = unheard_at(NOW), unheard_at(NOW + 3600) same_week = unheard_at(NOW), unheard_at(NOW + 3600)
assert same_week[0] == same_week[1] assert same_week[0] == same_week[1]
@@ -984,7 +982,7 @@ def test_playlists_are_group_readable(tmp_path):
music_curator.build_playlists(store, mirror, str(source), 100, NOW) music_curator.build_playlists(store, mirror, str(source), 100, NOW)
for playlist in (mirror / "_playlists").glob("*.m3u8"): for playlist in (mirror / "_playlists").glob("*.m3u"):
assert playlist.stat().st_mode & stat.S_IRGRP, playlist assert playlist.stat().st_mode & stat.S_IRGRP, playlist
@@ -1061,7 +1059,7 @@ def test_a_vibe_selects_by_tag(tmp_path):
music_curator.build_vibe_playlists(store, vibes, mirror, str(source), 100, NOW) music_curator.build_vibe_playlists(store, vibes, mirror, str(source), 100, NOW)
written = (mirror / "_playlists" / "screamo.m3u8").read_text() written = (mirror / "_playlists" / "screamo.m3u").read_text()
assert "Played Band" in written assert "Played Band" in written
assert "Silent Band" not in written assert "Silent Band" not in written
@@ -1075,7 +1073,7 @@ def test_a_weakly_tagged_artist_is_below_the_threshold(tmp_path):
music_curator.build_vibe_playlists(store, vibes, mirror, str(source), 100, NOW) music_curator.build_vibe_playlists(store, vibes, mirror, str(source), 100, NOW)
assert (mirror / "_playlists" / "screamo.m3u8").read_text() == "#EXTM3U\n" assert (mirror / "_playlists" / "screamo.m3u").read_text() == "#EXTM3U\n"
def test_a_vibe_can_be_restricted_by_release_year(tmp_path): def test_a_vibe_can_be_restricted_by_release_year(tmp_path):
@@ -1090,12 +1088,12 @@ def test_a_vibe_can_be_restricted_by_release_year(tmp_path):
# FakeLidarr dates every album 2019, so neither window should catch it. # FakeLidarr dates every album 2019, so neither window should catch it.
music_curator.build_vibe_playlists(store, inside, mirror, str(source), 100, NOW) music_curator.build_vibe_playlists(store, inside, mirror, str(source), 100, NOW)
music_curator.build_vibe_playlists(store, outside, mirror, str(source), 100, NOW) music_curator.build_vibe_playlists(store, outside, mirror, str(source), 100, NOW)
assert (mirror / "_playlists" / "eighties.m3u8").read_text() == "#EXTM3U\n" assert (mirror / "_playlists" / "eighties.m3u").read_text() == "#EXTM3U\n"
assert (mirror / "_playlists" / "nineties.m3u8").read_text() == "#EXTM3U\n" assert (mirror / "_playlists" / "nineties.m3u").read_text() == "#EXTM3U\n"
modern = [{"name": "modern", "tags": ["synthpop"], "years": [2000, 2030]}] modern = [{"name": "modern", "tags": ["synthpop"], "years": [2000, 2030]}]
music_curator.build_vibe_playlists(store, modern, mirror, str(source), 100, NOW) music_curator.build_vibe_playlists(store, modern, mirror, str(source), 100, NOW)
assert "Played Band" in (mirror / "_playlists" / "modern.m3u8").read_text() assert "Played Band" in (mirror / "_playlists" / "modern.m3u").read_text()
def test_the_built_in_vibes_are_all_usable_filenames(): def test_the_built_in_vibes_are_all_usable_filenames():
@@ -1281,7 +1279,7 @@ def test_an_excluded_tag_drops_the_artist(tmp_path):
music_curator.build_vibe_playlists(store, vibes, mirror, str(source), 100, NOW) music_curator.build_vibe_playlists(store, vibes, mirror, str(source), 100, NOW)
written = (mirror / "_playlists" / "eighties.m3u8").read_text() written = (mirror / "_playlists" / "eighties.m3u").read_text()
assert "Silent Band" in written assert "Silent Band" in written
assert "Played Band" not in written assert "Played Band" not in written
@@ -1344,195 +1342,3 @@ def test_a_playlist_is_chowned_to_match_the_mirror(tmp_path, monkeypatch):
assert all(tuple(owner) == (568, 568) for _, *owner in attempted) assert all(tuple(owner) == (568, 568) for _, *owner in attempted)
# The temporary file, before the rename, never the finished playlist. # The temporary file, before the rename, never the finished playlist.
assert all(path.endswith(".part") or path.endswith("_playlists") for path, *_ in attempted) assert all(path.endswith(".part") or path.endswith("_playlists") for path, *_ in attempted)
def test_a_renamed_mood_does_not_leave_its_old_playlist_behind(tmp_path):
"""Otherwise the dead file stays on the device for ever."""
api, source, mirror = playlist_library(tmp_path)
store = store_at(tmp_path)
directory = mirror / "_playlists"
directory.mkdir(parents=True, exist_ok=True)
(directory / "high-energy-rock.m3u").write_text("#EXTM3U\n")
(directory / "screamo.m3u8").write_text("#EXTM3U\n")
store.set_state(
"playlist_files", json.dumps(["high-energy-rock.m3u", "screamo.m3u8"])
)
music_curator.prune_playlists(directory, ["screamo.m3u8"], store)
assert not (directory / "high-energy-rock.m3u").exists()
assert (directory / "screamo.m3u8").exists()
def test_pruning_leaves_a_playlist_it_never_wrote(tmp_path):
"""A hand-made playlist in that directory is not ours to delete."""
api, source, mirror = playlist_library(tmp_path)
store = store_at(tmp_path)
directory = mirror / "_playlists"
directory.mkdir(parents=True, exist_ok=True)
(directory / "lyras-own-mix.m3u").write_text("#EXTM3U\n")
store.set_state("playlist_files", json.dumps(["screamo.m3u8"]))
music_curator.prune_playlists(directory, [], store)
assert (directory / "lyras-own-mix.m3u").exists()
def cold_store(tmp_path, played=(), added="2020-05-01T12:00:00Z"):
"""A library where nothing is played unless named, indexed and matched."""
library = [
{
"name": "Played Band",
"albums": [{"title": "Known", "tracks": [{"title": "Hit", "added": added}]}],
},
{
"name": "Silent Band",
"albums": [
{"title": "Unknown", "tracks": [{"title": "Never Heard", "added": added}]}
],
},
]
store = store_at(tmp_path)
ingest(store, FakeLastfm([scrobble_of(a, t) for a, t in played]))
music_curator.index_library(
music_curator.Lidarr("http://lidarr", "key", transport=FakeLidarr(library)), store
)
music_curator.match_library(store)
return store
def test_an_album_with_no_plays_is_cold(tmp_path):
store = cold_store(tmp_path, played=[("Played Band", "Hit")])
cold = music_curator.cold_report(store, NOW)
assert [row["artist"] for row in cold] == ["Silent Band"]
def test_an_album_with_any_play_is_not_cold(tmp_path):
"""A record with one played track is a record that gets played; picking the
rest off it leaves gaps rather than reclaiming anything."""
store = cold_store(tmp_path, played=[("Played Band", "Hit")])
cold = music_curator.cold_report(store, NOW)
assert "Played Band" not in [row["artist"] for row in cold]
def test_a_recent_arrival_is_too_young_to_judge(tmp_path):
"""It has not had a chance to be played yet."""
recent = datetime.fromtimestamp(NOW - 30 * 86400, tz=timezone.utc).isoformat()
store = cold_store(tmp_path, played=[("Played Band", "Hit")], added=recent)
assert music_curator.cold_report(store, NOW) == []
def test_the_cull_refuses_to_run_on_an_incomplete_index(tmp_path):
"""A missing artist makes their played music look unplayed, which is
precisely how this would delete something you like."""
store = cold_store(tmp_path, played=[("Played Band", "Hit")])
store.set_state("index_skipped", "1")
assert music_curator.cull_is_safe(store) is not None
assert music_curator.cold_report(store, NOW) == []
def test_the_cull_refuses_while_the_backfill_is_unfinished(tmp_path):
store = cold_store(tmp_path, played=[("Played Band", "Hit")])
store.set_state("backfill_complete", "no")
assert music_curator.cold_report(store, NOW) == []
def test_the_cold_report_writes_nothing_to_lidarr(tmp_path):
"""Stage four is read-only until the list it produces has been looked at."""
store = cold_store(tmp_path, played=[("Played Band", "Hit")])
before = store.scalar("SELECT COUNT(*) FROM lidarr_album WHERE monitored = 1")
music_curator.cold_report(store, NOW)
assert store.scalar("SELECT COUNT(*) FROM lidarr_album WHERE monitored = 1") == before
def test_human_bytes_reads_at_a_glance():
assert music_curator.human_bytes(0) == "0.0 B"
assert music_curator.human_bytes(1536) == "1.5 KiB"
assert music_curator.human_bytes(3 * 1024**3) == "3.0 GiB"
def test_playlists_are_written_as_m3u8(tmp_path):
"""Rockbox's is_m3u8_name() treats every extension as UTF-8 except an
explicit ".m3u", which it decodes through the configured codepage instead.
A library with accented names needs the other extension."""
api, source, mirror = playlist_library(tmp_path)
store = store_at(tmp_path)
music_curator.index_library(
music_curator.Lidarr("http://lidarr", "key", transport=api), store
)
music_curator.match_library(store)
music_curator.build_playlists(store, mirror, str(source), 100, NOW)
written = sorted(p.suffix for p in (mirror / "_playlists").iterdir())
assert set(written) == {".m3u8"}
def test_playlists_are_written_as_utf8(tmp_path):
playlist = tmp_path / "_playlists" / "accents.m3u8"
music_curator.write_playlist(
playlist,
[{"artist": "Mötley Crüe", "title": "Kickstart My Heart",
"duration": 283000, "mirror": tmp_path / "x.mp3"}],
)
# Decodes as UTF-8, and carries no BOM: Rockbox does not need one at this
# extension, and a BOM confuses players that do not expect it.
raw = playlist.read_bytes()
assert not raw.startswith(b"\xef\xbb\xbf")
assert "Mötley Crüe" in raw.decode("utf-8")
def test_the_old_m3u_playlists_are_pruned_after_the_rename(tmp_path):
"""Without this the seventeen dead .m3u files stay on the device for ever."""
api, source, mirror = playlist_library(tmp_path)
store = store_at(tmp_path)
directory = mirror / "_playlists"
directory.mkdir(parents=True, exist_ok=True)
(directory / "screamo.m3u").write_text("#EXTM3U\n")
store.set_state("playlist_files", json.dumps(["screamo.m3u"]))
music_curator.prune_playlists(directory, ["screamo.m3u8"], store)
assert not (directory / "screamo.m3u").exists()
def test_a_playlist_under_a_superseded_extension_is_removed(tmp_path):
"""Switching from .m3u to .m3u8 left every playlist on the device twice.
The old files predated the record of what had been written, so there was
nothing to prune them by."""
api, source, mirror = playlist_library(tmp_path)
store = store_at(tmp_path)
directory = mirror / "_playlists"
directory.mkdir(parents=True, exist_ok=True)
(directory / "screamo.m3u").write_text("#EXTM3U\n")
(directory / "screamo.m3u8").write_text("#EXTM3U\n")
# No record at all, as on the first run after the rename.
music_curator.prune_playlists(directory, ["screamo.m3u8"], store)
assert not (directory / "screamo.m3u").exists()
assert (directory / "screamo.m3u8").is_file()
def test_an_unrelated_m3u_is_still_left_alone(tmp_path):
"""Only a name this run is writing anyway is matched, so a playlist made by
hand survives whatever its extension."""
api, source, mirror = playlist_library(tmp_path)
store = store_at(tmp_path)
directory = mirror / "_playlists"
directory.mkdir(parents=True, exist_ok=True)
(directory / "lyras-own-mix.m3u").write_text("#EXTM3U\n")
music_curator.prune_playlists(directory, ["screamo.m3u8"], store)
assert (directory / "lyras-own-mix.m3u").is_file()
+115
View File
@@ -0,0 +1,115 @@
#!/usr/bin/env python3
"""Report tracks in a Music library whose files are no longer on disk.
Reads the XML from Music's File > Library > Export Library. Asking Music itself
does not work: a broken track makes AppleScript's `location` raise rather than
return a value, so a bulk query dies on the first one with error -1728, and a
per-track loop costs an Apple event apiece.
The library is checked against a single directory walk rather than by testing
each file. Over SMB a per-file test is one network round trip per track --
fifty thousand of them -- where a walk reads each directory once and gets every
name in it back at once.
"""
import argparse
import os
import plistlib
import sys
import unicodedata
import urllib.parse
from pathlib import Path
def location_of(track):
"""Return the filesystem path a track points at, or None if it has none."""
location = track.get("Location")
if not location or not location.startswith("file://"):
return None
return Path(urllib.parse.unquote(urllib.parse.urlparse(location).path))
def key_for(path):
"""Return a comparison key for a path.
Normalised to NFC because macOS stores filenames decomposed and most
everything else composes them, so "Motley Crue" with its umlauts is two
different byte strings depending on which side wrote it. Casefolded because
the share is very likely case-insensitive and a file is not missing merely
because someone capitalised it differently.
"""
return unicodedata.normalize("NFC", str(path)).casefold()
def existing_under(roots):
"""Return every file below the given roots, keyed for comparison."""
found = set()
for root in roots:
for base, _, names in os.walk(root):
for name in names:
found.add(key_for(os.path.join(base, name)))
return found
def roots_of(paths, given):
"""Return the directories worth walking."""
if given:
return [Path(root) for root in given]
try:
return [Path(os.path.commonpath([str(path) for path in paths]))]
except ValueError:
# Tracks spread across separate volumes have no common parent.
return sorted({path.parents[-2] for path in paths if len(path.parents) > 1})
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("library", help="Library.xml exported from Music")
parser.add_argument("--root", action="append", help="directory to scan; repeatable")
parser.add_argument("--limit", type=int, default=0, help="show at most this many")
args = parser.parse_args(argv)
with open(args.library, "rb") as handle:
library = plistlib.load(handle)
tracks = [
(track, location_of(track)) for track in library.get("Tracks", {}).values()
]
local = [(track, path) for track, path in tracks if path is not None]
if not local:
print("no local files in this library", file=sys.stderr)
return 0
roots = roots_of([path for _, path in local], args.root)
print(f"scanning {', '.join(str(root) for root in roots)}", file=sys.stderr)
present = existing_under(roots)
gone = [(track, path) for track, path in local if key_for(path) not in present]
for track, path in gone[: args.limit or None]:
print(
"\t".join(
(
track.get("Artist", ""),
track.get("Album", ""),
track.get("Name", ""),
str(path),
)
)
)
print(
f"\n{len(gone)} of {len(local)} local tracks are missing their file"
f" ({len(tracks) - len(local)} have no file at all)",
file=sys.stderr,
)
if gone and len(gone) == len(local):
print(
"Every single one is missing, which almost certainly means the share"
" is not mounted rather than that the library is empty.",
file=sys.stderr,
)
return 0
if __name__ == "__main__":
sys.exit(main())