"""HTTP server and request handler for IPTV endpoints."""
import mimetypes
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import parse_qs, urlparse
from .schedule import build_playlist_data
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):
"""Handles M3U playlists, EPG, live streams, and on-demand video."""
def __init__(self, request, client_address, server):
# Dependencies injected by server_factory
self._channel_repo = getattr(server, "channel_repo", None)
self._schedule_builder = getattr(server, "schedule_builder", None)
self._epg_builder = getattr(server, "epg_builder", None)
self._cache_manager = getattr(server, "cache_manager", None)
super().__init__(request, client_address, server)
def log_message(self, format, *args):
pass
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 = self._channel_repo.get_all()
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 = self._channel_repo.get_all()
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."""
channel = self._channel_repo.get_by_id(channel_id)
if not channel:
self.send_error(404, "Channel not found")
return
path_keys = channel.get("paths", [])
base = get_base_url(self)
playlist_data = None
if self._cache_manager:
playlist_data = self._cache_manager.get_playlist(channel_id)
if playlist_data is not None:
playlist = [(name, f"{base}/stream?path={path_str}") for name, path_str in playlist_data]
else:
playlist_data = build_playlist_data(path_keys)
playlist = [(name, f"{base}/stream?path={path_str}") for name, path_str in playlist_data]
if self._cache_manager:
self._cache_manager.set_playlist(channel_id, playlist_data)
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 = self._epg_builder.build_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)."""
from iptv.streamer import stream_live_channel
schedule = self._schedule_builder.get_or_build(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."""
from iptv.streamer import (
path_to_absolute,
get_preferred_audio_stream_index,
stream_file_with_preferred_audio,
)
abs_path = path_to_absolute(path_param)
if abs_path is None:
self.send_error(404, "File not found")
return
audio_idx = get_preferred_audio_stream_index(abs_path)
if audio_idx is not None:
self.send_response(200)
self.send_header("Content-Type", "video/mp4")
self.send_header("Cache-Control", "no-cache")
self.end_headers()
stream_file_with_preferred_audio(abs_path, self.wfile)
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 create_server(
channel_repo,
schedule_builder,
epg_builder,
host: str = "0.0.0.0",
port: int = 8080,
cache_manager=None,
) -> HTTPServer:
"""Create HTTP server with injected dependencies."""
server = HTTPServer((host, port), IPTVHandler)
server.channel_repo = channel_repo
server.schedule_builder = schedule_builder
server.epg_builder = epg_builder
server.cache_manager = cache_manager
return server