tom/ytstream
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
ytstream/proxy/ytstream_proxy.py
T
Claude 22d8828080 Never serve a partial file: it makes Jellyfin transcode 
The 12s first-byte grace was the wrong trade and a real play found it
within the hour. A 2-hour upload took 3 minutes to start, played 6
seconds, and stalled. Jellyfin had run ffmpeg with -probesize 1G against
the growing stream and then transcoded to HLS with libx264.

The cause is the container. A fragmented MP4 with empty_moov has no
duration in its header, so the only way to get one is to sum every
fragment -- probing a growing file reads all of it. Jellyfin cannot
establish duration, codec or bitrate, so it abandons direct play and
transcodes a stream it also cannot seek. It was targeting 4.83 Mbps
against a source measured at 3.29: re-encoding a stream that already fit,
because it could not measure it.

The same video once complete reports SupportsDirectPlay with the exact
runtime and bitrate.

So FIRST_BYTE_GRACE defaults to infinite again, with --wait-timeout raised
to 600s for a 2-hour upload. A cold long video is slow to start, which is
accepted: the fetch outlives the request so a retry is instant, and a
retryable stall beats a transcode that wastes a gigabyte and cannot work.
Both failure modes are recorded at the constant in the order measured so
the 12s cap is not reintroduced.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-13 11:32:12 +01:00

