From db39c420c0dc47dad46dbfd0bb90d5fdb9292bfb Mon Sep 17 00:00:00 2001 From: Bram Date: Tue, 10 Feb 2026 23:44:22 +0100 Subject: [PATCH] lets try this --- Dockers/self-hosted-iptv/Dockerfile | 13 + Dockers/self-hosted-iptv/README.md | 41 +++ .../data/channels.json.example | 14 + Dockers/self-hosted-iptv/main.py | 257 ++++++++++++++++++ 4 files changed, 325 insertions(+) create mode 100644 Dockers/self-hosted-iptv/Dockerfile create mode 100644 Dockers/self-hosted-iptv/README.md create mode 100644 Dockers/self-hosted-iptv/data/channels.json.example create mode 100644 Dockers/self-hosted-iptv/main.py diff --git a/Dockers/self-hosted-iptv/Dockerfile b/Dockers/self-hosted-iptv/Dockerfile new file mode 100644 index 0000000..c39e322 --- /dev/null +++ b/Dockers/self-hosted-iptv/Dockerfile @@ -0,0 +1,13 @@ +FROM python:3.12-slim + +WORKDIR /app + +COPY main.py . + +# /movies and /series are mounted by the user (video libraries) +# /data is mounted for channels.json +VOLUME ["/movies", "/series", "/data"] + +EXPOSE 8080 + +CMD ["python", "-u", "main.py"] diff --git a/Dockers/self-hosted-iptv/README.md b/Dockers/self-hosted-iptv/README.md new file mode 100644 index 0000000..10e4fd8 --- /dev/null +++ b/Dockers/self-hosted-iptv/README.md @@ -0,0 +1,41 @@ +# Self-hosted IPTV + +Serves M3U playlists that randomly schedule video files from `/movies` and `/series`, with the same show never appearing twice in a row. + +## Setup + +1. Create `/data/channels.json` (see `data/channels.json.example`). + +**channels.json format:** + +- `channels`: array of channel objects +- Each channel: + - `id`: unique id (used in URLs) + - `name`: display name + - `paths`: list of path keys to include: + - `"movies"` — all videos under `/movies` + - `"series/Show Name"` — all videos under `/series/Show Name` + +2. Run the container with volumes: + +```bash +docker build -t self-hosted-iptv . +docker run -d --name iptv -p 8080:8080 \ + -v /path/to/movies:/movies:ro \ + -v /path/to/series:/series:ro \ + -v /path/to/data:/data:ro \ + self-hosted-iptv +``` + +3. In your IPTV client, add playlist URL: `http://:8080/playlist.m3u` + +## Endpoints + +- `GET /` — simple web index with playlist link +- `GET /playlist.m3u` — master M3U (all channels) +- `GET /channel//playlist.m3u` — single channel M3U (randomized, no adjacent same show) +- `GET /stream?path=movies/...` or `path=series/...` — stream a video file + +## Shuffle behaviour + +Videos are grouped by path (e.g. one group per show or the whole movies folder). Each group is shuffled, then items are interleaved round-robin so the same show never plays back-to-back. diff --git a/Dockers/self-hosted-iptv/data/channels.json.example b/Dockers/self-hosted-iptv/data/channels.json.example new file mode 100644 index 0000000..ad5e085 --- /dev/null +++ b/Dockers/self-hosted-iptv/data/channels.json.example @@ -0,0 +1,14 @@ +{ + "channels": [ + { + "id": "comedy", + "name": "Comedy", + "paths": ["movies", "series/The Office", "series/Parks and Recreation"] + }, + { + "id": "drama", + "name": "Drama", + "paths": ["movies", "series/Breaking Bad", "series/Better Call Saul"] + } + ] +} diff --git a/Dockers/self-hosted-iptv/main.py b/Dockers/self-hosted-iptv/main.py new file mode 100644 index 0000000..367bd79 --- /dev/null +++ b/Dockers/self-hosted-iptv/main.py @@ -0,0 +1,257 @@ +#!/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 random +from collections import defaultdict +from pathlib import Path +from http.server import HTTPServer, BaseHTTPRequestHandler +from urllib.parse import urlparse, parse_qs, unquote +import mimetypes + +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"} + + +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_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). + """ + items = collect_channel_videos(path_keys) + paths = shuffle_no_adjacent_same_show(items) + 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}" + + +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 + self.send_error(404, "Not found") + + def send_index(self): + channels = get_channels() + body = "Self-hosted IPTV" + body += "

Self-hosted IPTV

Add this playlist URL to your IPTV client:

" + body += f"

{get_base_url(self)}/playlist.m3u

" + body += "

Channels

    " + for ch in channels: + cid = ch.get("id", ch.get("name", "")) + body += f"
  • {ch.get('name', cid)}
    • " + body += "
" + 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 listing all channels (one URL per channel playlist).""" + 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}/channel/{cid}/playlist.m3u") + 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_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(): + if not CHANNELS_FILE.exists(): + print(f"Warning: {CHANNELS_FILE} not found. Create it from data/channels.json.example") + 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()