#!/usr/bin/env python3 """ EUNI -> Plex: druhy pruchod, doplni metadata k uz naskenovanym videim. Paruje polozky v Plex sekci EUNI s kurzy v Mongu (podle nazvu souboru = build_plan z plex_export.py) a zapisuje: Originally Available, Studio, Summary, Genre, Label, a u multi-video kurzu Collection (sloucí díly k sobe). Pole zamyka (.locked=1), takze je sken neprepise. Idempotentni - lze poustet opakovane. Zavislosti: pip install pymongo requests Pouziti: set PLEX_TOKEN=... # (Windows) nebo export PLEX_TOKEN=... python plex_meta.py --dry-run # vypise co by zapsal, nic nemeni python plex_meta.py # zapise python plex_meta.py --token XXXX --section 23 """ import argparse import os import sys from datetime import datetime import requests from pymongo import MongoClient import plex_export as px # sdili build_plan() -> stejne nazvy souboru for _s in (sys.stdout, sys.stderr): try: _s.reconfigure(encoding="utf-8", errors="replace") except Exception: pass DEF_PLEX = "http://192.168.1.76:32400" DEF_SECTION = 23 # EUNI (Other Videos) DEF_TOKEN = "Em6_tQ7DizF2s36-9_Jx" # natvrdo; lze prebit env PLEX_TOKEN nebo --token # Mapovani profese kodu EUNI -> nazev (nezname kody se vynechaji). # Cely soucasny batch je profese 2 = Lékař. Dalsi profese doplnit az se dotahnou. PROFESE = { 2: "Lékař", # 1: "...", 3: "Farmaceut", ... # doplnit pri dalsich profesich } def build_kurz_index(db): """stem nazvu souboru (bez .mp4) -> (kurz_doc, multi?).""" kurzy = {k["_id"]: k for k in db.kurzy.find({})} _, plan = px.build_plan(db) # kolik souboru ma kurz -> multi? from collections import Counter # plan nezna kurz_id; znovu spocteme pres materialy vids = list(db.materialy.find( {"druh": "video", "seaweed_fids": {"$exists": True, "$ne": []}}, {"kurz_id": 1, "soubor": 1, "seaweed_path": 1, "seaweed_size": 1, "klic": 1})) by_course = {} for v in vids: by_course.setdefault(v["kurz_id"], []).append(v) idx = {} for kid, items in by_course.items(): k = kurzy.get(kid, {}) nazev = px.sanitize(k.get("nazev") or items[0].get("soubor", kid)) autor = px.surname(k.get("autor")) dp = k.get("datum_publikace") rok = dp.year if isinstance(dp, datetime) else None ystr = f" ({rok})" if rok else "" multi = len(items) > 1 if not multi: who = f" - {autor}" if autor else "" stem = px.clip(px.sanitize(f"{nazev}{who}{ystr}")) idx[stem] = (k, False) else: for i, v in enumerate(px.order_items(items), 1): lbl = px.seg_label(v.get("soubor", ""), k.get("nazev") or "") mid = f" - {i:02d} {lbl}" if lbl else f" - {i:02d}" stem = px.clip(px.sanitize(f"{nazev}{mid}{ystr}")) idx[stem] = (k, True) return idx def make_summary(k): lines = [] if k.get("autor"): lines.append(f"Autor: {k['autor']}") bits = [] if k.get("akreditace"): bits.append(f"Akreditace {k['akreditace']}") if k.get("kredity"): bits.append(f"{k['kredity']} kreditů") if bits: lines.append(" · ".join(bits)) if k.get("url"): lines.append(f"Zdroj: {k['url']}") return "\n".join(lines) def fix_poster(base, token, rk): """Vybere nejvetsi auto-generovany nahled jako poster (vyhne se cernemu snimku). Vraci True kdyz nejaky nastavil.""" import xml.etree.ElementTree as ET r = requests.get(f"{base}/library/metadata/{rk}/posters", params={"X-Plex-Token": token}, timeout=15) r.raise_for_status() root = ET.fromstring(r.content) cands = [p.get("ratingKey") for p in root.findall("Photo") if (p.get("ratingKey") or "").startswith("media://")] if not cands: return False best, bsz = None, -1 for u in cands: d = requests.get(f"{base}/library/metadata/{rk}/file", params={"url": u, "X-Plex-Token": token}, timeout=15) if len(d.content) > bsz: bsz, best = len(d.content), u requests.put(f"{base}/library/metadata/{rk}/poster", params={"url": best, "X-Plex-Token": token}, timeout=15) return True def plex_items(base, section, token): r = requests.get(f"{base}/library/sections/{section}/all", params={"X-Plex-Token": token}, timeout=15) r.raise_for_status() import xml.etree.ElementTree as ET root = ET.fromstring(r.content) return [(v.get("ratingKey"), v.get("title")) for v in root.findall("Video")] def push(base, section, token, rating_key, k, multi, nazev, dry): params = [("type", "1"), ("id", rating_key), ("X-Plex-Token", token), ("studio.value", "EUNI"), ("studio.locked", "1"), ("label[0].tag.tag", "EUNI"), ("label.locked", "1"), ("genre[0].tag.tag", "EUNI kurz"), ("genre.locked", "1")] dp = k.get("datum_publikace") if isinstance(dp, datetime): params += [("originallyAvailableAt.value", dp.strftime("%Y-%m-%d")), ("originallyAvailableAt.locked", "1")] summ = make_summary(k) if summ: params += [("summary.value", summ), ("summary.locked", "1")] # profese -> dalsi genre (jen znamne kody) gi = 1 for code in (k.get("profese") or []): name = PROFESE.get(code) if name: params.append((f"genre[{gi}].tag.tag", name)) gi += 1 if multi: params += [("collection[0].tag.tag", nazev), ("collection.locked", "1")] if dry: return "DRY" r = requests.put(f"{base}/library/sections/{section}/all", params=params, timeout=15) return r.status_code def main(): ap = argparse.ArgumentParser() ap.add_argument("--dry-run", action="store_true") ap.add_argument("--plex", default=DEF_PLEX) ap.add_argument("--section", type=int, default=DEF_SECTION) ap.add_argument("--mongo", default=px.DEF_MONGO) ap.add_argument("--token", default=os.environ.get("PLEX_TOKEN") or DEF_TOKEN) ap.add_argument("--no-poster", action="store_true", help="nevybirat nahled (jen metadata)") args = ap.parse_args() if not args.token: sys.exit("Chybi PLEX_TOKEN (env nebo --token).") db = MongoClient(args.mongo, serverSelectionTimeoutMS=5000)["EUNI"] idx = build_kurz_index(db) items = plex_items(args.plex, args.section, args.token) print(f"Plex polozek: {len(items)} | namapovano kurzu: {len(idx)}\n") ok = miss = fail = 0 for rk, title in items: hit = idx.get(title) if not hit: miss += 1 print(f" ? bez parovani: {title}") continue k, multi = hit try: code = push(args.plex, args.section, args.token, rk, k, multi, px.sanitize(k.get("nazev", "")), args.dry_run) if code in (200, "DRY"): ok += 1 if code == 200 and not args.no_poster: try: fix_poster(args.plex, args.token, rk) except Exception as e: print(f" poster? {title} :: {e}") else: fail += 1 print(f" FAIL {code}: {title}") except Exception as e: fail += 1 print(f" FAIL {title} :: {e}") tag = "[DRY-RUN] " if args.dry_run else "" print(f"\n{tag}OK {ok}, nenamapovano {miss}, chyb {fail}") if __name__ == "__main__": main()