diff --git a/backend/app/plex/subtitles.py b/backend/app/plex/subtitles.py index 770c10b0..ea685cfd 100644 --- a/backend/app/plex/subtitles.py +++ b/backend/app/plex/subtitles.py @@ -1,15 +1,57 @@ -"""WebVTT conversion, HLS-clock time-shift, and on-disk cache for Plex subtitle streams. +"""WebVTT conversion, HLS-clock time-shift, and the on-disk cache + background extractor for Plex +subtitle streams. Text subtitles (embedded or external) are turned into WebVTT so the browser can render them; bitmap subtitles (Blu-ray/DVD) can't be (would need burn-in). Converted VTTs are cached on the shared Plex on-disk cache root (``cache.CACHE_DIR``). + +WHY THERE IS A BACKGROUND EXTRACTOR HERE — measured on prod, 2026-09-10, against a 17.06 GB 1080p +MKV with two embedded SubRip tracks on the 81 TB array: + +=========================================================== ======== + full pass, ONE track (what this module used to do) 154 s + full pass, BOTH text tracks in a SINGLE ffmpeg run 111 s + `-c:s copy` instead of converting to WebVTT 143 s + from 1:00:00 to the end of the file 66 s + a 630 s WINDOW at 1:00:00, both tracks 8 s +=========================================================== ======== + +The conversion is not the cost (dropping it saved 7%): ffmpeg has to stream every byte of the +container between its seek point and its stop point, and the array delivers ~140 MB/s — so the time +is (bytes read) / 140 MB/s, and the ONLY lever is reading less. Hence: + +* extract a WINDOW around the playhead first — the viewer sees subtitles in seconds — then complete + the whole track in the background so every later seek is instant and the next session is free; +* extract EVERY embedded text track in ONE pass, because the read is shared: picking Hungarian used + to cost a full read and then switching to English cost another one (the "hektikus" report); +* never do any of it inside a request (it froze the with no feedback AND held an API worker + thread for the whole read). + +`-copyts` CANNOT be used for a window: it makes ffmpeg ignore `-t`/`-to` (measured — a [0, 600 s] +window still took the full 145 s and emitted the whole track), so windows are extracted on a +zero-based clock and shifted back to absolute time here with `shift_vtt`. """ +import json +import logging import re import subprocess +import tempfile +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from sqlalchemy.orm import Session + +from app.db import SessionLocal +from app.models import PlexItem +from app.plex import paths as plex_paths from app.plex.cache import CACHE_DIR -from app.plex.client import PlexError +from app.plex.client import PlexClient, PlexError +from app.plex.lookups import by_rating_key + +log = logging.getLogger("siftlode.plex") # Image-based subtitle codecs (Blu-ray/DVD) — can't be turned into text WebVTT (would need burn-in). BITMAP_SUBS = { @@ -109,7 +151,7 @@ def ffmpeg_to_vtt(src: str, sub_ord: int | None) -> str: "webvtt", "pipe:1", ] - r = subprocess.run(cmd, capture_output=True) + r = subprocess.run(cmd, capture_output=True, timeout=EXTRACT_TIMEOUT_S) if r.returncode != 0 or not r.stdout: raise PlexError( (r.stderr or b"").decode("utf-8", "replace")[:200] or "ffmpeg subtitle convert failed" @@ -133,3 +175,329 @@ def cache_write(key: str, vtt: str) -> None: tmp.replace(CACHE_DIR / f"{key}.vtt") # atomic publish except OSError: pass + + +# --- cache identity ----------------------------------------------------------------------------- + +# The cached VTT holds ABSOLUTE cue times (per rating_key + subtitle ordinal); the per-session shift +# is applied on the way out, so a resume/seek to any offset reuses the same file. Beside it, a small +# JSON sidecar records WHICH absolute-time ranges the file actually covers — a windowed extraction +# publishes a usable but incomplete track, and "incomplete" must never be mistaken for "all there is". +_FULL_END_S = 1_000_000_000.0 # stand-in for "to the end of the file" (JSON can't hold infinity) + + +def cache_key(rating_key: str, sub_ord: int) -> str: + return f"sub_{rating_key}_{sub_ord}" + + +def _vtt_path(key: str) -> Path: + return CACHE_DIR / f"{key}.vtt" + + +def _meta_path(key: str) -> Path: + return CACHE_DIR / f"{key}.json" + + +def read_meta(key: str) -> tuple[list[list[float]], bool]: + """Covered absolute-time ranges + whether the track is fully extracted. A VTT with no sidecar + (written by an older build, which only ever cached complete tracks) counts as complete.""" + try: + raw = json.loads(_meta_path(key).read_text(encoding="utf-8")) + return [[float(a), float(b)] for a, b in raw.get("ranges", [])], bool(raw.get("complete")) + except (OSError, ValueError, TypeError): + return ([[0.0, _FULL_END_S]], True) if _vtt_path(key).exists() else ([], False) + + +def _write_meta(key: str, ranges: list[list[float]], complete: bool) -> None: + try: + CACHE_DIR.mkdir(parents=True, exist_ok=True) + tmp = CACHE_DIR / f"{key}.json.tmp" + tmp.write_text(json.dumps({"ranges": ranges, "complete": complete}), encoding="utf-8") + tmp.replace(_meta_path(key)) + except OSError: + pass + + +def generation(key: str) -> int: + """Monotonic id of the cached file's current content. It goes into the URL so that when a + window is superseded by a wider one the browser refetches instead of keeping the short VTT.""" + try: + return _vtt_path(key).stat().st_mtime_ns + except OSError: + return 0 + + +def merge_ranges(ranges: list[list[float]]) -> list[list[float]]: + """Sort + coalesce touching/overlapping covered ranges.""" + out: list[list[float]] = [] + for a, b in sorted((r for r in ranges if r[1] > r[0]), key=lambda r: r[0]): + if out and a <= out[-1][1]: + out[-1][1] = max(out[-1][1], b) + else: + out.append([a, b]) + return out + + +def covers(ranges: list[list[float]], at: float) -> bool: + return any(a <= at <= b for a, b in ranges) + + +def merge_vtt(*sources: str) -> str: + """Union of the cue blocks of several absolute-time VTTs, ordered by start time. + + Two extraction passes over the same file emit identical cues for the region they share, so + de-duplication keys off the TIMING line — the one part of a cue that identifies it uniquely and + that neither pass can renumber (ffmpeg's webvtt muxer writes no cue identifiers).""" + seen: set[str] = set() + cues: list[tuple[float, str]] = [] + for src in sources: + for block in re.split(r"\n[ \t]*\n", src or ""): + b = block.strip() + if not b or "-->" not in b: + continue # header / NOTE / STYLE — the merged file gets its own single header + timing = next(ln for ln in b.split("\n") if "-->" in ln) + if timing in seen: + continue + seen.add(timing) + try: + start = _vtt_ts_to_s(timing.partition("-->")[0].strip()) + except (ValueError, IndexError): + continue + cues.append((start, b)) + cues.sort(key=lambda c: c[0]) + return "WEBVTT\n\n" + "\n\n".join(b for _, b in cues) + "\n" + + +def publish(key: str, vtt: str, covered: tuple[float, float], complete: bool) -> None: + """Merge a freshly extracted region into the cached track and widen its covered ranges.""" + old = cache_read(key) + old_ranges, was_complete = read_meta(key) + cache_write(key, merge_vtt(old or "", vtt)) + _write_meta(key, merge_ranges(old_ranges + [list(covered)]), complete or was_complete) + + +# --- background extraction ---------------------------------------------------------------------- + +# How much of the track to extract before the viewer sees anything. WINDOW_PRE_S starts a little +# before the playhead so a cue that is already on screen is not missed; WINDOW_LEN_S is the span +# published as the first, provisional VTT (~4 s of work on the measured file). It only has to hold +# until the full pass lands behind it — 111 s for that file — so it is deliberately much longer than +# any realistic full pass. +WINDOW_PRE_S = 30.0 +WINDOW_LEN_S = 300.0 +# Ceiling on one ffmpeg run so a wedged extraction can't hold a pool thread forever (the old +# in-request extraction had no timeout at all). +EXTRACT_TIMEOUT_S = 900.0 +# A failed track is remembered briefly so a 1 s status poll doesn't respawn ffmpeg every second, and +# forgotten again soon after so fixing the file (or Plex coming back) doesn't need a restart. +FAIL_TTL_S = 60.0 +# Two items' full passes plus a window for a third: enough that a seek into an un-extracted region is +# served while a full pass runs, bounded so a burst can't thrash the array. +_POOL = ThreadPoolExecutor(max_workers=3, thread_name_prefix="plex-subs") + +_LOCK = threading.Lock() +_INFLIGHT: set[tuple] = set() +_FAILURES: dict[str, tuple[str, float]] = {} + + +class SubtitleUnavailable(Exception): + """A track that cannot be served as text, carrying the stable code the route maps to a status.""" + + def __init__(self, code: str): + super().__init__(code) + self.code = code + + +# code -> HTTP status. An extraction/upstream failure is 422 with a structured detail (502 is +# reserved for the raw byte proxies); the rest are genuine "no such thing" / "not text". +UNAVAILABLE_STATUS = { + "no_item": 404, + "no_track": 404, + "no_file": 404, + "not_fetchable": 404, + "bitmap": 415, + "extract_failed": 422, + "plex_error": 422, +} + + +def reset_state() -> None: + """Drop the in-flight/failure registries (tests; nothing in the app needs it).""" + with _LOCK: + _INFLIGHT.clear() + _FAILURES.clear() + + +def _fail(key: str, code: str) -> None: + with _LOCK: + _FAILURES[key] = (code, time.monotonic()) + + +def _recent_failure(key: str) -> str | None: + with _LOCK: + hit = _FAILURES.get(key) + if hit and time.monotonic() - hit[1] < FAIL_TTL_S: + return hit[0] + _FAILURES.pop(key, None) + return None + + +def ensure(rating_key: str, sub_ord: int, at: float = 0.0) -> dict: + """State of one subtitle track, scheduling whatever extraction is still missing. NEVER blocks. + + ``ready`` = fully extracted · ``partial`` = usable, covers ``at``, still growing · ``preparing`` + = nothing to show for ``at`` yet · ``failed`` = ``code`` says why. ``gen`` changes whenever the + cached file grows, so the client can refetch a window that has been superseded.""" + key = cache_key(rating_key, sub_ord) + ranges, complete = read_meta(key) + have = cache_read(key) is not None + covered = have and covers(ranges, at) + if complete and have: + return {"state": "ready", "code": None, "gen": generation(key)} + code = _recent_failure(key) + if code and not covered: + return {"state": "failed", "code": code, "gen": generation(key)} + if not code: + _schedule(rating_key, sub_ord, at, want_window=not covered) + return {"state": "partial" if covered else "preparing", "code": None, "gen": generation(key)} + + +def _schedule(rating_key: str, sub_ord: int, at: float, want_window: bool) -> None: + """Kick off the jobs this track still needs: the window that makes it watchable now, and the one + full pass that makes it complete. Each is deduplicated by identity, so a 1 s status poll (or a + second viewer) joins the running job instead of starting another ffmpeg.""" + jobs: list[tuple[tuple, float | None]] = [] + if want_window: + w_start = max(0.0, at - WINDOW_PRE_S) + jobs.append((("win", rating_key, round(w_start)), w_start)) + jobs.append((("full", rating_key), None)) + for ident, window_start in jobs: + with _LOCK: + if ident in _INFLIGHT: + continue + _INFLIGHT.add(ident) + _POOL.submit(_run, ident, rating_key, sub_ord, window_start) + + +def _run(ident: tuple, rating_key: str, sub_ord: int, window_start: float | None) -> None: + try: + _extract_and_publish(rating_key, sub_ord, window_start) + except SubtitleUnavailable as e: + _fail(cache_key(rating_key, sub_ord), e.code) + except PlexError as e: + log.warning("subtitle extraction failed for %s/%s: %s", rating_key, sub_ord, e) + _fail(cache_key(rating_key, sub_ord), "plex_error") + except ( + Exception + ): # a worker thread has nobody to raise to — record it, let the next poll see it + log.exception("subtitle extraction crashed for %s/%s", rating_key, sub_ord) + _fail(cache_key(rating_key, sub_ord), "extract_failed") + finally: + with _LOCK: + _INFLIGHT.discard(ident) + + +def _text_streams(db: Session, rating_key: str) -> tuple[PlexItem, list[dict]]: + """The item plus its subtitle streams, in the ordinal order the API and the UI both use.""" + it = by_rating_key(db, PlexItem, rating_key) + if it is None: + raise SubtitleUnavailable("no_item") + with PlexClient(db) as plex: + part = ((plex.metadata(it.rating_key) or {}).get("Media") or [{}])[0].get("Part") or [{}] + return it, [s for s in (part[0].get("Stream") or []) if s.get("streamType") == 3] + + +def _is_bitmap(stream: dict) -> bool: + return (stream.get("codec") or stream.get("format") or "").lower() in BITMAP_SUBS + + +def _extract_and_publish(rating_key: str, sub_ord: int, window_start: float | None) -> None: + """One extraction job. A sidecar track is fetched whole from Plex (cheap, no ffmpeg); an embedded + one is extracted together with EVERY other embedded text track in the file, because they all ride + on the same single read — the whole point of the measurement at the top of this module.""" + with SessionLocal() as db: + it, subs = _text_streams(db, rating_key) + if sub_ord < 0 or sub_ord >= len(subs): + raise SubtitleUnavailable("no_track") + want = subs[sub_ord] + if _is_bitmap(want): + raise SubtitleUnavailable("bitmap") + if want.get("key"): # external sidecar → one fetch, always the complete track + if window_start is not None: + return # nothing to window — the full job serves it in a single fetch + vtt = _sidecar_vtt(db, want) + publish(cache_key(rating_key, sub_ord), vtt, (0.0, _FULL_END_S), True) + return + if want.get("index") is None: + raise SubtitleUnavailable("not_fetchable") + src = plex_paths.local_media_path(db, it.file_path) + if src is None: + raise SubtitleUnavailable("no_file") + # Ordinal among the file's EMBEDDED subtitle streams — what `-map 0:s:N` counts. Sidecars sit + # in the same list but not in the file, so they must not shift the mapping. + targets = { + sum(1 for x in subs[:i] if x.get("index") is not None): i + for i, s in enumerate(subs) + if s.get("index") is not None and not _is_bitmap(s) + } + start = window_start or 0.0 + length = (WINDOW_LEN_S + WINDOW_PRE_S) if window_start is not None else None + end = _FULL_END_S if length is None else start + length + for emb_ord, vtt in _ffmpeg_extract(str(src), sorted(targets), window_start, length).items(): + publish( + cache_key(rating_key, targets[emb_ord]), + shift_vtt(vtt, -start), # zero-based window clock → absolute cue times + (start, end), + length is None, + ) + + +def _sidecar_vtt(db: Session, stream: dict) -> str: + """An external subtitle file, fetched through Plex and converted to WebVTT.""" + with PlexClient(db) as plex: + raw = plex.raw_get(stream["key"]) + codec = (stream.get("codec") or stream.get("format") or "").lower() + if codec in ("srt", "subrip", ""): + return srt_to_vtt(raw.decode("utf-8", "replace")) + with tempfile.NamedTemporaryFile(suffix="." + (codec or "sub"), delete=False) as fh: + fh.write(raw) + tmp = fh.name + try: + return ffmpeg_to_vtt(tmp, None) + finally: + Path(tmp).unlink(missing_ok=True) + + +def _ffmpeg_extract( + src: str, emb_ords: list[int], start: float | None, length: float | None +) -> dict[int, str]: + """Extract several embedded subtitle streams to WebVTT in ONE ffmpeg run (= one read of the file). + + `start`/`length` bound the read to a window. Without `-copyts` the cues come out on a zero-based + clock (the caller shifts them back) — which is exactly why `-copyts` is absent: with it ffmpeg + ignores `-t` and reads the whole container anyway (measured).""" + if not emb_ords: + return {} + with tempfile.TemporaryDirectory(prefix="plexsub-") as tmpdir: + cmd = ["ffmpeg", "-nostdin", "-loglevel", "error"] + if start: + cmd += ["-ss", f"{start:.3f}"] + cmd += ["-i", src] + outs = {o: Path(tmpdir) / f"{o}.vtt" for o in emb_ords} + for emb_ord, out in outs.items(): + cmd += ["-map", f"0:s:{emb_ord}"] + if length is not None: + cmd += ["-t", f"{length:.3f}"] + cmd += ["-f", "webvtt", "-y", str(out)] + r = subprocess.run(cmd, capture_output=True, timeout=EXTRACT_TIMEOUT_S) + got = { + o: p.read_text(encoding="utf-8", errors="replace") + for o, p in outs.items() + if p.exists() + } + if not got: + raise SubtitleUnavailable("extract_failed") + if r.returncode != 0: + # Partial output with a non-zero exit still beats nothing (a later pass widens it) — say so. + log.warning("ffmpeg subtitle pass exited %s: %s", r.returncode, (r.stderr or b"")[:200]) + return got diff --git a/backend/app/routes/plex/playback.py b/backend/app/routes/plex/playback.py index 1a25750c..c6e6d1dd 100644 --- a/backend/app/routes/plex/playback.py +++ b/backend/app/routes/plex/playback.py @@ -1,8 +1,6 @@ """Plex playback: image/subtitle proxies, HLS streaming, and watch-state writes.""" import hashlib -import os -import tempfile import threading from datetime import datetime, timedelta, timezone @@ -89,62 +87,45 @@ def image(rating_key: str, request: Request, variant: str = "thumb") -> Response return imgcache.response(data, ct) +@router.get("/subtitle/{rating_key}/{sub_ord}/status") +def subtitle_status(rating_key: str, sub_ord: int, request: Request, at: float = 0.0) -> dict: + """Is this subtitle track showable yet — and if not, get it moving. Extraction runs in the + background (see plex/subtitles.py for the measurements that forced it there), so the player polls + this while it waits instead of hanging on a that will not load for a minute or more. + + `at` is the viewer's absolute position: the extractor windows around it, so what gets read first + is the part actually about to be watched.""" + _require_session(request) + return subtitles.ensure(rating_key, sub_ord, at) + + @router.get("/subtitle/{rating_key}/{sub_ord}") -def subtitle(rating_key: str, sub_ord: int, request: Request, offset: float = 0.0) -> Response: +def subtitle( + rating_key: str, sub_ord: int, request: Request, offset: float = 0.0, at: float = 0.0 +) -> Response: """Serve the ord-th subtitle track as WebVTT for the browser's — so choosing a subtitle overlays it directly (no HLS re-mux / session restart). Handles EXTERNAL sidecar subs (fetched from Plex via the stream `key` — the old `-map 0:s` broke on these since they're not in the file) - AND embedded text subs (extracted from the local file). Image subs (PGS/VobSub) → 415. Disk-cached.""" + AND embedded text subs (extracted from the local file). Image subs (PGS/VobSub) → 415. Disk-cached. + + NEVER extracts inline. A cold track answers 202 and the extraction proceeds in the background; the + player mounts the only once /status says there is something to show, so in practice this + is only ever hit warm (or on a cache eviction race, which the 202 sends back to polling).""" _require_session(request) - # Cache the ABSOLUTE-time VTT once (per rk+ord); the per-session cue-shift is applied on the way - # out, so a resume/seek to any offset reuses the same cached track (no cache-per-offset explosion). - cache_key = f"sub_{rating_key}_{sub_ord}" - hit = subtitles.cache_read(cache_key) - if hit is not None: - return Response(subtitles.shift_vtt(hit, offset), media_type="text/vtt") - try: - with SessionLocal() as db: - it = item_or_404(db, rating_key) - with PlexClient(db) as plex: - part = ((plex.metadata(it.rating_key) or {}).get("Media") or [{}])[0].get( - "Part" - ) or [{}] - subs = [s for s in (part[0].get("Stream") or []) if s.get("streamType") == 3] - if sub_ord < 0 or sub_ord >= len(subs): - raise HTTPException(status_code=404, detail="No such subtitle") - s = subs[sub_ord] - codec = (s.get("codec") or s.get("format") or "").lower() - if codec in subtitles.BITMAP_SUBS: - raise HTTPException( - status_code=415, - detail="Image-based subtitle can't be shown as text", - ) - if s.get("key"): # external sidecar → fetch from Plex, convert to VTT - raw = plex.raw_get(s["key"]) - if codec in ("srt", "subrip", ""): - vtt = subtitles.srt_to_vtt(raw.decode("utf-8", "replace")) - else: - fd, tmp = tempfile.mkstemp(suffix="." + (codec or "sub")) - try: - os.write(fd, raw) - os.close(fd) - vtt = subtitles.ffmpeg_to_vtt(tmp, None) - finally: - os.unlink(tmp) - elif s.get("index") is not None: # embedded → extract from the local file - src = plex_paths.local_media_path(db, it.file_path) - if src is None: - raise HTTPException(status_code=404, detail="Media file not found on disk") - emb_ord = sum(1 for x in subs[:sub_ord] if x.get("index") is not None) - vtt = subtitles.ffmpeg_to_vtt(str(src), emb_ord) - else: - raise HTTPException(status_code=404, detail="Subtitle not fetchable") - except PlexNotConfigured as e: - raise HTTPException(status_code=400, detail=str(e)) - except PlexError as e: - raise HTTPException(status_code=502, detail=f"Subtitle fetch failed: {e}") - subtitles.cache_write(cache_key, vtt) - return Response(subtitles.shift_vtt(vtt, offset), media_type="text/vtt") + # The cached VTT holds ABSOLUTE cue times (per rk+ord); the per-session cue-shift is applied on + # the way out, so a resume/seek to any offset reuses the same track (no cache-per-offset explosion). + st = subtitles.ensure(rating_key, sub_ord, at) + if st["state"] == "failed": + raise HTTPException( + status_code=subtitles.UNAVAILABLE_STATUS.get(st["code"], 422), + detail={"code": "subtitle", "reason": st["code"]}, + ) + hit = subtitles.cache_read(subtitles.cache_key(rating_key, sub_ord)) + if hit is None: + return Response( + status_code=202, media_type="application/json", content='{"state":"preparing"}' + ) + return Response(subtitles.shift_vtt(hit, offset), media_type="text/vtt") # Resume/watched thresholds. _PROGRESS_MIN_S matches the feed's PROGRESS_MIN_SECONDS (5), but diff --git a/backend/tests/test_plex_subtitle_prepare.py b/backend/tests/test_plex_subtitle_prepare.py new file mode 100644 index 00000000..128a84eb --- /dev/null +++ b/backend/tests/test_plex_subtitle_prepare.py @@ -0,0 +1,263 @@ +"""Subtitle extraction must not happen inside the request, and must read as little as possible. + +Bug hunt 2026-09, item 8 — "switching the subtitle track shows nothing; after a few stop/starts it +suddenly works". Measured on prod against a 17.06 GB MKV with two embedded SubRip tracks: + + full pass, one track 154 s | both tracks in ONE run 111 s | a 630 s window at 1:00:00 8 s + +The cost is not the WebVTT conversion, it is that ffmpeg streams every byte of the container between +its seek point and its stop point. The old endpoint did that pass INSIDE the 's request, once +per track, with no timeout — so the browser sat on an empty track for minutes (and an API worker +thread sat with it), and picking the second language cost the whole read again. + +These tests pin the three things that fixes it: the request never waits for ffmpeg, the extraction +publishes a window around the playhead before the whole track, and one ffmpeg run covers every +embedded text track. Pure — no Postgres, no ffmpeg. +""" + +import subprocess +import threading +import time +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +import app.state as app_state +from app.main import app +from app.plex import subtitles +from app.routes.plex import playback + + +@pytest.fixture(autouse=True) +def _isolated_cache(tmp_path, monkeypatch): + """A throwaway cache root + empty registries: this module's state is process-global.""" + monkeypatch.setattr(subtitles, "CACHE_DIR", tmp_path) + subtitles.reset_state() + yield + subtitles.reset_state() + + +@pytest.fixture +def client(monkeypatch) -> TestClient: + # The subtitle endpoints authenticate off the signed session cookie directly (not a FastAPI + # dependency), so there is no dependency_overrides hook to use. + monkeypatch.setattr(playback, "resolved_user_id", lambda request: (1, None)) + monkeypatch.setattr(app_state, "_configured_cache", True) # pass the instance-setup gate, no DB + return TestClient(app) + + +class _ManualPool: + """Deterministic stand-in for the extraction thread pool: jobs run when the test says so.""" + + def __init__(self) -> None: + self.jobs: list[tuple] = [] + + def submit(self, fn, *args) -> None: + self.jobs.append((fn, args)) + + def run_next(self) -> None: + fn, args = self.jobs.pop(0) + fn(*args) + + def run_all(self) -> int: + jobs, self.jobs = self.jobs, [] + for fn, args in jobs: + fn(*args) + return len(jobs) + + +def _cue(start: str, end: str, text: str) -> str: + return f"{start} --> {end}\n{text}" + + +def _vtt(*cues: str) -> str: + return "WEBVTT\n\n" + "\n\n".join(cues) + "\n" + + +# --- the request must never wait for the extraction --------------------------------------------- + + +def test_a_cold_track_answers_at_once_instead_of_waiting_for_ffmpeg(client, monkeypatch): + """THE regression. A cold track used to block the request for the length of a full file read; + now it answers 202 and the extraction runs behind it. Timed, because "does it block?" is the + whole question — the fake extraction holds a worker for 5 s, which the request must not feel.""" + started = threading.Event() + release = threading.Event() + + def _slow(rating_key, sub_ord, window_start): + started.set() + release.wait(timeout=5) + + monkeypatch.setattr(subtitles, "_extract_and_publish", _slow) + try: + t0 = time.monotonic() + resp = client.get("/api/plex/subtitle/98614/0") + elapsed = time.monotonic() - t0 + assert resp.status_code == 202, resp.text + assert elapsed < 2.0, f"the request waited {elapsed:.1f}s on the extraction" + assert started.wait(timeout=5), "the extraction never started in the background" + # …and the status endpoint answers just as fast while that extraction is still running. + t0 = time.monotonic() + st = client.get("/api/plex/subtitle/98614/0/status") + assert st.status_code == 200 and st.json()["state"] == "preparing" + assert time.monotonic() - t0 < 2.0 + finally: + release.set() + + +def test_a_burst_of_polls_starts_one_window_and_one_full_pass(monkeypatch): + """The player polls once a second while it waits. Each poll must JOIN the running jobs, not + start another ffmpeg over the same 17 GB.""" + pool = _ManualPool() + monkeypatch.setattr(subtitles, "_POOL", pool) + for _ in range(5): + subtitles.ensure("98614", 0, at=600.0) + assert len(pool.jobs) == 2 # one window (at 600 s) + the one full pass + # submit(_run, ident, rating_key, sub_ord, window_start) — the last arg is what distinguishes them + assert [a[3] for _, a in pool.jobs] == [570.0, None] # the window first, then the full pass + + +# --- window first, whole track after ------------------------------------------------------------ + + +def test_status_walks_preparing_to_partial_to_ready(client, monkeypatch): + """A window around the playhead makes the track showable long before it is complete, and the + generation changes when it grows so the browser refetches instead of keeping the short VTT.""" + pool = _ManualPool() + monkeypatch.setattr(subtitles, "_POOL", pool) + key = subtitles.cache_key("98614", 0) + + def _publish(rating_key, sub_ord, window_start): + if window_start is not None: + subtitles.publish( + key, _vtt(_cue("59:54.187", "59:58.650", "window")), (3570.0, 3900.0), False + ) + else: + subtitles.publish(key, _vtt(_cue("01:12.000", "01:15.000", "whole")), (0.0, 1e9), True) + + monkeypatch.setattr(subtitles, "_extract_and_publish", _publish) + + first = client.get("/api/plex/subtitle/98614/0/status?at=3600").json() + assert first["state"] == "preparing" and first["gen"] == 0 + + pool.run_next() # the window job, submitted first — the full pass stays queued behind it + windowed = client.get("/api/plex/subtitle/98614/0/status?at=3600").json() + assert windowed["state"] == "partial", "a window around the playhead is showable" + assert windowed["gen"] > 0 + # …and the request now gets real WebVTT rather than another 202. + served = client.get("/api/plex/subtitle/98614/0?at=3600") + assert served.status_code == 200 and "window" in served.text + + pool.run_next() # the full pass, scheduled by that same first poll + done = client.get("/api/plex/subtitle/98614/0/status?at=3600").json() + assert done["state"] == "ready" + assert done["gen"] != windowed["gen"], "a widened track must not reuse the window's generation" + assert "whole" in client.get("/api/plex/subtitle/98614/0").text + + +def test_a_position_outside_the_extracted_window_is_not_called_showable(client, monkeypatch): + """Honest states: cues exist, but none for where the viewer just seeked to. Saying "ready" there + would put an empty track on screen with no explanation — the original complaint.""" + pool = _ManualPool() + monkeypatch.setattr(subtitles, "_POOL", pool) + subtitles.publish( + subtitles.cache_key("98614", 0), + _vtt(_cue("00:10.000", "00:12.000", "x")), + (0.0, 330.0), + False, + ) + assert client.get("/api/plex/subtitle/98614/0/status?at=100").json()["state"] == "partial" + assert client.get("/api/plex/subtitle/98614/0/status?at=5000").json()["state"] == "preparing" + + +def test_an_unshowable_track_says_why_and_maps_to_a_status(client, monkeypatch): + pool = _ManualPool() + monkeypatch.setattr(subtitles, "_POOL", pool) + + def _bitmap(rating_key, sub_ord, window_start): + raise subtitles.SubtitleUnavailable("bitmap") + + monkeypatch.setattr(subtitles, "_extract_and_publish", _bitmap) + client.get("/api/plex/subtitle/98614/0/status") + pool.run_all() + st = client.get("/api/plex/subtitle/98614/0/status").json() + assert st["state"] == "failed" and st["code"] == "bitmap" + resp = client.get("/api/plex/subtitle/98614/0") + assert resp.status_code == 415 # a structured detail, so the UI can translate it + assert resp.json()["detail"] == {"code": "subtitle", "reason": "bitmap"} + + +# --- the ffmpeg pass itself --------------------------------------------------------------------- + + +def _capture_ffmpeg(monkeypatch) -> list[list[str]]: + """Record the argv of each ffmpeg run and satisfy it by writing the outputs it asked for.""" + runs: list[list[str]] = [] + + def _fake_run(cmd, **kwargs): + runs.append(cmd) + for i, arg in enumerate(cmd): + if arg == "-y": + Path(cmd[i + 1]).write_text(_vtt(_cue("00:01.000", "00:02.000", "hi")), "utf-8") + return subprocess.CompletedProcess(cmd, 0, b"", b"") + + monkeypatch.setattr(subtitles.subprocess, "run", _fake_run) + return runs + + +def test_one_ffmpeg_run_extracts_every_embedded_text_track(monkeypatch): + """The read is the cost and it is SHARED. Extracting one track at a time is what made picking + Hungarian cost a full pass and then switching to English cost another one.""" + runs = _capture_ffmpeg(monkeypatch) + out = subtitles._ffmpeg_extract("/plex-media/f.mkv", [0, 1], None, None) + assert len(runs) == 1, "each extra ffmpeg run is another full read of the file" + assert set(out) == {0, 1} + assert runs[0].count("-map") == 2 + + +def test_a_windowed_pass_bounds_the_read_and_never_uses_copyts(monkeypatch): + """`-copyts` is the trap: with it ffmpeg IGNORES -t and reads the whole container anyway + (measured — a [0, 600 s] window still took the full 145 s). So the window is extracted on a + zero-based clock and shifted back to absolute time by the caller.""" + runs = _capture_ffmpeg(monkeypatch) + subtitles._ffmpeg_extract("/plex-media/f.mkv", [0], 3570.0, 330.0) + cmd = runs[0] + assert "-copyts" not in cmd + assert cmd[cmd.index("-ss") + 1].startswith( + "3570" + ) # input seek — before -i, so it is a real seek + assert cmd.index("-ss") < cmd.index("-i") + assert cmd[cmd.index("-t") + 1].startswith("330") + + +def test_the_full_pass_is_not_bounded(monkeypatch): + runs = _capture_ffmpeg(monkeypatch) + subtitles._ffmpeg_extract("/plex-media/f.mkv", [0], None, None) + assert "-t" not in runs[0] and "-ss" not in runs[0] + + +# --- the pure pieces the above rests on --------------------------------------------------------- + + +def test_merge_vtt_unions_windows_in_time_order(): + late = _vtt(_cue("59:54.187", "59:58.650", "late")) + early = _vtt(_cue("01:12.000", "01:15.000", "early"), _cue("59:54.187", "59:58.650", "late")) + merged = subtitles.merge_vtt(late, early) + assert merged.startswith("WEBVTT\n\n") + assert merged.count("-->") == 2, "the shared region must not be duplicated" + assert merged.index("early") < merged.index("late") + + +def test_covered_ranges_coalesce_and_answer_membership(): + ranges = subtitles.merge_ranges([[300.0, 600.0], [0.0, 330.0], [900.0, 1200.0]]) + assert ranges == [[0.0, 600.0], [900.0, 1200.0]] + assert subtitles.covers(ranges, 450.0) + assert not subtitles.covers(ranges, 700.0) + + +def test_a_cached_vtt_from_an_older_build_counts_as_complete(tmp_path): + """Before this change the cache only ever held whole tracks, and none of them has a sidecar. + Reading one as "covers nothing" would re-extract every already-cached track after the upgrade.""" + (tmp_path / "sub_98614_0.vtt").write_text(_vtt(_cue("00:01.000", "00:02.000", "x")), "utf-8") + assert subtitles.read_meta("sub_98614_0") == ([[0.0, 1_000_000_000.0]], True) diff --git a/frontend/src/components/PlexPlayer.tsx b/frontend/src/components/PlexPlayer.tsx index ff076f6e..43c7a747 100644 --- a/frontend/src/components/PlexPlayer.tsx +++ b/frontend/src/components/PlexPlayer.tsx @@ -41,6 +41,7 @@ import { import { Ctrl, PanelHead, Skip10 } from "./plexPlayerControls"; import { PlexPlayerSettings } from "./PlexPlayerSettings"; import { PlexStatsOverlay } from "./PlexStatsOverlay"; +import { createFragMeter } from "../lib/plexStats"; import { UpNextCard } from "./UpNextCard"; import { useAutoSkip, @@ -114,6 +115,10 @@ export default function PlexPlayer({ itemId, onClose, queue }: Props) { // stats mode row's "· HDR→SDR" tag; null/false for an 8-bit HW, software, or non-transcode session. const tonemapRef = useRef(null); const durationRef = useRef(0); + // Rolling per-fragment bitrate meter for the stats overlay. Fed from hls.js FRAG_LOADED below and + // reset with each session, because the master playlist's own BANDWIDTH is not a measurement — for a + // remux it is the AAC track's nominal 192k + 10%, i.e. a constant 211 kbps (see plexStats.ts). + const fragMeterRef = useRef(createFragMeter()); // ≥2 audio tracks → multi-rendition HLS: all tracks ship in one session and hls.js switches audio // client-side (no restart). Ref so the stable loadSession/changeAudio callbacks read it fresh. const multiAudioRef = useRef(false); @@ -303,6 +308,17 @@ export default function PlexPlayer({ itemId, onClose, queue }: Props) { // buffer forward up to maxBufferLength. Applies to remux + transcode alike (both use copyts). const hls = new Hls({ maxBufferLength: 30, enableWorker: true, startPosition: 0 }); hlsRef.current = hls; + // Measure the real delivered bitrate: bytes per second of MEDIA, per fragment. A new session + // is a new stream (different start offset, possibly a different audio rendition), so the + // rolling window starts empty rather than carrying the previous session's numbers over. + fragMeterRef.current.reset(); + hls.on(Hls.Events.FRAG_LOADED, (_e, data) => { + fragMeterRef.current.add({ + type: data.frag.type, + bytes: data.frag.stats?.total || data.frag.stats?.loaded || 0, + durationS: data.frag.duration, + }); + }); hls.loadSource(`${sess.url}?t=${Math.floor(startAt)}`); hls.attachMedia(video); // Subtitles are native s (not muxed into the HLS), so nothing subtitle-related here. @@ -361,6 +377,10 @@ export default function PlexPlayer({ itemId, onClose, queue }: Props) { textSubs, selectedSub, subTrackOffset, + subShowable, + subFail, + subGen, + onSubTrackError, } = usePlexTracks({ id, detail, @@ -851,15 +871,17 @@ export default function PlexPlayer({ itemId, onClose, queue }: Props) { > {/* Text subtitles as native tracks (same-origin → sent with the session cookie). The effect above flips modes; the browser fetches a track's VTT the first time it's shown. */} - {selectedSub && ( + {selectedSub && subShowable && ( )} @@ -894,6 +916,29 @@ export default function PlexPlayer({ itemId, onClose, queue }: Props) { )} + {/* Subtitle-extraction notice. Pulling a track out of a big MKV is a full read of the container + (154 s for a 17 GB film, measured), so it happens in the background — around the playhead + first. Until then the is deliberately NOT mounted, and this says why instead of + leaving the viewer to conclude that switching subtitles "does nothing" (bug hunt 2026-09, + item 8). It sits one rung above the track-switch OSD so the two never overlap. */} +
+ {selectedSub && !subShowable && ( +
+ {subFail ? ( + {t(`plex.player.subUnavailable.${subFail}`)} + ) : ( + <> + + {t("plex.player.subPreparing", { label: selectedSub.label })} + + )} +
+ )} +
+ {/* Keyboard track-switch OSD. Sits ABOVE the control bar's own space so it reads even while the bar is showing, and inside the player element so it survives fullscreen. `aria-live` because the whole point is confirming something you cannot otherwise see. @@ -1279,6 +1324,15 @@ export default function PlexPlayer({ itemId, onClose, queue }: Props) { a.ord === (audioOrd ?? 0))?.label ?? undefined + } + subLabel={ + subOrd == null + ? t("plex.player.subOff") + : (detail?.subtitle_streams.find((x) => x.ord === subOrd)?.label ?? undefined) + } playable={detail?.playable ?? ""} hwRef={hwRef} tonemapRef={tonemapRef} diff --git a/frontend/src/components/PlexStatsOverlay.tsx b/frontend/src/components/PlexStatsOverlay.tsx index 5fdd1029..0cadd95f 100644 --- a/frontend/src/components/PlexStatsOverlay.tsx +++ b/frontend/src/components/PlexStatsOverlay.tsx @@ -2,7 +2,7 @@ import { type RefObject, useEffect, useState } from "react"; import { useTranslation } from "react-i18next"; import type Hls from "hls.js"; import { X } from "lucide-react"; -import { buildStatRows, type HlsLevelInfo, type StatRow } from "../lib/plexStats"; +import { buildStatRows, type FragMeter, type HlsLevelInfo, type StatRow } from "../lib/plexStats"; // "Stats for nerds" debug overlay (E6 S2). Samples the live HTML5