tmdb integration

Dockers/self-hosted-iptv/main.py

@@ -5,16 +5,21 @@ with randomized order so the same show never appears twice in a row.
 """
 
 import json
+import os
 import random
+import re
 import subprocess
 import tempfile
+import threading
 import time
+import urllib.error
+import urllib.request
 import xml.etree.ElementTree as ET
-from collections import defaultdict
+from collections import defaultdict, deque
 from datetime import datetime, timezone, timedelta
 from pathlib import Path
 from http.server import HTTPServer, BaseHTTPRequestHandler
-from urllib.parse import urlparse, parse_qs, unquote
+from urllib.parse import urlparse, parse_qs, unquote, urlencode
 import mimetypes
 
 # Per-channel schedule cache: channel_id -> (paths, durations). TTL 24h so EPG and live stay in sync.
@@ -29,6 +34,22 @@ DATA_ROOT = Path("/data")
 CHANNELS_FILE = DATA_ROOT / "channels.json"
 VIDEO_EXTENSIONS = {".mkv", ".mp4", ".avi", ".mov", ".m4v", ".webm", ".wmv"}
 
+# TMDB (The Movie Database) for EPG metadata — Dutch (nl). Set TMDB_API_KEY.
+TMDB_API_BASE = "https://api.themoviedb.org/3"
+TMDB_LANGUAGE = "nl"
+TMDB_CACHE_TTL_SEC = 24 * 3600
+TMDB_RATE_LIMIT_REQUESTS = 40
+TMDB_RATE_LIMIT_WINDOW_SEC = 10.0
+TMDB_CACHE_DIR = DATA_ROOT / "tmdb_cache"
+_tmdb_series_cache: dict[str, tuple[int, str, float]] = {}  # programme_name -> (series_id, series_name_nl, ts)
+_tmdb_series_details_cache: dict[int, tuple[dict, float]] = {}  # series_id -> ({genres, country}, ts)
+_tmdb_episode_cache: dict[tuple[int, int, int], tuple[dict, float]] = {}  # (series_id, s, e) -> (meta, ts)
+_tmdb_episode_credits_cache: dict[tuple[int, int, int], tuple[dict, float]] = {}  # (series_id, s, e) -> (credits, ts)
+_tmdb_request_times: deque = deque(maxlen=TMDB_RATE_LIMIT_REQUESTS + 10)
+_tmdb_rate_limit_lock = threading.Lock()
+# S01E01-style pattern for matching episode numbers
+_episode_pattern = re.compile(r"(?i)s(\d+)e(\d+)")
+
 
 def get_channels():
     """Load channels from /data/channels.json."""
@@ -236,6 +257,264 @@ def get_programme_name_from_path(path: Path) -> str:
     return path.stem
 
 
+def _parse_season_episode(stem: str) -> tuple[int, int] | None:
+    """Parse S01E01-style from filename stem. Returns (season, episode) or None."""
+    m = _episode_pattern.search(stem)
+    if m:
+        return int(m.group(1)), int(m.group(2))
+    return None
+
+
+def _tmdb_sanitize_key(name: str) -> str:
+    """Sanitize a string for use as a cache filename (no path separators or problematic chars)."""
+    s = re.sub(r'[^\w\-.\s]', "", str(name))
+    return re.sub(r"\s+", "_", s).strip("_") or "unknown"
+
+
+def _tmdb_cache_path(subdir: str, key: str) -> Path:
+    """Path to a cache file: TMDB_CACHE_DIR/subdir/key.json."""
+    TMDB_CACHE_DIR.mkdir(parents=True, exist_ok=True)
+    (TMDB_CACHE_DIR / subdir).mkdir(exist_ok=True)
+    return TMDB_CACHE_DIR / subdir / f"{key}.json"
+
+
+def _tmdb_read_disk_cache(subdir: str, key: str) -> dict | None:
+    """Read cached JSON from disk. Returns None if missing or invalid."""
+    path = _tmdb_cache_path(subdir, key)
+    if not path.exists():
+        return None
+    try:
+        with open(path, encoding="utf-8") as f:
+            return json.load(f)
+    except (OSError, json.JSONDecodeError):
+        return None
+
+
+def _tmdb_write_disk_cache(subdir: str, key: str, data: dict) -> None:
+    """Write JSON to disk cache."""
+    path = _tmdb_cache_path(subdir, key)
+    try:
+        with open(path, "w", encoding="utf-8") as f:
+            json.dump(data, f, ensure_ascii=False, indent=0)
+    except OSError:
+        pass
+
+
+def _tmdb_wait_rate_limit() -> None:
+    """Block until we are allowed to make another TMDB request (40 per 10s)."""
+    with _tmdb_rate_limit_lock:
+        while True:
+            now = time.monotonic()
+            while _tmdb_request_times and now - _tmdb_request_times[0] >= TMDB_RATE_LIMIT_WINDOW_SEC:
+                _tmdb_request_times.popleft()
+            if len(_tmdb_request_times) < TMDB_RATE_LIMIT_REQUESTS:
+                break
+            wait = TMDB_RATE_LIMIT_WINDOW_SEC - (now - _tmdb_request_times[0])
+            if wait > 0:
+                time.sleep(min(0.5, wait))
+
+
+def _tmdb_record_request() -> None:
+    """Record that a TMDB request was just made."""
+    with _tmdb_rate_limit_lock:
+        _tmdb_request_times.append(time.monotonic())
+
+
+def _tmdb_request(path: str, params: dict | None = None) -> dict | None:
+    """GET TMDB API with API key and Dutch language. Rate-limited to 40 requests per 10s. Returns parsed JSON or None."""
+    api_key = os.environ.get("TMDB_API_KEY", "").strip()
+    if not api_key:
+        return None
+    _tmdb_wait_rate_limit()
+    q = {"api_key": api_key, "language": TMDB_LANGUAGE}
+    if params:
+        q.update(params)
+    url = f"{TMDB_API_BASE}{path}?{urlencode(q)}"
+    req = urllib.request.Request(url)
+    try:
+        with urllib.request.urlopen(req, timeout=15) as resp:
+            data = json.load(resp)
+            _tmdb_record_request()
+            return data
+    except (urllib.error.HTTPError, urllib.error.URLError, OSError, json.JSONDecodeError):
+        return None
+
+
+def _tmdb_search_series(name: str) -> tuple[int, str] | None:
+    """Search for a TV series by name. Returns (series_id, series_name_nl) or None. Uses memory, disk, then API (rate-limited)."""
+    now = time.time()
+    if name in _tmdb_series_cache:
+        sid, sname, ts = _tmdb_series_cache[name]
+        if now - ts < TMDB_CACHE_TTL_SEC:
+            return (sid, sname)
+    key = _tmdb_sanitize_key(name)
+    cached = _tmdb_read_disk_cache("series_search", key)
+    if cached is not None:
+        sid, sname = cached.get("series_id"), cached.get("series_name_nl")
+        if sid is not None and sname is not None:
+            _tmdb_series_cache[name] = (int(sid), sname, now)
+            return (int(sid), sname)
+    data = _tmdb_request("/search/tv", {"query": name})
+    if not data:
+        return None
+    results = (data.get("results") or [])
+    if not results:
+        return None
+    first = results[0]
+    sid = first.get("id")
+    sname = (first.get("name") or name).strip() or name
+    if sid is not None:
+        _tmdb_series_cache[name] = (int(sid), sname, now)
+        _tmdb_write_disk_cache("series_search", key, {"series_id": sid, "series_name_nl": sname})
+        return (int(sid), sname)
+    return None
+
+
+def _tmdb_get_series_details(series_id: int) -> dict | None:
+    """Fetch series details in Dutch. Returns {genres: [names], country: iso} or None. Uses memory, disk, then API."""
+    now = time.time()
+    if series_id in _tmdb_series_details_cache:
+        details, ts = _tmdb_series_details_cache[series_id]
+        if now - ts < TMDB_CACHE_TTL_SEC:
+            return details
+    key = str(series_id)
+    cached = _tmdb_read_disk_cache("series_details", key)
+    if cached is not None:
+        _tmdb_series_details_cache[series_id] = (cached, now)
+        return cached
+    data = _tmdb_request(f"/tv/{series_id}")
+    if not data:
+        return None
+    genres = [g.get("name") for g in (data.get("genres") or []) if g.get("name")]
+    countries = data.get("production_countries") or []
+    country = (countries[0].get("iso_3166_1") or "").upper() if countries else ""
+    details = {"genres": genres, "country": country}
+    _tmdb_series_details_cache[series_id] = (details, now)
+    _tmdb_write_disk_cache("series_details", key, details)
+    return details
+
+
+def _tmdb_get_episode(series_id: int, season: int, episode: int) -> dict | None:
+    """Fetch one episode in Dutch. Returns dict with name, overview, air_date or None. Uses memory, disk, then API."""
+    now = time.time()
+    key_tuple = (series_id, season, episode)
+    if key_tuple in _tmdb_episode_cache:
+        meta, ts = _tmdb_episode_cache[key_tuple]
+        if now - ts < TMDB_CACHE_TTL_SEC:
+            return meta
+    key = f"{series_id}_{season}_{episode}"
+    cached = _tmdb_read_disk_cache("episode", key)
+    if cached is not None:
+        _tmdb_episode_cache[key_tuple] = (cached, now)
+        return cached
+    data = _tmdb_request(f"/tv/{series_id}/season/{season}/episode/{episode}")
+    if not data:
+        return None
+    name = (data.get("name") or "").strip()
+    overview = (data.get("overview") or "").strip()
+    air_date = (data.get("air_date") or "").strip()  # YYYY-MM-DD
+    meta = {"name": name, "overview": overview, "air_date": air_date}
+    _tmdb_episode_cache[key_tuple] = (meta, now)
+    _tmdb_write_disk_cache("episode", key, meta)
+    return meta
+
+
+def _tmdb_get_episode_credits(series_id: int, season: int, episode: int) -> dict | None:
+    """Fetch episode credits. Returns {directors, actors, producers} or None. Uses memory, disk, then API."""
+    now = time.time()
+    key_tuple = (series_id, season, episode)
+    if key_tuple in _tmdb_episode_credits_cache:
+        cred, ts = _tmdb_episode_credits_cache[key_tuple]
+        if now - ts < TMDB_CACHE_TTL_SEC:
+            return cred
+    key = f"{series_id}_{season}_{episode}"
+    cached = _tmdb_read_disk_cache("episode_credits", key)
+    if cached is not None:
+        _tmdb_episode_credits_cache[key_tuple] = (cached, now)
+        return cached
+    data = _tmdb_request(f"/tv/{series_id}/season/{season}/episode/{episode}/credits")
+    if not data:
+        return None
+    directors = []
+    producers = []
+    for c in data.get("crew") or []:
+        job = (c.get("job") or "").strip()
+        name = (c.get("name") or "").strip()
+        if not name:
+            continue
+        if job == "Director":
+            directors.append(name)
+        elif job in ("Producer", "Executive Producer", "Co-Executive Producer", "Supervising Producer", "Co-Producer"):
+            producers.append(name)
+    actors = [(c.get("name") or "").strip() for c in (data.get("cast") or []) if (c.get("name") or "").strip()]
+    directors = list(dict.fromkeys(directors))
+    producers = list(dict.fromkeys(producers))
+    cred = {"directors": directors, "actors": actors, "producers": producers}
+    _tmdb_episode_credits_cache[key_tuple] = (cred, now)
+    _tmdb_write_disk_cache("episode_credits", key, cred)
+    return cred
+
+
+def _air_date_to_xmltv_date(air_date: str) -> str:
+    """Convert TMDB air_date (YYYY-MM-DD) to XMLTV date (YYYYMMDD)."""
+    if not air_date or len(air_date) < 10:
+        return ""
+    return air_date.replace("-", "")[:8]  # YYYYMMDD
+
+
+def get_episode_metadata(programme_name: str, path: Path) -> dict | None:
+    """
+    Return TMDB metadata for this series episode in Dutch: title, sub_title, desc, credits, date, categories, country, episode_nums.
+    Only for series under SERIES_ROOT; parses S01E01 from path.stem. Returns None if disabled or not found.
+    """
+    if SERIES_ROOT not in path.parents:
+        return None
+    if not os.environ.get("TMDB_API_KEY", "").strip():
+        return None
+    se = _parse_season_episode(path.stem)
+    if not se:
+        return None
+    season_num, episode_num = se
+    hit = _tmdb_search_series(programme_name)
+    if not hit:
+        return None
+    series_id, series_name_nl = hit
+    ep = _tmdb_get_episode(series_id, season_num, episode_num)
+    if not ep:
+        return None
+    sub_title = ep.get("name") or path.stem
+    overview = (ep.get("overview") or "").strip() or None
+    air_date = ep.get("air_date") or ""
+    date_str = _air_date_to_xmltv_date(air_date) or "20090101"  # fallback if missing
+
+    details = _tmdb_get_series_details(series_id)
+    genres = (details.get("genres") or []) if details else []
+    country = (details.get("country") or "") if details else ""
+
+    credits = _tmdb_get_episode_credits(series_id, season_num, episode_num)
+    directors = (credits.get("directors") or []) if credits else []
+    actors = (credits.get("actors") or []) if credits else []
+    producers = (credits.get("producers") or []) if credits else []
+
+    # xmltv_ns: season.episode.part/parts (single part = 0/1)
+    xmltv_ns = f"{season_num}.{episode_num}.0/1"
+    onscreen = f"S{season_num:02d}E{episode_num:02d}"
+
+    return {
+        "title": series_name_nl,
+        "sub_title": sub_title,
+        "desc": overview,
+        "lang": TMDB_LANGUAGE,
+        "directors": directors,
+        "actors": actors,
+        "producers": producers,
+        "date": date_str,
+        "categories": genres,
+        "country": country,
+        "episode_nums": {"xmltv_ns": xmltv_ns, "onscreen": onscreen},
+    }
+
+
 def build_epg_xml() -> str:
     """
     Build XMLTV EPG for all channels. Schedule repeats from midnight UTC (same
@@ -288,8 +567,43 @@ def build_epg_xml() -> str:
                     channel=cid,
                 )
                 programme_name = get_programme_name_from_path(path)
