scripts/media-census.py

@@ -1,13 +1,14 @@
 #!/usr/bin/env python3
-r"""Read-only census of a movie library — the inputs a relayout plan is built from.
+r"""Read-only census of a media library — the inputs a relayout plan is built from.
 
   media-census.py enumerate [--root DIR] [--out TSV]
   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]
 
 Run it on the machine that mounts the library. Nothing here writes inside the
-library and nothing is deleted; the only outputs are the three TSVs below, and
+library and nothing is deleted; the only outputs are the TSVs below, and the
-scripts/media-plan.py reads them instead of touching media.
+scripts/media-plan.py and scripts/media-shows.py read them instead of touching media.
 
   enum    bytes <TAB> ext <TAB> path                  every file under the root
   probed  bytes <TAB> path <TAB> dur <TAB> vcodec <TAB> WxH <TAB> fps <TAB> vbit
@@ -18,10 +19,17 @@ scripts/media-plan.py reads them instead of touching media.
   plex    file <TAB> leaf_title <TAB> … (see PLEX_COLS)
           the titles Plex already matched — the authority over any filename guess
 
+  jellyfin file <TAB> series <TAB> year <TAB> season <TAB> episode <TAB> extra
+          <TAB> tmdb      (see JF_COLS)
+          one row per episode Jellyfin has matched and numbered, its path mapped
+          from the container's view back to the host; `year` and `tmdb` are the
+          SERIES' values, `extra` is Jellyfin's own extra type when the row is
+          bonus material rather than an episode
+
 `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
-before a plan is cheap — and a plan is only ever as good as these three files.
+before a plan is cheap — and a plan is only ever as good as these files.
 """
 import argparse
 import csv
@@ -57,6 +65,14 @@ join metadata_items mi on mi.id = m.metadata_item_id
 left join metadata_items par on par.id = mi.parent_id
 where p.file like ? and p.deleted_at is null and mi.deleted_at is null
 """
+JF_DB = "/home/dank/Containers/jellyfin/config/data/jellyfin.db"
+JF_COLS = ["file", "series", "year", "season", "episode", "extra", "tmdb"]
+# Jellyfin stores the entity's .NET class as its type; the TV tree is these three.
+JF_TV = ("MediaBrowser.Controller.Entities.TV.Episode",
+         "MediaBrowser.Controller.Entities.TV.Season",
+         "MediaBrowser.Controller.Entities.TV.Series")
+JF_SQL = ("""select Id, Path, IndexNumber, ParentIndexNumber, ExtraType, IsVirtualItem
+              from BaseItems where Type = ? and Path like ?""")
 
 
 def enumerate_lib(root, out):
@@ -165,7 +181,7 @@ def plex(root, db, out):
                  "Plex runs rootless here: run this as the container's owner, never "
                  "with sudo — rootless storage is per-user, so root sees nothing.")
     with open(out, "w", newline="") as fh:
-        w = csv.writer(fh, delimiter="\t")
+        w = csv.writer(fh, delimiter="\t", lineterminator="\n")
         w.writerow(PLEX_COLS)
         n = 0
         for r in con.execute(PLEX_SQL, (root + "/%",)):
@@ -174,6 +190,79 @@ def plex(root, db, out):
     print(f"plex-matched={n} -> {out}")
 
 
+def jellyfin(root, db, container, out):
+    """The episodes Jellyfin has already numbered, for the library at root.
+
+    Jellyfin stores each path as the CONTAINER sees it (its quadlet binds the
+    share at /data), so `container` is the in-container prefix that stands for
+    root and every exported path is mapped back to the host before it is written
+    — the planner reads host paths, because it plans moves on the array.
+
+    The numbering is what matters, not the name: a release group's file name says
+    nothing about which episode a file is, and Jellyfin has already answered that
+    question against a provider. Season/episode come from the episode row; the
+    series name, its year and its TMDb id come from the Series row the episode
+    climbs to, because those are the only place Jellyfin keeps them.
+    """
+    try:
+        con = sqlite3.connect(f"file:{db}?mode=ro", uri=True)
+    except sqlite3.Error as e:
+        sys.exit(f"cannot open the Jellyfin database read-only ({db}): {e}\n"
+                 "It is a WAL database: copy it with its -wal and -shm beside it and"
+                 " point --db at the copy. Never sudo it — the data dir belongs to"
+                 " the container's owner, and rootless storage is per-user.")
+    have = {r[1] for r in con.execute("pragma table_info(BaseItems)")}
+    need = {"Id", "ParentId", "Type", "Path", "IndexNumber", "ParentIndexNumber",
+            "ExtraType", "IsVirtualItem", "ProductionYear", "Data"}
+    if not need <= have:
+        sys.exit(f"{db} is not a Jellyfin library database (BaseItems has no"
+                 f" {','.join(sorted(need - have))})")
+    node = {}
+    for i, pid, typ, name, year, data in con.execute(
+            "select Id, ParentId, Type, Name, ProductionYear, Data from BaseItems"
+            " where Type in (?,?,?)", JF_TV):
+        node[i] = (pid, typ, name, year, data)
+
+    def series_of(item_id):
+        """Climb episode -> season -> series; empty when the chain is broken."""
+        for _ in range(4):
+            if item_id not in node:
+                return "", "", ""
+            pid, typ, name, year, data = node[item_id]
+            if typ.endswith(".Series"):
+                try:
+                    tmdb = (json.loads(data or "{}")
+                            .get("ProviderIds") or {}).get("Tmdb", "")
+                except ValueError:
+                    tmdb = ""
+                return name or "", year or "", tmdb
+            item_id = pid
+        return "", "", ""
+
+    # The fleet's quadlets bind the share at /data, so the container's view of
+    # this library is /data/<the library's own folder name> unless told otherwise.
+    if not container:
+        container = "/data/" + os.path.basename(os.path.normpath(root))
+    prefix = container.rstrip("/") + "/"
+    written = skipped = 0
+    with open(out, "w", newline="") as fh:
+        w = csv.writer(fh, delimiter="\t", lineterminator="\n")
+        w.writerow(JF_COLS)
+        for iid, path, idx, pidx, extra, virt in con.execute(JF_SQL,
+                                                             (JF_TV[0], prefix + "%")):
+            if virt or not path or not path.startswith(prefix):
+                skipped += 1          # a virtual item is a variant, not a file
+                continue
+            sname, syear, tmdb = series_of(iid)
+            w.writerow([root + "/" + path[len(prefix):],
+                        sname.replace("\t", " "), syear,
+                        "" if pidx is None else pidx,
+                        "" if idx is None else idx,
+                        (extra or "").replace("\t", " "), tmdb])
+            written += 1
+    print(f"episodes={written} virtual-or-blank={skipped} -> {out}")
+
+
 def main():
     ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
     sub = ap.add_subparsers(dest="cmd", required=True)
@@ -190,13 +279,25 @@ def main():
     x.add_argument("--root", **root)
     x.add_argument("--db", default=PLEX_DB)
     x.add_argument("--out", default="/tmp/plex-movies.tsv")
+    j = sub.add_parser("jellyfin",
+                       help="export the episodes Jellyfin has numbered for the root")
+    j.add_argument("--root", default=os.environ.get("MEDIA_ROOT",
+                                                    "/mnt/entropy/TVShows"),
+                   help="TV library root on the host")
+    j.add_argument("--db", default=JF_DB)
+    j.add_argument("--container", default=None,
+                   help="the in-container prefix that stands for --root"
+                        " (default /data/<root's folder name>)")
+    j.add_argument("--out", default="/tmp/jellyfin-tv.tsv")
     a = ap.parse_args()
     if a.cmd == "enumerate":
         enumerate_lib(a.root, a.out)
     elif a.cmd == "probe":
         probe(a.enum, a.out, a.jobs)
-    else:
+    elif a.cmd == "plex":
         plex(a.root, a.db, a.out)
+    else:
+        jellyfin(a.root, a.db, a.container, a.out)
 
 
 if __name__ == "__main__":

scripts/media-shows.py

@@ -0,0 +1,656 @@
+#!/usr/bin/env python3
+r"""Plan a Jellyfin-format relayout of a TV library. DRY RUN ONLY: nothing moves.
+
+ media-shows.py [--root DIR] [--enum TSV] [--titles TSV] [--probed TSV]
+                [--tmdb-cache JSON] [--tmdb] [--movies DIR] [--park DIR]
+                [--out TSV] [--exceptions TXT]
+
+Inputs are the TSVs scripts/media-census.py exports: `enumerate` lists what
+exists, `jellyfin` says which episode each file IS, and `probe` is optional
+duration evidence for the movie rule. The TMDb cache is a plain JSON file so a
+plan stays reproducible: answers are read from it, and a miss is looked up only
+with --tmdb and TMDB_API_KEY in the environment (the key never enters argv, and
+never the cache).
+
+Target contract (jellyfin.org/docs/general/server/media/shows):
+
+    <root>/<Series> (<year>)/Season <NN>/<Series> SxxEyy.ext
+    one file holding two episodes   SxxE01-E02      specials       Season 00
+    a documented part of an episode SxxE03 Part 1   sidecars       <base>.<lang>.srt
+    bonus material                  <series>/trailers|featurettes|… — never a season
+
+PLAN is the TSV scripts/media-apply.py replays: `action src dst kind`, every
+mkdir row before any move, and `keep` for a file already where the contract puts
+it. EXCEPTIONS is the review list: everything parked, and why.
+
+The authority chain, strongest first, for one video file:
+
+  1. a provider row — Jellyfin matched this exact path and numbered it.
+  2. the file's own SxxEyy / 1x04 marker, the series taken from a sibling's
+     provider row, else from the nearest ancestor folder that has one, else from
+     the release folder's own words.
+  3. nothing. An unnumbered file is PARKED under --park with its relative path
+     kept: an invented `Season 01` steals an episode number from the file that
+     really is S01E01, and a wrong number is worse than no number.
+
+Guards, and why each is here:
+
+  * One series, one folder, one year: the year ANY provider row of a series
+    states is shared by all of them, so a box set split across five release
+    folders lands inside ONE series folder instead of five.
+  * Two files claiming one episode: the bigger keeps the standard name, the other
+    parks with the reason. Nothing is overwritten and nothing is deleted — which
+    encode survives is Frank's call, and a parked file is one rename from home.
+  * A file with no episode evidence that runs long and that the MOVIE provider
+    names leaves the TV library for --movies under the movie naming contract.
+    Long and unnamed parks; short and unnumbered is bonus material.
+  * Series art and the show nfo follow the SERIES, a season image follows its
+    SEASON, a subtitle or episode nfo follows the episode's new base — an episode
+    moved without its subtitles is a half-move.
+  * Hidden files, macOS forks, qBittorrent resume parts and the park pile itself
+    are never moved. Skipping the park is what makes re-planning stable.
+
+The plan crosses three trees (the library, its park, the movie library), so the
+executor runs with --root at the share that holds them, on its own journal —
+one journal is one undo:
+
+    python3 scripts/media-shows.py --root /mnt/entropy/TVShows \
+        --enum /tmp/tv-enum.tsv --titles /tmp/jellyfin-tv.tsv \
+        --out /mnt/entropy/media-tv-plan.tsv
+    python3 scripts/media-apply.py apply --plan /mnt/entropy/media-tv-plan.tsv \
+        --root /mnt/entropy --journal /mnt/entropy/media-tv-relayout.jsonl
+"""
+import argparse
+import collections
+import importlib.util
+import json
+import os
+import re
+import sys
+import urllib.parse
+import urllib.request
+
+ARGS = argparse.ArgumentParser(description=__doc__.splitlines()[0],
+                               formatter_class=argparse.RawDescriptionHelpFormatter)
+ARGS.add_argument("--root", default=os.environ.get("TV_ROOT", "/mnt/entropy/TVShows"),
+                  help="the TV library the plan reorganises")
+ARGS.add_argument("--enum", default="/tmp/tv-enum.tsv",
+                  help="file census (media-census.py enumerate)")
+ARGS.add_argument("--titles", default="/tmp/jellyfin-tv.tsv",
+                  help="the provider's numbering (media-census.py jellyfin)")
+ARGS.add_argument("--probed", default="",
+                  help="ffprobe census for the movie rule; absent = no durations")
+ARGS.add_argument("--tmdb-cache", default="/tmp/tmdb-cache.json",
+                  help="TMDb answers as an input file, so a plan is reproducible")
+ARGS.add_argument("--tmdb", action="store_true",
+                  help="look up cache misses (needs TMDB_API_KEY)")
+ARGS.add_argument("--movies", default="/mnt/entropy/Movies",
+                  help="movie library a mis-filed film moves into")
+ARGS.add_argument("--park", default="",
+                  help="review pile (default: <root>/_unsorted)")
+ARGS.add_argument("--out", default="/tmp/tv-plan.tsv")
+ARGS.add_argument("--exceptions", default="/tmp/tv-exceptions.txt")
+ARGS = ARGS.parse_args()
+
+ROOT = os.path.abspath(ARGS.root)
+PARK = os.path.abspath(ARGS.park or os.path.join(ROOT, "_unsorted"))
+MOVIES = os.path.abspath(ARGS.movies)
+ENGINE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "media-titles.py")
+_spec = importlib.util.spec_from_file_location("eng", ENGINE)
+eng = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(eng)
+
+try:
+    CACHE = json.load(open(ARGS.tmdb_cache))
+except (OSError, ValueError):
+    CACHE = {}
+API_KEY = os.environ.get("TMDB_API_KEY", "")
+
+VIDEO_EXT = {".mkv", ".mp4", ".avi", ".m4v", ".mov", ".ts", ".m2ts", ".mpg",
+             ".mpeg", ".webm", ".wmv", ".divx", ".ogm", ".flv", ".vob", ".rm",
+             ".rmvb", ".iso", ".img"}
+SUB_EXT = {".srt", ".ass", ".ssa", ".sub", ".vtt"}
+IMG_EXT = {".jpg", ".jpeg", ".png", ".webp", ".bmp", ".gif", ".tbn"}
+NFO_EXT = {".nfo", ".xml"}
+# Files that are never anybody's business: hidden, forks, resume state, thumbnails.
+SKIP_NAME = re.compile(r"^(\.|#|dl_state|@eaDir|Thumbs\.db|desktop\.ini)|\.!q$")
+# Show-level metadata keeps its own name at the series root; season-level at the
+# season root; folder.jpg stays with the folder it was written for.
+SERIES_META = re.compile(r"^(tvshow|series|show)(\.nfo|\.xml)?$"
+                         r"|^(poster|tvposter|banner|clearlogo|clearart|logo|art"
+                         r"|landscape|fanart|showbackdrop|backdrop)\d*\."
+                         r"(jpg|jpeg|png|webp|bmp)$", re.I)
+SEASON_META = re.compile(r"^season\.nfo$", re.I)
+FOLDER_IMAGE = re.compile(r"^folder\.(jpg|jpeg|png|webp|bmp)$", re.I)
+SEASON_DIR_RE = re.compile(r"(?i)^\s*(?:season[\s._-]*(\d{1,3})|s(\d{1,2}))\s*$")
+# Release words a folder name carries that are not part of the show's name.
+FOLDER_JUNK = re.compile(r"(?i)^(complete|collection|box|set|batch|pack|series|season"
+                         r"|multigroup|dvd|remux|uncut|extended|dual|audio|multi|dub"
+                         r"|dubbed|sub|subbed|internal|repack|proper|webrip)$")
+STOP_WORDS = {"the", "a", "an", "of", "and", "to", "in", "on", "for", "with", "de",
+              "la", "le", "il", "des", "der", "das", "y", "e", "el", "les"}
+BAD_CHARS = re.compile(r'[<>:"/|?*]')
+# SxxEyy with an optional second episode, plus the 1x04 form. The second number
+# is guarded so a resolution never reads as an episode: "S01E01-1080p" is ONE
+# episode, and the longest-match backtracking of \d{1,3} is what the (?![\dpP])
+# after it closes off.
+EP_RANGE = re.compile(r"(?i)(?<![a-z0-9])s(?:eason[\s._-]*)?(\d{1,2})[\s._-]*"
+                      r"e(?:p(?:isode)[\s._-]*)?(\d{1,3})"
+                      r"(?:[\s._-]*(?:-|to[\s._-]*|ep)[\s._-]*(?:s\d{1,2}[\s._-]*)?e?"
+                      r"(\d{1,3})(?![\dpP]))?")
+NX_RE = re.compile(r"(?i)(?<![a-z0-9])(\d{1,2})x(\d{2,3})(?![a-z0-9])")
+PART_RE = re.compile(r"(?i)(?:^|[\s._-])(?:part|pt|cd|dvd|disc|disk)[\s._-]?"
+                     r"([1-9]|i{1,3})(?:$|[\s._-])")
+# Jellyfin reads these subfolder names as bonus material. The planner emits only
+# these spellings, so 40k extras end up in nine folders instead of nine hundred.
+EXTRA_DIRS = [
+    ("trailers", r"trailer|teaser|promo|advert|commercial|tv[\s._-]spot|preview"),
+    ("deleted scenes", r"deleted"),
+    ("behind the scenes", r"behind[\s._-]the[\s._-]scenes|making[\s._-]of|bts"
+                          r"|b[\s._-]roll|gag[\s._-]reel|blooper|outtake|archival"
+                          r"|video[\s._-]archive|scene[\s._-]selection|menu|gallery"
+                          r"|stills?|visual[\s._-]effects|workprint"),
+    ("interviews", r"interview|cast[\s._-]and[\s._-]crew|audition|commentary"),
+    ("scene cuts", r"scene[\s._-]cuts?|scenecuts"),
+    ("featurettes", r"featurettes?"),
+    ("shorts", r"shorts?"),
+    ("clips", r"clips?"),
+]
+# Jellyfin's own ExtraType values, folded onto the same folder names.
+EXTRA_KIND = {"trailer": "trailers", "tmbd": "trailers", "intro": "other",
+              "movie": "other", "scene": "scene cuts", "clip": "clips",
+              "interview": "interviews", "behindthescenes": "behind the scenes",
+              "deleted": "deleted scenes", "other": "other", "epilogue": "other",
+              "opening": "other", "featurette": "featurettes", "short": "shorts"}
+SHORT_MAX = 45 * 60        # over this, a file with no episode evidence is film-shaped
+
+
+def rel(p):
+    return p[len(ROOT) + 1:] if p.startswith(ROOT + "/") else p
+
+
+def clean(name):
+    """A name the filesystem and both matchers can read."""
+    name = BAD_CHARS.sub("-", name).replace("\\", "-").strip(" .")
+    return name[:180] or "untitled"
+
+
+def title_of(text):
+    """The engine's parser, applied to a name that has dots but no extension:
+    splitext would read "The.Examples" as stem "The", so a sentinel extension is
+    added and the parser sees an ordinary media name."""
+    return eng.title_of(text + ".mkv")
+
+
+def folder_series(name):
+    """The show's own words in a release folder's name: bracketed junk, years,
+    season stamps and release vocabulary all fall away."""
+    text = re.sub(r"[\[(][^\])]*[\])]", " ", name)
+    text = re.sub(r"(?i)\b(19|20)\d\d\b", " ", text)
+    out = []
+    for w in re.split(r"[.\s_\-\[\](){}+!~,;:&@#/*'\"`´|]+", text):
+        c = eng.norm(re.sub(r"[^A-Za-z0-9']", "", w))
+        if not c or FOLDER_JUNK.match(c) or eng.JUNK_RE.match(c) or eng.RES_RE.match(c) \
+                or eng.folder_is_bonus(c):
+            break
+        if c.isdigit() and out:
+            break
+        out.append(w)
+    while out and eng.norm(out[-1]) in STOP_WORDS:
+        out.pop()
+    return clean(" ".join(out)) if out else ""
+
+
+def markers(name):
+    """(season, [episodes], position) the episode markers a name states; None when
+    it states none. The position is where the series' own words stop."""
+    stem = os.path.splitext(name)[0]
+    m = EP_RANGE.search(stem)
+    if m:
+        eps = [int(m.group(2))] + ([int(m.group(3))] if m.group(3) else [])
+        return int(m.group(1)), eps, m.start()
+    m = NX_RE.search(stem)
+    if m:
+        return int(m.group(1)), [int(m.group(2))], m.start()
+    return None
+
+
+def tmdb(kind, query):
+    """A provider answer for a query: the cache first, TMDb once, then never again."""
+    ck = f"{kind}:{query}"
+    if ck in CACHE:
+        hit = CACHE[ck]
+        # A cache an operator edited by hand can carry the year as a number; the
+        # comparison below is text, so the wire form is imposed on every hit.
+        return None if hit is None else {**hit, "year": str(hit.get("year") or "")}
+    if not (ARGS.tmdb and API_KEY):
+        return None
+    path = "movie" if kind == "movie" else "tv"
+    url = ("https://api.themoviedb.org/3/search/" + path + "?query="
+           + urllib.parse.quote(query) + "&include_adult=false")
+    req = urllib.request.Request(url, headers={
+        # The key rides a header, never the URL: a token in argv is world-readable.
+        "Authorization": "Bearer " + API_KEY,
+        "Accept": "application/json"})
+    try:
+        with urllib.request.urlopen(req, timeout=20) as r:
+            found = json.load(r).get("results") or []
+    except Exception as e:                                # noqa: BLE001
+        print(f"tmdb {kind} lookup failed for {query!r}: {e}", file=sys.stderr)
+        return None
+    if not found:
+        CACHE[ck] = None
+        return None
+    top = found[0]
+    CACHE[ck] = {"name": (top.get("title") or top.get("name") or "").strip(),
+                 "year": str(top.get("year") or "") or (top.get("first_air_date") or "")[:4],
+                 "id": top.get("id")}
+    return CACHE[ck]
+
+
+# ------------------------------------------------------------------- the inputs
+rows = []
+for line in open(ARGS.enum):
+    size, ext, path = line.rstrip("\n").split("\t")
+    p = os.path.abspath(path)
+    base = os.path.basename(p)
+    if not p.startswith(ROOT + "/") or SKIP_NAME.search(base):
+        continue
+    if p.startswith(PARK + "/"):
+        continue                    # the park is already reviewed; re-parking is churn
+    rows.append({"bytes": int(size), "ext": ext.lower(), "path": p, "base": base})
+held = {r["path"]: r for r in rows}
+
+titles = {}
+with open(ARGS.titles) as fh:
+    cols = fh.readline().rstrip("\n").split("\t")
+    for line in fh:
+        f = line.rstrip("\n").split("\t")
+        f += [""] * (len(cols) - len(f))
+        row = dict(zip(cols, f))
+        p = os.path.abspath(row.get("file", ""))
+        if p in held:
+            titles[p] = row
+
+dur = {}
+if ARGS.probed and os.path.exists(ARGS.probed):
+    for line in open(ARGS.probed):
+        f = line.rstrip("\n").split("\t")
+        if len(f) > 2:
+            try:
+                dur[os.path.abspath(f[1])] = float(f[2] or 0)
+            except ValueError:
+                pass
+
+video = [r for r in rows if r["ext"] in VIDEO_EXT]
+other = [r for r in rows if r["ext"] not in VIDEO_EXT]
+
+# What a folder already knows: one provider row names the series for every sibling
+# that has no row of its own — that is how an anime pack gets its show.
+by_dir = collections.defaultdict(list)
+for r in video:
+    by_dir[os.path.dirname(r["path"])].append(r)
+dir_series = {}
+for d, rs in by_dir.items():
+    names = collections.Counter(titles[r["path"]]["series"] for r in rs
+                                if r["path"] in titles and titles[r["path"]]["series"])
+    if names:
+        dir_series[d] = names.most_common(1)[0][0]
+
+
+# A series folder that sits above a folder of provider rows names its own show, and
+# that is what keeps an already-filed library stable when it is re-planned: extras in
+# "The Examples (2010)/trailers" belong to "The Examples", not to a series invented
+# out of the word "trailers".
+for d, name in sorted(dir_series.items()):
+    a = os.path.dirname(d)
+    while a and a != ROOT and len(a) > len(ROOT):
+        if eng.norm(folder_series(os.path.basename(a))) == eng.norm(name):
+            dir_series.setdefault(a, name)
+        a = os.path.dirname(a)
+
+
+def series_of_dir(d):
+    """The series an ancestor folder belongs to, nearest first."""
+    while d and d != ROOT and len(d) > len(ROOT):
+        if d in dir_series:
+            return dir_series[d]
+        d = os.path.dirname(d)
+    return ""
+
+
+def head_series(path):
+    """The folder evidence for a file's series: its own folder's words, then the
+    top-level release folder's."""
+    parts = rel(path).split("/")[:-1]
+    for d in reversed(parts):
+        name = folder_series(d)
+        if name:
+            return name
+    return ""
+
+
+def extras_kind(path, extra_type):
+    """The canonical bonus subfolder, from Jellyfin's own extra type first."""
+    if extra_type:
+        return EXTRA_KIND.get(eng.norm(extra_type).replace(" ", ""), "other")
+    for d in rel(path).split("/")[:-1]:
+        if eng.folder_is_bonus(d):
+            for name, rx in EXTRA_DIRS:
+                if re.search(rx, d, re.I):
+                    return name
+            return "extras"
+    stem = os.path.splitext(os.path.basename(path))[0]
+    for name, rx in EXTRA_DIRS:
+        if re.search(rx, stem, re.I):
+            return name
+    return "extras"
+
+
+# ----------------------------------------------------------- classify every file
+series = {}                       # norm(name) -> {name, year, tmdb}
+claimants = []                    # (row, name, season, [episodes])
+extras = []                       # (row, name, bonus folder)
+films = []                        # film-shaped, no episode evidence
+unknown = []                      # (row, reason)
+
+
+def note(name, year="", tmdb_id=""):
+    """Register a series; the first name wins, the first year and id stick."""
+    k = eng.norm(name)
+    s = series.setdefault(k, {"name": clean(name), "year": "", "tmdb": ""})
+    if year and not s["year"]:
+        s["year"] = str(year)
+    if tmdb_id and not s["tmdb"]:
+        s["tmdb"] = str(tmdb_id)
+
+
+
+def as_extra(r, owner, extra_type):
+    """Bonus material needs a series to file under; with none it is parked."""
+    if not owner:
+        unknown.append((r, "bonus material with no series to own it"))
+        return
+    note(owner)
+    extras.append((r, owner, extras_kind(r["path"], extra_type)))
+
+for r in video:
+    p, t = r["path"], titles.get(r["path"])
+    marked = markers(r["base"])
+    if t and t["extra"]:
+        as_extra(r, t["series"] or series_of_dir(os.path.dirname(p)) or head_series(p),
+                 t["extra"])
+        continue
+    if t and t["series"] and t["season"] and t["episode"]:
+        season, eps = int(t["season"]), [int(t["episode"])]
+        # One file can hold two episodes; the row gives the first, the name the range.
+        if marked and marked[0] == season and len(marked[1]) > 1:
+            eps = marked[1]
+        claimants.append((r, t["series"], season, eps))
+        note(t["series"], t["year"], t["tmdb"])
+        continue
+    if marked:
+        owner = (series_of_dir(os.path.dirname(p)) or head_series(p)
+                 or title_of(r["base"][:marked[2]])[0])
+        if not eng.norm(owner).replace(" ", ""):
+            unknown.append((r, "an episode marker with no series name to hang it on"))
+            continue
+        claimants.append((r, owner, marked[0], marked[1]))
+        note(owner)
+        continue
+    if eng.is_extra(p, title_of(r["base"])[0], dur.get(p, 0.0), ROOT):
+        as_extra(r, series_of_dir(os.path.dirname(p)) or head_series(p), "")
+        continue
+    if dur.get(p, 0.0) >= SHORT_MAX or (p not in dur and title_of(r["base"])[1] != "?"):
+        films.append(r)
+        continue
+    unknown.append((r, "no episode marker, no provider row, nothing long enough to be a film"))
+
+# A provider that never states a year still has one: the search is exact, so a
+# miss parks nothing — the folder simply carries no year, which Jellyfin accepts.
+for k, s in series.items():
+    if not s["year"]:
+        got = tmdb("tv", s["name"])
+        if got and eng.norm(got["name"]) == k:
+            s["year"] = got["year"]
+
+if ARGS.tmdb:
+    with open(ARGS.tmdb_cache, "w") as fh:
+        json.dump(CACHE, fh, indent=1, sort_keys=True)
+
+
+def series_dir(name):
+    s = series.get(eng.norm(name))
+    if not s:
+        s = {"name": clean(name), "year": ""}
+    label = f"{s['name']} ({s['year']})" if s["year"] else s["name"]
+    return f"{ROOT}/{clean(label)}"
+
+
+# ------------------------------------------------------------------- the plan
+moves, keeps, parked = [], [], []
+reasons = collections.defaultdict(list)
+used = {}                          # dst -> the file that claimed it
+
+
+def book(r, dst_dir, name, kind):
+    """Claim one destination for one file, or park the file when it is taken."""
+    dst = f"{dst_dir}/{name}"
+    if dst == r["path"]:
+        keeps.append((r["path"], dst, kind))
+        return
+    if dst in used:
+        park(r, "something else in the plan already files there")
+        return
+    if dst in held:
+        park(r, f"{rel(dst)} already occupies that name")
+        return
+    used[dst] = r["path"]
+    moves.append((r["path"], dst, kind))
+
+
+def park(r, reason):
+    parked.append((r, reason))
+    reasons[reason].append(r)
+    dst = f"{PARK}/{rel(r['path'])}"
+    if dst in used or dst in held:
+        dst = f"{PARK}/{rel(r['path'])}.{len(reasons)}"
+    used[dst] = r["path"]
+    moves.append((r["path"], dst, "park"))
+
+
+# Episodes first: two files claiming one number is a decision Frank makes, not one
+# the plan sneaks — the bigger copy takes the standard name, the other waits.
+by_ep = collections.defaultdict(list)
+for r, name, season, eps in claimants:
+    for e in eps:
+        by_ep[(eng.norm(name), season, e)].append(r)
+winners, losers = set(), {}
+for (key, season, e), rs in by_ep.items():
+    rs.sort(key=lambda r: (-r["bytes"], r["path"]))
+    winners.add(rs[0]["path"])
+    for r in rs[1:]:
+        losers.setdefault(r["path"],
+                          f"duplicate of {series[key]['name']} S{season:02d}E{e:02d}"
+                          " (the bigger copy keeps the standard name)")
+for r, name, season, eps in claimants:
+    if r["path"] in losers:
+        park(r, losers[r["path"]])
+        continue
+    s = series[eng.norm(name)]
+    label = f" S{season:02d}E" + "-E".join(f"{e:02d}" for e in eps)
+    part = PART_RE.search(os.path.splitext(r["base"])[0])
+    tail = f" Part {part.group(1).upper()}" if part and len(eps) == 1 else ""
+    book(r, f"{series_dir(name)}/Season {season:02d}", s["name"] + label + tail + r["ext"],
+         "tv")
+
+for r, name, kind in extras:
+    book(r, f"{series_dir(name)}/{kind}",
+         clean(os.path.splitext(r["base"])[0]) + r["ext"], "extra")
+
+# A film inside the TV library leaves for the movie library only when the movie
+# provider answers with a title that matches the name and, when the name states a
+# year, that year too.
+for r in films:
+    parsed, year = title_of(r["base"])
+    got = tmdb("movie", parsed)
+    words = eng.norm(got["name"]).split() if got else []
+    if not got or not eng.mentions(parsed.split(), words) \
+            or (year != "?" and got["year"] and got["year"] != year):
+        park(r, "long file with no episode evidence, and no movie provider names it")
+        continue
+    label = f"{got['name']} ({got['year']})" if got["year"] else got["name"]
+    book(r, f"{MOVIES}/{clean(label)}", clean(label) + r["ext"], "movie")
+
+# ------------------------------------------------- sidecars and folder metadata
+moved = {}                                          # src -> dst, for every booked file
+for src, dst, _kind in moves:
+    moved[src] = dst
+for src, dst, _kind in keeps:
+    moved[src] = dst
+dst_dirs = collections.defaultdict(collections.Counter)
+video_dst = collections.defaultdict(list)
+for src, dst in moved.items():
+    if src in {r["path"] for r in video}:
+        video_dst[os.path.dirname(src)].append((src, dst))
+    dst_dirs[os.path.dirname(src)][os.path.dirname(dst)] += 1
+
+
+def series_name_at(d):
+    """The series the files of a folder belong to, else the folder's own words."""
+    names = [titles[src]["series"] for src, _ in video_dst.get(d, [])
+             if src in titles and titles[src]["series"]]
+    if names:
+        return collections.Counter(names).most_common(1)[0][0]
+    return series_of_dir(d) or head_series(d + "/x")
+
+
+def season_at(d):
+    """The season number a folder carries: its own name, else its files' rows."""
+    m = SEASON_DIR_RE.match(os.path.basename(d))
+    if m:
+        return int(m.group(1) or m.group(2))
+    nums = [int(titles[src]["season"]) for src, _ in video_dst.get(d, [])
+            if src in titles and titles[src]["season"]]
+    return collections.Counter(nums).most_common(1)[0][0] if nums else None
+
+
+def folder_dst(d):
+    """Where a folder's own images go: the one place its files landed, or the
+    series root when they scattered across several seasons."""
+    c = dst_dirs.get(d)
+    if not c:
+        return None
+    if len(c) == 1:
+        return c.most_common(1)[0][0]
+    owner = series_name_at(d)
+    return series_dir(owner) if owner else None
+
+
+def anchor_of(path, stem):
+    """The video file this sidecar names: the longest file name it extends, in its
+    own folder first, then the parent's (a Subs/ or Extras/ folder beside them)."""
+    for d in (os.path.dirname(path), os.path.dirname(os.path.dirname(path))):
+        best = None
+        for src, dst in video_dst.get(d, []):
+            base = os.path.basename(src)
+            for prefix in (os.path.splitext(base)[0], base):
+                if stem == prefix or stem.startswith(prefix + "."):
+                    if best is None or len(prefix) > best[1]:
+                        best = (src, dst, len(prefix))
+        if best:
+            return best[0], best[1]
+    return None
+
+
+def sidecar_name(r, src, dst):
+    """The sidecar's own tail — language, flag, the media extension an nfo must
+    carry — hung on the video file's new base."""
+    stem = os.path.splitext(os.path.basename(src))[0]
+    tail = os.path.splitext(r["base"])[0][len(stem):]
+    if r["ext"] in NFO_EXT:
+        tail = os.path.splitext(os.path.basename(src))[1] + tail
+    return clean(os.path.splitext(os.path.basename(dst))[0] + tail + r["ext"])
+
+
+for r in other:
+    p, base = r["path"], r["base"]
+    stem = os.path.splitext(base)[0]
+    d = os.path.dirname(p)
+    if r["ext"] in SUB_EXT or r["ext"] in NFO_EXT or r["ext"] in IMG_EXT:
+        a = anchor_of(p, stem)
+        if a:
+            book(r, os.path.dirname(a[1]), sidecar_name(r, a[0], a[1]),
+                 "subtitle" if r["ext"] in SUB_EXT else
+                 ("nfo" if r["ext"] in NFO_EXT else "image"))
+            continue
+    if SERIES_META.match(base):
+        owner = series_name_at(d)
+        if owner:
+            book(r, series_dir(owner), base, "meta")
+            continue
+    elif SEASON_META.match(base):
+        owner, season = series_name_at(d), season_at(d)
+        if owner and season is not None:
+            book(r, f"{series_dir(owner)}/Season {season:02d}", base, "meta")
+            continue
+    elif r["ext"] in IMG_EXT or r["ext"] in NFO_EXT:
+        dest = folder_dst(d)
+        if dest:
+            book(r, dest, base, "meta")
+            continue
+    unknown.append((r, "not media, or media with nothing beside it to belong to"))
+
+for r, reason in unknown:
+    park(r, reason)
+
+
+# --------------------------------------------------------------------- output
+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 not in (ROOT, MOVIES, PARK, 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")
+    sized = sum(held[s]["bytes"] 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():
+    no_year = sorted(s["name"] for s in series.values() if not s["year"])
+    with open(ARGS.exceptions, "w") as fh:
+        fh.write(f"TV relayout exceptions for {ROOT}\n\n")
+        fh.write(f"Everything below is parked under {PARK}, keeping the relative path it\n")
+        fh.write("has today. Nothing is deleted and nothing is overwritten: a parked file\n")
+        fh.write("is one rename from wherever the plan is pointed next.\n\n")
+        for reason in sorted(reasons, key=lambda r: -len(reasons[r])):
+            rs = reasons[reason]
+            fh.write(f"{len(rs):>6} files  {reason}\n")
+            for r in sorted(rs, key=lambda r: rel(r["path"]))[:400]:
+                fh.write(f"          {rel(r['path'])}\n")
+            if len(rs) > 400:
+                fh.write(f"          … and {len(rs) - 400} more\n")
+            fh.write("\n")
+        if no_year:
+            fh.write(f"{len(no_year):>6} series carry no year from any provider, so their\n")
+            fh.write("       folders are named without one; Jellyfin still matches them.\n")
+            fh.write("       " + ", ".join(no_year[:300]) + "\n")
+    print(f"exceptions -> {ARGS.exceptions}")
+
+
+write_plan()
+write_exceptions()

tests/media_shows.bats

@@ -0,0 +1,271 @@
+#!/usr/bin/env bats
+# Seam: the TV-library relayout — scripts/media-census.py jellyfin (the metadata
+# authority export) and scripts/media-shows.py (the plan).
+#
+# The TV library is 52k files written in five different release layouts, and the
+# one thing that makes a rename of that size survivable is that the plan is
+# derived from what a metadata provider already matched, not from a guess at the
+# file name. So every test here states which authority a file had — name only,
+# provider only, both, neither — and asserts where the PLAN files it, never how
+# the planner decided. A fixture Jellyfin database stands in for the live one and
+# the TMDb cache is an input file, so no test touches the network.
+#
+# The two outcomes that cost a re-run of a 50k-file plan are a file filed as an
+# episode that is not one (it steals an episode number from its real owner) and a
+# move that lands on an existing file. Everything below is written to catch
+# exactly those, plus the gate that matters most: the plan must be replayable by
+# scripts/media-apply.py, whose preflight is the fleet's only overwrite guard.
+
+load helpers
+
+PLAN="$REPO_ROOT/scripts/media-shows.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"; }
+
+# title_row FILE SERIES YEAR SEASON EPISODE [EXTRA] — one row of the authority
+# export scripts/media-census.py jellyfin writes.
+title_row() {
+    printf '%s\t%s\t%s\t%s\t%s\t%s\t\n' "$1" "$2" "$3" "${4:-}" "${5:-}" "${6:-}" \
+        >> "$TITLES"
+}
+
+# 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"
+}
+
+setup() {
+    LIB="$BATS_TEST_TMPDIR/TVShows"
+    PARK="$LIB/_unsorted"
+    MOVIES="$BATS_TEST_TMPDIR/Movies"
+    ENUM="$BATS_TEST_TMPDIR/enum.tsv"
+    PROBED="$BATS_TEST_TMPDIR/probed.tsv"
+    TITLES="$BATS_TEST_TMPDIR/titles.tsv"
+    CACHE="$BATS_TEST_TMPDIR/tmdb.json"
+    OUT="$BATS_TEST_TMPDIR/plan.tsv"
+    EXC="$BATS_TEST_TMPDIR/exceptions.txt"
+    : > "$PROBED"
+
+    SEASON="$LIB/The Examples (2010) Season 1 S01 (1080p BluRay x265 Silence)"
+    # An episode its own name states, with a subtitle, a season image and a
+    # trailer: four things that have to end up in four different places.
+    at "$SEASON/The.Examples.S01E01.Pilot.1080p.x265.mkv" 9000
+    at "$SEASON/The.Examples.S01E01.Pilot.1080p.x265.en.srt" 200
+    at "$SEASON/folder.jpg" 300
+    at "$SEASON/Trailers/teaser.mkv" 1000
+    at "$SEASON/tvshow.nfo" 400
+    # An anime pack: no episode marker anywhere in its name, numbered by the
+    # provider only. Its year is the one thing no authority states.
+    ANIME="$LIB/Anime Pack [grp] (1080p HEVC)"
+    at "$ANIME/[grp] Anime Pack - 05.mkv" 5000
+    # Two releases claiming one episode, and a file holding two episodes.
+    at "$LIB/Dupes/Show Two S01E01.mkv" 4000
+    at "$LIB/Dupes/Show.Two.S01E01.720p.other.mkv" 7000
+    at "$LIB/Dupes/Show.Two.S01E02-E03.720p.mkv" 7000
+    # A file no authority numbers, and a long file the movie provider knows.
+    at "$LIB/Orphan Tapes/clip001.mp4" 3000
+    at "$LIB/Longform/Some Documentary Special (2015).mkv" 8000
+    # One episode-shaped file loose at the library root.
+    at "$LIB/Loose.Show.S02E03.720p.WEB-DL.mkv" 4000
+
+    # The enum census lists every file; the planner reads that instead of walking
+    # the tree, so the fixture writes the same three columns.
+    python3 - "$LIB" "$ENUM" <<'PY'
+import os, sys
+root, out = sys.argv[1], sys.argv[2]
+with open(out, "w") as fh:
+    for d, _, fs in os.walk(root):
+        for f in fs:
+            p = os.path.join(d, f)
+            fh.write(f"{os.stat(p).st_size}\t{os.path.splitext(f)[1]}\t{p}\n")
+PY
+
+    printf 'file\tseries\tyear\tseason\tepisode\textra\ttmdb\n' > "$TITLES"
+    printf '{}\n' > "$CACHE"
+    title_row "$SEASON/The.Examples.S01E01.Pilot.1080p.x265.mkv" \
+        "The Examples" 2010 1 1
+    title_row "$ANIME/[grp] Anime Pack - 05.mkv" "Anime Pack" "" 1 5
+    title_row "$LIB/Dupes/Show Two S01E01.mkv" "Show Two" 2016 1 1
+    title_row "$LIB/Dupes/Show.Two.S01E01.720p.other.mkv" "Show Two" 2016 1 1
+    title_row "$LIB/Dupes/Show.Two.S01E02-E03.720p.mkv" "Show Two" 2016 1 2
+    title_row "$LIB/Loose.Show.S02E03.720p.WEB-DL.mkv" "Loose Show" 2019 2 3
+}
+
+plan() {
+    run python3 "$PLAN" --root "$LIB" --enum "$ENUM" --titles "$TITLES" \
+        --probed "$PROBED" --tmdb-cache "$CACHE" --movies "$MOVIES" \
+        --out "$OUT" --exceptions "$EXC"
+}
+
+@test "the census exports the provider's episode numbering as a TSV" {
+    DB="$BATS_TEST_TMPDIR/jellyfin.db"
+    python3 - "$DB" "$SEASON" <<'PY'
+import os, sqlite3, sys, uuid
+db, season = sys.argv[1], sys.argv[2]
+c = sqlite3.connect(db)
+pre = "/data/TVShows/" + os.path.basename(season)
+c.execute("""create table BaseItems (Id, ParentId, Type, Name, Path, IndexNumber,
+                                      ParentIndexNumber, ProductionYear, ExtraType,
+                                      IsVirtualItem, Data)""")
+sid = str(uuid.uuid4())
+c.execute("insert into BaseItems values (?,?,?,?,?,?,?,?,?,?,?)",
+          (sid, None, "MediaBrowser.Controller.Entities.TV.Series", "The Examples",
+           pre, None, None, 2010, None, 0,
+           '{"ProviderIds":{"Tmdb":"1234"}}'))
+c.execute("insert into BaseItems values (?,?,?,?,?,?,?,?,?,?,?)",
+          (str(uuid.uuid4()), sid, "MediaBrowser.Controller.Entities.TV.Episode",
+           "Pilot", pre + "/The.Examples.S01E01.Pilot.1080p.x265.mkv", 1, 1,
+           None, None, 0, '{}'))
+c.commit()
+PY
+    run python3 "$CENSUS" jellyfin --root "$LIB" --db "$DB" \
+        --out "$BATS_TEST_TMPDIR/exp.tsv"
+    [ "$status" -eq 0 ]
+    run head -1 "$BATS_TEST_TMPDIR/exp.tsv"
+    [ "${lines[0]}" = "file	series	year	season	episode	extra	tmdb" ]
+    tail -1 "$BATS_TEST_TMPDIR/exp.tsv" \
+        | grep -q "^$SEASON/The\.Examples\.S01E01\.Pilot\.1080p\.x265\.mkv	The Examples	2010	1	1		1234$"
+}
+
+@test "an episode its own name states files as Show SxxEyy under Season NN" {
+    plan
+    [ "$status" -eq 0 ]
+    [ "$(dst "$SEASON/The.Examples.S01E01.Pilot.1080p.x265.mkv")" \
+        = "$LIB/The Examples (2010)/Season 01/The Examples S01E01.mkv" ]
+}
+
+@test "a subtitle follows its episode and keeps its language flag" {
+    plan
+    [ "$(dst "$SEASON/The.Examples.S01E01.Pilot.1080p.x265.en.srt")" \
+        = "$LIB/The Examples (2010)/Season 01/The Examples S01E01.en.srt" ]
+}
+
+@test "a season image keeps its own name; a series nfo keeps the series folder" {
+    plan
+    [ "$(dst "$SEASON/folder.jpg")" = "$LIB/The Examples (2010)/Season 01/folder.jpg" ]
+    [ "$(dst "$SEASON/tvshow.nfo")" = "$LIB/The Examples (2010)/tvshow.nfo" ]
+}
+
+@test "bonus material files as an extra, never as an episode" {
+    plan
+    [ "$(dst "$SEASON/Trailers/teaser.mkv")" = "$LIB/The Examples (2010)/trailers/teaser.mkv" ]
+    [ -z "$(awk -F '\t' '$3 ~ /Season .*teaser/ {print $3}' "$OUT")" ]
+}
+
+@test "a file only the provider numbers files at the episode it is matched to" {
+    plan
+    [ "$(dst "$ANIME/[grp] Anime Pack - 05.mkv")" \
+        = "$LIB/Anime Pack/Season 01/Anime Pack S01E05.mkv" ]
+}
+
+@test "one file holding two episodes keeps both numbers in its name" {
+    plan
+    [ "$(dst "$LIB/Dupes/Show.Two.S01E02-E03.720p.mkv")" \
+        = "$LIB/Show Two (2016)/Season 01/Show Two S01E02-E03.mkv" ]
+}
+
+@test "two files claiming one episode: one keeps the name, the other parks" {
+    plan
+    n=$(grep -c "	$LIB/Show Two (2016)/Season 01/Show Two S01E01.mkv	" "$OUT")
+    [ "$n" -eq 1 ]
+    run dst "$LIB/Dupes/Show Two S01E01.mkv"
+    [[ "$output" == "$PARK/"* ]]
+    run dst "$LIB/Dupes/Show.Two.S01E01.720p.other.mkv"
+    [[ "$output" == "$LIB/Show Two (2016)/Season 01/"* ]]
+}
+
+@test "a file no authority numbers is parked, never renamed into the library" {
+    plan
+    run dst "$LIB/Orphan Tapes/clip001.mp4"
+    [[ "$output" == "$PARK/Orphan Tapes/clip001.mp4" ]]
+    grep -q "clip001.mp4" "$EXC"
+}
+
+@test "a long file the movie provider knows leaves the TV library for the movie one" {
+    printf '8000\t%s\t7200\thevc\t1920x1080\t23.976\t4000000\tyuv420p10le\taac:eng\t0\t0\tmatroska,webm\n' \
+        "$LIB/Longform/Some Documentary Special (2015).mkv" > "$PROBED"
+    python3 -c 'import json, sys
+json.dump({"movie:some documentary":
+           {"name": "Some Documentary Special", "year": 2015, "id": 55}},
+          open(sys.argv[1], "w"))' "$CACHE"
+    plan
+    [ "$(dst "$LIB/Longform/Some Documentary Special (2015).mkv")" \
+        = "$MOVIES/Some Documentary Special (2015)/Some Documentary Special (2015).mkv" ]
+}
+
+@test "a long file no movie provider knows is parked, not moved into Movies" {
+    printf '8000\t%s\t7200\thevc\t1920x1080\t23.976\t4000000\tyuv420p10le\taac:eng\t0\t0\tmatroska,webm\n' \
+        "$LIB/Longform/Some Documentary Special (2015).mkv" > "$PROBED"
+    plan
+    run dst "$LIB/Longform/Some Documentary Special (2015).mkv"
+    [[ "$output" == "$PARK/"* ]]
+}
+
+@test "the plan creates every folder before the move that needs it" {
+    plan
+    python3 - "$OUT" "$LIB" "$MOVIES" "$PARK" <<'PY'
+import os, sys
+plan, known = sys.argv[1], set(sys.argv[2:])
+made, bad = set(), []
+for line in open(plan).read().splitlines()[1:]:
+    action, _src, dest, _kind = line.split("\t")
+    if action == "mkdir":
+        made.add(dest)
+    elif action == "move":
+        d = os.path.dirname(dest)
+        while d not in known and d != os.path.dirname(d):
+            if d not in made:
+                bad.append(f"{dest}: {d} was never made")
+            d = os.path.dirname(d)
+print("\n".join(bad[:5]))
+sys.exit(1 if bad else 0)
+PY
+}
+
+@test "no move lands outside the library or the movie library" {
+    plan
+    run awk -F '\t' -v l="$LIB" -v m="$MOVIES" \
+        '$1 == "move" && $3 !~ "^" l "/" && $3 !~ "^" m "/" {print $3}' "$OUT"
+    [ -z "$output" ]
+}
+
+@test "the executor preflights the plan and no target is claimed twice" {
+    plan
+    run python3 "$APPLY" apply --plan "$OUT" --root "$BATS_TEST_TMPDIR" \
+        --journal "$BATS_TEST_TMPDIR/journal.jsonl"
+    [ "$status" -eq 0 ]
+    dup=$(awk -F '\t' '$1 == "move" {print $3}' "$OUT" | sort | uniq -d)
+    [ -z "$dup" ]
+}
+
+@test "re-planning a library the plan already satisfies moves nothing" {
+    plan
+    # Applying the plan and re-censusing is exactly what the operator does next:
+    # every authority file is keyed by path, so all three are replayed through the
+    # plan the way a fresh census would see them. (The enum has no header; the
+    # titles export does, and its header row is not a path so it survives.)
+    python3 - "$OUT" "$ENUM" "$TITLES" "$PROBED" <<'PY'
+import os, sys
+plan = sys.argv[1]
+moves = {}
+for line in open(plan).read().splitlines()[1:]:
+    action, src, dest, _kind = line.split("\t")
+    if action == "move":
+        moves[src] = dest
+        os.makedirs(os.path.dirname(dest), exist_ok=True)
+        os.rename(src, dest)
+for path, col in ((sys.argv[2], 2), (sys.argv[3], 0), (sys.argv[4], 1)):
+    if not os.path.getsize(path):
+        continue
+    rows = [line.rstrip("\n").split("\t") for line in open(path)]
+    with open(path, "w") as fh:
+        for f in rows:
+            f[col] = moves.get(f[col], f[col])
+            fh.write("\t".join(f) + "\n")
+PY
+    plan
+    [ -z "$(awk -F '\t' '$1 == "move" {print $2}' "$OUT")" ]
+}