peter
31dc4184f5
Deep-review round 4, backend half. - The periodic HLS reaper (round 3) applied the "finished and unfetched for 30 s" rule that only a NEW session request may apply. A remux finishes long before the film, and a paused player with a full buffer fetches nothing, so a one-minute pause killed the session and playback died ~30 s after resuming. The periodic pass is `reap_abandoned` now (the long idle limit, finished or not). - A subtitle `publish` that could not keep its output (full / read-only cache volume) returned quietly, so every poll queued the window and the whole-container pass again. It reports failure and the pass is backed off like a failed read; a cached VTT that cannot be READ is answered `unavailable` (`cache_unreadable`) instead of scheduling reads that can never land. - The subtitle routes bound `rating_key` / `sub_ord` (they become a cache file name: a 250-digit one was ENAMETOOLONG, which pathlib's exists() re-raises -> a 500), and the cache helpers no longer throw on such a name. - One app-wide handler turns psycopg's "text fields cannot contain NUL" DataError into a 422: round 3 guarded only the by-rating_key lookup, and `contains=`, `contains_group=`, reorder/bulk key lists and the collection search still 500'd. - The hand-written `db.get(...) is None -> 404 "Unknown ..."` siblings use `get_or_404` with a `what` (channel / video / user), so the Hungarian UI no longer shows English for a gone row.
852 lines
40 KiB
Python
852 lines
40 KiB
Python
"""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 <track> 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 <track> 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 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 = {
|
||
"pgs",
|
||
"hdmv_pgs_subtitle",
|
||
"dvd_subtitle",
|
||
"dvdsub",
|
||
"vobsub",
|
||
"dvbsub",
|
||
"dvb_subtitle",
|
||
}
|
||
|
||
|
||
def srt_to_vtt(text: str) -> str:
|
||
"""Minimal SubRip→WebVTT: add the header and switch the ms separator (`,`→`.`). SRT cue-index
|
||
lines are valid in WebVTT, so they're left as-is."""
|
||
body = re.sub(r"(\d\d:\d\d:\d\d),(\d\d\d)", r"\1.\2", text.replace("\r\n", "\n").lstrip(""))
|
||
return "WEBVTT\n\n" + body.strip() + "\n"
|
||
|
||
|
||
def _vtt_ts_to_s(ts: str) -> float:
|
||
# Also reads ffmpeg's spelling of a NEGATIVE time (a windowed pass, see `_ffmpeg_extract`): its
|
||
# webvtt muxer splits the milliseconds with C's truncating division, so every field carries the
|
||
# sign — -5.005 s is `00:-5.-05`, -65.3 s is `-1:-5.-300` — and summing the signed fields is exact.
|
||
parts = ts.split(":")
|
||
h = "0" if len(parts) == 2 else parts[0]
|
||
m, rest = parts[-2], parts[-1]
|
||
sec, _, ms = rest.partition(".")
|
||
return int(h) * 3600 + int(m) * 60 + int(sec) + int(ms or 0) / 1000.0
|
||
|
||
|
||
def _vtt_s_to_ts(x: float) -> str:
|
||
# Rounded to whole milliseconds FIRST, then split: rounding the fraction last carried a rounded-up
|
||
# millisecond into the seconds and no further, so 59.9996 came out as `00:00:60.000` — a timestamp
|
||
# the WebVTT parser rejects, silently dropping that cue.
|
||
h, rest = divmod(int(round(max(0.0, x) * 1000)), 3_600_000)
|
||
m, rest = divmod(rest, 60_000)
|
||
s, ms = divmod(rest, 1000)
|
||
return f"{h:02d}:{m:02d}:{s:02d}.{ms:03d}"
|
||
|
||
|
||
def shift_vtt(vtt: str, offset: float) -> str:
|
||
"""Subtract `offset` seconds from every cue's start/end. WebVTT cues carry ABSOLUTE content times,
|
||
but an HLS remux session's clock is zero-based at the real keyframe (see stream.py media_start_s),
|
||
so a subtitle overlaid on that session must be shifted by that offset to line up with speech.
|
||
|
||
Processes cue BLOCKS so a cue that shifts fully into the past (end <= 0) can be DROPPED entirely.
|
||
(Collapsing them to 00:00:00.000-->00:00:00.000 made EVERY past cue count as active at currentTime 0
|
||
— where a resume/seek lands — piling all of them on screen until playback advanced past 0.) A cue
|
||
straddling 0 is clamped to start at 0."""
|
||
if not offset:
|
||
return vtt
|
||
out_blocks: list[str] = []
|
||
for block in re.split(r"\n[ \t]*\n", vtt):
|
||
if "-->" not in block:
|
||
out_blocks.append(block) # header / NOTE / STYLE — keep verbatim
|
||
continue
|
||
lines = block.split("\n")
|
||
keep = True
|
||
for i, ln in enumerate(lines):
|
||
if "-->" not in ln:
|
||
continue
|
||
left, _, right = ln.partition("-->")
|
||
rparts = right.strip().split(None, 1)
|
||
settings = (" " + rparts[1]) if len(rparts) > 1 else ""
|
||
try:
|
||
st = _vtt_ts_to_s(left.strip()) - offset
|
||
en = _vtt_ts_to_s(rparts[0]) - offset
|
||
except (ValueError, IndexError):
|
||
break # unparseable timing → leave the block as-is
|
||
if en <= 0:
|
||
keep = False # whole cue is before the session start → drop the block
|
||
else:
|
||
lines[i] = (
|
||
f"{_vtt_s_to_ts(st)} --> {_vtt_s_to_ts(en)}{settings}" # _vtt_s_to_ts clamps st>=0
|
||
)
|
||
break
|
||
if keep:
|
||
out_blocks.append("\n".join(lines))
|
||
return "\n\n".join(out_blocks)
|
||
|
||
|
||
def ffmpeg_to_vtt(src: str, sub_ord: int | None) -> str:
|
||
"""Convert one subtitle stream of a file to WebVTT via ffmpeg (for embedded subs, and non-SRT
|
||
external subs written to a temp file). `sub_ord` = ordinal among the file's subtitle streams."""
|
||
cmd = [
|
||
"ffmpeg",
|
||
"-nostdin",
|
||
"-loglevel",
|
||
"error",
|
||
"-i",
|
||
src,
|
||
"-map",
|
||
f"0:s:{sub_ord if sub_ord is not None else 0}",
|
||
"-f",
|
||
"webvtt",
|
||
"pipe:1",
|
||
]
|
||
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"
|
||
)
|
||
return r.stdout.decode("utf-8", "replace")
|
||
|
||
|
||
def cache_read(key: str) -> str | None:
|
||
try:
|
||
p = CACHE_DIR / f"{key}.vtt"
|
||
return p.read_text(encoding="utf-8") if p.exists() else None
|
||
except OSError:
|
||
return None
|
||
|
||
|
||
def cache_write(key: str, vtt: str) -> bool:
|
||
"""Atomically replace the cached VTT; False when it could not be written (`publish` must then
|
||
not claim the region it was carrying)."""
|
||
try:
|
||
CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
||
tmp = CACHE_DIR / f"{key}.vtt.tmp"
|
||
tmp.write_text(vtt, encoding="utf-8")
|
||
tmp.replace(CACHE_DIR / f"{key}.vtt") # atomic publish
|
||
return True
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
# --- 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 _exists(_vtt_path(key)) else ([], False)
|
||
|
||
|
||
def _exists(p: Path) -> bool:
|
||
"""`Path.exists` that answers False for a path the OS refuses outright: pathlib swallows only
|
||
ENOENT-like errors, so an over-long name (ENAMETOOLONG, from an absurd rating_key or ordinal)
|
||
escaped as a 500 from the status route."""
|
||
try:
|
||
return p.exists()
|
||
except (OSError, ValueError):
|
||
return 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 <track> 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, ValueError): # ValueError: a NUL in the key — `stat` refuses the path outright
|
||
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 they
|
||
are de-duplicated — but NOT on the raw timing text. The two passes do not spell a timestamp the
|
||
same way: a window comes back through `shift_vtt`, which always writes `HH:MM:SS.mmm`, while a
|
||
full pass is ffmpeg's own output, and its webvtt muxer drops the hour below one hour
|
||
(`10:00.123`). Keyed on the text, every cue of a window in the first hour survived twice and the
|
||
viewer saw each line stacked on itself. So each timing line is parsed and re-emitted in one
|
||
canonical form, and the key is that canonical timing PLUS the cue text: two different cues that
|
||
happen to share a time (two speakers, a sign over dialogue) are both kept."""
|
||
seen: set[str] = set()
|
||
cues: list[tuple[float, str]] = []
|
||
for src in sources:
|
||
for block in re.split(r"\n[ \t]*\n", src or ""):
|
||
canon = _canonical_cue(block.strip())
|
||
if canon is None:
|
||
continue # header / NOTE / STYLE — the merged file gets its own single header
|
||
start, key, text = canon
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
cues.append((start, text))
|
||
cues.sort(key=lambda c: c[0])
|
||
return "WEBVTT\n\n" + "\n\n".join(b for _, b in cues) + "\n"
|
||
|
||
|
||
def _canonical_cue(block: str) -> tuple[float, str, str] | None:
|
||
"""(start seconds, identity, the block with its timing line canonicalised) — or None for a block
|
||
that is not a cue. The identity leaves out anything before the timing line (an SRT-style cue
|
||
number), which is a label, not content."""
|
||
lines = block.split("\n")
|
||
for i, ln in enumerate(lines):
|
||
if "-->" not in ln:
|
||
continue
|
||
left, _, right = ln.partition("-->")
|
||
rparts = right.strip().split(None, 1)
|
||
settings = (" " + rparts[1]) if len(rparts) > 1 else ""
|
||
try:
|
||
st = _vtt_ts_to_s(left.strip())
|
||
en = _vtt_ts_to_s(rparts[0])
|
||
except (ValueError, IndexError):
|
||
return None
|
||
lines[i] = f"{_vtt_s_to_ts(st)} --> {_vtt_s_to_ts(en)}{settings}"
|
||
return st, "\n".join(lines[i:]), "\n".join(lines)
|
||
return None
|
||
|
||
|
||
def _last_cue_end(vtt: str) -> float:
|
||
"""Where the last cue of a VTT ends — how far a pass that died early provably got."""
|
||
end = 0.0
|
||
for ln in vtt.split("\n"):
|
||
if "-->" not in ln:
|
||
continue
|
||
try:
|
||
end = max(end, _vtt_ts_to_s(ln.partition("-->")[2].strip().split(None, 1)[0]))
|
||
except (ValueError, IndexError):
|
||
continue
|
||
return end
|
||
|
||
|
||
def publish(key: str, vtt: str, covered: tuple[float, float], complete: bool) -> bool:
|
||
"""Merge a freshly extracted region into the cached track and widen its covered ranges. False
|
||
when the region did not land (the cache is unwritable, or the VTT there cannot be read) — the
|
||
caller must then record a FAILED pass: returning quietly let the next poll schedule the same read
|
||
again, a whole-container pass every time, while the player said "preparing" for good.
|
||
|
||
A read-merge-write, so it runs under `_PUBLISH_LOCK`: two jobs publish into the same keys (a
|
||
window and the full pass, and every embedded track of the file rides each of them), and without
|
||
the lock the later writer drops whatever the earlier one merged — permanently, once `complete` is
|
||
set. It is also the only writer of the cache files, which is what keeps their fixed `.tmp`
|
||
staging names safe."""
|
||
with _PUBLISH_LOCK:
|
||
old = cache_read(key)
|
||
if old is None and _vtt_path(key).exists():
|
||
# There, but unreadable (`cache_read` swallows the OSError). Merging into "" would replace
|
||
# a whole track with this one region while its sidecar still claimed all of it.
|
||
log.warning("subtitle cache %s unreadable; not publishing into it", key)
|
||
return False
|
||
old_ranges, was_complete = read_meta(key)
|
||
if old is None:
|
||
# A sidecar with no VTT behind it claims cues that are not there, so it is not merged
|
||
# into. And a VTT with no sidecar reads as a COMPLETE track (see read_meta — how older
|
||
# builds cached), while a publish is two renames: so before a key's first VTT lands, its
|
||
# sidecar is made to claim nothing. Dying between the two then leaves a window that is
|
||
# extracted again, never one taken for the whole track for good.
|
||
old_ranges, was_complete = [], False
|
||
_write_meta(key, [], False)
|
||
if not cache_write(key, merge_vtt(old or "", vtt)):
|
||
log.warning("subtitle cache %s could not be written", key)
|
||
return False # the region never landed, so the sidecar must not claim it
|
||
_write_meta(key, merge_ranges(old_ranges + [list(covered)]), complete or was_complete)
|
||
return True
|
||
|
||
|
||
# --- 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
|
||
# …but it does not always land: a full pass can be backed off (FULL_RETRY_S) or simply slower than
|
||
# the window lasts (a file past ~40 GB, or several passes sharing the array). A window was only asked
|
||
# for once the playhead had LEFT coverage, so every edge unmounted the track for a poll plus an
|
||
# extraction. Within this distance of a covered stretch's end, the next window is read ahead.
|
||
LOOKAHEAD_S = 60.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
|
||
# How long a timed-out ffmpeg gets, after a TERM, to close its outputs before it is killed.
|
||
_TERM_GRACE_S = 10.0
|
||
# A failed track (or window) 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
|
||
# …except a FULL pass that failed. That is a read of the whole container (minutes), a failure that
|
||
# reproduces does so every time, and the viewer is typically inside what it did publish — so the
|
||
# short memo re-armed a whole-file read every minute for as long as they watched. Windows around the
|
||
# playhead still run meanwhile; only the full pass waits this out.
|
||
FULL_RETRY_S = 3600.0
|
||
# The furthest position the status poll accepts. No film is this long, and an absurd one used to
|
||
# reach ffmpeg's `-ss` (which rejects anything past its int64 microseconds).
|
||
MAX_POSITION_S = 100_000.0
|
||
# Windows and full passes run on SEPARATE pools, so a window never waits behind a full pass. They
|
||
# shared one FIFO pool of three whose comment promised "room for a window" that nothing enforced:
|
||
# stepping through three episodes with subtitles on left three full passes (minutes each) holding every
|
||
# worker, and a seek window on the third queued behind them. Two of each: bounded, so a burst can't
|
||
# thrash the array.
|
||
_WIN_POOL = ThreadPoolExecutor(max_workers=2, thread_name_prefix="plex-subs-win")
|
||
_FULL_POOL = ThreadPoolExecutor(max_workers=2, thread_name_prefix="plex-subs-full")
|
||
# Bounds on what may wait for those pools — separate too, so held full-pass slots (minutes each)
|
||
# cannot refuse a window (seconds). The queues are unbounded and every job is a read of the library:
|
||
# a viewer scrubbing through a film queued a window per stop (the one where they stopped ran LAST),
|
||
# and a script could queue thousands. Past a bound nothing is queued — the next poll asks again, by
|
||
# which time something has finished.
|
||
_MAX_WINDOWS_INFLIGHT = 8
|
||
_MAX_FULL_INFLIGHT = 6
|
||
_MAX_TRACK_WINDOWS = 2
|
||
|
||
_LOCK = threading.Lock()
|
||
# Queued or running jobs → the absolute span a window job reads (None for a full pass).
|
||
_INFLIGHT: dict[tuple, tuple[float, float] | None] = {}
|
||
# Track-level failures: a track that cannot be served at all (no item, a bitmap track, Plex down…).
|
||
_FAILURES: dict[str, tuple[str, float]] = {}
|
||
# rating_key → when its full pass last failed (see FULL_RETRY_S).
|
||
_FULL_FAILED: dict[str, float] = {}
|
||
# cache key → the spans windows failed to read, with when. A failed window says nothing about the rest
|
||
# of the track — the full pass may be about to deliver it — so it is remembered as a SPAN: it must be
|
||
# neither respawned every poll nor taken as a failure of positions it never tried. (It used to set the
|
||
# track-level memo, which told every viewer at any uncovered position "failed" — a final answer that
|
||
# ends the player's poll — including one planted by a request for an absurd position.)
|
||
_WIN_FAILED: dict[str, list[tuple[float, float, float]]] = {}
|
||
_PUBLISH_LOCK = threading.Lock()
|
||
# The ffmpeg children currently running, so a shutdown can kill them (see `shutdown`), and whether it
|
||
# has: a worker that reaches its Popen after the kill sweep must not start a read.
|
||
_PROCS: set[subprocess.Popen] = set()
|
||
_STOPPING = False
|
||
|
||
|
||
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,
|
||
}
|
||
# The codes that are a FINAL answer about the track: it can never be served as text, so the player
|
||
# stops asking. The others (`plex_error`, `extract_failed`) are about now or about one stretch — a
|
||
# memo that lapses, a window that failed where the rest is fine — and answered as `unavailable`: said,
|
||
# but asked again. Given as `failed`, one Plex hiccup ended the subtitles for the rest of the session,
|
||
# at cached positions too, and after the full pass had completed behind it.
|
||
FINAL_CODES = frozenset({"no_item", "no_track", "no_file", "not_fetchable", "bitmap"})
|
||
|
||
|
||
def reset_state() -> None:
|
||
"""Drop the in-flight/failure registries (tests; nothing in the app needs it)."""
|
||
global _STOPPING
|
||
with _LOCK:
|
||
_INFLIGHT.clear()
|
||
_FAILURES.clear()
|
||
_FULL_FAILED.clear()
|
||
_WIN_FAILED.clear()
|
||
_STOPPING = False
|
||
|
||
|
||
def shutdown() -> None:
|
||
"""Stop extracting, for the app's shutdown. The pool's workers are non-daemon threads and Python
|
||
joins them at interpreter exit — so a worker still inside a full pass (minutes on a big file, up
|
||
to EXTRACT_TIMEOUT_S) held the process open after uvicorn had stopped, until the container runtime
|
||
SIGKILLed it. Queued jobs are dropped and running ffmpeg children are killed, which returns their
|
||
workers at once — and a worker still BEFORE its ffmpeg (the Plex lookup, the DB) finds the flag
|
||
and never starts one. Nothing that matters is lost: an interrupted publish cannot pass for a whole
|
||
track (see `publish`), and an interrupted track is simply extracted again on the next poll."""
|
||
global _STOPPING
|
||
with _LOCK:
|
||
_STOPPING = True
|
||
procs = list(_PROCS)
|
||
for pool in (_WIN_POOL, _FULL_POOL):
|
||
pool.shutdown(wait=False, cancel_futures=True)
|
||
for p in procs:
|
||
p.kill()
|
||
|
||
|
||
def _sweep_memos() -> None:
|
||
"""(Caller holds _LOCK.) Drop every lapsed failure memo. They were only ever pruned when the SAME
|
||
key was read again, and the status route takes any rating_key/ordinal — so a script polling fresh
|
||
keys grew the API process without bound. Swept on every write, they hold at most a TTL's worth."""
|
||
now = time.monotonic()
|
||
for k in [k for k, (_, t) in _FAILURES.items() if now - t >= FAIL_TTL_S]:
|
||
del _FAILURES[k]
|
||
for k in [k for k, t in _FULL_FAILED.items() if now - t >= FULL_RETRY_S]:
|
||
del _FULL_FAILED[k]
|
||
for k in list(_WIN_FAILED):
|
||
_live_window_failures(k)
|
||
|
||
|
||
def _fail(key: str, code: str) -> None:
|
||
with _LOCK:
|
||
_sweep_memos()
|
||
_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 _full_failed(rating_key: str) -> bool:
|
||
with _LOCK:
|
||
when = _FULL_FAILED.get(rating_key)
|
||
if when is not None and time.monotonic() - when < FULL_RETRY_S:
|
||
return True
|
||
_FULL_FAILED.pop(rating_key, None)
|
||
return False
|
||
|
||
|
||
def _live_window_failures(key: str) -> list[tuple[float, float, float]]:
|
||
"""(Caller holds _LOCK.) The key's failed-window spans that are still remembered."""
|
||
now = time.monotonic()
|
||
live = [f for f in _WIN_FAILED.get(key, []) if now - f[2] < FAIL_TTL_S]
|
||
if live:
|
||
_WIN_FAILED[key] = live
|
||
else:
|
||
_WIN_FAILED.pop(key, None)
|
||
return live
|
||
|
||
|
||
def _window_fail(key: str, span: tuple[float, float]) -> None:
|
||
with _LOCK:
|
||
_sweep_memos()
|
||
_WIN_FAILED[key] = [*_live_window_failures(key), (span[0], span[1], time.monotonic())]
|
||
|
||
|
||
def _window_failed(key: str, at: float) -> bool:
|
||
with _LOCK:
|
||
return any(a <= at <= b for a, b, _ in _live_window_failures(key))
|
||
|
||
|
||
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 · ``unavailable`` = nothing for ``at`` and nothing being read
|
||
for it right now (``code`` says why), ask again · ``failed`` = the track can never be shown
|
||
(``code`` says why) — the only final answer. ``gen`` changes whenever the cached file grows, so
|
||
the client can refetch a window that has been superseded; 0 = nothing is cached at all."""
|
||
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)
|
||
gen = generation(key)
|
||
if complete and have:
|
||
return {"state": "ready", "code": None, "gen": gen}
|
||
if not have and _exists(_vtt_path(key)):
|
||
# On disk but unreadable (permissions, a bad volume): nothing can serve it, and no extraction
|
||
# can help — `publish` refuses to merge into it, and a job whose sidecar covers the position
|
||
# skips as already done — so scheduling only kept the player on "preparing" for good. `gen` 0:
|
||
# there is nothing a <track> could fetch.
|
||
return {"state": "unavailable", "code": "cache_unreadable", "gen": 0}
|
||
code = _recent_failure(key)
|
||
if code and not covered:
|
||
return {
|
||
"state": "failed" if code in FINAL_CODES else "unavailable",
|
||
"code": code,
|
||
"gen": gen,
|
||
}
|
||
# Nothing left that could cover the position for now: the full pass failed, AND so did a window
|
||
# here. Still not final — the memos lapse, and the rest of the track may be right there.
|
||
full_failed = _full_failed(rating_key)
|
||
window_failed = not covered and _window_failed(key, at)
|
||
if full_failed and window_failed:
|
||
return {"state": "unavailable", "code": "extract_failed", "gen": gen}
|
||
if not code:
|
||
probe, want_window = at, not covered and not window_failed
|
||
if covered:
|
||
# Read ahead of an edge instead of waiting to fall off it (see LOOKAHEAD_S). The probe is
|
||
# just PAST the edge: `covers` is inclusive, so the edge itself reads as already done.
|
||
edge = max(b for a, b in ranges if a <= at <= b)
|
||
if edge - at <= LOOKAHEAD_S and edge < _FULL_END_S:
|
||
probe = edge + 1.0
|
||
want_window = not _window_failed(key, probe)
|
||
_schedule(rating_key, sub_ord, probe, want_window=want_window, want_full=not full_failed)
|
||
return {"state": "partial" if covered else "preparing", "code": None, "gen": gen}
|
||
|
||
|
||
def cached_track(rating_key: str, sub_ord: int) -> tuple[str | None, str | None]:
|
||
"""What the <track> fetch can serve RIGHT NOW, scheduling nothing: the cached VTT (whatever it
|
||
covers), else the recent failure that explains why there is none.
|
||
|
||
Scheduling belongs to `ensure` — the status poll, which knows where the viewer is. The fetch does
|
||
not: when it scheduled too, it judged readiness at position 0, so a viewer who resumed an hour in
|
||
got an unrequested window over the head of the file (plus a remount when it landed), and with a
|
||
failure memo alive it was refused a window that WAS on disk while the poll called it showable."""
|
||
key = cache_key(rating_key, sub_ord)
|
||
hit = cache_read(key)
|
||
return (hit, None) if hit is not None else (None, _recent_failure(key))
|
||
|
||
|
||
def _schedule(
|
||
rating_key: str, sub_ord: int, at: float, want_window: bool, want_full: bool = True
|
||
) -> 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. A 1 s status poll (or a second viewer) joins a job already
|
||
queued or running instead of starting another ffmpeg.
|
||
|
||
A window is joined when one of this track's in-flight windows will COVER the position asked about
|
||
— decided by what it reads, not by an identity derived from the position. The poll sends the live
|
||
playhead, so a position-keyed window was a new job every second; bucketing the position instead
|
||
still queued a second read when the playhead crossed a bucket edge, and joined a backward seek to
|
||
a window that could not cover it. Nor may it be one window per track: a window cut around the
|
||
wrong spot (the first poll after a resume can still report 0) made the right one wait for it. The
|
||
full pass is per ITEM, because one read fans out to every embedded text track of the file."""
|
||
jobs: list[tuple[tuple, float | None]] = []
|
||
with _LOCK:
|
||
if _STOPPING:
|
||
return
|
||
n_windows = sum(1 for span in _INFLIGHT.values() if span is not None)
|
||
n_full = len(_INFLIGHT) - n_windows
|
||
if want_window and n_windows < _MAX_WINDOWS_INFLIGHT:
|
||
spans = [
|
||
span
|
||
for ident, span in _INFLIGHT.items()
|
||
if span is not None and ident[1:3] == (rating_key, sub_ord)
|
||
]
|
||
if not any(a <= at <= b for a, b in spans) and len(spans) < _MAX_TRACK_WINDOWS:
|
||
# Rounded to what ffmpeg is told (`-ss %.3f`): the cues come back on ffmpeg's zero,
|
||
# and shifting them by an unrounded start put them up to half a millisecond off —
|
||
# enough to spell a timestamp differently from the full pass and double the cue.
|
||
w_start = float(f"{max(0.0, at - WINDOW_PRE_S):.3f}")
|
||
ident = ("win", rating_key, sub_ord, w_start)
|
||
_INFLIGHT[ident] = (w_start, w_start + WINDOW_PRE_S + WINDOW_LEN_S)
|
||
jobs.append((ident, w_start))
|
||
if want_full and n_full < _MAX_FULL_INFLIGHT and ("full", rating_key) not in _INFLIGHT:
|
||
_INFLIGHT[("full", rating_key)] = None
|
||
jobs.append((("full", rating_key), None))
|
||
for ident, window_start in jobs:
|
||
pool = _FULL_POOL if window_start is None else _WIN_POOL
|
||
pool.submit(_run, ident, rating_key, sub_ord, window_start, at)
|
||
|
||
|
||
def _already_done(key: str, window_start: float | None, at: float) -> bool:
|
||
"""A queued job looks again before it reads anything. While it waited, another job may have
|
||
covered it: every embedded track rides each pass over its file, so a window for one language also
|
||
publishes the others."""
|
||
if not _exists(_vtt_path(key)):
|
||
return False
|
||
ranges, complete = read_meta(key)
|
||
return complete or (window_start is not None and covers(ranges, at))
|
||
|
||
|
||
def _run(
|
||
ident: tuple, rating_key: str, sub_ord: int, window_start: float | None, at: float
|
||
) -> None:
|
||
key = cache_key(rating_key, sub_ord)
|
||
try:
|
||
if _STOPPING or _already_done(key, window_start, at):
|
||
return
|
||
_extract_and_publish(rating_key, sub_ord, window_start)
|
||
except SubtitleUnavailable as e:
|
||
if e.code == "extract_failed":
|
||
_pass_failed(rating_key, [key], window_start)
|
||
else:
|
||
_fail(key, e.code)
|
||
except PlexError as e:
|
||
log.warning("subtitle extraction failed for %s/%s: %s", rating_key, sub_ord, e)
|
||
_fail(key, "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)
|
||
_pass_failed(rating_key, [key], window_start)
|
||
finally:
|
||
with _LOCK:
|
||
_INFLIGHT.pop(ident, None)
|
||
|
||
|
||
def _pass_failed(
|
||
rating_key: str,
|
||
keys: list[str],
|
||
window_start: float | None,
|
||
span: tuple[float, float] | None = None,
|
||
) -> None:
|
||
"""Record an ffmpeg pass that failed: a full pass backs off the whole item (FULL_RETRY_S); a
|
||
window marks only the stretch it failed to read, for the tracks it was reading."""
|
||
if window_start is None:
|
||
with _LOCK:
|
||
_sweep_memos()
|
||
_FULL_FAILED[rating_key] = time.monotonic()
|
||
return
|
||
for key in keys:
|
||
_window_fail(key, span or (window_start, window_start + WINDOW_PRE_S + WINDOW_LEN_S))
|
||
|
||
|
||
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"):
|
||
# An external sidecar is one small fetch from Plex, never a read of the media file, so a
|
||
# window job fetches the whole thing too. It must not be left to the full job: that
|
||
# identity is per ITEM (the embedded fan-out shares it), so a sidecar picked while an
|
||
# embedded track's full pass was running sat behind that pass for minutes.
|
||
vtt = _sidecar_vtt(db, want)
|
||
if not publish(cache_key(rating_key, sub_ord), vtt, (0.0, _FULL_END_S), True):
|
||
raise SubtitleUnavailable("extract_failed")
|
||
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
|
||
texts, clean = _ffmpeg_extract(str(src), sorted(targets), window_start, length)
|
||
for emb_ord, vtt in texts.items():
|
||
key = cache_key(rating_key, targets[emb_ord])
|
||
# A pass that exited non-zero holds only what ffmpeg reached before it died. Claim no more
|
||
# than its last cue, never call it complete, and remember the failure: published as complete,
|
||
# the track read "ready" for good while its subtitles stopped wherever ffmpeg did (the cache
|
||
# has no eviction, so nothing would ever have extracted it again).
|
||
covered_end = end if clean else start + _last_cue_end(vtt)
|
||
landed = publish(
|
||
key,
|
||
shift_vtt(vtt, -start), # zero-based window clock → absolute cue times
|
||
(start, covered_end),
|
||
clean and length is None,
|
||
)
|
||
if not landed:
|
||
# A read that could not be kept is a failed read (see publish) — backed off like one.
|
||
_pass_failed(rating_key, [key], window_start, (start, end))
|
||
elif not clean:
|
||
_pass_failed(rating_key, [key], window_start, (covered_end, end))
|
||
|
||
|
||
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
|
||
) -> tuple[dict[int, str], bool]:
|
||
"""Extract several embedded subtitle streams to WebVTT in ONE ffmpeg run (= one read of the file).
|
||
Returns the texts and whether ffmpeg exited cleanly — a non-zero exit can still leave partial
|
||
output, which is worth publishing but must never be taken for the whole track.
|
||
|
||
`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 {}, True
|
||
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}"]
|
||
# The input seek lands on the keyframe at or BEFORE `start`, so a cue already running
|
||
# at `start` comes out at a negative time — and by default the muxer then shifts the
|
||
# WHOLE output so that cue sits at 0: every cue of the window late by the distance to
|
||
# that keyframe (1–5 s measured on a real library file, a whole GOP on a synthetic
|
||
# one), and doubled for good once the full pass merged in beside it. Kept negative, the
|
||
# times are exact on the zero-based clock (`_vtt_ts_to_s` reads the sign).
|
||
cmd += ["-avoid_negative_ts", "disabled"]
|
||
cmd += ["-f", "webvtt", "-y", str(out)]
|
||
# Popen rather than run(): the child is registered so `shutdown` can kill it — and one that
|
||
# would start after the shutdown's kill sweep never does, or is killed right here.
|
||
with _LOCK:
|
||
if _STOPPING:
|
||
raise SubtitleUnavailable("extract_failed")
|
||
proc = subprocess.Popen(
|
||
cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE
|
||
)
|
||
with _LOCK:
|
||
_PROCS.add(proc)
|
||
if _STOPPING:
|
||
proc.kill()
|
||
timed_out = False
|
||
try:
|
||
try:
|
||
_, err = proc.communicate(timeout=EXTRACT_TIMEOUT_S)
|
||
except subprocess.TimeoutExpired:
|
||
# A pass that ran out of time is a pass that died partway: what it reached is kept and
|
||
# published as partial (the caller's unclean path), like any other non-zero exit. It
|
||
# used to be killed and its outputs deleted unread — and with the full-pass backoff, the
|
||
# same doomed read came back every hour. TERM first: ffmpeg closes its outputs on it,
|
||
# whereas a KILL leaves only what was already flushed in whole blocks (measured: none).
|
||
timed_out = True
|
||
proc.terminate()
|
||
try:
|
||
_, err = proc.communicate(timeout=_TERM_GRACE_S)
|
||
except subprocess.TimeoutExpired:
|
||
proc.kill()
|
||
_, err = proc.communicate()
|
||
finally:
|
||
with _LOCK:
|
||
_PROCS.discard(proc)
|
||
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")
|
||
clean = proc.returncode == 0 and not timed_out
|
||
if not clean:
|
||
log.warning(
|
||
"ffmpeg subtitle pass %s: %s",
|
||
"timed out" if timed_out else f"exited {proc.returncode}",
|
||
(err or b"")[:200],
|
||
)
|
||
return got, clean
|