projects/Dockers/self-hosted-iptv/main.py

#!/usr/bin/env python3
"""
Self-hosted IPTV: serves M3U playlists that schedule video files from /movies and /series
with randomized order so the same show never appears twice in a row.
"""

import json
import logging
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, 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, urlencode
import mimetypes

logger = logging.getLogger(__name__)

# Per-channel schedule cache: channel_id -> (paths, durations). TTL = (timespan - 3h) so a new span is ready before the current one ends.
_schedule_cache: dict[str, tuple[list[Path], list[float], float]] = {}
# Parsed from EPG_TIMESPAN (e.g. "2d", "14d", "48h"). Default 14d. Schedule refreshes after (timespan - 3h).
EPG_TIMESPAN_DAYS = 14.0
EPG_TIMESPAN_SEC = 14.0 * 24 * 3600
SCHEDULE_CACHE_TTL_SEC = (14 * 24 - 3) * 3600  # 14d - 3h
DEFAULT_DURATION_SEC = 3600


def _parse_epg_timespan(value: str) -> tuple[float, float]:
    """Parse EPG_TIMESPAN env (e.g. '2d', '14d', '48h'). Returns (days, seconds). Default (14, 14*24*3600) on error."""
    value = (value or "").strip().lower()
    if not value:
        return 14.0, 14.0 * 24 * 3600
    m = re.match(r"^(\d+(?:\.\d+)?)\s*([dh])?$", value)
    if not m:
        return 14.0, 14.0 * 24 * 3600
    num = float(m.group(1))
    unit = (m.group(2) or "d")
    if unit == "h":
        days = num / 24.0
    else:
        days = num
    sec = days * 24 * 3600
    return days, sec


def _init_epg_timespan() -> None:
    """Load EPG_TIMESPAN from env and set EPG_* and SCHEDULE_CACHE_TTL_SEC."""
    global EPG_TIMESPAN_DAYS, EPG_TIMESPAN_SEC, SCHEDULE_CACHE_TTL_SEC
    days, sec = _parse_epg_timespan(os.environ.get("EPG_TIMESPAN", ""))
    EPG_TIMESPAN_DAYS = days
    EPG_TIMESPAN_SEC = sec
    # New schedule is generated after (timespan - 3 hours); minimum 1 hour TTL
    schedule_ttl = sec - 3 * 3600
    SCHEDULE_CACHE_TTL_SEC = max(3600, schedule_ttl)


_init_epg_timespan()

