Files
serverAppConfigs/scripts/media-shows.py
T
Frank c3ab565d9b feat(media): tvdb names the show — episode titles, id-backed folders, park outside
The TV relayout that landed copied Jellyfin's Series names verbatim, and the
naming census showed what that launders in: one Series item per release folder
(1,107 items for 1,107 top-level dirs), the same show spelled six ways, and
box sets living in 346 separate folders for 821 identities. Jellyfin names a
Series after the folder it scanned, so `24 (2001) Season 1` is a folder echo,
not a show name.

The name authority is now tvdb, reached through the id the server matched:

  * media-census.py jellyfin emits two more columns — `title` (the episode row's
    own Name) and `tvdb` (the series id, read from BaseItemProviders where
    Jellyfin 12 keeps provider ids and from the row's Data blob where older
    builds do, so the census does not care which shape the live DB has);
  * media-shows.py resolves each series to tvdb — the row's tvdb id, else its
    TMDb id through TMDb's external ids, else an exact tvdb name search — and
    writes tvdb's name and year. That also collapses the provider's spellings
    into ONE identity, which is what turns a box set split over five release
    folders into one folder and what makes two files claiming one episode MEET
    in the duplicate rule;
  * the episode file becomes `<Series> (<year>) - SxxEyy - <Title>`: the title
    comes from tvdb's episode list, else the row's own title, else nowhere — no
    title means no separator, never an invented one.

Two refusals come with it. A row whose series is the hidden top-level folder the
file itself sits in is NOT evidence: the review pile parked inside the library
was indexed as one show named `_unsorted`, and its rows would have renamed real
shows to it. And inside such a folder a file's own words may MATCH a registered
show but never INVENT one — invention is what filled the pile. `--park`
therefore belongs outside the library, which is also what makes the next plan
re-file the pile instead of skipping it.

Both providers answer offline-reproducibly: every answer is cached in one JSON
file, so a plan can be re-planned without a key. Lookups stay behind --tmdb and
--tvdb, and the tvdb token is read from a mode-600 file, never argv.

tests/media_shows.bats is 23 tests, none of which touch the network — the
fixture Jellyfin DB carries the providers table and the name cache is an input
file. ./tests/run.sh: 753 green.
2026-10-31 07:01:50 -07:00

952 lines
40 KiB
Python
Executable File

#!/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):
<root>/<Series> (<year>)/Season <NN>/<Series> (<year>) - SxxEyy - <Title>.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", "")
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(kind, query):
"""A provider answer for a query: the cache first, TMDb once, then never again."""
ck = f"{kind}:{query}"
if ck in CACHE:
hit = CACHE[ck]
# A cache an operator edited by hand can carry the year as a number; the
# comparison below is text, so the wire form is imposed on every hit.
return None if hit is None else {**hit, "year": str(hit.get("year") or "")}
if not (ARGS.tmdb and API_KEY):
return None
path = "movie" if kind == "movie" else "tv"
url = ("https://api.themoviedb.org/3/search/" + path + "?query="
+ urllib.parse.quote(query) + "&include_adult=false")
req = urllib.request.Request(url, headers={
# The key rides a header, never the URL: a token in argv is world-readable.
"Authorization": "Bearer " + API_KEY,
"Accept": "application/json"})
try:
with urllib.request.urlopen(req, timeout=20) as r:
found = json.load(r).get("results") or []
except Exception as e: # noqa: BLE001
print(f"tmdb {kind} lookup failed for {query!r}: {e}", file=sys.stderr)
return None
if not found:
CACHE[ck] = None
return None
top = found[0]
CACHE[ck] = {"name": (top.get("title") or top.get("name") or "").strip(),
"year": str(top.get("year") or "") or (top.get("first_air_date") or "")[:4],
"id": top.get("id")}
return CACHE[ck]
# ---------------------------------------------------------------- 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 = http_json(f"https://api.themoviedb.org/3/tv/{mid}/external_ids",
headers={"Authorization": "Bearer " + API_KEY})
except Exception as e: # noqa: BLE001
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 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 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")
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()