notebook
This commit is contained in:
@@ -0,0 +1,206 @@
|
||||
#!/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)
|
||||
|
||||
# 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", ""))
|
||||
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()
|
||||
Reference in New Issue
Block a user