#!/usr/bin/env python3 """Save.TV Film-Enricher v2 — reichert Archiv-Filme per OpenRouter (Grok 4.1 Fast) an. Unterschiede zu v1: * Nutzt OpenRouter statt lokalen Ollama (KI-Server ist abgeschaltet). * Iteriert ueber /mnt/savetv/*.mp4 (lokale Dateien) statt ueber das 25-Tage Save.TV-Archiv — damit rutschen auch aeltere Filme nicht durch. * Ein Film gilt als "angereichert" sobald ein Cache-Eintrag MIT Land vorhanden ist (Description optional, kein hartes Kriterium mehr). * --dry-run / --limit / --only-missing (Default) / --force / --verbose Prompt und JSON-Schema bleiben gleich -> voll rueckwaertskompatibel mit dem existierenden .filminfo_cache.json, das savetv_web.py und savetv_extra_routes.py nutzen. Laeuft idempotent und kann jederzeit erneut gestartet werden. """ from __future__ import annotations import argparse import fcntl import json import logging import os import re import sys import time import unicodedata from pathlib import Path from typing import Iterable import requests sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, "/opt/homelab-ai-bot") from core import config as homelab_config OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions" DEFAULT_MODEL = "x-ai/grok-4.3" FALLBACK_MODEL = "qwen/qwen3-next-80b-a3b-instruct" SAVETV_DIR = Path("/mnt/savetv") FILMINFO_CACHE = SAVETV_DIR / ".filminfo_cache.json" LOCKFILE = Path("/tmp/savetv_enrich.lock") LOG_FILE = Path("/var/log/savetv-enrich.log") BATCH_SIZE = 10 # Cache alle N Enrichments persistieren SLEEP_BETWEEN = 0.3 # Rate-Limit-Schutz (OpenRouter vertraegt viel) HTTP_TIMEOUT = 45 # Raw-Format: "Titel_12345678.mp4"; Good: "Titel (YYYY).mp4" RAW_RE = re.compile(r"^(.+)_(\d{6,})\.mp4$") GOOD_RE = re.compile(r"^(.+?)\s*\((19|20)\d{2}\)\.mp4$") log = logging.getLogger("savetv_enrich_v2") # --------------------------------------------------------------------- # Cache I/O # --------------------------------------------------------------------- def _load_cache() -> dict: if FILMINFO_CACHE.exists(): try: return json.loads(FILMINFO_CACHE.read_text(encoding="utf-8")) except Exception as e: log.error("Cache-Load fehlgeschlagen: %s — starte mit leerem Cache", e) return {} def _save_cache(cache: dict) -> None: tmp = FILMINFO_CACHE.with_suffix(".tmp") tmp.write_text(json.dumps(cache, ensure_ascii=False, indent=1), encoding="utf-8") tmp.rename(FILMINFO_CACHE) def _is_enriched(entry: dict | None) -> bool: """Mindestkriterium: es gibt einen Eintrag mit country ODER description.""" if not entry: return False if entry.get("countries"): return True if entry.get("description"): return True return False # --------------------------------------------------------------------- # Titel-Extraktion aus Dateinamen # --------------------------------------------------------------------- def _normalize_key(s: str) -> str: s = unicodedata.normalize("NFKD", s).encode("ascii", "ignore").decode("ascii") s = s.lower() s = re.sub(r"[^a-z0-9]+", " ", s) return s.strip() def _filename_to_candidate_title(name: str) -> str: """Extrahiert einen Klartext-Titel aus einem mp4-Dateinamen.""" m = RAW_RE.match(name) if m: return m.group(1).replace("_", " ").strip() m = GOOD_RE.match(name) if m: return m.group(1).strip() return re.sub(r"\.mp4$", "", name).replace("_", " ").strip() def _collect_local_titles() -> list[tuple[str, str]]: """Gibt Liste [(dateiname, pseudo_titel)] aller lokalen mp4 zurueck. Duplikate nach normalisiertem Titel werden auf den ersten Treffer reduziert. """ seen_norm: set[str] = set() result: list[tuple[str, str]] = [] for f in sorted(SAVETV_DIR.glob("*.mp4")): pseudo = _filename_to_candidate_title(f.name) nk = _normalize_key(pseudo) if not nk or nk in seen_norm: continue seen_norm.add(nk) result.append((f.name, pseudo)) return result def _cache_key_for(pseudo: str, cache: dict) -> str | None: """Findet vorhandenen Cache-Schluessel fuer einen Titel (case-/diakritika- insensitiv). Erstbelegung gewinnt.""" nk = _normalize_key(pseudo) for k in cache.keys(): if _normalize_key(k) == nk: return k return None # --------------------------------------------------------------------- # OpenRouter-Call # --------------------------------------------------------------------- def _get_api_key() -> str: cfg = homelab_config.parse_config() key = cfg.api_keys.get("openrouter_key") or "" if not key: raise RuntimeError("openrouter_key fehlt in homelab.conf") return key def _call_openrouter(prompt: str, api_key: str, model: str = DEFAULT_MODEL) -> str: payload = { "model": model, "messages": [ {"role": "system", "content": ( "Du bist eine Filmdatenbank. Antworte AUSSCHLIESSLICH mit validem JSON. " "Kein Markdown, keine Erklärungen, kein Denken. Nur JSON. Sprache: Deutsch." )}, {"role": "user", "content": prompt}, ], "temperature": 0.2, "max_tokens": 700, "response_format": {"type": "json_object"}, } headers = { "Authorization": f"Bearer {api_key}", "HTTP-Referer": "https://homelab.orbitalo.net/savetv-enrich", "X-Title": "savetv-enrich", "Content-Type": "application/json", } try: r = requests.post(OPENROUTER_URL, json=payload, headers=headers, timeout=HTTP_TIMEOUT) r.raise_for_status() data = r.json() choice = (data.get("choices") or [{}])[0] msg = choice.get("message", {}).get("content", "") or "" return msg.strip() except requests.exceptions.ReadTimeout: if model != FALLBACK_MODEL: log.warning("Timeout mit %s — Fallback auf %s", model, FALLBACK_MODEL) return _call_openrouter(prompt, api_key, model=FALLBACK_MODEL) raise except requests.exceptions.HTTPError as e: body = "" try: body = e.response.text[:300] if e.response is not None else "" except Exception: pass log.error("OpenRouter HTTP-Fehler %s: %s", e, body) return "" except Exception as e: log.error("OpenRouter Call-Fehler: %s", e) return "" def _parse_json_block(raw: str) -> dict | None: if not raw: return None try: return json.loads(raw) except json.JSONDecodeError: pass # manche Modelle packen Code-Fences drumrum stripped = raw if stripped.startswith("```"): stripped = re.sub(r"^```\w*\n?", "", stripped) stripped = re.sub(r"\n?```$", "", stripped) try: return json.loads(stripped) except json.JSONDecodeError: pass m = re.search(r"\{[\s\S]*\}", raw) if m: try: return json.loads(m.group()) except json.JSONDecodeError: return None return None def _is_mostly_latin(text: str) -> bool: if not text: return False latin = sum(1 for c in text if c.isascii() or '\u00C0' <= c <= '\u024F') return latin / max(len(text), 1) > 0.7 def _normalize_actors(actors_raw) -> list[str]: if not isinstance(actors_raw, list): return [] out: list[str] = [] for a in actors_raw[:4]: if isinstance(a, str): out.append(a.strip()) elif isinstance(a, dict): name = a.get("name") or a.get("actor") or "" if name: out.append(str(name).strip()) return [x for x in out if x] def _enrich_one(title: str, api_key: str, model: str) -> dict: """Fragt OpenRouter nach Film-Metadaten. Gibt Dict mit year / countries / genres / actors / director / description zurueck. Bei Fehlern: leere Felder.""" clean_title = re.sub(r"\s*[-\u2013\u2014]\s*.+$", "", title).strip() or title prompt = f"""Gib mir Informationen zum Film "{clean_title}". Antworte als JSON mit exakt diesen Feldern (alle Texte auf Deutsch): {{ "year": "Erscheinungsjahr als String", "countries": ["Produktionsländer als Strings"], "genres": ["bis zu 3 Genres als Strings"], "actors": ["bis zu 4 Hauptdarsteller als Strings"], "director": "Regisseur als String", "description": "3-5 Sätze auf Deutsch: Worum geht es, für wen geeignet. Keine Spoiler." }} Wichtig: - Alle Werte müssen Strings sein. - Schreibe die description komplett auf Deutsch. - Falls du den Film nicht eindeutig kennst, setze description auf leeren String und lasse andere Felder leer. - Bei Produktionsländern bevorzuge "Vereinigte Staaten" (nicht "USA").""" raw = _call_openrouter(prompt, api_key, model=model) data = _parse_json_block(raw) or {} desc = str(data.get("description") or "")[:600] if desc and not _is_mostly_latin(desc): log.info(" Nicht-lateinische Beschreibung gefiltert: %.80s", desc) desc = "" return { "year": str(data.get("year") or "")[:4], "countries": [str(c) for c in (data.get("countries") or [])[:3] if c], "genres": [str(g) for g in (data.get("genres") or [])[:3] if g], "actors": _normalize_actors(data.get("actors")), "director": str(data.get("director") or "")[:60], "description": desc, } # --------------------------------------------------------------------- # Main # --------------------------------------------------------------------- def run(args) -> int: log.info("=== savetv_enrich v2 (OpenRouter %s) ===", args.model) local = _collect_local_titles() log.info("Lokale mp4: %d einzigartige Titel (nach Normalisierung)", len(local)) cache = _load_cache() log.info("Cache: %d bestehende Eintraege", len(cache)) # Plan bauen plan: list[tuple[str, str, str | None]] = [] # (filename, pseudo, existing_cache_key) for fname, pseudo in local: ck = _cache_key_for(pseudo, cache) entry = cache.get(ck) if ck else None if args.force or not _is_enriched(entry): plan.append((fname, pseudo, ck)) log.info("Zu enrichen: %d Titel (%s)", len(plan), "force=alle" if args.force else "nur fehlende") if args.limit: plan = plan[: args.limit] log.info("Limit aktiv: Plan auf %d begrenzt", len(plan)) if args.dry_run: log.info("--- DRY-RUN, keine API-Calls ---") for fname, pseudo, ck in plan[:50]: log.info(" would-enrich: %s [cache_key=%s]", pseudo, ck or "NEU") if len(plan) > 50: log.info(" ... und %d weitere", len(plan) - 50) return 0 api_key = _get_api_key() enriched = 0 basic = 0 errors = 0 for i, (fname, pseudo, ck) in enumerate(plan, 1): title_for_api = ck or pseudo log.info("[%d/%d] %s", i, len(plan), title_for_api) try: info = _enrich_one(title_for_api, api_key, model=args.model) except Exception as e: log.error(" Fehler: %s", e) errors += 1 time.sleep(SLEEP_BETWEEN) continue cache_key = ck or pseudo old = cache.get(cache_key, {}) if isinstance(cache.get(cache_key), dict) else {} merged = dict(old) for k in ("year", "countries", "genres", "actors", "director", "description"): v = info.get(k) if v: merged[k] = v cache[cache_key] = merged has_country = bool(merged.get("countries")) has_desc = bool(merged.get("description")) if has_desc and has_country: enriched += 1 log.info(" OK: %s, %s, genres=%s", merged.get("year", "?"), ",".join(merged.get("countries", []))[:40], ",".join(merged.get("genres", []))[:40]) elif has_country or merged.get("year"): basic += 1 log.info(" Teil-OK (ohne Beschreibung): year=%s countries=%s", merged.get("year", ""), merged.get("countries", [])) else: log.info(" Film unbekannt / leere Antwort") if (enriched + basic) and (enriched + basic) % BATCH_SIZE == 0: _save_cache(cache) log.info(" Cache-Checkpoint (%d full, %d basic, %d err)", enriched, basic, errors) time.sleep(SLEEP_BETWEEN) _save_cache(cache) log.info("=== Fertig: full=%d basic=%d errors=%d von %d geplant ===", enriched, basic, errors, len(plan)) return 0 def setup_logging(verbose: bool) -> None: level = logging.DEBUG if verbose else logging.INFO fmt = "%(asctime)s %(levelname)s %(message)s" datefmt = "%H:%M:%S" handlers: list[logging.Handler] = [logging.StreamHandler(sys.stdout)] try: handlers.append(logging.FileHandler(LOG_FILE, encoding="utf-8")) except Exception: pass logging.basicConfig(level=level, format=fmt, datefmt=datefmt, handlers=handlers) def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--dry-run", action="store_true") ap.add_argument("--limit", type=int, default=0) ap.add_argument("--force", action="store_true", help="auch bereits angereicherte Filme erneut abfragen") ap.add_argument("--model", default=DEFAULT_MODEL) ap.add_argument("--verbose", action="store_true") args = ap.parse_args() setup_logging(args.verbose) lock_fd = open(LOCKFILE, "w") try: fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: print("Enricher laeuft bereits — Abbruch.") return 0 try: return run(args) finally: fcntl.flock(lock_fd, fcntl.LOCK_UN) lock_fd.close() if __name__ == "__main__": sys.exit(main())