scripts/media-audiobooks.py

@@ -0,0 +1,379 @@
+#!/usr/bin/env python3
+r"""media-audiobooks.py — a plan that re-files the audiobook library into one folder per book.
+
+    media-audiobooks.py --root DIR --enum TSV --rows TSV [--out TSV] [--exceptions FILE]
+
+Plex reads this library and Jellyfin does not: its music scanner has no audiobook
+concept, and the tree is indexed entirely by Plex, which matched 58,975 of the
+62,879 files here into its Artist/Album/Track shape. So the authority is
+`scripts/media-census.py audiobooks` — the BOOK (album row), its NARRATOR (that
+book's single artist row) and its CHAPTERS (track rows) — and never the folder a
+release pack made. The plan is:
+
+    Audiobooks/<Narrator>/<Book> (<year>)/<the file's own name>.<ext>
+                     ... the same folder, " (2)" on the name, when two chapters
+                         of one book carry the same name
+
+What that buys is the one property both media servers' book scanners assume: a
+folder holds exactly one book. Today 212 top-level folders are a mix of
+author-named shelves and collection packs, and the books inside the packs are
+split across `… [Part 1]`, `… [Part 2]` folders three or four deep.
+
+Three laws here are the opposite of what a music shelf does, and each is forced
+by what the rows actually hold:
+
+ * A chapter is NEVER renumbered. 7,556 of the 10,485 books contain at least one
+   chapter whose index Plex does not know, so a number written into a name would
+   be invented — and it would buy nothing, because both servers order playback by
+   the embedded track tag, not by the file name. A wrong number in a name is worse
+   than none: it is the number a human later trusts. A file already carrying its
+   own number keeps it, untouched.
+ * The shelf is the NARRATOR, not the author. Every book has exactly one artist row
+   and it is the narrator; the author survives only inside 10% of the album titles
+   (`Kryptos by Graham McNeill`) and in no column Plex stores, so deriving the
+   author would mean parsing titles for one book in ten and inventing it for the
+   rest. One rule that always resolves beats a shelf that is half author and half
+   narrator; the author stays visible in the book folder's own name, which is the
+   title Plex shows.
+ * Nothing is guessed at. A chapter row with no book or no narrator parks, an audio
+   file Plex never matched parks, a video container parks, and a loose file in a
+   folder holding two books parks: nothing says which book owns it.
+
+Nothing is deleted and nothing is overwritten. The output is a list of rename(2)
+intents for `scripts/media-apply.py`, whose preflight refuses any target that is
+already claimed and whose journal is the rollback.
+"""
+import argparse
+import csv
+import os
+import re
+import sys
+from collections import Counter, defaultdict
+
+# What a chapter file can be. .m4b is the audiobook container; the rest are the
+# audio a pack ships when it was ripped as music.
+AUDIO_EXT = {".mp3", ".m4a", ".m4b", ".ogg", ".oga", ".opus", ".wav", ".flac",
+             ".wma", ".aiff", ".aif", ".m4v", ".aac", ".mka"}
+# Never an audiobook, whatever the library it sits in says: Plex reads these as
+# video, and renaming one into an audio extension would claim it is a chapter.
+VIDEO_EXT = {".mkv", ".mp4", ".webm", ".avi", ".mov", ".mpg", ".mpeg", ".ts",
+             ".m2ts", ".vob", ".ifo", ".wmv", ".iso", ".img", ".bdmv", ".ogm",
+             ".divx"}
+# The characters Plex documents as breaking playback. They are dropped, not
+# substituted: a replacement character is a guess at what the title meant.
+BAD_CHARS = set('<>:"/\\|?*')
+TRASH_DIR = re.compile(r"(^|/)(\.AGENTTRASH|\.thumbnails|\.Trash-\d+|@eaDir"
+                       r"|lost\+found)(/|$)")
+
+ARGS = None
+HELD = {}                            # path -> (bytes, ext)
+ROWS = {}                            # path -> the chapter row Plex matched
+PLACED = {}                          # chapter path -> the destination the plan gave it
+CLAIMED = set()                      # every destination path the plan has handed out
+MOVES, KEEPS = [], []
+PARKED = defaultdict(list)           # reason -> [path]
+NOTES = defaultdict(list)            # section -> [line]
+AUDIO_BY_DIR = defaultdict(set)      # source dir -> chapter files a row names
+CHILD_OF = defaultdict(set)          # dir -> the dirs directly below it holding chapters
+FOLDERS_SEEN = {}                    # (dir, depth) -> book folders placed below it
+SUBTREE_DEPTH = 2                    # how far below a folder its chapters may sit
+BOOKS = {}                           # (narrator, book, year) -> the book group
+
+AMBIGUOUS = object()
+
+
+def dominant(values):
+    """The spelling the rows mostly agree on: the most frequent value, ties broken
+    by sort so the answer never depends on the order rows arrived in."""
+    counts = Counter(values)
+    return min(counts.items(), key=lambda kv: (-kv[1], kv[0]))[0]
+
+
+def rel(path):
+    """The path as seen from the library root."""
+    return os.path.relpath(path, ARGS.root)
+
+
+def safe(name):
+    """A name every client can carry: no reserved character, nothing unprintable,
+    no trailing dot or space, and short enough to rename."""
+    out = "".join(c for c in (name or "")
+                  if c not in BAD_CHARS and (c.isprintable() or c == " "))
+    out = re.sub(r"\s+", " ", out).strip().rstrip(". ")
+    while out and len(out.encode("utf-8", "ignore")) > 200:
+        out = out[:-4].rstrip()
+    return out
+
+
+def assign(src, folder, name, ext, kind, note):
+    """Book one file into `folder` under `name`, renamed if that exact destination
+    is already taken — by another file in the plan, or by a file standing there
+    today. Two files claiming one path is the fault that costs a re-run of a 60k
+    plan, and so is a plan that only works if the renames happen in the right
+    order: the executor refuses any target that is occupied, so an occupied path
+    is never a destination, even when its occupant is scheduled to leave.
+    A file already standing where the contract puts it is a keep, not a move."""
+    stem = safe(name) or "file"
+    dst = os.path.join(folder, stem + ext)
+    if dst == src:
+        CLAIMED.add(dst.lower())          # it is standing on its own target
+        KEEPS.append((src, dst, kind))
+        return dst
+    # A file already in this folder under this name — or under the numbered
+    # variant the collision rule gives it — keeps the name it has. Renumbering it
+    # every run would turn the plan into a shuffle instead of a fixed point.
+    base = os.path.splitext(os.path.basename(src))[0]
+    if os.path.dirname(src) == folder and (base == stem
+                                           or base.startswith(stem + " (")):
+        CLAIMED.add(src.lower())
+        KEEPS.append((src, src, kind))
+        return src
+    n = 1
+    while dst.lower() in CLAIMED:
+        n += 1
+        dst = os.path.join(folder, f"{stem} ({n}){ext}")
+    if n > 1:
+        NOTES[note].append(rel(dst))
+    CLAIMED.add(dst.lower())
+    MOVES.append((src, dst, kind))
+    return dst
+
+
+def park(path, reason):
+    """Hold a file where it is, under the park folder, keeping its relative path."""
+    PARKED[reason].append(path)
+    dst = os.path.join(ARGS.park_dir, rel(path))
+    if dst != path:
+        CLAIMED.add(dst.lower())
+        MOVES.append((path, dst, "park"))
+
+
+def under_park(path):
+    """True for a file already parked: that pile is for a human to review, and
+    re-parking it every run would only push it deeper."""
+    return path.startswith(ARGS.park_dir + os.sep)
+
+
+# ------------------------------------------------------------------------ books
+def group_books():
+    """Every matched chapter into one group per (narrator, book, year).
+
+    The key is the identity Plex asserts in its own database, not the folder the
+    file sits in: one book arrives as four part folders inside a collection pack,
+    and two copies of one book in one pack are one book to Plex, whose album item
+    IS that narrator, title and year. The year is part of the key so a 1965
+    recording and a 2020 recording that share a title stay two books.
+    """
+    for path, r in sorted(ROWS.items()):
+        if path not in HELD or under_park(path) or TRASH_DIR.search(path):
+            continue
+        d = os.path.dirname(path)
+        AUDIO_BY_DIR[d].add(path)
+        CHILD_OF[os.path.dirname(d)].add(d)   # so a cover above Part 1/Part 2 sees both
+        narrator = (r["narrator"] or "").strip()
+        book = (r["book"] or "").strip()
+        if not book:
+            park(path, "a chapter row no book names — Plex has the file but no "
+                       "album item to put it in")
+            continue
+        if not narrator:
+            park(path, "a chapter row no narrator names — the book has no artist "
+                       "row, so no shelf owns it")
+            continue
+        key = (narrator.casefold(), book.casefold(), (r["year"] or "").strip())
+        g = BOOKS.setdefault(key, {"narrators": [], "books": [], "year": key[2],
+                                   "members": [], "indexed": 0})
+        g["members"].append(path)
+        g["narrators"].append(narrator)
+        g["books"].append(book)
+        if str(r["track"] or "").isdigit():
+            g["indexed"] += 1
+
+
+def book_folder(g):
+    name = safe(dominant(g["books"])) or "Untitled"
+    return os.path.join(ARGS.root, safe(dominant(g["narrators"])) or "Unknown Narrator",
+                        f"{name} ({g['year']})" if g["year"] else name)
+
+
+def place_books():
+    """One folder per book; every chapter keeps the name it has today."""
+    for key in sorted(BOOKS):
+        g = BOOKS[key]
+        folder = book_folder(g)
+        for path in sorted(g["members"]):
+            ext = HELD[path][1]
+            base = os.path.splitext(os.path.basename(path))[0]
+            # The name the ripper wrote, unchanged: the track tag inside the file
+            # is what orders playback, and 72% of the books here have chapters no
+            # row numbers, so any number added here would be invented.
+            PLACED[path] = assign(path, folder, base, ext, "chapter",
+                                  "two chapters of one book carried the same name")
+
+
+# --------------------------------------------------------------------- sidecars
+def folders_below(d, depth):
+    """The book folders the chapters in `d`, or within `depth` directories below
+    it, were placed in. Memoised, because every loose file in a big collection
+    folder asks the same question about it."""
+    if (d, depth) not in FOLDERS_SEEN:
+        seen = {os.path.dirname(PLACED[a]) for a in AUDIO_BY_DIR.get(d, ())
+                if a in PLACED}
+        if depth:
+            for child in CHILD_OF.get(d, ()):
+                seen |= folders_below(child, depth - 1)
+        FOLDERS_SEEN[(d, depth)] = seen
+    return FOLDERS_SEEN[(d, depth)]
+
+
+def owner_folder(d):
+    """The book folder that owns this directory — the first place up the tree
+    where chapters sit, in the directory itself or within SUBTREE_DEPTH levels
+    below it, which is how a pack whose parts are `… [Part 1]`/`… [Part 2]` owns
+    the cover art lying above them. None when no chapter is near enough to say,
+    AMBIGUOUS when the chapters there belong to more than one book."""
+    while d.startswith(ARGS.root + os.sep) and d != ARGS.root:
+        folders = folders_below(d, SUBTREE_DEPTH)
+        if folders:
+            return next(iter(folders)) if len(folders) == 1 else AMBIGUOUS
+        d = os.path.dirname(d)
+    return None
+
+
+def place_other():
+    """Everything that is not a chapter Plex matched: covers, booklets, EPUBs,
+    logs, checksums, playlists — and the clutter that is none of those."""
+    for path, (_size, ext) in sorted(HELD.items()):
+        if path in ROWS or under_park(path) or TRASH_DIR.search(path):
+            continue
+        d = os.path.dirname(path)
+        base = os.path.splitext(os.path.basename(path))[0]
+        if ext in VIDEO_EXT:
+            park(path, "a video container inside the audiobook library: Plex reads "
+                       "it as video, and renaming it would claim it is a chapter")
+        elif ext in AUDIO_EXT:
+            park(path, "an audio file no chapter row names — it is not in the "
+                       "library Plex matched, so nothing says which book it is")
+        else:
+            owner = owner_folder(d)
+            if owner is AMBIGUOUS:
+                park(path, "a file whose folder holds more than one book, so "
+                           "nothing says which one it belongs to")
+            elif owner:
+                assign(path, owner, base, ext,
+                       "playlist" if ext in (".m3u", ".m3u8") else "sidecar",
+                       "one book folder would have carried two files of the same"
+                       " sidecar name")
+            else:
+                park(path, "not an audiobook file, and no book's folder lies with"
+                           " it to carry it")
+
+
+# ---------------------------------------------------------------------- outputs
+def write_plan():
+    """mkdir rows for every destination folder and its whole chain, then the moves.
+    Parents first: the executor creates directories in the order they are listed."""
+    chain = set()
+    for _s, dst, _k in MOVES:
+        d = os.path.dirname(dst)
+        while d and d != ARGS.root and d != os.path.dirname(d):
+            chain.add(d)
+            d = os.path.dirname(d)
+    dirs = sorted(chain, key=lambda p: (p.count("/"), p))
+    with open(ARGS.out, "w") as fh:
+        fh.write("action\tsrc\tdst\tkind\n")
+        for d in dirs:
+            fh.write(f"mkdir\t\t{d}\tdir\n")
+        for src, dst, kind in sorted(MOVES, key=lambda m: (m[1], m[0])):
+            fh.write(f"move\t{src}\t{dst}\t{kind}\n")
+        for src, dst, kind in sorted(KEEPS, key=lambda m: (m[1], m[0])):
+            fh.write(f"keep\t{src}\t{dst}\t{kind}\n")
+    parked = [m for m in MOVES if m[2] == "park"]
+    sized = sum(HELD[s][0] for s, _d, _k in MOVES if s in HELD)
+    print(f"plan: {len(MOVES) - len(parked)} moves, {len(parked)} parked"
+          f" ({sized / 2**40:.2f} TiB touched), {len(KEEPS)} already placed,"
+          f" {len(dirs)} folders -> {ARGS.out}")
+
+
+def write_exceptions():
+    with open(ARGS.exceptions, "w") as fh:
+        fh.write(f"Audiobook relayout exceptions for {ARGS.root}\n\n")
+        fh.write(f"Everything below is parked under {ARGS.park_dir}, keeping the\n"
+                 "relative path it has today. Nothing is deleted and nothing is\n"
+                 "overwritten: a parked file is one rename from wherever the plan is\n"
+                 "pointed next.\n\n")
+        for reason in sorted(PARKED, key=lambda r: -len(PARKED[r])):
+            rs = PARKED[reason]
+            fh.write(f"{len(rs):>6} files  {reason}\n")
+            for p in sorted(rs, key=rel)[:400]:
+                fh.write(f"          {rel(p)}\n")
+            if len(rs) > 400:
+                fh.write(f"          … and {len(rs) - 400} more\n")
+            fh.write("\n")
+        for section in sorted(NOTES, key=lambda s: -len(NOTES[s])):
+            lines = sorted(set(NOTES[section]))
+            fh.write(f"{len(lines):>6}  {section}\n")
+            for line in lines[:200]:
+                fh.write(f"          {line}\n")
+            if len(lines) > 200:
+                fh.write(f"          … and {len(lines) - 200} more\n")
+            fh.write("\n")
+        unindexed = sorted(f"{safe(dominant(g['narrators']))}/{safe(dominant(g['books']))}"
+                           for g in BOOKS.values()
+                           if 0 < g["indexed"] < len(g["members"]))
+        if unindexed:
+            fh.write(f"{len(unindexed):>6} books have chapters Plex does not number. Their\n")
+            fh.write("       files keep the names they carry: playback order comes from the\n"
+                     "       track tag inside each file, and a number written here would be\n"
+                     "       invented for the chapters with no index. Fix it in the tags, not\n"
+                     "       in the names — Plex re-reads tags on a scan and never names files.\n\n")
+        fh.write("       Playlists move with their book and their contents are never\n"
+                 "       rewritten: the paths inside an .m3u go stale.\n")
+    print(f"exceptions -> {ARGS.exceptions}")
+
+
+def main():
+    global ARGS
+    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    ap.add_argument("--root", default=os.environ.get("MEDIA_AUDIOBOOKS_ROOT",
+                                                     "/mnt/entropy/Audiobooks"))
+    ap.add_argument("--enum", default="/tmp/audiobooks-enum.tsv",
+                    help="the enumerate TSV (bytes, ext, path)")
+    ap.add_argument("--rows", default="/tmp/plex-audiobooks.tsv",
+                    help="the chapter rows scripts/media-census.py audiobooks wrote")
+    ap.add_argument("--out", default="/tmp/audiobooks-plan.tsv")
+    ap.add_argument("--exceptions", default="/tmp/audiobooks-exceptions.txt")
+    ap.add_argument("--park", default="_unsorted",
+                    help="the folder inside the root that holds what cannot be filed")
+    ARGS = ap.parse_args()
+    ARGS.root = os.path.normpath(ARGS.root)
+    ARGS.park_dir = os.path.join(ARGS.root, ARGS.park)
+    read_inputs()
+    # Every path a file occupies today is claimed before a single destination is
+    # handed out: the executor refuses an occupied target, so a plan that aimed at
+    # one would only fail after the renames that were supposed to clear it.
+    CLAIMED.update(p.lower() for p in HELD)
+    group_books()
+    place_books()
+    place_other()
+    write_plan()
+    write_exceptions()
+
+
+def read_inputs():
+    with open(ARGS.enum) as fh:
+        for line in fh:
+            parts = line.rstrip("\n").split("\t")
+            if len(parts) >= 3:
+                HELD[parts[2]] = (int(parts[0]), parts[1].lower())
+    with open(ARGS.rows, newline="") as fh:
+        for row in csv.DictReader(fh, delimiter="\t"):
+            if row.get("file"):
+                ROWS[row["file"]] = row
+    gone = [p for p in ROWS if p not in HELD]
+    if gone:
+        NOTES["a chapter row whose file the census no longer sees"].extend(gone)
+
+
+if __name__ == "__main__":
+    main()