MOVIES_ROOT = Path("/movies")
SERIES_ROOT = Path("/series")
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_IMAGE_BASE = "https://image.tmdb.org/t/p"
TMDB_IMAGE_STILL_SIZE = "w500"  # episode still width for EPG icon
_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."""
    if not CHANNELS_FILE.exists():
        return []
    with open(CHANNELS_FILE, encoding="utf-8") as f:
        data = json.load(f)
    return data.get("channels", [])


def resolve_path(path_key: str) -> Path | None:
    """Resolve a channel path key to an absolute Path. Returns None if invalid."""
    path_key = path_key.strip("/")
    if path_key == "movies":
        return MOVIES_ROOT
    if path_key.startswith("series/"):
        subpath = path_key[7:]  # len("series/")
        return SERIES_ROOT / subpath
    return None


def collect_videos(root: Path) -> list[tuple[str, Path]]:
    """Collect all video files under root. Returns list of (path_key, Path)."""
    if not root.exists() or not root.is_dir():
        return []
    videos = []
    try:
        for p in root.rglob("*"):
            if p.is_file() and p.suffix.lower() in VIDEO_EXTENSIONS:
                videos.append(p)
    except OSError:
        pass
    return videos


def collect_channel_videos(path_keys: list[str]) -> list[tuple[str, Path]]:
    """
    For each path key (e.g. 'movies', 'series/Breaking Bad'), collect videos.
    Returns list of (path_key, absolute_path) so we can group by path_key (show).
    """
    out = []
    for key in path_keys:
        root = resolve_path(key)
        if root is None:
            continue
        for vid in collect_videos(root):
            out.append((key, vid))
    return out


def shuffle_no_adjacent_same_show(items: list[tuple[str, Path]]) -> list[Path]:
    """
    Shuffle so that the same path_key (show) never appears twice in a row.
    Groups by path_key, shuffles each group, then interleaves round-robin.
    """
    if not items:
        return []
    by_key = defaultdict(list)
    for key, path in items:
        by_key[key].append(path)
    for key in by_key:
        random.shuffle(by_key[key])
    # Round-robin interleave so no two from same key are adjacent
    groups = list(by_key.values())
    result = []
    n = max(len(g) for g in groups)
    for i in range(n):
        for g in groups:
            if i < len(g):
                result.append(g[i])
    return result


def build_channel_path_list(path_keys: list[str]) -> list[Path]:
    """Build randomized list of absolute video paths for a channel (no same show adjacent)."""
    items = collect_channel_videos(path_keys)
    return shuffle_no_adjacent_same_show(items)


def get_duration(path: Path) -> float:
    """Return duration in seconds via ffprobe. Uses DEFAULT_DURATION_SEC on failure."""
    t0 = time.monotonic()
    try:
        out = subprocess.run(
            [
                "ffprobe",
                "-v", "error",
                "-show_entries", "format=duration",
                "-of", "default=noprint_wrappers=1:nokey=1",
                str(path),
            ],
            capture_output=True,
            text=True,
            timeout=30,
        )
        if out.returncode == 0 and out.stdout.strip():
            d = float(out.stdout.strip())
            logger.debug("ffprobe %s -> %.0fs in %.2fs", path.name, d, time.monotonic() - t0)
            return d
    except (subprocess.TimeoutExpired, ValueError, FileNotFoundError) as e:
        logger.debug("ffprobe %s failed: %s", path.name, e)
    return DEFAULT_DURATION_SEC


def get_or_build_schedule(channel_id: str) -> tuple[list[Path], list[float]] | None:
    """
    Return (paths, durations) for the channel, so live stream and EPG use the same schedule.
    Cached for SCHEDULE_CACHE_TTL_SEC.
    """
    now = time.time()
    if channel_id in _schedule_cache:
        paths, durations, ts = _schedule_cache[channel_id]
        if now - ts < SCHEDULE_CACHE_TTL_SEC:
            logger.debug("schedule cache hit for channel %s (%s videos)", channel_id, len(paths))
            return paths, durations
    logger.info("Building schedule for channel %s ...", channel_id)
    t0 = time.monotonic()
    channels = get_channels()
    channel = next((c for c in channels if c.get("id") == channel_id), None)
    if not channel:
        return None
    paths = build_channel_path_list(channel.get("paths", []))
    if not paths:
        return None
    durations = [get_duration(p) for p in paths]
    elapsed = time.monotonic() - t0
    _schedule_cache[channel_id] = (paths, durations, now)
    logger.info("Schedule for channel %s built in %.1fs (%s videos)", channel_id, elapsed, len(paths))
    return paths, durations


def build_playlist(path_keys: list[str], base_url: str) -> list[tuple[str, str]]:
    """
    Build playlist as list of (display_name, stream_url).
    base_url is the base URL for stream links (e.g. http://host:port).
    """
    paths = build_channel_path_list(path_keys)
    playlist = []
    for p in paths:
        try:
            if MOVIES_ROOT in p.parents:
                rel = p.relative_to(MOVIES_ROOT)
                path_str = "movies/" + str(rel).replace("\\", "/")
            else:
                rel = p.relative_to(SERIES_ROOT)
                path_str = "series/" + str(rel).replace("\\", "/")
        except ValueError:
            path_str = "movies/" + p.name
        stream_url = f"{base_url}/stream?path={path_str}"
        display = p.stem
        playlist.append((display, stream_url))
    return playlist


# Preferred audio languages for streaming: Dutch/Flemish first, then English, then default (first audio).
PREFERRED_AUDIO_LANGUAGES = ("nld", "dut", "eng")  # ISO 639-2: nld/dut = Dutch, eng = English


def get_preferred_audio_stream_index(path: Path) -> int | None:
    """
    Probe the file with ffprobe and return the 0-based audio stream index to use:
    first Dutch/Flemish (nld/dut), else first English (eng), else first audio stream.
    Returns None on probe failure or no audio streams.
    """
    try:
        out = subprocess.run(
            [
                "ffprobe",
                "-v", "quiet",
                "-print_format", "json",
                "-show_entries", "stream=index,codec_type",
                "-show_entries", "stream_tags=language",
                str(path),
            ],
            capture_output=True,
            text=True,
            timeout=10,
        )
        if out.returncode != 0 or not out.stdout:
            return None
        data = json.loads(out.stdout)
        streams = data.get("streams") or []
        audio_streams = []  # (index, language or "")
        for s in streams:
            if s.get("codec_type") != "audio":
                continue
            idx = s.get("index")
            if idx is None:
                continue
            tags = s.get("tags") or {}
            lang = (tags.get("language") or "").strip().lower()[:3]
            audio_streams.append((idx, lang))
        if not audio_streams:
            return None
        for pref in PREFERRED_AUDIO_LANGUAGES:
            for idx, lang in audio_streams:
                if lang == pref:
                    return idx
        return audio_streams[0][0]
    except (subprocess.TimeoutExpired, ValueError, FileNotFoundError, json.JSONDecodeError):
        return None


def stream_file_with_preferred_audio(path: Path, wfile) -> bool:
    """
    Stream the file via FFmpeg with only video and the preferred audio track (Dutch/Flemish, else English, else first).
    Returns True if streaming was started, False if caller should fall back to raw file.
    """
    audio_idx = get_preferred_audio_stream_index(path)
    if audio_idx is None:
        return False
    try:
        proc = subprocess.Popen(
            [
                "ffmpeg",
                "-i", str(path),
                "-map", "0:v",
                "-map", f"0:{audio_idx}",
                "-c", "copy",
                "-f", "mp4",
                "-movflags", "frag_keyframe+empty_moov+default_base_moof",
                "-",
            ],
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )
        while True:
            chunk = proc.stdout.read(65536)
            if not chunk:
                break
            wfile.write(chunk)
            wfile.flush()
        proc.wait(timeout=1)
    except (OSError, subprocess.TimeoutExpired):
        try:
            proc.terminate()
            proc.wait(timeout=2)
        except Exception:
            pass
        return False
    return True


def path_to_absolute(path_param: str) -> Path | None:
    """
    Convert a path query parameter to an absolute Path under /movies or /series.
    Path must start with 'movies/' or 'series/'. Prevents path traversal.
    """
    path_param = unquote(path_param).strip("/")
    if ".." in path_param or path_param.startswith("/"):
        return None
    if path_param.startswith("movies/"):
        sub = path_param[7:]
        root = MOVIES_ROOT
    elif path_param.startswith("series/"):
        sub = path_param[7:]
        root = SERIES_ROOT
    else:
        return None
    candidate = (root / sub).resolve()
    try:
        candidate.relative_to(root)
    except ValueError:
        return None
    if candidate.exists() and candidate.is_file() and candidate.suffix.lower() in VIDEO_EXTENSIONS:
        return candidate
    return None


def get_base_url(handler: BaseHTTPRequestHandler) -> str:
    """Build base URL for playlist links from request."""
    host = handler.headers.get("Host", "localhost:8080")
    return f"http://{host}"


def _format_xmltv_time(dt: datetime) -> str:
    """Format as XMLTV: YYYYMMDDHHmmss +0000"""
    tz = dt.strftime("%z") if dt.tzinfo else "+0000"
    return dt.strftime("%Y%m%d%H%M%S ") + tz


def get_programme_name_from_path(path: Path) -> str:
    """
    Return the series/movie name from the folder (e.g. series/W817/... -> W817,
    movies/MyMovie/... -> MyMovie). If the file is directly under root, use stem.
    """
    try:
        if SERIES_ROOT in path.parents:
            rel = path.relative_to(SERIES_ROOT)
            parts = rel.parts
            if len(parts) > 1:
                return parts[0]
            return path.stem
        if MOVIES_ROOT in path.parents:
            rel = path.relative_to(MOVIES_ROOT)
            parts = rel.parts
            if len(parts) > 1:
                return parts[0]
            return path.stem
    except ValueError:
        pass
    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:
                logger.debug("TMDB rate limit: waiting %.1fs (%s/40 in window)", wait, len(_tmdb_request_times))
                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)
    logger.debug("TMDB request: %s", path)
    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) as e:
        logger.debug("TMDB request failed %s: %s", path, e)
        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:
            logger.debug("TMDB series_search %s: memory hit -> %s", name, sid)
            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)
            logger.debug("TMDB series_search %s: disk hit -> %s", name, sid)
            return (int(sid), sname)
    logger.debug("TMDB series_search %s: API request", name)
    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:
            logger.debug("TMDB series_details %s: memory hit", series_id)
            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)
        logger.debug("TMDB series_details %s: disk hit", series_id)
        return cached
    logger.debug("TMDB series_details %s: API request", series_id)
    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:
            logger.debug("TMDB episode %s s%s e%s: memory hit", series_id, season, episode)
            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)
        logger.debug("TMDB episode %s s%s e%s: disk hit", series_id, season, episode)
        return cached
    logger.debug("TMDB episode %s s%s e%s: API request", series_id, season, episode)
    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
    still_path = (data.get("still_path") or "").strip()
    if still_path and not still_path.startswith("/"):
        still_path = "/" + still_path
    meta = {"name": name, "overview": overview, "air_date": air_date, "still_path": still_path or None}
    _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:
            logger.debug("TMDB episode_credits %s s%s e%s: memory hit", series_id, season, episode)
            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)
        logger.debug("TMDB episode_credits %s s%s e%s: disk hit", series_id, season, episode)
        return cached
    logger.debug("TMDB episode_credits %s s%s e%s: API request", series_id, season, episode)
    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

    still_path = ep.get("still_path") or ""
    if still_path and still_path.startswith("/"):
        icon_url = f"{TMDB_IMAGE_BASE}/{TMDB_IMAGE_STILL_SIZE}{still_path}"
    else:
        icon_url = None

    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,
        "icon": icon_url,
        "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
    epoch as live stream). Programmes generated for EPG_TIMESPAN_DAYS. Uses cached schedule.
    """
    epg_start = time.monotonic()
    channels = get_channels()
    if not channels:
        return '<?xml version="1.0" encoding="UTF-8"?>\n<tv></tv>'

    logger.info("EPG build started (%s channels)", len(channels))
    now = datetime.now(timezone.utc)
    epoch = _schedule_epoch_utc()
    end_epoch = epoch + timedelta(days=EPG_TIMESPAN_DAYS)

    root = ET.Element("tv")
    root.set("source-info-name", "self-hosted-iptv")
    root.set("generator-info-name", "self-hosted-iptv")

    for ch in channels:
        cid = ch.get("id", ch.get("name", ""))
        name = ch.get("name", cid)
        chan_el = ET.SubElement(root, "channel", id=cid)
        ET.SubElement(chan_el, "display-name").text = name

    total_programmes = 0
    for ch in channels:
        cid = ch.get("id", ch.get("name", ""))
        ch_start = time.monotonic()
        schedule = get_or_build_schedule(cid)
        if not schedule:
            logger.debug("EPG channel %s: no schedule", cid)
            continue
        paths, durations = schedule
        cycle_duration_sec = sum(durations)
        if cycle_duration_sec <= 0:
            continue
        cycle_delta = timedelta(seconds=cycle_duration_sec)
        offsets_sec = [0.0]
        for d in durations[:-1]:
            offsets_sec.append(offsets_sec[-1] + d)

        channel_programmes = 0
        cycle_start = epoch
        while cycle_start < end_epoch:
            for i, (path, dur) in enumerate(zip(paths, durations)):
                start_dt = cycle_start + timedelta(seconds=offsets_sec[i])
                stop_dt = start_dt + timedelta(seconds=dur)
                if stop_dt <= now:
                    continue
                if start_dt >= end_epoch:
                    break
                channel_programmes += 1
                prog = ET.SubElement(root, "programme",
                    start=_format_xmltv_time(start_dt),
                    stop=_format_xmltv_time(stop_dt),
                    channel=cid,
                )
                programme_name = get_programme_name_from_path(path)
                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"]
                    if meta.get("icon"):
                        ET.SubElement(prog, "icon", attrib={"src": meta["icon"]})
                    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

        ch_elapsed = time.monotonic() - ch_start
        total_programmes += channel_programmes
        logger.info("EPG channel %s: %s programmes in %.1fs", cid, channel_programmes, ch_elapsed)

    ET.indent(root, space="  ")
    epg_elapsed = time.monotonic() - epg_start
    logger.info("EPG build finished in %.1fs (%s programmes total)", epg_elapsed, total_programmes)
    return '<?xml version="1.0" encoding="UTF-8"?>\n' + ET.tostring(root, encoding="unicode", default_namespace=None)


def _escape_path(p: Path) -> str:
    return str(p.resolve()).replace("\\", "\\\\").replace("'", "'\\''")


def write_concat_list(paths: list[Path], fd) -> None:
    """Write FFmpeg concat demuxer list to file. Escapes paths for safety."""
    for p in paths:
        fd.write(f"file '{_escape_path(p)}'\n")
    fd.flush()


def _schedule_epoch_utc() -> datetime:
    """Midnight UTC today — same reference used by EPG and live stream."""
    return datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)


def _current_position_in_cycle(durations: list[float]) -> tuple[int, float]:
    """
    Return (file_index, offset_sec_in_file) for 'now' in the schedule cycle.
    Schedule epoch = midnight UTC; cycle = one full playlist.
    """
    epoch = _schedule_epoch_utc()
    now = datetime.now(timezone.utc)
    cycle_duration = sum(durations)
    if cycle_duration <= 0:
        return 0, 0.0
    offset_sec = (now - epoch).total_seconds() % cycle_duration
    cumul = 0.0
    for i, dur in enumerate(durations):
        if offset_sec < cumul + dur:
            return i, offset_sec - cumul
        cumul += dur
    return len(durations) - 1, durations[-1]


def write_concat_list_from_current(paths: list[Path], durations: list[float], fd) -> None:
    """
    Write a concat list that starts at the current position in the schedule and
    loops seamlessly. Uses inpoint/outpoint so we join mid-file, then repeat.
    Cycle: current file (offset→end), next files…, start…current-1, current file (0→offset).
    """
    n = len(paths)
    if n == 0:
        return
    idx, offset_in_file = _current_position_in_cycle(durations)
    for i in range(n):
        j = (idx + i) % n
        fd.write(f"file '{_escape_path(paths[j])}'\n")
        if i == 0 and offset_in_file >= 0.5:
            fd.write(f"inpoint {offset_in_file:.2f}\n")
    if offset_in_file >= 0.5:
        fd.write(f"file '{_escape_path(paths[idx])}'\n")
        fd.write(f"outpoint {offset_in_file:.2f}\n")
    fd.flush()


