Add initial Restreamer implementation for YouTube-to-Jellyfin live restreaming

src/restreamer/broadcaster.py

@@ -0,0 +1,42 @@
+import asyncio
+from contextlib import suppress
+
+from loguru import logger
+
+from restreamer.constants import CLIENT_QUEUE_MAXSIZE
+
+
+class ChannelBroadcaster:
+    """Fans a single upstream byte stream out to any number of connected clients."""
+
+    def __init__(self, channel_id: str) -> None:
+        self._channel_id = channel_id
+        self._subscribers: set[asyncio.Queue[bytes | None]] = set()
+
+    def subscribe(self) -> asyncio.Queue[bytes | None]:
+        queue: asyncio.Queue[bytes | None] = asyncio.Queue(maxsize=CLIENT_QUEUE_MAXSIZE)
+        self._subscribers.add(queue)
+        logger.debug("Channel '{}' gained a subscriber ({} total)", self._channel_id, len(self._subscribers))
+        return queue
+
+    @property
+    def subscriber_count(self) -> int:
+        return len(self._subscribers)
+
+    def unsubscribe(self, queue: asyncio.Queue[bytes | None]) -> None:
+        self._subscribers.discard(queue)
+        logger.debug("Channel '{}' lost a subscriber ({} total)", self._channel_id, len(self._subscribers))
+
+    def publish(self, chunk: bytes) -> None:
+        for queue in list(self._subscribers):
+            try:
+                queue.put_nowait(chunk)
+            except asyncio.QueueFull:
+                logger.warning("Dropping slow subscriber on channel '{}'", self._channel_id)
+                self._subscribers.discard(queue)
+
+    def close(self) -> None:
+        for queue in list(self._subscribers):
+            with suppress(asyncio.QueueFull):
+                queue.put_nowait(None)
+        self._subscribers.clear()

src/restreamer/channel_worker.py

@@ -0,0 +1,72 @@
+import asyncio
+import time
+from datetime import UTC, datetime
+
+from loguru import logger
+
+from restreamer.broadcaster import ChannelBroadcaster
+from restreamer.config import ChannelConfig
+from restreamer.constants import (
+    BACKOFF_RESET_AFTER_HEALTHY_SECONDS,
+    MAX_RESTART_BACKOFF_SECONDS,
+    MIN_RESTART_BACKOFF_SECONDS,
+)
+from restreamer.ffmpeg_runner import FfmpegStreamProcess
+from restreamer.resolver import StreamResolver
+
+
+class ChannelWorker:
+    """Keeps one channel's stream alive: resolve -> run ffmpeg -> refresh/restart forever."""
+
+    def __init__(self, channel: ChannelConfig, resolver: StreamResolver, broadcaster: ChannelBroadcaster) -> None:
+        self._channel = channel
+        self._resolver = resolver
+        self._broadcaster = broadcaster
+        self._stopped = asyncio.Event()
+
+    async def run(self) -> None:
+        backoff = MIN_RESTART_BACKOFF_SECONDS
+
+        while not self._stopped.is_set():
+            started_at = time.monotonic()
+            try:
+                await self._run_once()
+                backoff = MIN_RESTART_BACKOFF_SECONDS
+            except Exception:  # noqa: BLE001 -- one channel's failure must never take down the others
+                logger.exception("Channel '{}' failed, will retry", self._channel.id)
+
+            if self._stopped.is_set():
+                break
+
+            if time.monotonic() - started_at >= BACKOFF_RESET_AFTER_HEALTHY_SECONDS:
+                backoff = MIN_RESTART_BACKOFF_SECONDS
+
+            logger.info("Channel '{}' restarting in {:.0f}s", self._channel.id, backoff)
+            await self._sleep_unless_stopped(backoff)
+            backoff = min(backoff * 2, MAX_RESTART_BACKOFF_SECONDS)
+
+    def stop(self) -> None:
+        self._stopped.set()
+
+    async def _run_once(self) -> None:
+        resolved = await self._resolver.resolve(str(self._channel.url))
+        refresh_deadline = resolved.expires_at
+
+        process = FfmpegStreamProcess(resolved, self._channel.id)
+        await process.start()
+        logger.info("Channel '{}' streaming (refresh due {})", self._channel.id, refresh_deadline)
+
+        try:
+            async for chunk in process.iter_chunks():
+                self._broadcaster.publish(chunk)
+                if refresh_deadline is not None and datetime.now(tz=UTC) >= refresh_deadline:
+                    logger.info("Channel '{}' proactively refreshing before URL expiry", self._channel.id)
+                    break
+        finally:
+            await process.stop()
+
+    async def _sleep_unless_stopped(self, seconds: float) -> None:
+        try:
+            await asyncio.wait_for(self._stopped.wait(), timeout=seconds)
+        except TimeoutError:
+            pass