scripts/media-census.py

@@ -5,6 +5,7 @@ r"""Read-only census of a media library — the inputs a relayout plan is built
   media-census.py probe     [--enum TSV] [--out TSV] [--jobs N]
   media-census.py plex      [--root DIR] [--db PATH] [--out TSV]
   media-census.py jellyfin  [--root DIR] [--db PATH] [--container PREFIX] [--out TSV]
+  media-census.py audiobooks [--root DIR] [--db PATH] [--out TSV]
 
 Run it on the machine that mounts the library. Nothing here writes inside the
 library and nothing is deleted; the only outputs are the TSVs below, and the
@@ -34,6 +35,14 @@ scripts/media-plan.py and scripts/media-shows.py read them instead of touching m
          album item and the tags' otherwise; `mb` is the album's MusicBrainz
          release group, `performer` the track artist the tags carry
 
+  audiobooks file <TAB> narrator <TAB> book <TAB> year <TAB> track <TAB> chapter
+         (see AB_COLS)
+         one row per chapter Plex matched under the audiobook root: `book` is the
+         album row Plex built from the files' own tags, `narrator` that book's
+         artist row — Plex keeps exactly one artist per book, which is why it, and
+         not an author parsed out of a title, names the shelf. `track` is empty
+         wherever Plex has no chapter number, which is most of this library.
+
 `probe` is the slow step (every container is opened for its header, ~10 min for
 4,800 files on a spindle array at 6 workers) so it is `nice`d: a census beside a
 live service must not starve it. `enumerate` is a stat pass, so re-censusing
@@ -99,6 +108,26 @@ MUSIC_SQL = ("""select b.Path, b.AlbumArtists, b.Artists, b.Album, b.ProductionY
 # release id is one reissue of it.
 MUSIC_PROVIDERS = ("MusicBrainzReleaseGroup", "MusicBrainzAlbum")
 
