208 lines
7.5 KiB
Python
208 lines
7.5 KiB
Python
#!/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()
|