#!/usr/bin/env python3 r"""Ask tvdb what it calls the shows a plan renamed WITHOUT a tvdb id. media-shows-tvdb.py [--plan TSV] [--root DIR] [--park DIR] [--map TSV] [--out TSV] [--credentials FILE] `media-shows.py` renames a series to tvdb's own words only when an id backs the name; everything else keeps the metadata provider's words and no episode title, and the plan's mkdir rows are where those folders show up. This script turns that list into a decision sheet: one row per destination series folder that carries no `[tvdbid-NNNN]`, with what tvdb's search answers for it and a verdict that says HOW MUCH was guessed to get there. It changes nothing. The sheet is for the operator; the file that changes behaviour is `scripts/media-shows-tvdb-ids.tsv`, whose rows he approves. That is the loop: plan -> this sheet -> approved map rows -> re-plan. Verdicts, weakest last: EXACT tvdb's own name equals the folder's words as they stand. PUNCT it equals them only after release junk is stripped (pack numbers, codecs, group names, `Complete Series`); the difference is spelling, not identity. NEAR tvdb answered something whose name is NOT the query. The best hit is printed so a human can judge it, and it is never applied by anything. NONE nothing left to ask with, or no answer at all. Acceptance is deliberately unforgiving — a hit counts only when tvdb's name equals the query with every non-alphanumeric dropped from both sides, and the series' year still has to fall inside any year span the folder names. The cleaner is the only thing that got smarter over the passes this tool replaces; the acceptance test never moved, because a wrong id renames a whole pack to the wrong show. A query that needs the cleaner is reported as PUNCT for exactly that reason: the machine stripped words, so a human reads the row. Read-only against tvdb. The credential file is the planner's own (`api_key=`, optional `pin=`), read here and never printed. """ import argparse import collections import importlib.util import json import os import re import sys import time import urllib.error import urllib.parse import urllib.request ARGS = argparse.ArgumentParser(description=__doc__.splitlines()[0], formatter_class=argparse.RawDescriptionHelpFormatter) ARGS.add_argument("--plan", default="/tmp/tv-plan.tsv", help="the plan media-shows.py just wrote (its mkdir rows are " "the folders this sheet asks about)") ARGS.add_argument("--root", default=os.environ.get("TV_ROOT", "/mnt/entropy/TVShows"), help="the TV library the plan renames") ARGS.add_argument("--park", default="", help="the review pile inside it (its folders are not shows)") ARGS.add_argument("--map", default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "media-shows-tvdb-ids.tsv"), help="the operator's approved map; its folders are reported as " "MAPPED and never asked again") ARGS.add_argument("--credentials", default=os.path.expanduser("~/.config/secrets/tvdb"), help="file with an api_key= line (and a pin= for a user-issued " "key); mode 600, and never printed") ARGS.add_argument("--out", default="/tmp/tvdb-proposal.tsv", help="the decision sheet") ARGS = ARGS.parse_args() ROOT = os.path.abspath(ARGS.root) PARK = os.path.abspath(ARGS.park or os.path.join(ROOT, "_unsorted")) _spec = importlib.util.spec_from_file_location( "eng", os.path.join(os.path.dirname(os.path.abspath(__file__)), "media-titles.py")) eng = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(eng) # Release metadata, not the show: everything a scene folder name says about the # ENCODING, the packaging or the languages, which tvdb has never heard of. JUNK = re.compile( r"complete|series|dual|multi|audio|sub|subs|multisub|dualaudio|" r"\d{3,4}p|1080p|720p|2160p|480p|4k|hd|hdtv|cam|hdcam|telesync|x264|x265|hevc|avc|" r"\d+bit|bit|bluray|blu-?ray|bdrip|web-?rip|web-?dl|webrip|hdtvrip|" r"dvdrip|dvdr|dvdscr|tsr|tsv|mkv|avi|mp4|aac|ac3|dts|eac3|flac|ddp?5[.1]1|" r"bd|dvd|rip|boxset|box|season|seasons|episodes?|specials?|shorts|ost|" r"ova|ovas|oad|repack|proper|internal|mi?xed|budgetbits|deluxe|remux|" r"remastered|uncut|collector|edition|extras?|volume|volumes|part|" r"disc|disk|cd|ntsc|pal|region|fixed|patched|proof|english|esp|" r"espagnol|french|vostfr|subtitul\w*|spanish|german|italian|ita|eng|fre|spa|lat|" r"upscaled?|ai|dd|movie|movies", re.I) GROUP = re.compile(r"KONTRAST|TGx|Judas|Silence|FreetheFilms|NTb|DEADPOOL|Xploit|" r"SM737|TSA|3OISL3R|PLAYWEB|EPiC|FLUX|HONEY|Zaxton|Cca3rW|" r"monikeres|frank|hdh|psi|rtfpemaker|cbm|web|asser|astro|" r"yfn|uindex|www|org|com|net|mesc|ataraxiaprime|seph\d*|" r"[a-z]+subs?|edge", re.I) BRACKET = re.compile(r"[\[({]([^\])}]*)[\])}]") NUM_RANGE = re.compile(r"(?= 3 or (not out and after >= 2): continue out.append(t) while out and out[-1].lower() in ("and", "with", "the", "a", "an", "to", "of"): out.pop() return " ".join(out).strip() def years(name): return [int(y) for y in re.findall(r"\b(19\d\d|20\d\d)\b", name)] def bare(s): return re.sub(r"[^a-z0-9]", "", s.lower()) CRED = {} try: for _line in open(ARGS.credentials): if "=" in _line: k, v = _line.strip().split("=", 1) CRED[k.strip()] = v.strip() except OSError: print(f"tvdb: no credentials at {ARGS.credentials} — the sheet would be all " "NEAR and NONE, which is not a decision sheet", file=sys.stderr) sys.exit(2) if not CRED.get("api_key"): print(f"tvdb: no api_key in {ARGS.credentials}", file=sys.stderr) sys.exit(2) TOKEN = {"jwt": ""} def call(path): """One tvdb v4 GET, logging in again once when the JWT expires mid-run.""" for attempt in (1, 2): if not TOKEN["jwt"]: body = {"apiKey": CRED["api_key"]} if CRED.get("pin"): body["token"] = CRED["pin"] req = urllib.request.Request("https://api4.thetvdb.com/v4/login", data=json.dumps(body).encode(), headers={"Content-Type": "application/json"}) TOKEN["jwt"] = json.load( urllib.request.urlopen(req, timeout=30))["data"]["token"] try: req = urllib.request.Request("https://api4.thetvdb.com/v4" + path, headers={"Authorization": "Bearer " + TOKEN["jwt"]}) return json.load(urllib.request.urlopen(req, timeout=30))["data"] except urllib.error.HTTPError as e: if e.code == 401 and attempt == 1: TOKEN["jwt"] = "" continue raise MAPPED = {} if ARGS.map and os.path.exists(ARGS.map): for _line in open(ARGS.map): _line = _line.rstrip("\n") if not _line.strip() or _line.startswith("#"): continue cols = _line.split("\t") if len(cols) > 1 and cols[1].strip().isdigit(): MAPPED[cols[0].strip()] = cols[1].strip() # The folders the plan created inside the library, with the file count each one # will hold. The review pile is not a show: its files belong to no identity, and # naming the pile would put one junk series on the sheet for all of them. shows = collections.Counter() for line in open(ARGS.plan).read().splitlines()[1:]: parts = line.split("\t") if len(parts) < 4 or parts[0] != "mkdir": continue dst = os.path.abspath(parts[2]) if not dst.startswith(ROOT + os.sep) or dst.startswith(PARK + os.sep): continue top = os.path.relpath(dst, ROOT).split(os.sep)[0] if top.startswith((".", "_")): continue if "[tvdbid-" in top: continue # an id already backs it shows.setdefault(top, 0) for line in open(ARGS.plan).read().splitlines()[1:]: parts = line.split("\t") if len(parts) < 4 or parts[0] != "move": continue dst = os.path.abspath(parts[2]) if not dst.startswith(ROOT + os.sep) or dst.startswith(PARK + os.sep): continue top = os.path.relpath(dst, ROOT).split(os.sep)[0] if "[tvdbid-" in top: continue shows[top] += 1 tally = collections.Counter() rows = [] for folder in sorted(shows, key=lambda f: (-shows[f], f)): files, tag, note = shows[folder], "NONE", "" if folder in MAPPED: rows.append([files, "MAPPED", folder, "", "", "", MAPPED[folder], "the operator's map already states this id"]) tally["MAPPED"] += 1 continue query, untouched = clean(folder), clean(folder) == folder ys = years(folder) if not query: rows.append([files, "NONE", folder, "", "", "", "", "nothing is left of the folder's name once release junk" " is gone"]) else: try: hits = call("/search?query=" + urllib.parse.quote(query) + "&type=series") or [] except Exception as e: # noqa: BLE001 hits = [] note = f"tvdb did not answer: {e}" want = bare(query) pick = None for h in hits: if bare(h.get("name") or "") != want: continue hy = str(h.get("year") or "") if ys and hy.isdigit() and not (min(ys) <= int(hy) <= max(ys)): continue pick = h break if pick: tag = "EXACT" if untouched else "PUNCT" rows.append([files, tag, folder, query, pick.get("name", ""), str(pick.get("year", "") or ""), str(pick.get("tvdb_id", "") or ""), ""]) elif hits: best = hits[0] rows.append([files, "NEAR", folder, query, best.get("name", ""), str(best.get("year", "") or ""), str(best.get("tvdb_id", "") or ""), f"{len(hits)} answer(s); tvdb's own name is not the " "query, so nothing would take it"]) else: rows.append([files, "NONE", folder, query, "", "", "", note or "tvdb has no series by that name"]) tally[rows[-1][1]] += 1 time.sleep(0.12) # Two folders mapping to one record is not two shows; it is one show the release # folders split. Say it on every row involved, or the operator approves the same # id twice and reads the second as a mistake. by_id = collections.Counter(r[6] for r in rows if r[6] and r[1] != "MAPPED") for r in rows: if r[6] and r[1] != "MAPPED" and by_id[r[6]] > 1: r[7] = (f"MERGE: {by_id[r[6]]} folders map to tvdb {r[6]} — one show, " "the releases split it; " + r[7]).strip("; ") with open(ARGS.out, "w") as fh: fh.write("files\tverdict\tfolder\tquery\ttvdb_name\ttvdb_year\ttvdb_id\tnote\n") for r in rows: fh.write("\t".join(str(c) for c in r) + "\n") print(f"sheet: {len(rows)} folders" + "".join(f" {t}={n}" for t, n in sorted(tally.items())) + f" -> {ARGS.out}")