+# Plex's audiobook tree is the same Artist/Album/Track shape as its music tree:
+# the leaf (10) is a CHAPTER, its parent (9) is the BOOK, and that book's artist
+# (8) is the NARRATOR. The track's own `index` is exported because the exceptions
+# report counts how many books have chapters Plex does not number — the reason the
+# relayout refuses to invent numbers into file names.
+PLEX_ARTIST, PLEX_ALBUM, PLEX_TRACK = 8, 9, 10
+AB_COLS = ["file", "narrator", "book", "year", "track", "chapter"]
+AB_SQL = f"""
+select p.file, coalesce(ar.title,''), coalesce(al.title,''), coalesce(al.year,0),
+       coalesce(mi."index",-1), coalesce(mi.title,'')
+from media_parts p
+join media_items m on m.id = p.media_item_id
+join metadata_items mi on mi.id = m.metadata_item_id
+join metadata_items al on al.id = mi.parent_id and al.metadata_type = {PLEX_ALBUM}
+left join metadata_items ar on ar.id = al.parent_id
+                          and ar.metadata_type = {PLEX_ARTIST}
+where p.file like ? and p.deleted_at is null and mi.deleted_at is null
+  and al.deleted_at is null
+"""
+
 
 def enumerate_lib(root, out):
     """One row per real file under root: size, extension, path."""
