with epg

Dockers/self-hosted-iptv/main.py

@@ -8,12 +8,21 @@ import json
 import random
 import subprocess
 import tempfile
+import time
+import xml.etree.ElementTree as ET
 from collections import defaultdict
+from datetime import datetime, timezone, timedelta
 from pathlib import Path
 from http.server import HTTPServer, BaseHTTPRequestHandler
 from urllib.parse import urlparse, parse_qs, unquote
 import mimetypes
 
+# Per-channel schedule cache: channel_id -> (paths, durations). TTL 24h so EPG and live stay in sync.
+_schedule_cache: dict[str, tuple[list[Path], list[float], float]] = {}
+SCHEDULE_CACHE_TTL_SEC = 24 * 3600
+EPG_DAYS = 14
+DEFAULT_DURATION_SEC = 3600
+
 MOVIES_ROOT = Path("/movies")
 SERIES_ROOT = Path("/series")
 DATA_ROOT = Path("/data")
@@ -99,6 +108,50 @@ def build_channel_path_list(path_keys: list[str]) -> list[Path]:
     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."""
+    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():
+            return float(out.stdout.strip())
+    except (subprocess.TimeoutExpired, ValueError, FileNotFoundError):
+        pass
+    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:
+            return paths, durations
+    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]
+    _schedule_cache[channel_id] = (paths, durations, now)
+    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).
@@ -154,6 +207,71 @@ def get_base_url(handler: BaseHTTPRequestHandler) -> str:
     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 build_epg_xml() -> str:
+    """
+    Build XMLTV EPG for all channels. Schedule repeats from midnight UTC;
+    programmes generated for EPG_DAYS. Uses cached schedule per channel.
+    """
+    channels = get_channels()
+    if not channels:
+        return '<?xml version="1.0" encoding="UTF-8"?>\n<tv></tv>'
+
+    now = datetime.now(timezone.utc)
+    epoch = now.replace(hour=0, minute=0, second=0, microsecond=0)
+    end_epoch = epoch + timedelta(days=EPG_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
+
+    for ch in channels:
+        cid = ch.get("id", ch.get("name", ""))
+        schedule = get_or_build_schedule(cid)
+        if not schedule:
+            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)
+
+        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
+                prog = ET.SubElement(root, "programme",
+                    start=_format_xmltv_time(start_dt),
+                    stop=_format_xmltv_time(stop_dt),
+                    channel=cid,
+                )
+                title = path.stem
+                ET.SubElement(prog, "title", lang="en").text = title
+            cycle_start += cycle_delta
+
+    ET.indent(root, space="  ")
+    return '<?xml version="1.0" encoding="UTF-8"?>\n' + ET.tostring(root, encoding="unicode", default_namespace=None)
+
+
 def write_concat_list(paths: list[Path], fd) -> None:
     """Write FFmpeg concat demuxer list to file. Escapes paths for safety."""
     for p in paths:
@@ -234,13 +352,18 @@ class IPTVHandler(BaseHTTPRequestHandler):
             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>Add this playlist URL to your IPTV client:</p>"
+        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", ""))
@@ -290,14 +413,23 @@ class IPTVHandler(BaseHTTPRequestHandler):
         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)."""
-        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")
+        """Stream channel as continuous MPEG-TS (for Plex DVR). Uses cached 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 = build_channel_path_list(channel.get("paths", []))
+        paths, _ = schedule
         if not paths:
             self.send_error(404, "No videos in channel")
             return