def stream_live_channel(paths: list[Path], wfile, durations: list[float]) -> None:
    """
    Run FFmpeg to output a continuous MPEG-TS stream. Concat list starts at
    current position in schedule (midnight UTC cycle) so each new client joins
    where the 'channel' is now; then the list loops. EPG and stream stay in sync.
    """
    if not paths:
        return

    with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
        write_concat_list_from_current(paths, durations, f)
        list_path = f.name
    try:
        proc = subprocess.Popen(
            [
                "ffmpeg",
                "-stream_loop", "-1",
                "-f", "concat", "-safe", "0", "-i", list_path,
                "-c", "copy",
                "-f", "mpegts",
                "-",
            ],
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )
        try:
            while True:
                chunk = proc.stdout.read(65536)
                if not chunk:
                    break
                wfile.write(chunk)
                wfile.flush()
        except (BrokenPipeError, ConnectionResetError, OSError):
            pass
        finally:
            proc.terminate()
            try:
                proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                proc.kill()
    finally:
        Path(list_path).unlink(missing_ok=True)


class IPTVHandler(BaseHTTPRequestHandler):
    def log_message(self, format, *args):
        pass  # quiet by default; set to super().log_message for debugging

    def do_GET(self):
        parsed = urlparse(self.path)
        path = parsed.path.rstrip("/") or "/"
        qs = parse_qs(parsed.query)

        if path == "/" or path == "/index.html":
            self.send_index()
            return
        if path == "/playlist.m3u" or path == "/channels.m3u":
            self.send_channels_m3u()
            return
        if path.startswith("/channel/") and path.endswith("/playlist.m3u"):
            channel_id = path.split("/")[2]
            self.send_channel_m3u(channel_id)
            return
        if path == "/stream":
            path_param = qs.get("path", [""])[0]
            self.send_stream(path_param)
            return
        if path.startswith("/live/"):
            parts = path.split("/")
            if len(parts) == 3 and parts[1] == "live":
                self.send_live_stream(parts[2])
                return
        if path == "/epg.xml" or path == "/xmltv.xml":
            self.send_epg()
            return
        self.send_error(404, "Not found")

    def send_index(self):
        channels = get_channels()
        body = "<!DOCTYPE html><html><head><meta charset='utf-8'><title>Self-hosted IPTV</title></head><body>"
        body += "<h1>Self-hosted IPTV</h1><p>Playlist (M3U):</p>"
        body += f"<p><code>{get_base_url(self)}/playlist.m3u</code></p>"
        body += "<p>EPG (XMLTV, for Plex DVR):</p>"
        body += f"<p><code>{get_base_url(self)}/epg.xml</code></p>"
        body += "<h2>Channels</h2><ul>"
        for ch in channels:
            cid = ch.get("id", ch.get("name", ""))
            body += f"<li><a href='/channel/{cid}/playlist.m3u'>{ch.get('name', cid)}</a></li>"
        body += "</ul></body></html>"
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(body.encode("utf-8"))))
        self.end_headers()
        self.wfile.write(body.encode("utf-8"))

    def send_channels_m3u(self):
        """Master M3U for Plex DVR: one live stream URL per channel."""
        base = get_base_url(self)
        channels = get_channels()
        lines = ["#EXTM3U"]
        for ch in channels:
            cid = ch.get("id", ch.get("name", ""))
            name = ch.get("name", cid)
            lines.append(f'#EXTINF:-1,{name}')
            lines.append(f"{base}/live/{cid}")
        body = "\n".join(lines) + "\n"
        self.send_response(200)
        self.send_header("Content-Type", "application/x-mpegURL; charset=utf-8")
        self.send_header("Content-Length", str(len(body.encode("utf-8"))))
        self.end_headers()
        self.wfile.write(body.encode("utf-8"))

    def send_channel_m3u(self, channel_id: str):
        """Single channel M3U with randomized video entries (no same show adjacent)."""
        channels = get_channels()
        channel = next((c for c in channels if c.get("id") == channel_id), None)
        if not channel:
            self.send_error(404, "Channel not found")
            return
        path_keys = channel.get("paths", [])
        base = get_base_url(self)
        playlist = build_playlist(path_keys, base)
        lines = ["#EXTM3U"]
        for name, url in playlist:
            lines.append(f"#EXTINF:-1,{name}")
            lines.append(url)
        body = "\n".join(lines) + "\n"
        self.send_response(200)
        self.send_header("Content-Type", "application/x-mpegURL; charset=utf-8")
        self.send_header("Content-Length", str(len(body.encode("utf-8"))))
        self.end_headers()
        self.wfile.write(body.encode("utf-8"))

    def send_epg(self):
        """Serve XMLTV EPG (programme schedule for Plex DVR)."""
        body = build_epg_xml()
        self.send_response(200)
        self.send_header("Content-Type", "application/xml; charset=utf-8")
        self.send_header("Content-Length", str(len(body.encode("utf-8"))))
        self.send_header("Cache-Control", "public, max-age=300")
        self.end_headers()
        self.wfile.write(body.encode("utf-8"))

    def send_live_stream(self, channel_id: str):
        """Stream channel as continuous MPEG-TS (for Plex DVR). Starts at current position in schedule so EPG matches."""
        schedule = get_or_build_schedule(channel_id)
        if not schedule:
            self.send_error(404, "Channel not found or no videos")
            return
        paths, durations = schedule
        if not paths:
            self.send_error(404, "No videos in channel")
            return
        self.send_response(200)
        self.send_header("Content-Type", "video/MP2T")
        self.send_header("Cache-Control", "no-cache, no-store")
        self.end_headers()
        stream_live_channel(paths, self.wfile, durations)

    def send_stream(self, path_param: str):
        """Stream a video file. path is relative to /movies or /series. Prefers Dutch/Flemish audio, then English, then default."""
        abs_path = path_to_absolute(path_param)
        if abs_path is None:
            self.send_error(404, "File not found")
            return
        audio_idx = get_preferred_audio_stream_index(abs_path)
        if audio_idx is not None:
            self.send_response(200)
            self.send_header("Content-Type", "video/mp4")
            self.send_header("Cache-Control", "no-cache")
            self.end_headers()
            stream_file_with_preferred_audio(abs_path, self.wfile)
            return
        content_type, _ = mimetypes.guess_type(str(abs_path))
        content_type = content_type or "video/mp4"
        size = abs_path.stat().st_size
        self.send_response(200)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(size))
        self.send_header("Accept-Ranges", "bytes")
        self.end_headers()
        with open(abs_path, "rb") as f:
            self.wfile.write(f.read())


def main():
    log_level = getattr(logging, os.environ.get("LOG_LEVEL", "INFO").upper(), logging.INFO)
    logging.basicConfig(
        level=log_level,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )
    if not CHANNELS_FILE.exists():
        print(f"Warning: {CHANNELS_FILE} not found. Create it from data/channels.json.example")
    logger.info(
        "EPG timespan: %s days, schedule refresh after: %.1f h",
        EPG_TIMESPAN_DAYS,
        SCHEDULE_CACHE_TTL_SEC / 3600,
    )
    port = 8080
    server = HTTPServer(("0.0.0.0", port), IPTVHandler)
    print(f"Serving at http://0.0.0.0:{port}/  (playlist: http://<host>:{port}/playlist.m3u)")
    server.serve_forever()


if __name__ == "__main__":
    main()