@@ -363,6 +392,49 @@ def music(root, db, container, out):
     print(f"tracks={written} outside-root-or-blank={skipped} -> {out}")
 
 
+def audiobooks(root, db, out):
+    """The chapters Plex matched inside the audiobook library, read-only.
+
+    Plex stores HOST paths here (unlike Jellyfin, which stores the container's
+    view), so no prefix mapping is needed: the part's `file` is the path the
+    operator has today. Rows for files outside `root` are skipped, because one
+    Plex library DB carries every library on the box.
+
+    The chapter's `index` is exported as an empty field when Plex has none. That
+    is not a gap to fill in here: three quarters of the books on this box contain
+    at least one un-indexed chapter, so the relayout that reads this file must
+    not renumber anything from it.
+    """
+    uri = f"file:{db}?mode=ro&immutable=1"
+    try:
+        con = sqlite3.connect(uri, uri=True)
+    except sqlite3.Error as e:
+        sys.exit(f"cannot open the Plex database read-only ({db}): {e}\n"
+                 "Plex runs rootless here: run this as the container's owner, never "
+                 "with sudo — rootless storage is per-user, so root sees nothing.")
+    tables = {r[0] for r in con.execute(
+        "select name from sqlite_master where type='table'")}
+    missing = {"media_parts", "media_items", "metadata_items"} - tables
+    if missing:
+        sys.exit(f"{db} is not a Plex library database (no "
+                 f"{', '.join(sorted(missing))})")
+    written = skipped = 0
+    with open(out, "w", newline="") as fh:
+        w = csv.writer(fh, delimiter="\t", lineterminator="\n")
+        w.writerow(AB_COLS)
+        for file, narrator, book, year, track, chapter in con.execute(
+                AB_SQL, (root.rstrip("/") + "/%",)):
+            if not file or not file.startswith(root.rstrip("/") + "/"):
+                skipped += 1              # a row for a file outside this library
+                continue
+            w.writerow([file.replace("\t", " "), narrator.replace("\t", " "),
+                        book.replace("\t", " "), "" if not year else year,
+                        "" if track is None or int(track) < 0 else track,
+                        chapter.replace("\t", " ")])
+            written += 1
+    print(f"chapters={written} outside-root={skipped} -> {out}")
+
+
 def main():
     ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
     sub = ap.add_subparsers(dest="cmd", required=True)