-                ET.SubElement(prog, "title", lang="en").text = programme_name
-                ET.SubElement(prog, "sub-title", lang="en").text = path.stem
+                meta = get_episode_metadata(programme_name, path)
+                lang = (meta.get("lang") or "en") if meta else "en"
+                if meta:
+                    ET.SubElement(prog, "title", lang=lang).text = meta["title"]
+                    ET.SubElement(prog, "sub-title", lang=lang).text = meta["sub_title"]
+                    if meta.get("desc"):
+                        ET.SubElement(prog, "desc", lang=lang).text = meta["desc"]
+                    directors = meta.get("directors") or []
+                    actors = meta.get("actors") or []
+                    producers = meta.get("producers") or []
+                    if directors or actors or producers:
+                        credits_el = ET.SubElement(prog, "credits")
+                        for d in directors:
+                            ET.SubElement(credits_el, "director").text = d
+                        for a in actors:
+                            ET.SubElement(credits_el, "actor").text = a
+                        for p in producers:
+                            ET.SubElement(credits_el, "producer").text = p
+                    if meta.get("date"):
+                        ET.SubElement(prog, "date").text = meta["date"]
+                    for cat in meta.get("categories") or []:
+                        ET.SubElement(prog, "category", lang=lang).text = cat
+                    if meta.get("country"):
+                        ET.SubElement(prog, "country", lang=lang).text = meta["country"]
+                    nums = meta.get("episode_nums") or {}
+                    if nums.get("xmltv_ns"):
+                        ET.SubElement(prog, "episode-num", system="xmltv_ns").text = nums["xmltv_ns"]
+                    if nums.get("onscreen"):
+                        ET.SubElement(prog, "episode-num", system="onscreen").text = nums["onscreen"]
+                else:
+                    ET.SubElement(prog, "title", lang="en").text = programme_name
+                    ET.SubElement(prog, "sub-title", lang="en").text = path.stem
+                    se = _parse_season_episode(path.stem)
+                    if se:
+                        s, e = se
+                        ET.SubElement(prog, "episode-num", system="xmltv_ns").text = f"{s}.{e}.0/1"
+                        ET.SubElement(prog, "episode-num", system="onscreen").text = f"S{s:02d}E{e:02d}"
             cycle_start += cycle_delta
 
     ET.indent(root, space="  ")