.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 The series name and the episode title are the ones the IDENTITY authority — tvdb — states for the id the metadata server matched, not the words a release folder or a server's folder-derived Series name happens to carry. 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. The row's NAME is only evidence once an id backs it: the server names a Series after the folder it scanned, so `24 (2001) Season 1` is a folder echo, and the series' tvdb id is what undoes 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 a registered show its own words name, else — outside the review pile — from the release folder's words. Inside a hidden top-level folder a file may MATCH a registered show but never INVENT one: invention is what filled the pile to begin with. 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 are never moved. A file already inside --park is never moved either: that is what makes re-planning stable. --park belongs OUTSIDE the library — parked files inside it get indexed as one show named after the park folder, and a row like that is a folder echo, never evidence (the guard in `row_names_the_park`). 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 \ --tvdb --tmdb --out /mnt/entropy/media-tv-plan.tsv \ --park /mnt/entropy/media-park/TVShows 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/tv-names.json", help="name answers (TMDb and Tvdb) as ONE input file, so a plan " "is reproducible; keys carry the provider they came from") ARGS.add_argument("--tmdb", action="store_true", help="look up cache misses at TMDb (needs TMDB_API_KEY)") ARGS.add_argument("--tvdb", action="store_true", help="look up cache misses at Tvdb (needs the credentials file)") ARGS.add_argument("--tvdb-credentials", default=os.path.expanduser("~/.config/secrets/tvdb"), help="file with an api_key= line (and a pin= line for a " "user-issued key); mode 600, and never the cache") 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 OUTSIDE the library (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", "") TMDB = {"tries": 0, "failed": 0} 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)(?<![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 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 LEAD_IDX_RE = re.compile(r"^\s*(\d{1,3})\s*[-\u2013.\s]\s*[A-Za-z]") PACK_IDX_RE = re.compile(r"^\s*\d+[a-z]?[.)]\s+") def num(x): try: return int(x) except (TypeError, ValueError): return 0 def folder_season(path): """The season the FOLDER states (`Season 3`, `S3`). A human wrote it, and where a scraped row disagrees this is the one that survives a re-plan.""" m = SEASON_DIR_RE.match(os.path.basename(os.path.dirname(path))) return int(next(g for g in m.groups() if g)) if m else None def leading_index(name): """The number a file leads with inside a season folder — `01 Kyle MacLachlan, Sinead OC.avi` is episode 1. Both matchers read it that way.""" m = LEAD_IDX_RE.match(os.path.splitext(name)[0]) return int(m.group(1)) if m else None def tmdb_request(url): """One TMDb call, authenticated. The two credential shapes are NOT interchangeable and the difference is a 401 on every single call: a v3 key travels as the `api_key` query parameter, and only a v4 request token is accepted as a Bearer header. A token in argv is world-readable in /proc/<pid>/cmdline — this URL is built inside the process and never reaches argv, so the query form is safe here.""" TMDB["tries"] += 1 if API_KEY.count(".") == 2: # three segments: a v4 JWT return http_json(url, headers={"Authorization": "Bearer " + API_KEY}) return http_json(url + ("&" if "?" in url else "?") + "api_key=" + urllib.parse.quote(API_KEY)) 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") try: found = tmdb_request(url).get("results") or [] except Exception as e: # noqa: BLE001 TMDB["failed"] += 1 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] # ---------------------------------------------------------------- tvdb, the identity # tvdb names the show and the episode: it is the only provider whose strings are # the ones the operator asked for, so its answer replaces a folder-derived Series # name. The login mints a JWT per run from a mode-600 file; neither the key nor # the token reaches argv, the cache, or the plan. A provider that will not answer # never invents a name: every lookup here returns None, and None keeps the # provider's own words and lands in the exceptions. TVDB = {"jwt": None, "dead": False} def http_json(url, body=None, headers=None): req = urllib.request.Request( url, data=json.dumps(body).encode() if body is not None else None, headers={"Accept": "application/json", "Content-Type": "application/json", **(headers or {})}) with urllib.request.urlopen(req, timeout=30) as r: return json.load(r) def tvdb_login(): """The run's JWT, or None: a project key needs only api_key=, a user-issued key also needs the pin.""" if TVDB["jwt"] is not None: return TVDB["jwt"] or None TVDB["jwt"] = "" try: lines = open(ARGS.tvdb_credentials).read().splitlines() except OSError: print(f"tvdb: no credentials at {ARGS.tvdb_credentials}", file=sys.stderr) return None sec = {} for line in lines: line = line.strip() if line and not line.startswith("#") and "=" in line: k, v = line.split("=", 1) sec[k.strip()] = v.strip() if not sec.get("api_key"): print(f"tvdb: no api_key in {ARGS.tvdb_credentials}", file=sys.stderr) return None body = {"apiKey": sec["api_key"]} if sec.get("pin"): body["token"] = sec["pin"] try: TVDB["jwt"] = (http_json("https://api4.thetvdb.com/v4/login", body) .get("data") or {}).get("token") or "" except Exception as e: # noqa: BLE001 print(f"tvdb login failed: {e}", file=sys.stderr) return None return TVDB["jwt"] or None def tvdb(path): """One Tvdb v4 GET, or None when it cannot be answered.""" if not ARGS.tvdb or TVDB["dead"]: return None jwt = tvdb_login() if not jwt: TVDB["dead"] = True return None try: return http_json("https://api4.thetvdb.com/v4" + path, headers={"Authorization": "Bearer " + jwt}).get("data") except Exception as e: # noqa: BLE001 print(f"tvdb {path} failed: {e}", file=sys.stderr) return None def tvdb_series(sid): """{name, year} for a tvdb series id.""" ck = f"tvdb:series:{sid}" if ck in CACHE: return CACHE[ck] data = tvdb(f"/series/{sid}") if not data: return None CACHE[ck] = {"name": (data.get("name") or "").strip(), "year": (data.get("firstAired") or "")[:4]} return CACHE[ck] def tvdb_id_for_tmdb(mid): """The tvdb id TMDb records for a series id — the bridge for the bulk of this library, which the server matched by TMDb.""" ck = f"tmdb:external:{mid}" if ck in CACHE: return CACHE[ck] or "" if not (ARGS.tmdb and API_KEY): return "" try: got = tmdb_request("https://api.themoviedb.org/3/tv/" + mid + "/external_ids") except Exception as e: # noqa: BLE001 TMDB["failed"] += 1 print(f"tmdb external ids failed for {mid}: {e}", file=sys.stderr) return "" CACHE[ck] = str(got.get("tvdb_id") or "") return CACHE[ck] def tvdb_search(name): """The tvdb series a provider name states, or None: only tvdb's own spelling counts, so a near-miss never renames a show.""" ck = f"tvdb:search:{name}" if ck in CACHE: return CACHE[ck] for hit in tvdb("/search?query=" + urllib.parse.quote(name) + "&type=series") or []: if eng.norm(hit.get("name") or "") != eng.norm(name): continue CACHE[ck] = {"name": (hit.get("name") or "").strip(), "year": str(hit.get("year") or ""), "id": str(hit.get("tvdb_id") or "")} return CACHE[ck] CACHE[ck] = None return None def tvdb_episodes(sid): """`season/episode` -> title for one series, in the one call tvdb answers.""" ck = f"tvdb:episodes:{sid}" if ck in CACHE: return CACHE[ck] data = tvdb(f"/series/{sid}/episodes/default") if not data: return None CACHE[ck] = {f"{e.get('seasonNumber')}/{e.get('number')}": (e.get("name") or "").strip() for e in data.get("episodes") or []} 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} def row_names_the_park(path, name): """The library's own review pile is not a show. When the park sat INSIDE the library the server indexed the whole folder as one series named after it, and a row like that would rename a real show to `_unsorted` — so a row whose series is the hidden top-level folder the file itself sits in is dropped.""" top = rel(path).split("/")[0] return top.startswith("_") and eng.norm(name) == eng.norm(top) def in_review_pile(path): """A file under the library's hidden top-level folder is review material by that folder's own convention: its filename may MATCH a show a provider names somewhere else, but it may never INVENT one — inventing one from a file's own words is what filled the pile to begin with.""" return rel(path).split("/")[0].startswith("_") 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 and not row_names_the_park(p, row.get("series", "")): 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] # 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="", tvdb_id=""): """Register a series; the first name wins, the first year and ids stick.""" k = eng.norm(name) s = series.setdefault(k, {"name": series_name(name), "year": "", "tmdb": "", "tvdb": ""}) if year and not s["year"]: s["year"] = str(year) if tmdb_id and not s["tmdb"]: s["tmdb"] = str(tmdb_id) if tvdb_id and not s["tvdb"]: s["tvdb"] = str(tvdb_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 known_in(r["base"]) or ("" if in_review_pile(p) else 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 "", t["tvdb"] 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"] # tvdb is the identity authority. The row's own tvdb id is the door; a TMDb id # crosses through TMDb's external ids; a bare name counts only when tvdb spells # it the same and the years do not disagree. A series nothing resolves keeps the # provider's words AND says so in the exceptions, which is where the operator # finds the folders the plan did not rename to authority. for s in series.values(): got = tvdb_series(s["tvdb"]) if s["tvdb"] else None if not got and s["tmdb"]: sid = tvdb_id_for_tmdb(s["tmdb"]) if sid: s["tvdb"] = sid got = tvdb_series(sid) if not got: hit = tvdb_search(s["name"]) if hit and hit["id"] and (not s["year"] or not hit["year"] or hit["year"] == s["year"]): s["tvdb"] = hit["id"] got = hit if got and got["name"]: s["canon"], s["canon_year"] = got["name"], got["year"] # One identity, one folder: the provider's spellings collapse to the shows tvdb # says they are, so a box set split over five release folders is ONE series and # two files claiming one episode MEET inside the duplicate rule below. alias, canon = {}, {} for k, s in series.items(): key = f"id:{s['tvdb']}" if s["tvdb"] else k alias[k] = key keep = canon.setdefault(key, s) if keep is not s: for f in ("year", "tvdb", "canon", "canon_year"): if not keep.get(f) and s.get(f): keep[f] = s[f] if ARGS.tmdb or ARGS.tvdb: with open(ARGS.tmdb_cache, "w") as fh: json.dump(CACHE, fh, indent=1, sort_keys=True) def identity(s): """(name, year) a folder and its files carry: tvdb's whenever it answered, the provider's own words only for a series nothing resolved.""" if s.get("canon"): return s["canon"], s["canon_year"] or s["year"] return s["name"], s["year"] def series_key(name): """The merged identity a provider spelling belongs to.""" return alias.get(eng.norm(name), eng.norm(name)) def label_of(s): """`<Series> (<year>)`, the form both matchers read.""" nm, yr = identity(s) nm = clean(nm) return f"{nm} ({yr})" if yr else nm def episode_title(r, s, season, ep): """The title this episode carries: tvdb's for the id the series resolved to, else the title the matched row states, else nothing at all — a dangling ` - ` is not a title, and an invented one is worse than none.""" if s["tvdb"]: t = ((tvdb_episodes(s["tvdb"]) or {}).get(f"{season}/{ep}") or "").strip() if t: return clean(t) t = ((titles.get(r["path"]) or {}).get("title") or "").strip() return clean(t) if t else "" def series_dir(name): s = canon.get(series_key(name)) return f"{ROOT}/{label_of(s)}" if s else f"{ROOT}/{clean(series_name(name))}" # ------------------------------------------------------------------- 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[(series_key(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 {canon[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 = canon[series_key(name)] stem = f"{label_of(s)} - S{season:02d}E" + "-E".join(f"{e:02d}" for e in eps) title = episode_title(r, s, season, eps[0]) if title: stem += f" - {title}" part = PART_RE.search(os.path.splitext(r["base"])[0]) if part and len(eps) == 1: stem += f" Part {part.group(1).upper()}" book(r, f"{series_dir(name)}/Season {season:02d}", stem + 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/<release-name>/<release>.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 provider_broken(): """TMDb was asked for names and did not answer: the plan is not the plan the operator asked for, whatever the move count says.""" return bool(ARGS.tmdb and (not API_KEY or TMDB["failed"])) def provider_note(): """Provider health on the summary line. The 401s are already on stderr, and a terminal that scrolled past them shows a plan that looks finished: the count belongs where it cannot be missed, because a half-resolved plan renames folders with the folder's own junk words.""" if not (ARGS.tmdb or ARGS.tvdb): return "" if not ARGS.tmdb: return ", tmdb off (names from the cache and tvdb only)" if not API_KEY: return ", tmdb NOT ASKED (TMDB_API_KEY unset): NOT provider-complete" if not TMDB["tries"]: return ", tmdb cache-only (every name was already answered)" done = TMDB["tries"] - TMDB["failed"] return (f", tmdb {done}/{TMDB['tries']} answered" + (" — NOT provider-complete" if TMDB["failed"] else "")) 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{provider_note()} -> {ARGS.out}") def write_exceptions(): no_year = sorted({s["name"] for s in canon.values() if not s["year"]}) no_tvdb = sorted({identity(s)[0] for s in canon.values() if not s.get("canon")}) 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") if provider_broken(): fh.write("WARNING: " + provider_note().strip(", ").replace(": NOT", " — NOT") + "\n Names and episode titles that should have come from TMDb are\n" " missing, so folders keep the metadata provider's words. Fix the\n" " credential and re-plan: the cache keeps every answer that did\n" " arrive, so the re-run costs only the lookups that failed.\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") if no_tvdb: fh.write(f"\n{len(no_tvdb):>6} series carry no tvdb identity, so their folders\n") fh.write(" keep the name the metadata provider gave them and no episode\n") fh.write(" title was taken. Nothing was invented: give these a resolvable\n") fh.write(" id (--tvdb) and re-plan; a rename is idempotent.\n") fh.write(" " + ", ".join(no_tvdb[:300]) + "\n") print(f"exceptions -> {ARGS.exceptions}") write_plan() write

#!/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 (its series' TMDb and Tvdb ids, and the episode title the row carries), and `probe` is optional duration evidence for the movie rule. Every name and title answer lives in the cache JSON so a plan stays reproducible: answers are read from it, and a miss is looked up only with --tmdb/--tvdb and their credentials (a 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 <base>.<lang>.srt bonus material <series>/trailers|featurettes|… — never a season The series name and the episode title are the ones the IDENTITY authority — tvdb — states for the id the metadata server matched, not the words a release folder or a server's folder-derived Series name happens to carry. 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. The row's NAME is only evidence once an id backs it: the server names a Series after the folder it scanned, so `24 (2001) Season 1` is a folder echo, and the series' tvdb id is what undoes 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 a registered show its own words name, else — outside the review pile — from the release folder's words. Inside a hidden top-level folder a file may MATCH a registered show but never INVENT one: invention is what filled the pile to begin with. 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 are never moved. A file already inside --park is never moved either: that is what makes re-planning stable. --park belongs OUTSIDE the library — parked files inside it get indexed as one show named after the park folder, and a row like that is a folder echo, never evidence (the guard in `row_names_the_park`). 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 \ --tvdb --tmdb --out /mnt/entropy/media-tv-plan.tsv \ --park /mnt/entropy/media-park/TVShows 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/tv-names.json", help="name answers (TMDb and Tvdb) as ONE input file, so a plan " "is reproducible; keys carry the provider they came from") ARGS.add_argument("--tmdb", action="store_true", help="look up cache misses at TMDb (needs TMDB_API_KEY)") ARGS.add_argument("--tvdb", action="store_true", help="look up cache misses at Tvdb (needs the credentials file)") ARGS.add_argument("--tvdb-credentials", default=os.path.expanduser("~/.config/secrets/tvdb"), help="file with an api_key= line (and a pin= line for a " "user-issued key); mode 600, and never the cache") 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 OUTSIDE the library (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", "") TMDB = {"tries": 0, "failed": 0} 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)(?<![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 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 LEAD_IDX_RE = re.compile(r"^\s*(\d{1,3})\s*[-\u2013.\s]\s*[A-Za-z]") PACK_IDX_RE = re.compile(r"^\s*\d+[a-z]?[.)]\s+") def num(x): try: return int(x) except (TypeError, ValueError): return 0 def folder_season(path): """The season the FOLDER states (`Season 3`, `S3`). A human wrote it, and where a scraped row disagrees this is the one that survives a re-plan.""" m = SEASON_DIR_RE.match(os.path.basename(os.path.dirname(path))) return int(next(g for g in m.groups() if g)) if m else None def leading_index(name): """The number a file leads with inside a season folder — `01 Kyle MacLachlan, Sinead OC.avi` is episode 1. Both matchers read it that way.""" m = LEAD_IDX_RE.match(os.path.splitext(name)[0]) return int(m.group(1)) if m else None def tmdb_request(url): """One TMDb call, authenticated. The two credential shapes are NOT interchangeable and the difference is a 401 on every single call: a v3 key travels as the `api_key` query parameter, and only a v4 request token is accepted as a Bearer header. A token in argv is world-readable in /proc/<pid>/cmdline — this URL is built inside the process and never reaches argv, so the query form is safe here.""" TMDB["tries"] += 1 if API_KEY.count(".") == 2: # three segments: a v4 JWT return http_json(url, headers={"Authorization": "Bearer " + API_KEY}) return http_json(url + ("&" if "?" in url else "?") + "api_key=" + urllib.parse.quote(API_KEY)) 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") try: found = tmdb_request(url).get("results") or [] except Exception as e: # noqa: BLE001 TMDB["failed"] += 1 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] # ---------------------------------------------------------------- tvdb, the identity # tvdb names the show and the episode: it is the only provider whose strings are # the ones the operator asked for, so its answer replaces a folder-derived Series # name. The login mints a JWT per run from a mode-600 file; neither the key nor # the token reaches argv, the cache, or the plan. A provider that will not answer # never invents a name: every lookup here returns None, and None keeps the # provider's own words and lands in the exceptions. TVDB = {"jwt": None, "dead": False} def http_json(url, body=None, headers=None): req = urllib.request.Request( url, data=json.dumps(body).encode() if body is not None else None, headers={"Accept": "application/json", "Content-Type": "application/json", **(headers or {})}) with urllib.request.urlopen(req, timeout=30) as r: return json.load(r) def tvdb_login(): """The run's JWT, or None: a project key needs only api_key=, a user-issued key also needs the pin.""" if TVDB["jwt"] is not None: return TVDB["jwt"] or None TVDB["jwt"] = "" try: lines = open(ARGS.tvdb_credentials).read().splitlines() except OSError: print(f"tvdb: no credentials at {ARGS.tvdb_credentials}", file=sys.stderr) return None sec = {} for line in lines: line = line.strip() if line and not line.startswith("#") and "=" in line: k, v = line.split("=", 1) sec[k.strip()] = v.strip() if not sec.get("api_key"): print(f"tvdb: no api_key in {ARGS.tvdb_credentials}", file=sys.stderr) return None body = {"apiKey": sec["api_key"]} if sec.get("pin"): body["token"] = sec["pin"] try: TVDB["jwt"] = (http_json("https://api4.thetvdb.com/v4/login", body) .get("data") or {}).get("token") or "" except Exception as e: # noqa: BLE001 print(f"tvdb login failed: {e}", file=sys.stderr) return None return TVDB["jwt"] or None def tvdb(path): """One Tvdb v4 GET, or None when it cannot be answered.""" if not ARGS.tvdb or TVDB["dead"]: return None jwt = tvdb_login() if not jwt: TVDB["dead"] = True return None try: return http_json("https://api4.thetvdb.com/v4" + path, headers={"Authorization": "Bearer " + jwt}).get("data") except Exception as e: # noqa: BLE001 print(f"tvdb {path} failed: {e}", file=sys.stderr) return None def tvdb_series(sid): """{name, year} for a tvdb series id.""" ck = f"tvdb:series:{sid}" if ck in CACHE: return CACHE[ck] data = tvdb(f"/series/{sid}") if not data: return None CACHE[ck] = {"name": (data.get("name") or "").strip(), "year": (data.get("firstAired") or "")[:4]} return CACHE[ck] def tvdb_id_for_tmdb(mid): """The tvdb id TMDb records for a series id — the bridge for the bulk of this library, which the server matched by TMDb.""" ck = f"tmdb:external:{mid}" if ck in CACHE: return CACHE[ck] or "" if not (ARGS.tmdb and API_KEY): return "" try: got = tmdb_request("https://api.themoviedb.org/3/tv/" + mid + "/external_ids") except Exception as e: # noqa: BLE001 TMDB["failed"] += 1 print(f"tmdb external ids failed for {mid}: {e}", file=sys.stderr) return "" CACHE[ck] = str(got.get("tvdb_id") or "") return CACHE[ck] def tvdb_search(name): """The tvdb series a provider name states, or None: only tvdb's own spelling counts, so a near-miss never renames a show.""" ck = f"tvdb:search:{name}" if ck in CACHE: return CACHE[ck] for hit in tvdb("/search?query=" + urllib.parse.quote(name) + "&type=series") or []: if eng.norm(hit.get("name") or "") != eng.norm(name): continue CACHE[ck] = {"name": (hit.get("name") or "").strip(), "year": str(hit.get("year") or ""), "id": str(hit.get("tvdb_id") or "")} return CACHE[ck] CACHE[ck] = None return None def tvdb_episodes(sid): """`season/episode` -> title for one series, in the one call tvdb answers.""" ck = f"tvdb:episodes:{sid}" if ck in CACHE: return CACHE[ck] data = tvdb(f"/series/{sid}/episodes/default") if not data: return None CACHE[ck] = {f"{e.get('seasonNumber')}/{e.get('number')}": (e.get("name") or "").strip() for e in data.get("episodes") or []} 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} def row_names_the_park(path, name): """The library's own review pile is not a show. When the park sat INSIDE the library the server indexed the whole folder as one series named after it, and a row like that would rename a real show to `_unsorted` — so a row whose series is the hidden top-level folder the file itself sits in is dropped.""" top = rel(path).split("/")[0] return top.startswith("_") and eng.norm(name) == eng.norm(top) def in_review_pile(path): """A file under the library's hidden top-level folder is review material by that folder's own convention: its filename may MATCH a show a provider names somewhere else, but it may never INVENT one — inventing one from a file's own words is what filled the pile to begin with.""" return rel(path).split("/")[0].startswith("_") 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 and not row_names_the_park(p, row.get("series", "")): 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] # 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="", tvdb_id=""): """Register a series; the first name wins, the first year and ids stick.""" k = eng.norm(name) s = series.setdefault(k, {"name": series_name(name), "year": "", "tmdb": "", "tvdb": ""}) if year and not s["year"]: s["year"] = str(year) if tmdb_id and not s["tmdb"]: s["tmdb"] = str(tmdb_id) if tvdb_id and not s["tvdb"]: s["tvdb"] = str(tvdb_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 known_in(r["base"]) or ("" if in_review_pile(p) else 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 "", t["tvdb"] 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"] # tvdb is the identity authority. The row's own tvdb id is the door; a TMDb id # crosses through TMDb's external ids; a bare name counts only when tvdb spells # it the same and the years do not disagree. A series nothing resolves keeps the # provider's words AND says so in the exceptions, which is where the operator # finds the folders the plan did not rename to authority. for s in series.values(): got = tvdb_series(s["tvdb"]) if s["tvdb"] else None if not got and s["tmdb"]: sid = tvdb_id_for_tmdb(s["tmdb"]) if sid: s["tvdb"] = sid got = tvdb_series(sid) if not got: hit = tvdb_search(s["name"]) if hit and hit["id"] and (not s["year"] or not hit["year"] or hit["year"] == s["year"]): s["tvdb"] = hit["id"] got = hit if got and got["name"]: s["canon"], s["canon_year"] = got["name"], got["year"] # One identity, one folder: the provider's spellings collapse to the shows tvdb # says they are, so a box set split over five release folders is ONE series and # two files claiming one episode MEET inside the duplicate rule below. alias, canon = {}, {} for k, s in series.items(): key = f"id:{s['tvdb']}" if s["tvdb"] else k alias[k] = key keep = canon.setdefault(key, s) if keep is not s: for f in ("year", "tvdb", "canon", "canon_year"): if not keep.get(f) and s.get(f): keep[f] = s[f] if ARGS.tmdb or ARGS.tvdb: with open(ARGS.tmdb_cache, "w") as fh: json.dump(CACHE, fh, indent=1, sort_keys=True) def identity(s): """(name, year) a folder and its files carry: tvdb's whenever it answered, the provider's own words only for a series nothing resolved.""" if s.get("canon"): return s["canon"], s["canon_year"] or s["year"] return s["name"], s["year"] def series_key(name): """The merged identity a provider spelling belongs to.""" return alias.get(eng.norm(name), eng.norm(name)) def label_of(s): """`<Series> (<year>)`, the form both matchers read.""" nm, yr = identity(s) nm = clean(nm) return f"{nm} ({yr})" if yr else nm def episode_title(r, s, season, ep): """The title this episode carries: tvdb's for the id the series resolved to, else the title the matched row states, else nothing at all — a dangling ` - ` is not a title, and an invented one is worse than none.""" if s["tvdb"]: t = ((tvdb_episodes(s["tvdb"]) or {}).get(f"{season}/{ep}") or "").strip() if t: return clean(t) t = ((titles.get(r["path"]) or {}).get("title") or "").strip() return clean(t) if t else "" def series_dir(name): s = canon.get(series_key(name)) return f"{ROOT}/{label_of(s)}" if s else f"{ROOT}/{clean(series_name(name))}" # ------------------------------------------------------------------- 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[(series_key(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 {canon[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 = canon[series_key(name)] stem = f"{label_of(s)} - S{season:02d}E" + "-E".join(f"{e:02d}" for e in eps) title = episode_title(r, s, season, eps[0]) if title: stem += f" - {title}" part = PART_RE.search(os.path.splitext(r["base"])[0]) if part and len(eps) == 1: stem += f" Part {part.group(1).upper()}" book(r, f"{series_dir(name)}/Season {season:02d}", stem + 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/<release-name>/<release>.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 provider_broken(): """TMDb was asked for names and did not answer: the plan is not the plan the operator asked for, whatever the move count says.""" return bool(ARGS.tmdb and (not API_KEY or TMDB["failed"])) def provider_note(): """Provider health on the summary line. The 401s are already on stderr, and a terminal that scrolled past them shows a plan that looks finished: the count belongs where it cannot be missed, because a half-resolved plan renames folders with the folder's own junk words.""" if not (ARGS.tmdb or ARGS.tvdb): return "" if not ARGS.tmdb: return ", tmdb off (names from the cache and tvdb only)" if not API_KEY: return ", tmdb NOT ASKED (TMDB_API_KEY unset): NOT provider-complete" if not TMDB["tries"]: return ", tmdb cache-only (every name was already answered)" done = TMDB["tries"] - TMDB["failed"] return (f", tmdb {done}/{TMDB['tries']} answered" + (" — NOT provider-complete" if TMDB["failed"] else "")) 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{provider_note()} -> {ARGS.out}") def write_exceptions(): no_year = sorted({s["name"] for s in canon.values() if not s["year"]}) no_tvdb = sorted({identity(s)[0] for s in canon.values() if not s.get("canon")}) 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") if provider_broken(): fh.write("WARNING: " + provider_note().strip(", ").replace(": NOT", " — NOT") + "\n Names and episode titles that should have come from TMDb are\n" " missing, so folders keep the metadata provider's words. Fix the\n" " credential and re-plan: the cache keeps every answer that did\n" " arrive, so the re-run costs only the lookups that failed.\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") if no_tvdb: fh.write(f"\n{len(no_tvdb):>6} series carry no tvdb identity, so their folders\n") fh.write(" keep the name the metadata provider gave them and no episode\n") fh.write(" title was taken. Nothing was invented: give these a resolvable\n") fh.write(" id (--tvdb) and re-plan; a rename is idempotent.\n") fh.write(" " + ", ".join(no_tvdb[:300]) + "\n") print(f"exceptions -> {ARGS.exceptions}") write_plan() write_exceptions()