src/restreamer/config.py

@@ -0,0 +1,43 @@
+from pathlib import Path
+
+import yaml
+from pydantic import BaseModel, Field, HttpUrl
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+from restreamer import constants
+
+
+class ChannelConfig(BaseModel):
+    id: str
+    name: str
+    url: HttpUrl
+    refresh_interval_seconds: int = constants.DEFAULT_REFRESH_INTERVAL_SECONDS
+
+
+class ChannelsFile(BaseModel):
+    channels: list[ChannelConfig] = Field(default_factory=list)
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_prefix="", extra="ignore")
+
+    env_debug: bool = Field(default=False, alias="ENV_DEBUG")
+    http_host: str = Field(default=constants.DEFAULT_HTTP_HOST, alias="HTTP_HOST")
+    http_port: int = Field(default=constants.DEFAULT_HTTP_PORT, alias="HTTP_PORT")
+    channels_file: Path = Field(default=Path(constants.DEFAULT_CHANNELS_FILE), alias="CHANNELS_FILE")
+    public_base_url: str | None = Field(default=None, alias="PUBLIC_BASE_URL")
+
+
+def load_channels(path: Path) -> list[ChannelConfig]:
+    if not path.is_file():
+        raise FileNotFoundError(f"Channels file not found: {path}")
+
+    raw = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
+    parsed = ChannelsFile.model_validate(raw)
+
+    ids = [channel.id for channel in parsed.channels]
+    duplicates = {channel_id for channel_id in ids if ids.count(channel_id) > 1}
+    if duplicates:
+        raise ValueError(f"Duplicate channel id(s) in {path}: {sorted(duplicates)}")
+
+    return parsed.channels

src/restreamer/constants.py

@@ -0,0 +1,28 @@
+import os
+
+APP_NAME = "Restreamer"
+
+DEFAULT_DEBUG = os.getenv("ENV_DEBUG", "false").lower() == "true"
+
+DEFAULT_HTTP_HOST = "0.0.0.0"
+DEFAULT_HTTP_PORT = 8000
+DEFAULT_CHANNELS_FILE = "channels.yaml"
+
+# Proactive refresh ceiling used when a resolved stream carries no expiry hint.
+DEFAULT_REFRESH_INTERVAL_SECONDS = 4 * 60 * 60
+
+# Restart backoff for a channel worker after a failed resolve/ffmpeg run.
+MIN_RESTART_BACKOFF_SECONDS = 5.0
+MAX_RESTART_BACKOFF_SECONDS = 120.0
+BACKOFF_RESET_AFTER_HEALTHY_SECONDS = 60.0
+
+# Grace period given to ffmpeg to exit on SIGTERM before SIGKILL.
+FFMPEG_TERMINATE_TIMEOUT_SECONDS = 5.0
+
+FFMPEG_READ_CHUNK_SIZE = 188 * 64  # multiple of the MPEG-TS packet size
+
+MPEGTS_CONTENT_TYPE = "video/mp2t"
+
+# Bounded per-client output queue; a slow client is dropped rather than
+# allowed to stall the broadcast for everyone else.
+CLIENT_QUEUE_MAXSIZE = 128

src/restreamer/ffmpeg_runner.py

@@ -0,0 +1,92 @@
+import asyncio
+from collections.abc import AsyncIterator
+
+from loguru import logger
+
+from restreamer.constants import FFMPEG_READ_CHUNK_SIZE, FFMPEG_TERMINATE_TIMEOUT_SECONDS
+from restreamer.resolver import ResolvedStream
+
+_RECONNECT_INPUT_ARGS = ["-reconnect", "1", "-reconnect_streamed", "1", "-reconnect_delay_max", "5"]
+
+
+def build_ffmpeg_args(stream: ResolvedStream) -> list[str]:
+    args = ["ffmpeg", "-loglevel", "warning", "-nostdin"]
+
+    header_lines = "".join(f"{key}: {value}\r\n" for key, value in stream.http_headers.items())
+
+    if header_lines:
+        args += ["-headers", header_lines]
+    args += [*_RECONNECT_INPUT_ARGS, "-i", stream.video_url]
+
+    if stream.audio_url:
+        if header_lines:
+            args += ["-headers", header_lines]
+        args += [*_RECONNECT_INPUT_ARGS, "-i", stream.audio_url]
+        args += ["-map", "0:v:0", "-map", "1:a:0"]
+
+    args += [
+        "-c",
+        "copy",
+        "-f",
+        "mpegts",
+        "-mpegts_flags",
+        "+resend_headers",
+        "pipe:1",
+    ]
+    return args
+
+
+class FfmpegStreamProcess:
+    """Wraps a running ffmpeg process that copies a resolved stream to stdout as MPEG-TS."""
+
+    def __init__(self, stream: ResolvedStream, channel_id: str) -> None:
+        self._stream = stream
+        self._channel_id = channel_id
+        self._process: asyncio.subprocess.Process | None = None
+        self._stderr_task: asyncio.Task[None] | None = None
+
+    async def start(self) -> None:
+        args = build_ffmpeg_args(self._stream)
+        logger.debug("Starting ffmpeg for channel '{}'", self._channel_id)
+        self._process = await asyncio.create_subprocess_exec(
+            *args,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        self._stderr_task = asyncio.create_task(self._drain_stderr())
+
+    async def iter_chunks(self) -> AsyncIterator[bytes]:
+        if self._process is None or self._process.stdout is None:
+            raise RuntimeError("ffmpeg process has not been started")
+
+        while True:
+            chunk = await self._process.stdout.read(FFMPEG_READ_CHUNK_SIZE)
+            if not chunk:
+                return
+            yield chunk
+
+    async def wait(self) -> int:
+        if self._process is None:
+            raise RuntimeError("ffmpeg process has not been started")
+        return await self._process.wait()
+
+    async def stop(self) -> None:
+        if self._process is None or self._process.returncode is not None:
+            return
+
+        self._process.terminate()
+        try:
+            await asyncio.wait_for(self._process.wait(), timeout=FFMPEG_TERMINATE_TIMEOUT_SECONDS)
+        except TimeoutError:
+            logger.warning("ffmpeg for channel '{}' did not exit in time, killing it", self._channel_id)
+            self._process.kill()
+            await self._process.wait()
+
+        if self._stderr_task is not None:
+            await self._stderr_task
+
+    async def _drain_stderr(self) -> None:
+        if self._process is None or self._process.stderr is None:
+            return
+        async for line in self._process.stderr:
+            logger.debug("[ffmpeg:{}] {}", self._channel_id, line.decode(errors="replace").rstrip())

src/restreamer/http_server.py

@@ -0,0 +1,64 @@
+from aiohttp import web
+from loguru import logger
+
+from restreamer.broadcaster import ChannelBroadcaster
+from restreamer.config import ChannelConfig
+from restreamer.constants import MPEGTS_CONTENT_TYPE
+
+
+def build_playlist(channels: list[ChannelConfig], base_url: str) -> str:
+    base_url = base_url.rstrip("/")
+    lines = ["#EXTM3U"]
+    for channel in channels:
+        lines.append(f'#EXTINF:-1 tvg-id="{channel.id}",{channel.name}')
+        lines.append(f"{base_url}/stream/{channel.id}")
+    return "\n".join(lines) + "\n"
+
+
+def create_app(channels: list[ChannelConfig], broadcasters: dict[str, ChannelBroadcaster], base_url: str) -> web.Application:
+    app = web.Application()
+    app["channels"] = channels
+    app["broadcasters"] = broadcasters
+    app["base_url"] = base_url
+
+    app.router.add_get("/playlist.m3u", _handle_playlist)
+    app.router.add_get("/stream/{channel_id}", _handle_stream)
+    app.router.add_get("/healthz", _handle_health)
+    return app
+
+
+async def _handle_playlist(request: web.Request) -> web.Response:
+    channels: list[ChannelConfig] = request.app["channels"]
+    base_url: str = request.app["base_url"]
+    body = build_playlist(channels, base_url)
+    return web.Response(text=body, content_type="audio/x-mpegurl")
+
+
+async def _handle_stream(request: web.Request) -> web.StreamResponse:
+    channel_id = request.match_info["channel_id"]
+    broadcasters: dict[str, ChannelBroadcaster] = request.app["broadcasters"]
+    broadcaster = broadcasters.get(channel_id)
+    if broadcaster is None:
+        raise web.HTTPNotFound(text=f"Unknown channel '{channel_id}'")
+
+    response = web.StreamResponse(status=200, headers={"Content-Type": MPEGTS_CONTENT_TYPE})
+    await response.prepare(request)
+
+    queue = broadcaster.subscribe()
+    try:
+        while True:
+            chunk = await queue.get()
+            if chunk is None:
+                break
+            await response.write(chunk)
+    except ConnectionResetError:
+        logger.debug("Client disconnected from channel '{}'", channel_id)
+    finally:
+        broadcaster.unsubscribe(queue)
+
+    return response
+
+
+async def _handle_health(request: web.Request) -> web.Response:
+    channels: list[ChannelConfig] = request.app["channels"]
+    return web.json_response({"status": "ok", "channels": [c.id for c in channels]})

src/restreamer/logging_setup.py

@@ -0,0 +1,23 @@
+import sys
+
+from loguru import logger
+
+from restreamer.constants import APP_NAME
+
+
+def configure_logging(debug: bool) -> None:
+    logger.remove()
+
+    logger.add(
+        sys.stdout,
+        level="INFO",
+        format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}",
+    )
+
+    if debug:
+        logger.add(
+            f"logs/{APP_NAME}_{{time}}.log",
+            level="DEBUG",
+            retention=10,
+            format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}",
+        )

