#!/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): / ()/Season / SxxEyy.ext one file holding two episodes SxxE01-E02 specials Season 00 a documented part of an episode SxxE03 Part 1 sidecars ..srt bonus material /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: /_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"} # The container's own word, for a sidecar named "episode.mkv.srt": that component # is not a language flag, and both matchers read the plain form better. CONTAINER = {e[1:] for e in VIDEO_EXT} # 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*$") 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)(? 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] # The registry: every series a provider names, judged before any folder is. Folder # words may resolve a file to one of these names; they may never invent one. That is # the whole difference between "NCOP 01.mkv" inside the Gurren Lagann pack, which the # pack names, and a new root folder called "01 NCOP". registry = {} for _p, _t in titles.items(): if _t["series"]: registry.setdefault(eng.norm(_t["series"]), _t["series"]) def known_in(text): """The registered series named inside this text: the longest registered name whose words appear in it as a run. "(CBB) Tengen Toppa Gurren Lagann - The Complete Series (BD 1080p)" names "Tengen Toppa Gurren Lagann".""" hay = " " + eng.norm(text) + " " best = "" for k, name in registry.items(): if len(k) >= 5 and f" {k} " in hay and len(k) > len(best): best = name return best # Rows also name their own folders' ancestors, so an already-filed library stays # where it is when it is re-planned. for d, name in sorted(dir_series.items()): a = os.path.dirname(d) while a and a != ROOT and len(a) > len(ROOT): if a not in dir_series: got = known_in(os.path.basename(a)) if got: dir_series[a] = got a = os.path.dirname(a) def series_of_dir(d): """The series an ancestor folder belongs to, nearest first: the folder holding provider rows, else a folder whose own words name a registered show.""" while d and d != ROOT and len(d) > len(ROOT): if d in dir_series: return dir_series[d] got = known_in(os.path.basename(d)) if got: return got d = os.path.dirname(d) 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 series_name(name): """The provider's name as a folder: the pack index a scene release leads with ("4a. Seed DESTINY") falls away, and the doubled spaces it leaves fold.""" return clean(re.sub(r"\s{2,}", " ", PACK_IDX_RE.sub("", name)).strip(" .")) 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": series_name(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"]) fs = folder_season(p) if t and t["extra"]: as_extra(r, t["series"] or series_of_dir(os.path.dirname(p)), t["extra"]) continue row = bool(t and t["series"] and (t["season"] or t["episode"])) if marked or row: # The provider row names the SHOW: its registered name, its year. The number # that orders the file comes from the file's own name first, then the folder # it sits in, then the row — scraped pack rows number episodes badly, where # every file under SNL 16-20/Season 16..19 carries season 16 episode 20. owner = (t["series"] if t and t["series"] else series_of_dir(os.path.dirname(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 if marked: season, eps = marked[0], marked[1] else: season = fs if fs is not None else num(t["season"]) ep = leading_index(r["base"]) or num(t["episode"]) if not 0 < season <= 200 or ep <= 0: unknown.append((r, f"a provider row of season {season} episode {ep}")) continue eps = [ep] claimants.append((r, owner, season, eps)) note(owner, t["year"] if t else "", t["tmdb"] if t else "") continue owner = series_of_dir(os.path.dirname(p)) if fs is not None and owner and leading_index(r["base"]): claimants.append((r, owner, fs, [leading_index(r["base"])])) note(owner) continue if eng.is_extra(p, title_of(r["base"])[0], dur.get(p, 0.0), ROOT): as_extra(r, owner, "") 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 — a registered name, or nothing. Folder words alone never name a show.""" 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) 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 the NEAREST folder that holds one. A scene pack buries its subtitles two levels down (`Pack/Subs//.jpn.srt`), so the search climbs to the library root instead of stopping at the parent.""" d = os.path.dirname(path) while len(d) >= len(ROOT) and d.startswith(ROOT): 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[2]: best = (src, dst, len(prefix)) if best: return best[0], best[1] up = os.path.dirname(d) if up == d: break d = up 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):] # "Show.S01E01.mkv.srt" carries the container's name, not a flag: Jellyfin and # Plex both read "Show S01E01.srt", so the container word is dropped here. parts = tail.split(".") if len(parts) > 1 and parts[1].lower() in CONTAINER: tail = "".join("." + p for p in parts[2:]) 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()