258 lines
9.0 KiB
Python
258 lines
9.0 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 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 = "<!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 += f"<p><code>{get_base_url(self)}/playlist.m3u</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 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://<host>:{port}/playlist.m3u)")
|
|
server.serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|