#!/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_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 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 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: 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 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 epoch as live stream). Programmes generated for EPG_TIMESPAN_DAYS. Uses cached schedule. """ epg_start = time.monotonic() channels = get_channels() if not channels: return '\n' 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"] 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 '\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 = "Self-hosted IPTV" body += "

Self-hosted IPTV

Playlist (M3U):

" body += f"

{get_base_url(self)}/playlist.m3u

" body += "

EPG (XMLTV, for Plex DVR):

" body += f"

{get_base_url(self)}/epg.xml

" body += "

Channels

" 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.""" abs_path = path_to_absolute(path_param) if abs_path is None: self.send_error(404, "File not found") 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://:{port}/playlist.m3u)") server.serve_forever() if __name__ == "__main__": main()