src/restreamer/manager.py

@@ -0,0 +1,48 @@
+import asyncio
+import signal
+
+from aiohttp import web
+from loguru import logger
+
+from restreamer.broadcaster import ChannelBroadcaster
+from restreamer.channel_worker import ChannelWorker
+from restreamer.config import ChannelConfig, Settings
+from restreamer.http_server import create_app
+from restreamer.resolver import StreamResolver
+
+
+class RestreamerManager:
+    def __init__(self, settings: Settings, channels: list[ChannelConfig], resolver: StreamResolver) -> None:
+        self._settings = settings
+        self._channels = channels
+        self._resolver = resolver
+        self._broadcasters = {channel.id: ChannelBroadcaster(channel.id) for channel in channels}
+        self._workers = [
+            ChannelWorker(channel, resolver, self._broadcasters[channel.id]) for channel in channels
+        ]
+
+    async def run(self) -> None:
+        base_url = self._settings.public_base_url or f"http://{self._settings.http_host}:{self._settings.http_port}"
+        app = create_app(self._channels, self._broadcasters, base_url)
+
+        runner = web.AppRunner(app)
+        await runner.setup()
+        site = web.TCPSite(runner, self._settings.http_host, self._settings.http_port)
+        await site.start()
+        logger.info("HTTP server listening on {}:{}", self._settings.http_host, self._settings.http_port)
+        logger.info("Playlist available at {}/playlist.m3u", base_url)
+
+        worker_tasks = [asyncio.create_task(worker.run()) for worker in self._workers]
+
+        stop_event = asyncio.Event()
+        loop = asyncio.get_running_loop()
+        for sig in (signal.SIGINT, signal.SIGTERM):
+            loop.add_signal_handler(sig, stop_event.set)
+
+        await stop_event.wait()
+        logger.info("Shutting down")
+
+        for worker in self._workers:
+            worker.stop()
+        await asyncio.gather(*worker_tasks, return_exceptions=True)
+        await runner.cleanup()