875 lines
37 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
"""
ytstream -- just-in-time YouTube streaming proxy for Jellyfin.
Serves one endpoint per video:
GET /watch/<video_id> muxed avc1+mp4a MP4, range-capable
GET /healthz JSON status (sessions, cache use, counters)
A `.strm` file whose contents are `http://127.0.0.1:8099/watch/<video_id>` plays
that video without any of its bytes ever having been stored on disk beforehand.
ffmpeg reads only local FIFOs; every piece of YouTube protocol handling stays
inside yt-dlp. See FINDINGS.md for what has and has not been verified.
Safety rails, because a Jellyfin library scan can ask for every episode at once:
--no-fetch never start a pipeline. Log the request and return 503.
Use this for a first library scan: it reveals whether
Jellyfin probes .strm targets with zero YouTube traffic.
--max-pipelines cap on CONCURRENT pipelines (default 2). Excess -> 503.
--max-starts cap on TOTAL cold starts per window (default 20/hour). This
is the one that bounds a runaway metadata refresh: concurrency
alone only slows a 1249-episode churn down, it does not stop
it. A person watching podcasts starts a handful of videos an
hour; a refresh storm hits the cap in seconds.
--cache-gb tmpfs budget; least-recently-used complete files are evicted.
Serving modes. The output is a fragmented MP4, so it is *readable* while being
written -- but do not serve it that way to Jellyfin. See FIRST_BYTE_GRACE below:
a growing fragmented MP4 has no duration in its header, so Jellyfin drags up to a
gigabyte through this proxy trying to probe one and then transcodes a stream it
cannot seek. A slow start beats that.
default wait for the complete mux, bounded by --wait-timeout, then
serve it with real ranges and the exact duration. A cold
long video is slow to start; the fetch continues after the
client gives up, so the next attempt is instant.
--first-byte-grace finite value: stop waiting after N seconds and stream the
partial file. Fast start, but only for a client that
tolerates an unseekable, duration-less stream.
--growing grace of zero: serve as soon as any byte exists.
"""
import argparse
import collections
import json
import os
import re
import shutil
import subprocess
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
POT_ARGS = "youtubepot-bgutilhttp:base_url=http://127.0.0.1:4416"
# player_client: `mweb` is deliberately NOT included. Measured 2026-08-12:
# * mweb formats 403 on every attempt, both directly and via --load-info-json,
# even though they are the only ones carrying a PO token in the URL.
# * mweb is also the ONLY source of the DRC and dubbed-language variants that
# caused the two original picker bugs. Dropping it removes both hazards
# before the picker ever sees them.
# * `default` resolves to android_vr for avc1+mp4a, which works without a PO
# token. web / ios / web_safari / tv are SABR-only and yield no usable
# formats at all. Left as `default` rather than pinned to `android_vr` so a
# yt-dlp update can follow YouTube if android_vr stops working.
CLIENT_ARGS = "youtube:player_client=default"
MAX_HEIGHT = 720
STALL_TIMEOUT = 45.0
# How long a request may block waiting for a *complete* mux before giving up and
# streaming the partial file instead. Infinite by default, i.e. wait for the whole
# thing (bounded by --wait-timeout), because serving a partial file to Jellyfin is
# WORSE than making it wait. Measured 2026-08-13, in this order:
#
# 1. Waiting for the complete file with no cap meant a 46-minute upload sent
# nothing for 79s and the client gave up -> "playback error".
# 2. So this was capped at 12s and the partial file streamed instead. A
# 2-hour upload then took 3 minutes to start, played 6 seconds, and stalled.
# Jellyfin had run:
# ffmpeg -analyzeduration 200M -probesize 1G -i .../watch/<id> ... libx264
# It TRANSCODED, after dragging up to a gigabyte through the proxy to probe.
#
# The reason is the container. A fragmented MP4 with empty_moov carries no
# duration in its header -- the only way to get one is to sum every fragment, so
# probing a growing file reads all of it. Jellyfin cannot establish duration,
# codec or bitrate, so it stops trusting direct play and transcodes a stream it
# also cannot seek. On the *complete* file ffmpeg patches the real duration into
# the moov on close, the probe is cheap, and Jellyfin reported SupportsDirectPlay
# with the exact runtime and a 3.29 Mbps bitrate -- comfortably under the
# 4.83 Mbps cap it had been transcoding down to.
#
# So a slow start is the correct failure: the fetch continues after the client
# gives up, and the next attempt is instant. A partial file is a fast start
# followed by a transcode that cannot work.
#
# Set a finite value only for a client that tolerates an unseekable, duration-less
# stream. Jellyfin does not.
FIRST_BYTE_GRACE = float("inf")
VIDEO_ID_RE = re.compile(r"^[A-Za-z0-9_-]{11}$")
_log_lock = threading.Lock()
_access_log = None
def log(msg):
print(f"[ytstream] {msg}", file=sys.stderr, flush=True)
def access(msg):
"""Append to the access log. This is the evidence trail for the scan-probe
question, so it is written unbuffered and never dropped."""
line = f"{time.strftime('%Y-%m-%dT%H:%M:%S')} {msg}"
with _log_lock:
print(line, file=sys.stderr, flush=True)
if _access_log:
with open(_access_log, "a") as f:
f.write(line + "\n")
# --------------------------------------------------------------------------
# Extraction and format selection
# --------------------------------------------------------------------------
def extract_info(url, work):
"""One extraction. Both downloads reuse it via --load-info-json, which is
verified not to re-extract."""
info_path = os.path.join(work, "info.json")
cmd = [
"yt-dlp", "-J", "--no-warnings",
"--extractor-args", CLIENT_ARGS,
"--extractor-args", POT_ARGS,
url,
]
t0 = time.monotonic()
out = subprocess.run(cmd, capture_output=True, text=True)
if out.returncode != 0:
raise RuntimeError(f"yt-dlp -J failed: {out.stderr.strip()[:500]}")
info = json.loads(out.stdout)
with open(info_path, "w") as f:
json.dump(info, f)
log(f"extracted {info.get('id')} in {time.monotonic() - t0:.1f}s -- "
f"{info.get('title')!r}")
return info, info_path
def pick_formats(info):
"""Prefer avc1 video and the original-language mp4a audio so the mux is a
pure copy and Jellyfin can direct-play."""
fmts = info.get("formats", [])
def usable(f):
return f.get("url") and f.get("protocol", "").startswith("http")
def is_drc(f):
# YouTube advertises DRC (dynamic-range-compressed) audio variants that
# carry the same abr as their plain counterparts but 403 on download.
# Verified against NH2MhBFQm9w: 140-drc -> 403, 140 -> fine.
return "-drc" in (f.get("format_id") or "") or "DRC" in (f.get("format_note") or "")
vids = [
f for f in fmts
if usable(f)
and f.get("vcodec") not in (None, "none")
and f.get("acodec") in (None, "none")
and (f.get("height") or 0) <= MAX_HEIGHT
]
auds = [
f for f in fmts
if usable(f)
and f.get("acodec") not in (None, "none")
and f.get("vcodec") in (None, "none")
]
# Ranking notes, all learned the hard way against real videos:
#
# * language_preference outranks everything. Multi-language uploads expose
# 140-0..140-N with IDENTICAL abr, so max() would otherwise tie-break on
# list order and silently pick whichever dub YouTube listed first --
# German audio on an English podcast (verified: J9O3sxoMs5U). The
# original track carries language_preference 10, dubs carry -1.
# * non-DRC next, because DRC variants 403. Language correctness ranks
# above this deliberately: a 403 is a loud failure, wrong-language audio
# is a silent one that would ship.
# * only then codec and bitrate.
def langpref(f):
return f.get("language_preference") or 0
def vkey(f):
return (
not is_drc(f),
f.get("vcodec", "").startswith("avc1"),
f.get("height") or 0,
f.get("tbr") or 0,
)
def akey(f):
return (
langpref(f),
not is_drc(f),
f.get("acodec", "").startswith("mp4a"),
f.get("abr") or 0,
)
if not vids or not auds:
raise RuntimeError(
"no separate video+audio pair; formats present: " + ", ".join(
f"{f.get('format_id')}({f.get('vcodec')}/{f.get('acodec')})"
for f in fmts[:15]))
v = max(vids, key=vkey)
a = max(auds, key=akey)
log(f" video {v['format_id']} {v.get('vcodec')} {v.get('height')}p | "
f"audio {a['format_id']} {a.get('acodec')} lang={a.get('language')}")
if not v.get("vcodec", "").startswith("avc1"):
log(" WARNING: no h264 at this height -- Jellyfin may transcode")
if len({f.get("language") for f in auds}) > 1 and langpref(a) <= 0:
log(f" WARNING: multi-language upload and the chosen track "
f"({a.get('language')}) is not the original -- expect a dub")
return v["format_id"], a["format_id"]
# --------------------------------------------------------------------------
# Producer: two yt-dlp -> FIFOs -> ffmpeg -c copy -> fragmented MP4
# --------------------------------------------------------------------------
def start_producer(info_path, vfmt, afmt, work):
vfifo = os.path.join(work, "v.fifo")
afifo = os.path.join(work, "a.fifo")
for p in (vfifo, afifo):
if os.path.exists(p):
os.unlink(p)
os.mkfifo(p)
out_path = os.path.join(work, "out.mp4")
done_path = out_path + ".done"
err_path = out_path + ".err"
procs = []
def feed(fmt, fifo, tag):
# Opening a FIFO for write blocks until a reader attaches, so ffmpeg
# must already be starting -- it is, just below.
fh = open(fifo, "wb")
p = subprocess.Popen(
[
"yt-dlp", "--load-info-json", info_path,
"-f", fmt, "-o", "-", "--quiet", "--no-warnings",
"--extractor-args", CLIENT_ARGS,
"--extractor-args", POT_ARGS,
],
stdout=fh, stderr=subprocess.PIPE,
)
procs.append((tag, p))
fh.close()
ff = subprocess.Popen(
[
"ffmpeg", "-y", "-loglevel", "error",
"-i", vfifo, "-i", afifo,
"-c", "copy",
"-movflags", "frag_keyframe+empty_moov+default_base_moof",
"-f", "mp4", out_path,
],
stderr=subprocess.PIPE,
)
threading.Thread(target=feed, args=(vfmt, vfifo, "video"), daemon=True).start()
threading.Thread(target=feed, args=(afmt, afifo, "audio"), daemon=True).start()
def reap():
rc = ff.wait()
err = ff.stderr.read().decode(errors="replace").strip()
problems = []
if rc != 0:
problems.append(f"ffmpeg exited {rc}: {err[:400]}")
for tag, p in procs:
if p.poll() not in (0, None):
problems.append(
f"yt-dlp[{tag}] exited {p.returncode}: "
f"{p.stderr.read().decode(errors='replace')[:300]}")
if problems:
# A single 403 on either stream kills the run and there is no retry
# yet; record it so /healthz and the caller can see why.
with open(err_path, "w") as f:
f.write("\n".join(problems))
log(f"producer FAILED in {work}: {problems[0]}")
else:
log(f"mux complete: {out_path} {os.path.getsize(out_path)} bytes")
open(done_path, "w").close()
threading.Thread(target=reap, daemon=True).start()
return out_path, done_path, err_path
# --------------------------------------------------------------------------
# Session manager
# --------------------------------------------------------------------------
class Session:
"""One video's pipeline and its output file.
`final` is the terminal signal: set once the session has either produced a
complete file or given up after retries. It is deliberately separate from
the producer's own done-marker, because a failed attempt writes that marker
too and a waiting request must not mistake a retry for a finished file.
"""
def __init__(self, video_id, work):
self.video_id = video_id
self.work = work
self.out_path = os.path.join(work, "out.mp4")
self.started = time.time()
self.last_used = time.time()
self.final = threading.Event()
self.failed = None
self.attempts = 0
self.readers = 0
@property
def complete(self):
"""True only when the file is finished AND good."""
return self.final.is_set() and not self.failed
def size(self):
try:
return os.path.getsize(self.out_path)
except OSError:
return 0
class Manager:
def __init__(self, work_root, max_pipelines, cache_bytes, no_fetch, growing,
max_retries=2, max_starts=20, starts_window=3600.0):
self.work_root = work_root
self.max_pipelines = max_pipelines
self.max_retries = max_retries
self.max_starts = max_starts
self.starts_window = starts_window
# Timestamps of cold starts, pruned to the window. Cache hits are not
# recorded: re-watching or resuming must never be rate limited.
self.start_log = collections.deque()
self.cache_bytes = cache_bytes
self.no_fetch = no_fetch
self.growing = growing
# Reported by /healthz so the serving mode is visible without reading the
# unit file. Set by main(); the handler holds the value it actually uses.
self.first_byte_grace = 0.0 if growing else FIRST_BYTE_GRACE
# True when a request waits for the whole mux rather than ever serving a
# partial file. Overwritten by main(); the default matches the default
# grace so a Manager built directly (tests) reports the same thing.
self.strict = not growing
self.lock = threading.Lock()
self.sessions = {}
self.counters = {"requests": 0, "started": 0, "reused": 0,
"refused_nofetch": 0, "refused_busy": 0,
"refused_ratelimit": 0, "retried": 0, "failed": 0,
"evicted": 0}
def _active(self):
return [s for s in self.sessions.values() if not s.complete]
def get(self, video_id):
"""Return (session, error_string). Never raises."""
with self.lock:
self.counters["requests"] += 1
s = self.sessions.get(video_id)
if s is not None:
s.last_used = time.time()
self.counters["reused"] += 1
return s, None
if self.no_fetch:
self.counters["refused_nofetch"] += 1
return None, "no-fetch mode: refusing to start a pipeline"
if len(self._active()) >= self.max_pipelines:
self.counters["refused_busy"] += 1
return None, (f"at pipeline cap ({self.max_pipelines}); "
f"refusing to start another")
# Cold-start budget. Refusing produces a short retry burst from
# libavformat and then aborts whatever refresh triggered it, which
# is the intended protective outcome: an aborted refresh costs some
# re-derivable metadata, a runaway one costs the whole catalogue in
# YouTube traffic.
now = time.monotonic()
while self.start_log and now - self.start_log[0] > self.starts_window:
self.start_log.popleft()
if len(self.start_log) >= self.max_starts:
self.counters["refused_ratelimit"] += 1
oldest = self.starts_window - (now - self.start_log[0])
log(f"RATE LIMIT: {len(self.start_log)} cold starts in the last "
f"{self.starts_window / 60:.0f}min, refusing {video_id}. "
f"Budget frees in {oldest:.0f}s. If this was a library "
f"refresh, that refresh is being stopped on purpose.")
return None, (f"cold-start budget exhausted "
f"({self.max_starts} per "
f"{self.starts_window / 60:.0f}min)")
self.start_log.append(now)
work = os.path.join(self.work_root, video_id)
shutil.rmtree(work, ignore_errors=True)
os.makedirs(work, exist_ok=True)
s = Session(video_id, work)
self.sessions[video_id] = s
self.counters["started"] += 1
threading.Thread(target=self._run, args=(s,), daemon=True).start()
return s, None
def _run(self, s):
"""Run the pipeline, retrying on failure with a fresh extraction.
Intermittent 403s do happen -- observed on both the video and the audio
stream of videos that succeeded minutes earlier, clustered after heavy
use, so most likely transient rate limiting. A single one used to kill
playback outright. Each retry re-extracts, because the resolved URLs and
their PO token binding are the most likely thing to have gone stale.
Only retried while nothing has been served yet (size == 0). If bytes
already went out we cannot rewind under a reader.
"""
url = f"https://www.youtube.com/watch?v={s.video_id}"
last = None
for attempt in range(1, self.max_retries + 2):
s.attempts = attempt
try:
shutil.rmtree(s.work, ignore_errors=True)
os.makedirs(s.work, exist_ok=True)
info, info_path = extract_info(url, s.work)
vfmt, afmt = pick_formats(info)
_out, done_path, err_path = start_producer(
info_path, vfmt, afmt, s.work)
while not os.path.exists(done_path):
time.sleep(0.2)
if not os.path.exists(err_path):
last = None
break
with open(err_path) as f:
last = f.read().strip()
except Exception as e: # noqa: BLE001 - report anything
last = str(e)[:500]
if s.size() > 0:
log(f"{s.video_id}: attempt {attempt} failed after serving "
f"bytes -- not retrying")
break
if attempt <= self.max_retries:
with self.lock:
self.counters["retried"] += 1
log(f"{s.video_id}: attempt {attempt} failed "
f"({(last or '')[:90]}) -- retrying")
time.sleep(2.0 * attempt)
s.failed = last
if last:
with self.lock:
self.counters["failed"] += 1
log(f"session {s.video_id} gave up after {s.attempts} attempt(s)")
s.final.set()
self.evict()
def evict(self):
"""Drop least-recently-used complete sessions until under budget."""
with self.lock:
done = [s for s in self.sessions.values()
if s.complete and s.readers == 0]
total = sum(s.size() for s in self.sessions.values())
for s in sorted(done, key=lambda x: x.last_used):
if total <= self.cache_bytes:
break
total -= s.size()
self.sessions.pop(s.video_id, None)
shutil.rmtree(s.work, ignore_errors=True)
self.counters["evicted"] += 1
log(f"evicted {s.video_id}")
def status(self):
with self.lock:
return {
# `null` rather than Infinity: json.dumps would happily emit the
# latter, and it is not valid JSON for whoever reads this.
"mode": ("growing" if self.growing
else "wait-for-complete" if self.strict
else "wait-then-stream"),
# `null` rather than a number when strict: there is no cap, and
# json.dumps would otherwise emit Infinity, which is not JSON.
"first_byte_grace_s": (
None if self.strict or self.first_byte_grace == float("inf")
else self.first_byte_grace),
"no_fetch": self.no_fetch,
"max_pipelines": self.max_pipelines,
"max_starts": self.max_starts,
"starts_in_window": len(self.start_log),
"starts_window_min": round(self.starts_window / 60),
"cache_gb": round(self.cache_bytes / 2**30, 2),
"cache_used_gb": round(
sum(s.size() for s in self.sessions.values()) / 2**30, 3),
"counters": dict(self.counters),
"sessions": [
{"video_id": s.video_id,
"complete": s.complete,
"bytes": s.size(),
"readers": s.readers,
"age_s": round(time.time() - s.started, 1),
"attempts": s.attempts,
"error": s.failed}
for s in self.sessions.values()
],
}
# --------------------------------------------------------------------------
# HTTP
# --------------------------------------------------------------------------
def make_handler(mgr, wait_timeout, first_byte_grace=FIRST_BYTE_GRACE):
class Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def log_message(self, *a):
pass # we do our own logging
# -- range helpers (verified by test_range.py) ----------------------
def _size(self, path):
return os.path.getsize(path) if os.path.exists(path) else 0
def _wait_for_bytes(self, path, offset, finished):
"""Block until the file grows past `offset`, or nothing more is coming.
`finished` must mean "the producer has stopped", not "the producer
succeeded" -- a run that died after writing some bytes never grows
again, and waiting on success would hang here for STALL_TIMEOUT and
then drop the connection.
"""
deadline = time.monotonic() + STALL_TIMEOUT
while True:
size = self._size(path)
if size > offset or finished():
return size
if time.monotonic() > deadline:
raise TimeoutError
time.sleep(0.05)
def _parse_range(self, rng, total):
"""Parse a single byte range against a known total size.
Returns (start, end) inclusive, "unsatisfiable", or None when there
is no usable range and the whole body should be sent. Multi-range
requests fall into the None case: answering with the whole body is
legal and beats mis-serving one part.
"""
if not rng or not rng.startswith("bytes=") or "," in rng:
return None
first, sep, last = rng[6:].strip().partition("-")
if not sep:
return None
try:
if not first: # bytes=-N -> final N bytes
n = int(last)
return (max(0, total - n), total - 1) if n > 0 else "unsatisfiable"
start = int(first)
end = int(last) if last else total - 1
except ValueError:
return None
if start >= total or start > end:
return "unsatisfiable"
return start, min(end, total - 1)
# -- routing --------------------------------------------------------
def do_HEAD(self):
self._route(True)
def do_GET(self):
self._route(False)
def _route(self, head_only):
rng = self.headers.get("Range", "-")
ua = (self.headers.get("User-Agent") or "-")[:60]
access(f"{self.command} {self.path} range={rng} ua={ua!r}")
if self.path == "/healthz":
body = json.dumps(mgr.status(), indent=2).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
if not head_only:
self.wfile.write(body)
return
if not self.path.startswith("/watch/"):
self._fail(404, "not found")
return
video_id = self.path[len("/watch/"):].split("?")[0]
if not VIDEO_ID_RE.match(video_id):
self._fail(400, "bad video id")
return
sess, refusal = mgr.get(video_id)
if refusal:
access(f" -> 503 {refusal}")
self._fail(503, refusal, retry_after=30)
return
# A finished file is worth a short wait -- it is the only way to get a
# correct duration and working seeks -- but the wait is proportional
# to the video's length, and until it ends this handler sends nothing
# at all, not even response headers. Bound it. Whatever has not muxed
# inside the grace gets served while it is still being written, which
# is possible because the container is fragmented.
grace = 0.0 if mgr.growing else min(first_byte_grace, wait_timeout)
strict = grace >= wait_timeout
if grace > 0 and not sess.final.is_set():
access(f" -> waiting up to {grace:.0f}s for {video_id} to mux")
sess.final.wait(grace)
if not sess.final.is_set():
if strict:
self._fail(504, "mux did not finish in time")
return
# Serving a partial file still needs its first bytes to exist: a
# retry underneath can leave the size at 0 for a while, so wait
# on both the size and the terminal signal.
access(f" -> serving {video_id} while it is still muxing")
deadline = time.monotonic() + STALL_TIMEOUT
while sess.size() == 0 and not sess.final.is_set():
if time.monotonic() > deadline:
self._fail(504, "producer wrote nothing")
return
time.sleep(0.1)
# `failed` is only ever set together with `final`, so this covers both
# paths. Strict callers asked for a good file and get an error
# instead; otherwise a partial file is better than nothing, and a
# failure with no bytes at all is still an error.
if sess.failed and (strict or sess.size() == 0):
access(f" -> 502 producer failed: {sess.failed[:120]}")
self._fail(502, f"producer failed: {sess.failed[:200]}")
return
with mgr.lock:
sess.readers += 1
sess.last_used = time.time()
try:
self._serve(sess, head_only)
finally:
with mgr.lock:
sess.readers -= 1
sess.last_used = time.time()
def _fail(self, code, msg, retry_after=None):
body = (msg + "\n").encode()
try:
self.send_response(code)
if retry_after:
self.send_header("Retry-After", str(retry_after))
self.send_header("Content-Type", "text/plain")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
except (BrokenPipeError, ConnectionResetError):
pass
# -- body -----------------------------------------------------------
def _serve(self, sess, head_only):
path = sess.out_path
# Two different questions, and conflating them hangs the connection:
# `finished` is "no more bytes are coming", which is what every wait
# below must test; `sess.complete` additionally means the file is
# good, which is what ranges require.
finished = lambda: sess.final.is_set() # noqa: E731
# Ranges are only honoured once the file is complete; while it is
# still growing there is no reliable time-to-byte mapping into a
# fragmented MP4, so we present a non-seekable stream instead.
chunked = not sess.complete
start, end, is_range = 0, None, False
if not chunked:
spec = self._parse_range(self.headers.get("Range"),
self._size(path))
if spec == "unsatisfiable":
total = self._size(path)
self.send_response(416)
self.send_header("Content-Range", f"bytes */{total}")
self.send_header("Content-Length", "0")
self.end_headers()
return
if spec:
start, end = spec
is_range = True
try:
self._wait_for_bytes(path, start, finished)
except TimeoutError:
self._fail(504, "producer stalled")
return
if chunked:
self.send_response(200)
self.send_header("Transfer-Encoding", "chunked")
else:
total = self._size(path)
if end is None:
end = total - 1
if is_range:
self.send_response(206)
self.send_header("Content-Range", f"bytes {start}-{end}/{total}")
else:
self.send_response(200)
self.send_header("Content-Length", str(end - start + 1))
self.send_header("Accept-Ranges", "bytes")
self.send_header("Content-Type", "video/mp4")
self.end_headers()
if head_only:
return
sent = start
# None while chunked: the growing case has no known end and stops
# when the producer does. A bounded range must send exactly the
# bytes it promised in Content-Length, no more.
remaining = None if end is None else end - start + 1
try:
with open(path, "rb") as f:
f.seek(start)
while True:
if remaining is not None and remaining <= 0:
break
buf = f.read(65536 if remaining is None
else min(65536, remaining))
if buf:
if chunked:
self.wfile.write(b"%X\r\n" % len(buf))
self.wfile.write(buf)
self.wfile.write(b"\r\n")
else:
self.wfile.write(buf)
sent += len(buf)
if remaining is not None:
remaining -= len(buf)
continue
if finished() and sent >= self._size(path):
break
try:
self._wait_for_bytes(path, sent, finished)
except TimeoutError:
break
if chunked:
self.wfile.write(b"0\r\n\r\n")
except (BrokenPipeError, ConnectionResetError):
# Normal: a probe reads the header then hangs up.
pass
return Handler
# --------------------------------------------------------------------------
def reset_work_root(path):
"""Clear leftover session directories at startup. Returns bytes reclaimed.
The session map is in memory only, so anything already in the work root is
unreachable after a restart: it can never be served (no session to find) and
never be evicted (the cache accounting only sums tracked sessions). Since the
work root is a tmpfs, leaving it there leaks RAM until the next reboot -- a
restart with 1.56 GB cached stranded exactly that much.
"""
reclaimed = 0
for name in os.listdir(path) if os.path.isdir(path) else []:
stale = os.path.join(path, name)
if not os.path.isdir(stale):
continue
out = os.path.join(stale, "out.mp4")
if os.path.exists(out):
reclaimed += os.path.getsize(out)
shutil.rmtree(stale, ignore_errors=True)
return reclaimed
def main():
ap = argparse.ArgumentParser(
description="just-in-time YouTube streaming proxy for Jellyfin")
ap.add_argument("--host", default="127.0.0.1")
ap.add_argument("--port", type=int, default=8099)
ap.add_argument("--work", default="/dev/shm/ytstream")
ap.add_argument("--cache-gb", type=float, default=8.0)
ap.add_argument("--max-pipelines", type=int, default=2)
ap.add_argument("--wait-timeout", type=float, default=300.0,
help="hard cap on blocking for the mux; only reachable when "
"--first-byte-grace is raised to meet it")
ap.add_argument("--first-byte-grace", type=float, default=FIRST_BYTE_GRACE,
help="seconds to wait for a complete mux before streaming the "
"partial file instead. Default is to wait for the whole "
"thing: a partial fragmented MP4 makes Jellyfin transcode "
"after a 1 GB probe. Only set this for a client that "
"tolerates an unseekable, duration-less stream")
ap.add_argument("--max-starts", type=int, default=20,
help="max cold starts per window; bounds a runaway library "
"refresh (default 20)")
ap.add_argument("--starts-window", type=float, default=3600.0,
help="rate-limit window in seconds (default 3600)")
ap.add_argument("--max-retries", type=int, default=2,
help="retries after a failed pipeline, each with a fresh "
"extraction (intermittent 403s do happen)")
ap.add_argument("--no-fetch", action="store_true",
help="never start a pipeline; log and 503. Use for a first "
"library scan to detect probing with no YouTube traffic")
ap.add_argument("--growing", action="store_true",
help="never wait: serve as soon as a byte exists, which is "
"--first-byte-grace 0")
ap.add_argument("--access-log", default=None)
args = ap.parse_args()
if not shutil.which("yt-dlp") or not shutil.which("ffmpeg"):
sys.exit("need yt-dlp and ffmpeg on PATH")
ver = subprocess.run(["yt-dlp", "--version"], capture_output=True, text=True)
version = ver.stdout.strip()
if version < "2025":
log(f"WARNING: yt-dlp {version} looks far too old, and a POT plugin is "
f"required. Expected ytstream's venv (/var/lib/ytstream/venv/bin) "
f"first on PATH -- the unit pins it, a manual run must too.")
else:
log(f"yt-dlp {version}")
global _access_log
_access_log = args.access_log
os.makedirs(args.work, exist_ok=True)
stranded = reset_work_root(args.work)
if stranded:
log(f"cleared {stranded / 2**30:.2f} GB of untracked cache from "
f"{args.work} left by a previous run")
mgr = Manager(args.work, args.max_pipelines,
int(args.cache_gb * 2**30), args.no_fetch, args.growing,
args.max_retries, args.max_starts, args.starts_window)
grace = 0.0 if args.growing else min(args.first_byte_grace, args.wait_timeout)
mgr.strict = not args.growing and grace >= args.wait_timeout
mgr.first_byte_grace = grace
if args.no_fetch:
log("NO-FETCH MODE: every /watch request will be logged and refused. "
"No YouTube traffic will be generated.")
if grace <= 0:
log("growing mode: a first play seeks badly and shows no duration until "
"the mux lands")
elif mgr.strict:
log(f"waiting for a complete mux, up to {grace:g}s. A cold long video is "
f"slow to start and the fetch outlives the request, so a retry is "
f"instant -- see FIRST_BYTE_GRACE for why partial is worse")
else:
log(f"waiting up to {grace:g}s for a complete mux, then streaming the "
f"partial file")
srv = ThreadingHTTPServer((args.host, args.port),
make_handler(mgr, args.wait_timeout, grace))
log(f"listening on http://{args.host}:{args.port} "
f"(/watch/<video_id>, /healthz)")
try:
srv.serve_forever()
except KeyboardInterrupt:
log("shutting down")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink