956 lines
37 KiB
Python
956 lines
37 KiB
Python
#!/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
|
|
|
|
|
|
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."""
|
|
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://<host>:{port}/playlist.m3u)")
|
|
server.serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|