src/restreamer/resolver.py

@@ -0,0 +1,98 @@
+import asyncio
+from dataclasses import dataclass, field
+from datetime import UTC, datetime, timedelta
+from typing import Any, Protocol
+from urllib.parse import parse_qs, urlparse
+
+import yt_dlp
+from loguru import logger
+
+from restreamer.constants import DEFAULT_REFRESH_INTERVAL_SECONDS
+
+
+class ResolveError(RuntimeError):
+    """Raised when a source URL cannot be resolved to a playable stream."""
+
+
+@dataclass(frozen=True)
+class ResolvedStream:
+    video_url: str
+    audio_url: str | None = None
+    http_headers: dict[str, str] = field(default_factory=dict)
+    expires_at: datetime | None = None
+
+
+class StreamResolver(Protocol):
+    async def resolve(self, source_url: str) -> ResolvedStream: ...
+
+
+def parse_expiry_from_url(url: str) -> datetime | None:
+    """Extract the `expire` query parameter googlevideo URLs carry, if present."""
+    query = parse_qs(urlparse(url).query)
+    values = query.get("expire")
+    if not values:
+        return None
+
+    try:
+        timestamp = int(values[0])
+    except ValueError:
+        return None
+
+    return datetime.fromtimestamp(timestamp, tz=UTC)
+
+
+class YtDlpResolver:
+    """Resolves a YouTube (live) URL to a direct, playable stream via yt-dlp."""
+
+    def __init__(self, ydl_opts: dict[str, Any] | None = None) -> None:
+        self._ydl_opts: dict[str, Any] = {
+            "quiet": True,
+            "no_warnings": True,
+            "noplaylist": True,
+            # Live YouTube streams are frequently served as separate video-only and
+            # audio-only HLS renditions rather than one muxed format.
+            "format": "bestvideo*+bestaudio/best",
+        }
+        if ydl_opts:
+            self._ydl_opts.update(ydl_opts)
+
+    async def resolve(self, source_url: str) -> ResolvedStream:
+        info = await asyncio.to_thread(self._extract_info, source_url)
+        return self._to_resolved_stream(info)
+
+    def _extract_info(self, source_url: str) -> dict[str, Any]:
+        # options are intentionally a plain dynamic dict, not the stub's strict TypedDict
+        with yt_dlp.YoutubeDL(self._ydl_opts) as ydl:  # type: ignore[arg-type]
+            info = ydl.extract_info(source_url, download=False)
+        if not info:
+            raise ResolveError(f"yt-dlp returned no information for {source_url}")
+        return dict(info)
+
+    @staticmethod
+    def _to_resolved_stream(info: dict[str, Any]) -> ResolvedStream:
+        video_url, audio_url = YtDlpResolver._select_urls(info)
+
+        headers = dict(info.get("http_headers") or {})
+        expires_at = parse_expiry_from_url(video_url)
+        if expires_at is None:
+            expires_at = datetime.now(tz=UTC) + timedelta(seconds=DEFAULT_REFRESH_INTERVAL_SECONDS)
+            logger.debug("No expiry hint in resolved URL, falling back to default refresh interval")
+
+        return ResolvedStream(video_url=video_url, audio_url=audio_url, http_headers=headers, expires_at=expires_at)
+
+    @staticmethod
+    def _select_urls(info: dict[str, Any]) -> tuple[str, str | None]:
+        requested_formats = info.get("requested_formats")
+        if requested_formats:
+            # Audio-only renditions are not consistently given an explicit acodec by
+            # yt-dlp, but a video rendition always has a real vcodec -- classify by that.
+            video_fmt = next((f for f in requested_formats if f.get("vcodec") not in (None, "none")), None)
+            audio_fmt = next((f for f in requested_formats if f is not video_fmt), None)
+            if video_fmt is None or audio_fmt is None:
+                raise ResolveError("yt-dlp did not return both a video and an audio format")
+            return video_fmt["url"], audio_fmt["url"]
+
+        video_url = info.get("url")
+        if not video_url:
+            raise ResolveError("yt-dlp did not return a playable direct URL")
+        return video_url, None