@@ -399,6 +471,13 @@ def main():
                    help="the in-container prefix that stands for --root"
                         " (default /data/<root's folder name>)")
     m.add_argument("--out", default="/tmp/jellyfin-music.tsv")
+    b = sub.add_parser("audiobooks",
+                       help="export the chapters Plex matched for the audiobook root")
+    b.add_argument("--root", default=os.environ.get("MEDIA_AUDIOBOOKS_ROOT",
+                                                    "/mnt/entropy/Audiobooks"),
+                   help="audiobook library root on the host")
+    b.add_argument("--db", default=PLEX_DB)
+    b.add_argument("--out", default="/tmp/plex-audiobooks.tsv")
     a = ap.parse_args()
     if a.cmd == "enumerate":
         enumerate_lib(a.root, a.out)
@@ -408,6 +487,8 @@ def main():
         plex(a.root, a.db, a.out)
     elif a.cmd == "music":
         music(a.root, a.db, a.container, a.out)
+    elif a.cmd == "audiobooks":
+        audiobooks(a.root, a.db, a.out)
     else:
         jellyfin(a.root, a.db, a.container, a.out)
 

tests/media_audiobooks.bats

@@ -0,0 +1,385 @@
+#!/usr/bin/env bats
+# Seam: the audiobook-library relayout — scripts/media-census.py audiobooks (the
+# export of the rows Plex matched) and scripts/media-audiobooks.py (the plan).
+#
+# The audiobook tree is the one library on this box that Jellyfin does not read
+# at all and Plex reads completely: every chapter here is a Plex Track row under
+# an Album row (the BOOK) under one Artist row (the NARRATOR). That is the
+# authority this seam files by — never the collection folder a release pack made.
+#
+# Two laws here are the opposite of the music tool's, and both come from what the
+# rows actually contain. Files are NEVER renumbered: 7,556 of 10,485 books carry
+# at least one chapter whose index Plex does not know, so a `00 -` invented from
+# a missing index would order a book wrong, and playback order comes from the
+# embedded track tag either way — the name buys nothing and can only lie. And the
+# folder artist is the NARRATOR even though a reader thinks in authors: every
+# book has exactly one artist row, while the author survives only in 10% of the
+# album titles (`Kryptos by Graham McNeill`) and in no column Plex stores, so one
+# rule beats a shelf that is half authors and half narrators.
+#
+# Every test states which authority a file has and asserts where the PLAN files
+# it, never how the planner decided.
+
+load helpers
+
+PLAN="$REPO_ROOT/scripts/media-audiobooks.py"
+CENSUS="$REPO_ROOT/scripts/media-census.py"
+APPLY="$REPO_ROOT/scripts/media-apply.py"
+
+# at PATH SIZE — a file with deterministic content.
+at() { mkdir -p "$(dirname "$1")"; head -c "$2" /dev/zero > "$1"; }
+
+# row FILE NARRATOR BOOK YEAR TRACK CHAPTER
+# One row of the authority export scripts/media-census.py audiobooks writes.
+row() {
+    printf '%s\t%s\t%s\t%s\t%s\t%s\n' \
+        "$1" "$2" "$3" "${4:-}" "${5:-}" "$6" >> "$ROWS"
+}
+
+# dst SRC — the destination the plan gives SRC, or nothing when it does not move.
+dst() {
+    awk -F '\t' -v s="$1" '$1 == "move" && $2 == s {print $3}' "$OUT"
+}
+
+# kind SRC — the kind the plan gave SRC's move row.
+kind() {
+    awk -F '\t' -v s="$1" '$1 == "move" && $2 == s {print $4}' "$OUT"
+}
+
+setup() {
+    AB="$BATS_TEST_TMPDIR/Audiobooks"
+    PARK="$AB/_unsorted"
+    ENUM="$BATS_TEST_TMPDIR/enum.tsv"
+    ROWS="$BATS_TEST_TMPDIR/rows.tsv"
+    OUT="$BATS_TEST_TMPDIR/plan.tsv"
+    EXC="$BATS_TEST_TMPDIR/exceptions.txt"
+    # The header is part of the authority export, so the fixture carries it.
+    printf 'file\tnarrator\tbook\tyear\ttrack\tchapter\n' > "$ROWS"
+
+    # One book of one author, split across the two part folders a pack shipped it
+    # in, with a cover above part 1 and a playlist beside the chapters.
+    A="$AB/Stephen King/It (1986)"
+    at "$A [Part 1]/01 - Chapter 1.mp3" 9000
+    at "$A [Part 1]/02 - Chapter 2.mp3" 9000
+    at "$A [Part 1]/cover.jpg" 20
+    at "$A [Part 1]/It.m3u" 30
+    at "$A [Part 2]/03 - Chapter 3.mp3" 9000
+    row "$A [Part 1]/01 - Chapter 1.mp3" "George Guidall" "It" 1986 1 "Chapter 1"
+    row "$A [Part 1]/02 - Chapter 2.mp3" "George Guidall" "It" 1986 2 "Chapter 2"
+    row "$A [Part 2]/03 - Chapter 3.mp3" "George Guidall" "It" 1986 3 "Chapter 3"
+
+    # A book whose album row carries no year at all.
+    B="$AB/Collected/Project Gold"
+    at "$B/01.ogg" 9000
+    row "$B/01.ogg" "Operator X" "Project Gold" "" 1 "Opening"
+
+    # One book title, two narrators: two different books.
+    C="$AB/Shared"
+    at "$C/Dune side a.m4b" 9000
+    at "$C/Dune side b.m4b" 9000
+    at "$C/Dune 2020.m4b" 9000
+    row "$C/Dune side a.m4b" "Scott Brick" "Dune" 1965 1 "Prologue"
+    row "$C/Dune side b.m4b" "Scott Brick" "Dune" 1965 2 "Dune"
+    row "$C/Dune 2020.m4b" "Malcolm Hillgartner" "Dune" 2020 1 "Book One"
+
+    # Two part folders of one book whose chapters share a file name.
+    D="$AB/Pulp"
+    at "$D/Book A/vol1.mp3" 9000
+    at "$D/Book A vol2/vol1.mp3" 9000
+    row "$D/Book A/vol1.mp3" "Nora Pace" "Hard Food" 2001 1 "Vol 1"
+    row "$D/Book A vol2/vol1.mp3" "Nora Pace" "Hard Food" 2001 2 "Vol 1"
+
+    # The same narrator spelled two ways across one book's chapters.
+    E="$AB/Spell"
+    at "$E/kharn 1.mp3" 9000
+    at "$E/kharn 2.mp3" 9000
+    at "$E/kharn 3.mp3" 9000
+    row "$E/kharn 1.mp3" "David Timson" "Kharn" 2013 1 "One"
+    row "$E/kharn 2.mp3" "David timson" "Kharn" 2013 2 "Two"
+    row "$E/kharn 3.mp3" "David Timson" "Kharn" 2013 3 "Three"
+
+    # A folder holding two different books: its loose note has no owner.
+    F="$AB/Mixed"
+    at "$F/book one chapter.mp3" 9000
+    at "$F/book two chapter.mp3" 9000
+    at "$F/notes.txt" 10
+    row "$F/book one chapter.mp3" "Ann Reader" "Book One" 2001 1 "One"
+    row "$F/book two chapter.mp3" "Bob Reader" "Book Two" 2002 1 "Two"
+
+    # A file already parked: the pile is for a human, so a row that would file it
+    # must not pull it back into the tree.
+    at "$PARK/Old/nobody.mp3" 9000
+    row "$PARK/Old/nobody.mp3" "Ann Reader" "Book One" 2001 1 "One"
+
+    # The odd cases: no row, a video container, a name carrying a character Plex
+    # says breaks playback, rows with a blank book and a blank narrator, and a
+    # matched chapter whose file has no extension.
+    G="$AB/Odd"
+    at "$G/loose.mp3" 9000
+    at "$G/concert.mp4" 9000
+    at "$G/Bad: Name/01 chapter.mp3" 9000
+    row "$G/Bad: Name/01 chapter.mp3" "Some Reader" "Trail: Marked" 1999 1 "Marked"
+    at "$G/blank book.mp3" 9000
+    row "$G/blank book.mp3" "Some Reader" "" 2000 1 "No Book"
+    at "$G/blank narrator.mp3" 9000
+    row "$G/blank narrator.mp3" "" "No Artist" 2000 1 "No Artist"
+    at "$G/NoExt/part01" 9000
+    row "$G/NoExt/part01" "Archivist" "Tapes" 1971 1 "Tape One"
+
+    # The enum TSV is what scripts/media-census.py enumerate writes: bytes, ext,
+    # path, no header.
+    python3 - "$AB" "$ENUM" <<'PY'
+import os, sys
+root, out = sys.argv[1], sys.argv[2]
+with open(out, "w") as fh:
+    for dirpath, _dirs, files in os.walk(root):
+        for fn in files:
+            p = os.path.join(dirpath, fn)
+            fh.write(f"{os.path.getsize(p)}\t{os.path.splitext(fn)[1].lower()}\t{p}\n")
+PY
+}
+
+plan() {
+    run python3 "$PLAN" --root "$AB" --enum "$ENUM" --rows "$ROWS" \
+        --out "$OUT" --exceptions "$EXC"
+    if [ "$status" -ne 0 ]; then
+        echo "planner said: $output"
+        return 1
+    fi
+}
+
+# db_with_part DB FILE NARRATOR BOOK YEAR TRACK CHAPTER — one Plex Track row
+# under an Album under an Artist, in the three tables the export joins.
+db_with_part() {
+    local db="$1" file="$2" narrator="$3" book="$4" year="$5" track="$6" chapter="$7"
+    local del="${8:-0}"
+    python3 - "$db" "$file" "$narrator" "$book" "$year" "$track" "$chapter" "$del" <<'PY'
+import sqlite3, sys
+db, file, narrator, book, year, track, chapter, deleted = sys.argv[1:9]
+c = sqlite3.connect(db)
+c.execute("create table if not exists media_parts (id integer primary key,"
+          " media_item_id, file, deleted_at)")
+c.execute("create table if not exists media_items (id integer primary key,"
+          " metadata_item_id, deleted_at)")
+c.execute("""create table if not exists metadata_items (id integer primary key,
+           parent_id, metadata_type, title, year, "index", deleted_at)""")
+now = None if deleted == "0" else "2026-01-01"
+ar = c.execute("insert into metadata_items (parent_id, metadata_type, title, year,"
+               " \"index\", deleted_at) VALUES (NULL, 8, ?, NULL, NULL, NULL)",
+               (narrator,)).lastrowid
+al = c.execute("insert into metadata_items (parent_id, metadata_type, title, year,"
+               " \"index\", deleted_at) VALUES (?, 9, ?, ?, NULL, NULL)",
+               (ar, book, int(year) if year else None)).lastrowid
+mi = c.execute("insert into metadata_items (parent_id, metadata_type, title, year,"
+               " \"index\", deleted_at) VALUES (?, 10, ?, NULL, ?, NULL)",
+               (al, chapter, int(track) if track else -1)).lastrowid
+it = c.execute("insert into media_items (metadata_item_id, deleted_at) VALUES (?, NULL)",
+               (mi,)).lastrowid
+c.execute("insert into media_parts (media_item_id, file, deleted_at) VALUES (?,?,?)",
+          (it, file, now))
+c.commit()
+PY
+}
+
+# ------------------------------------------------------------------ the export
+
+@test "the census exports the narrator, book and chapter Plex matched for a part" {
+    DB="$BATS_TEST_TMPDIR/plex.db"
+    db_with_part "$DB" "$BATS_TEST_TMPDIR/Audiobooks/Pack/It part 1/01.mp3" \
+        "George Guidall" "It" 1986 1 "Chapter 1"
+    run python3 "$CENSUS" audiobooks --root "$BATS_TEST_TMPDIR/Audiobooks" --db "$DB" \
+        --out "$BATS_TEST_TMPDIR/exp.tsv"
+    [ "$status" -eq 0 ]
+    run head -1 "$BATS_TEST_TMPDIR/exp.tsv"
+    [ "${lines[0]}" = "file	narrator	book	year	track	chapter" ]
+    tail -1 "$BATS_TEST_TMPDIR/exp.tsv" \
+        | grep -q "^$BATS_TEST_TMPDIR/Audiobooks/Pack/It part 1/01\.mp3	George Guidall	It	1986	1	Chapter 1$"
+}
+
+@test "the census exports nothing outside the root, and nothing Plex deleted" {
+    DB="$BATS_TEST_TMPDIR/plex.db"
+    db_with_part "$DB" "/mnt/other/Library/01.mp3" "Someone" "Elsewhere" 1990 1 "One"
+    db_with_part "$DB" "$BATS_TEST_TMPDIR/Audiobooks/Gone/01.mp3" "Someone" \
+        "Deleted" 1990 1 "One" 1
+    run python3 "$CENSUS" audiobooks --root "$BATS_TEST_TMPDIR/Audiobooks" --db "$DB" \
+        --out "$BATS_TEST_TMPDIR/exp.tsv"
+    [ "$status" -eq 0 ]
+    [ "$(wc -l < "$BATS_TEST_TMPDIR/exp.tsv")" -eq 1 ]   # header only
+}
+
+# -------------------------------------------------------------------- the plan
+
+@test "a matched chapter files under its narrator and its book, name untouched" {
+    plan
+    [ "$(dst "$A [Part 1]/01 - Chapter 1.mp3")" = \
+        "$AB/George Guidall/It (1986)/01 - Chapter 1.mp3" ]
+}
+
+@test "one book arrives in one folder however many part folders it shipped in" {
+    plan
+    [ "$(dst "$A [Part 2]/03 - Chapter 3.mp3")" = \
+        "$AB/George Guidall/It (1986)/03 - Chapter 3.mp3" ]
+    [ "$(dst "$A [Part 1]/02 - Chapter 2.mp3")" = \
+        "$AB/George Guidall/It (1986)/02 - Chapter 2.mp3" ]
+}
+
+@test "a chapter is never renumbered, whether or not Plex knows its index" {
+    # The law this guards: playback order comes from the embedded track tag, and
+    # three quarters of the books have chapters with no index at all, so a
+    # number written into a name would be invented and would lie.
+    plan
+    [ "$(dst "$B/01.ogg")" = "$AB/Operator X/Project Gold/01.ogg" ]
+    [ "$(dst "$A [Part 1]/01 - Chapter 1.mp3")" = \
+        "$AB/George Guidall/It (1986)/01 - Chapter 1.mp3" ]
+}
+
+@test "a book whose album row has no year gets no parentheses" {
+    plan
+    [ "$(dst "$B/01.ogg")" = "$AB/Operator X/Project Gold/01.ogg" ]
+}
+
+@test "two chapters claiming one name inside one book are named apart" {
+    plan
+    first="$(dst "$D/Book A/vol1.mp3")"
+    second="$(dst "$D/Book A vol2/vol1.mp3")"
+    [ "$first" != "$second" ]
+    [ -n "$first" ] && [ -n "$second" ]
+    # and the plan still passes the executor's preflight, which refuses a target
+    # two sources claim.
+    run python3 "$APPLY" apply --plan "$OUT" --root "$AB"
+    [ "$status" -eq 0 ]
+    [[ "$output" != *"double-claim"* ]]
+}
+
+@test "a cover follows the book whose folder it lies in" {
+    plan
+    [ "$(dst "$A [Part 1]/cover.jpg")" = "$AB/George Guidall/It (1986)/cover.jpg" ]
+}
+
+@test "a playlist moves with its book and its bytes are never rewritten" {
+    plan
+    [ "$(dst "$A [Part 1]/It.m3u")" = "$AB/George Guidall/It (1986)/It.m3u" ]
+    before=$(md5sum "$A [Part 1]/It.m3u" | cut -d" " -f1)
+    run python3 "$APPLY" apply --plan "$OUT" --root "$AB" --journal "$BATS_TEST_TMPDIR/j.jsonl" --apply
+    [ "$status" -eq 0 ]
+    [ "$(md5sum "$AB/George Guidall/It (1986)/It.m3u" | cut -d" " -f1)" = "$before" ]
+    # the stored paths inside it still name the old file names, which the tool
+    # leaves alone rather than editing a file the operator owns.
+    [ "$(kind "$A [Part 1]/It.m3u")" != "chapter" ]
+}
+
+@test "a note in a folder holding two books has no owner and parks" {
+    plan
+    [ "$(kind "$F/notes.txt")" = "park" ]
+    [ "$(dst "$F/notes.txt")" = "$PARK/Mixed/notes.txt" ]
+}
+
+@test "an audio file no row names parks with the path it has today" {
+    plan
+    [ "$(kind "$G/loose.mp3")" = "park" ]
+    [ "$(dst "$G/loose.mp3")" = "$PARK/Odd/loose.mp3" ]
+}
+
+@test "a video container parks instead of being renamed into audio" {
+    plan
+    [ "$(kind "$G/concert.mp4")" = "park" ]
+    run grep -c "concert\.mp4" "$EXC"
+    [ "$output" != "0" ]
+}
+
+@test "a character Plex says breaks playback never reaches a destination" {
+    plan
+    [ "$(dst "$G/Bad: Name/01 chapter.mp3")" = \
+        "$AB/Some Reader/Trail Marked (1999)/01 chapter.mp3" ]
+    run cut -f3 "$OUT"
+    [[ "$output" != *":"* ]]
+}
+
+@test "one book title with two narrators stays two books" {
+    plan
+    [ "$(dst "$C/Dune side a.m4b")" = "$AB/Scott Brick/Dune (1965)/Dune side a.m4b" ]
+    [ "$(dst "$C/Dune 2020.m4b")" = \
+        "$AB/Malcolm Hillgartner/Dune (2020)/Dune 2020.m4b" ]
+}
+
+@test "the spelling most chapters carry names the folder, and they share one folder" {
+    plan
+    [ "$(dst "$E/kharn 1.mp3")" = "$AB/David Timson/Kharn (2013)/kharn 1.mp3" ]
+    [ "$(dst "$E/kharn 2.mp3")" = "$AB/David Timson/Kharn (2013)/kharn 2.mp3" ]
+    [ "$(dst "$E/kharn 3.mp3")" = "$AB/David Timson/Kharn (2013)/kharn 3.mp3" ]
+}
+
+@test "a row with no book title parks rather than inventing a folder" {
+    plan
+    [ "$(kind "$G/blank book.mp3")" = "park" ]
+}
+
+@test "a row with no narrator parks rather than inventing a shelf" {
+    plan
+    [ "$(kind "$G/blank narrator.mp3")" = "park" ]
+}
+
+@test "the row, not the extension, decides what a chapter is" {
+    plan
+    [ "$(kind "$G/NoExt/part01")" = "chapter" ]
+    [ "$(dst "$G/NoExt/part01")" = "$AB/Archivist/Tapes (1971)/part01" ]
+}
+
+@test "the plan replays through the executor with nothing refused" {
+    plan
+    run python3 "$APPLY" apply --plan "$OUT" --root "$AB" \
+        --journal "$BATS_TEST_TMPDIR/journal.jsonl"
+    [ "$status" -eq 0 ]
+    [[ "$output" != *"refusing"* ]]
+    # The refusal this guards is two sources claiming one target: the executor
+    # stops the whole run on it, so the plan must never hand a path out twice.
+    [ -z "$(awk -F '\t' '$1 == "move" {print $3}' "$OUT" | sort | uniq -d)" ]
+}
+
+@test "an occupied destination is refused rather than overwritten" {
+    plan
+    mkdir -p "$AB/George Guidall/It (1986)"
+    at "$AB/George Guidall/It (1986)/01 - Chapter 1.mp3" 1
+    run python3 "$APPLY" apply --plan "$OUT" --root "$AB"
+    [ "$status" -ne 0 ]
+    [[ "$output" == *"occupied"* ]]
+    [ -e "$A [Part 1]/01 - Chapter 1.mp3" ]
+}
+
+@test "applying the plan and re-planning moves nothing" {
+    plan
+    run python3 "$APPLY" apply --plan "$OUT" --root "$AB" \
+        --journal "$BATS_TEST_TMPDIR/j.jsonl" --apply
+    [ "$status" -eq 0 ]
+    python3 - "$AB" "$ENUM" <<'PY'
+import os, sys
+root, out = sys.argv[1], sys.argv[2]
+with open(out, "w") as fh:
+    for dirpath, _dirs, files in os.walk(root):
+        for fn in files:
+            p = os.path.join(dirpath, fn)
+            fh.write(f"{os.path.getsize(p)}\t{os.path.splitext(fn)[1].lower()}\t{p}\n")
+PY
+    # The rows still name the paths the chapters were placed at: that is what a
+    # server re-scan of the moved tree reports.
+    python3 - "$OUT" "$ROWS" <<'PY'
+import csv, sys
+out, rows = sys.argv[1], sys.argv[2]
+moved = {r[1]: r[2] for r in csv.reader(open(out), delimiter="\t") if r[0] == "move"}
+with open(rows) as fh, open(rows + ".2", "w") as fh2:
+    fh2.write(fh.readline().rstrip("\n") + "\n")
+    for line in fh:
+        f = line.split("\t")[0]
+        fh2.write("\t".join([moved.get(f, f)] + line.rstrip("\n").split("\t")[1:]) + "\n")
+PY
+    run python3 "$PLAN" --root "$AB" --enum "$ENUM" --rows "$ROWS.2" \
+        --out "$OUT.2" --exceptions "$EXC.2"
+    [ "$status" -eq 0 ]
+    [ "$(awk -F '\t' '$1 == "move"' "$OUT.2" | wc -l)" -eq 0 ]
+}
+
+@test "nothing under the park root is planned again" {
+    plan
+    run awk -F '\t' '$1 == "move" && $2 ~ /_unsorted/' "$OUT"
+    [ -z "$output" ]
+}