peter
dc4e8b9fcd
HLS sessions were keyed by the item, so a second account opening the same film killed the first one's session; the displaced player reclaimed at its playhead and evicted the other in turn. The registry is now keyed by (viewer, rating_key), segment requests resolve by both, and the session directory carries the viewer (the orphan sweep still recognises the older shapes). Only transcodes counted toward a cap, so remuxes — each a whole segmented copy of the film on the scratch volume — were bounded by the library size. Three bounds now apply after every start, each dropping the least recently used: two sessions per account, the transcode cap, and a new admin knob, plex_max_sessions (default 8), over every session. A superseded session request returns at once, since it waits on the killed ffmpeg, so an account cannot park more request workers than it may hold sessions. The subtitle extractor's full-pass slots were global and minutes long each; one account opening subtitles on film after film could hold them all. An account may now hold three of the six, and still gets windows past that.
734 lines
34 KiB
Python
734 lines
34 KiB
Python
"""Plex playback: image/subtitle proxies, HLS streaming, and watch-state writes."""
|
|
|
|
import hashlib
|
|
import threading
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import httpx
|
|
from fastapi import (
|
|
APIRouter,
|
|
BackgroundTasks,
|
|
Depends,
|
|
HTTPException,
|
|
Path,
|
|
Query,
|
|
Request,
|
|
Response,
|
|
)
|
|
from fastapi.responses import FileResponse
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
from sqlalchemy import select, update
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth import current_user, resolved_user_id
|
|
from app.db import SessionLocal, get_db
|
|
from app.models import (
|
|
POSITION_MAX_S,
|
|
PlexItem,
|
|
PlexSeason,
|
|
PlexShow,
|
|
PlexState,
|
|
User,
|
|
)
|
|
from app.plex import imgcache, subtitles
|
|
from app.plex import paths as plex_paths
|
|
from app.plex.lookups import by_rating_key
|
|
from app.plex import people as plex_people
|
|
from app.plex import stream as plex_stream
|
|
from app.plex import watch_sync as plex_watch
|
|
from app.plex.client import PlexClient, PlexError, PlexNotConfigured
|
|
from app.routes.plex._shared import item_or_404
|
|
|
|
router = APIRouter(prefix="/api/plex", tags=["plex"])
|
|
|
|
|
|
def _require_session(request: Request) -> int:
|
|
"""Lightweight auth for the high-fan-out image proxies: validate the signed session cookie WITHOUT
|
|
a DB user-load, so these endpoints never hold a pool connection for the auth. Poster images are
|
|
low-sensitivity; a valid session is enough."""
|
|
uid, _ = resolved_user_id(request)
|
|
if not uid:
|
|
raise HTTPException(status_code=401, detail="Not authenticated")
|
|
return uid
|
|
|
|
|
|
@router.get("/image/{rating_key}")
|
|
def image(rating_key: str, request: Request, variant: str = "thumb") -> Response:
|
|
"""Proxy a Plex poster/art image (keeps the admin token server-side; no image duplication). A
|
|
thin on-disk cache serves repeat views without re-hitting Plex.
|
|
|
|
⚠️ This endpoint is called in BURSTS (a collections page fires 100+ concurrent poster requests),
|
|
so it must NOT hold a DB connection during the slow Plex fetch — doing so exhausted the pool
|
|
(QueuePool timeout, broken images on prod). Warm path (cache hit) touches NO DB; cold path opens a
|
|
brief session only to resolve the image key + read the Plex config, then RELEASES it before the
|
|
fetch. Auth is the signed-session check (no DB user-load)."""
|
|
_require_session(request)
|
|
v = "art" if variant == "art" else "thumb"
|
|
w, h = imgcache.SIZES[v]
|
|
# Size is part of the cache key: Plex resizes server-side (see PlexClient.image_bytes), so we cache
|
|
# the small variant. The width tag also means changing a target size just re-caches, never serves a
|
|
# stale differently-sized file (old full-size files under the un-tagged key are simply orphaned).
|
|
cache_key = f"item_{rating_key}_{v}_{w}"
|
|
hit = imgcache.read(cache_key)
|
|
if hit:
|
|
return imgcache.response(*hit)
|
|
# Cold miss: short-lived session for the key + Plex client config, released before the slow fetch.
|
|
try:
|
|
with SessionLocal() as db:
|
|
key = imgcache.key_for(db, str(rating_key), v)
|
|
if not key:
|
|
raise HTTPException(status_code=404, detail="No image")
|
|
plex = PlexClient(db) # reads base URL + token now; the fetch below needs no DB
|
|
except PlexNotConfigured as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
try:
|
|
with plex:
|
|
data, ct = plex.image_bytes(key, w, h)
|
|
except PlexError as e:
|
|
raise HTTPException(status_code=502, detail=f"Plex image fetch failed: {e}")
|
|
imgcache.write(cache_key, data, ct)
|
|
return imgcache.response(data, ct)
|
|
|
|
|
|
# Both subtitle routes authenticate with the full `current_user`, not the cookie-only
|
|
# `_require_session` the image proxies use. That lighter gate exists for bursts of low-sensitivity
|
|
# poster fetches and skips the is_active / is_suspended / session-epoch checks — acceptable for a
|
|
# poster, not for a route that starts reads of the media library: a suspended or "log out
|
|
# everywhere"-revoked cookie kept scheduling them. They are not bursty (one poll a second, one fetch
|
|
# per track), so the DB user-load costs nothing that matters.
|
|
#
|
|
# Both path parameters are bounded, the same way on both routes: they become a cache FILE name
|
|
# (`sub_<rating_key>_<ord>`), and an absurd one — a 250-digit key or ordinal — outgrew the OS's name
|
|
# limit and turned the status poll into a 500. A Plex rating_key is a short number; the bound only
|
|
# has to be generous.
|
|
_RatingKey = Path(max_length=32)
|
|
_SubOrd = Path(ge=0, le=9999)
|
|
|
|
|
|
@router.get("/subtitle/{rating_key}/{sub_ord}/status")
|
|
def subtitle_status(
|
|
rating_key: str = _RatingKey,
|
|
sub_ord: int = _SubOrd,
|
|
user: User = Depends(current_user),
|
|
# Bounded both ways: a non-finite position used to reach the scheduler's arithmetic and 500, and
|
|
# an absurd one reached ffmpeg's `-ss`, which refused it.
|
|
at: float = Query(0.0, ge=0.0, le=subtitles.MAX_POSITION_S, allow_inf_nan=False),
|
|
) -> 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 <track> 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."""
|
|
return subtitles.ensure(rating_key, sub_ord, at, viewer=user.id)
|
|
|
|
|
|
@router.get("/subtitle/{rating_key}/{sub_ord}")
|
|
def subtitle(
|
|
rating_key: str = _RatingKey,
|
|
sub_ord: int = _SubOrd,
|
|
user: User = Depends(current_user),
|
|
# Finite, like every float this router takes: `-inf` crashed the shift (a 500), `nan` stacked
|
|
# every cue at 0. It can legitimately be negative (a subtitle nudged later than the session start).
|
|
offset: float = Query(0.0, allow_inf_nan=False),
|
|
) -> Response:
|
|
"""Serve the ord-th subtitle track as WebVTT for the browser's <track> — 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.
|
|
|
|
A pure READ: it serves whatever is cached and schedules nothing — the status poll owns that,
|
|
because only it knows where the viewer is (see `subtitles.cached_track`). The player mounts the
|
|
<track> only once /status says there is something to show, so in practice this is only ever hit
|
|
warm; on a cache eviction race the 202 sends the player back to polling."""
|
|
# 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).
|
|
hit, code = subtitles.cached_track(rating_key, sub_ord)
|
|
if hit is None and code:
|
|
raise HTTPException(
|
|
status_code=subtitles.UNAVAILABLE_STATUS.get(code, 422),
|
|
detail={"code": "subtitle", "reason": code},
|
|
)
|
|
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
|
|
# _FINISH_MARGIN_S is DELIBERATELY larger than the feed's FINISH_MARGIN_SECONDS (30 vs 10): this is
|
|
# long-form content (movies/episodes with credits) and crossing it MARKS THE ITEM WATCHED, whereas
|
|
# the feed's margin only clears a short video's resume bar — different domains, intentional divergence,
|
|
# not drift. Keep them separate on purpose (see routes/feed.py).
|
|
_PROGRESS_MIN_S = 5
|
|
|
|
_FINISH_MARGIN_S = 30
|
|
|
|
# S2 write-back throttle: the resume position is now pushed to Plex on the periodic (non-final)
|
|
# checkpoint too — not only on the final pause/unmount flush — so Plex (the source of truth) stays
|
|
# current in near-real-time. To keep a single watch from spraying `/:/timeline` calls, a push is
|
|
# rate-limited to at most one per item per user every _PUSH_THROTTLE_S seconds; a `final` checkpoint
|
|
# (pause / seek / stop) always pushes immediately and resets the window. This is an in-memory, single-
|
|
# process rate-limiter (the deployment is one FastAPI process): after a restart the map is empty, so at
|
|
# most one extra push happens on the first checkpoint — harmless. It is NOT durable domain state.
|
|
_PUSH_THROTTLE_S = 10
|
|
_last_push_at: dict[tuple[int, int], datetime] = {}
|
|
_push_throttle_lock = threading.Lock()
|
|
|
|
|
|
def _should_push_resume(user_id: int, item_id: int, now: datetime, final: bool) -> bool:
|
|
"""Whether to mirror this resume checkpoint to Plex now. `final` always pushes; a periodic tick
|
|
pushes only when ≥_PUSH_THROTTLE_S has elapsed since the last push for this (user, item). Records
|
|
`now` as the last-push time whenever it returns True so the window is measured from the real push
|
|
(and so `_resume_push_superseded` can drop an older, still-pending push — see there)."""
|
|
with _push_throttle_lock:
|
|
last = _last_push_at.get((user_id, item_id))
|
|
if final or last is None or (now - last).total_seconds() >= _PUSH_THROTTLE_S:
|
|
_last_push_at[(user_id, item_id)] = now
|
|
# Opportunistic eviction so the map can't grow unbounded over a long-lived process (one
|
|
# entry per item ever checkpointed). Anything older than a few windows is dead — a push
|
|
# completes in seconds, so a stale entry protects nothing.
|
|
if len(_last_push_at) > 2048:
|
|
cutoff = now - timedelta(seconds=_PUSH_THROTTLE_S * 6)
|
|
for k in [k for k, t in _last_push_at.items() if t < cutoff]:
|
|
del _last_push_at[k]
|
|
return True
|
|
return False
|
|
|
|
|
|
def _resume_push_superseded(user_id: int, item_id: int, intent: datetime) -> bool:
|
|
"""True if a newer resume push for this (user, item) was scheduled after `intent`. Called at the
|
|
HTTP boundary of a background resume push: because periodic and seek/pause pushes are independent
|
|
background tasks with no ordering guard, an older still-pending periodic push could otherwise land
|
|
at Plex AFTER a newer backward-seek push and — with the Plex-authoritative pull — durably revive the
|
|
stale position. Skipping a superseded push keeps the last-scheduled position the one Plex settles on."""
|
|
with _push_throttle_lock:
|
|
return _last_push_at.get((user_id, item_id), intent) > intent
|
|
|
|
|
|
@router.get("/person-image")
|
|
def person_image(u: str, request: Request) -> Response:
|
|
"""Proxy a cast/crew photo from Plex's public metadata CDN (host-whitelisted; no token needed).
|
|
Signed-session auth (no DB) so it, like /image, never holds a pool connection during the fetch."""
|
|
_require_session(request)
|
|
if not u.startswith(plex_people.PERSON_IMG_HOST):
|
|
raise HTTPException(status_code=400, detail="Unsupported image host")
|
|
cache_key = "person_" + hashlib.sha1(u.encode()).hexdigest()
|
|
hit = imgcache.read(cache_key)
|
|
if hit:
|
|
return imgcache.response(*hit)
|
|
try:
|
|
with httpx.Client(timeout=10.0, follow_redirects=True) as c:
|
|
r = c.get(u)
|
|
r.raise_for_status()
|
|
except httpx.HTTPError:
|
|
raise HTTPException(status_code=502, detail="Image fetch failed")
|
|
ct = r.headers.get("content-type", "image/jpeg")
|
|
imgcache.write(cache_key, r.content, ct)
|
|
return imgcache.response(r.content, ct)
|
|
|
|
|
|
def _push_resume_guarded(
|
|
user_id: int,
|
|
item_id: int,
|
|
rating_key: str,
|
|
time_ms: int,
|
|
duration_ms: int,
|
|
expect_ts: datetime | None,
|
|
intent: datetime,
|
|
) -> None:
|
|
"""Background entrypoint for a resume push that drops itself if a newer resume push for the same
|
|
(user, item) was scheduled after it (see `_resume_push_superseded`), then delegates to the real
|
|
push. This narrows the periodic-vs-seek write-ordering race to the (rare) case where the older push
|
|
has already left the process before the newer one is scheduled."""
|
|
if _resume_push_superseded(user_id, item_id, intent):
|
|
return
|
|
plex_watch.push_state_to_plex(
|
|
user_id, item_id, rating_key, "resume", time_ms, duration_ms, expect_ts
|
|
)
|
|
|
|
|
|
def _push_watch(
|
|
background: BackgroundTasks,
|
|
db: Session,
|
|
user_id: int,
|
|
item_id: int,
|
|
rating_key: str,
|
|
action: str,
|
|
time_ms: int = 0,
|
|
duration_ms: int = 0,
|
|
intent: datetime | None = None,
|
|
) -> None:
|
|
"""Schedule a best-effort Siftlode→Plex watch push, but only for a user whose owner link has
|
|
push enabled — non-owner users never touch Plex. Guarding here (with the request session) avoids
|
|
spinning up a throwaway background session on every state change for the common no-link case.
|
|
|
|
A `resume` push with an `intent` timestamp is routed through `_push_resume_guarded` so a superseded
|
|
(older) push can't land out of order; `watched`/`unwatched` scrobbles go straight through."""
|
|
if plex_watch.link_for_push(db, user_id) is None:
|
|
return
|
|
# Capture the row's freshest timestamp NOW so the background push can tell if a newer local edit
|
|
# lands before it settles synced_to_plex (which would otherwise strand that newer value).
|
|
st = (
|
|
db.execute(
|
|
select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id)
|
|
)
|
|
.scalars()
|
|
.first()
|
|
)
|
|
marker = plex_watch._state_marker(st)
|
|
if action == "resume" and intent is not None:
|
|
background.add_task(
|
|
_push_resume_guarded, user_id, item_id, rating_key, time_ms, duration_ms, marker, intent
|
|
)
|
|
return
|
|
background.add_task(
|
|
plex_watch.push_state_to_plex,
|
|
user_id,
|
|
item_id,
|
|
rating_key,
|
|
action,
|
|
time_ms,
|
|
duration_ms,
|
|
marker,
|
|
)
|
|
|
|
|
|
def _mark_finished(db: Session, user_id: int, item_id: int, now: datetime) -> bool:
|
|
"""Mark the item watched (position cleared) and say whether THIS call made it so. Two near-end
|
|
checkpoints arrive together routinely (the 10 s tick and the debounced `seeked` flush), and a
|
|
read-then-write let both see "not watched" and both scrobble. The transition is decided by the
|
|
write itself: an insert that did not conflict, or an update conditional on the old status —
|
|
under READ COMMITTED the second writer re-reads the committed row and matches nothing."""
|
|
values = {
|
|
"status": "watched",
|
|
"watched_at": now,
|
|
"position_seconds": 0,
|
|
"progress_updated_at": now,
|
|
"synced_to_plex": False,
|
|
}
|
|
inserted = db.execute(
|
|
pg_insert(PlexState)
|
|
.values(user_id=user_id, item_id=item_id, **values)
|
|
.on_conflict_do_nothing(constraint="uq_plex_user_item")
|
|
.returning(PlexState.id)
|
|
).first()
|
|
if inserted is not None:
|
|
return True
|
|
mine = (PlexState.user_id == user_id) & (PlexState.item_id == item_id)
|
|
if db.execute(
|
|
update(PlexState).where(mine, PlexState.status != "watched").values(values)
|
|
).rowcount:
|
|
return True
|
|
# Already watched: a re-finish still restamps the row, as it always did — just no scrobble.
|
|
db.execute(update(PlexState).where(mine).values(values))
|
|
return False
|
|
|
|
|
|
class ItemProgressIn(BaseModel):
|
|
# Finite only: JSON parsing here accepts `Infinity`/`NaN`, and the handler's `int()` turned
|
|
# one into a 500.
|
|
model_config = ConfigDict(allow_inf_nan=False)
|
|
|
|
# float, truncated in the handler — the player may post a raw currentTime (a float); a plain
|
|
# `int` field would 422 it on the periodic progress checkpoint (the old `int(...)` truncated).
|
|
# Bounded above by the int4 column (see POSITION_MAX_S): past it the flush was a 500.
|
|
position_seconds: float = Field(0, le=POSITION_MAX_S)
|
|
duration_seconds: float = Field(0, le=POSITION_MAX_S)
|
|
final: bool = False
|
|
|
|
|
|
@router.post("/item/{rating_key}/progress")
|
|
def item_progress(
|
|
rating_key: str,
|
|
body: ItemProgressIn,
|
|
background: BackgroundTasks,
|
|
user: User = Depends(current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
"""Checkpoint the resume position while playing (near the end → mark watched + clear it).
|
|
|
|
S2 write-back: the resume position is mirrored to Plex (the source of truth) on the periodic
|
|
checkpoint too, not only on the final flush — throttled to ≥_PUSH_THROTTLE_S per item (see
|
|
`_should_push_resume`). `final` (sent on pause / seek / pagehide / unmount) pushes immediately and
|
|
resets the throttle window, so Plex sees a settled position the instant the user pauses or seeks."""
|
|
it = item_or_404(db, rating_key)
|
|
# A player parked mid-item — playing or PAUSED — checkpoints every 10 s, so this is the signal
|
|
# that such a viewer still wants their HLS session: the one thing `last_access` could not see,
|
|
# since it moved only on a file fetch and a paused player fetches nothing (see
|
|
# stream.reap_idle). It is NOT every open player: `usePlexProgress.flush` sends nothing once the
|
|
# finish latch is up, or before the first position is known. Those are the states in which
|
|
# reaping is the right answer anyway, and a viewer who scrubs back into a reaped session gets
|
|
# the same 404 → reclaim the player already handles.
|
|
plex_stream.touch_session(user.id, str(rating_key))
|
|
position = max(0, int(body.position_seconds))
|
|
duration = int(body.duration_seconds)
|
|
final = body.final
|
|
now = datetime.now(timezone.utc)
|
|
near_end = duration > 0 and position > duration - _FINISH_MARGIN_S
|
|
if near_end:
|
|
flipped = _mark_finished(db, user.id, it.id, now)
|
|
# The finish supersedes every resume push still in flight for this item: stamping it as the
|
|
# latest push makes `_resume_push_superseded` drop them. Popping the entry instead (to keep
|
|
# the map small) removed the very guard, and an older resume push could then land at Plex
|
|
# AFTER the scrobble and revive a resume point on a watched item. Eviction bounds the map.
|
|
with _push_throttle_lock:
|
|
_last_push_at[(user.id, it.id)] = now
|
|
db.commit()
|
|
# Only the checkpoint that flipped the row scrobbles: a later checkpoint of the same finished
|
|
# item — or a second one racing this one — would otherwise bump Plex's viewCount again.
|
|
if flipped:
|
|
_push_watch(background, db, user.id, it.id, rating_key, "watched")
|
|
return {"status": "watched", "position_seconds": 0}
|
|
|
|
st = (
|
|
db.execute(
|
|
select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == it.id)
|
|
)
|
|
.scalars()
|
|
.first()
|
|
)
|
|
|
|
if position < _PROGRESS_MIN_S:
|
|
if st is not None and st.status == "new":
|
|
db.delete(st)
|
|
elif st is not None:
|
|
st.position_seconds = 0
|
|
st.progress_updated_at = now
|
|
db.commit()
|
|
return {"position_seconds": 0}
|
|
|
|
if st is None:
|
|
st = PlexState(user_id=user.id, item_id=it.id)
|
|
db.add(st)
|
|
st.position_seconds = position
|
|
st.progress_updated_at = now
|
|
push_now = _should_push_resume(user.id, it.id, now, final)
|
|
if push_now:
|
|
# Mark dirty before the (background) push so the reconcile knows to re-push if it fails, and so
|
|
# `push_state_to_plex` can settle synced_to_plex once Plex confirms.
|
|
st.synced_to_plex = False
|
|
db.commit()
|
|
if push_now:
|
|
_push_watch(
|
|
background,
|
|
db,
|
|
user.id,
|
|
it.id,
|
|
rating_key,
|
|
"resume",
|
|
position * 1000,
|
|
duration * 1000,
|
|
intent=now,
|
|
)
|
|
return {"position_seconds": position}
|
|
|
|
|
|
class ItemStateIn(BaseModel):
|
|
status: str = ""
|
|
|
|
|
|
@router.post("/item/{rating_key}/state")
|
|
def item_state(
|
|
rating_key: str,
|
|
body: ItemStateIn,
|
|
background: BackgroundTasks,
|
|
user: User = Depends(current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
"""Set the per-user watch status (new|watched|hidden). watched→scrobble / →new→unscrobble push
|
|
to a linked Plex account immediately; `hidden` is a Siftlode-only concept with no Plex analogue,
|
|
so it never touches Plex."""
|
|
it = item_or_404(db, rating_key)
|
|
status = body.status
|
|
if status not in ("new", "watched", "hidden"):
|
|
raise HTTPException(status_code=400, detail="Invalid status")
|
|
st = (
|
|
db.execute(
|
|
select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == it.id)
|
|
)
|
|
.scalars()
|
|
.first()
|
|
)
|
|
prev_status = st.status if st is not None else "new"
|
|
if status == "new":
|
|
# Does Plex hold anything for this item that the un-mark has to clear? Two things qualify: a
|
|
# view count, and a resume point. An IN-PROGRESS item carries status "new" WITH a position —
|
|
# that is how the importer maps a bare `viewOffset` (`_plex_watch_to_state`) — so keying the
|
|
# push on the status alone silently skipped the single case "Clear resume point" exists for:
|
|
# the un-mark never reached Plex, the offset stayed, and the next pull imported it straight
|
|
# back (Protector, cleared twice by the user and back at 92s both times). `hidden` is
|
|
# Siftlode-only and still pushes nothing, exactly as this endpoint's docstring says.
|
|
mirrored_on_plex = st is not None and (
|
|
st.status == "watched" or (st.position_seconds or 0) > 0
|
|
)
|
|
if st is not None:
|
|
db.delete(st)
|
|
db.commit()
|
|
if mirrored_on_plex:
|
|
_push_watch(background, db, user.id, it.id, rating_key, "unwatched")
|
|
return {"status": "new"}
|
|
if st is None:
|
|
st = PlexState(user_id=user.id, item_id=it.id)
|
|
db.add(st)
|
|
st.status = status
|
|
if status == "watched":
|
|
st.watched_at = datetime.now(timezone.utc)
|
|
st.position_seconds = 0
|
|
st.synced_to_plex = False
|
|
db.commit()
|
|
if status == "watched" and prev_status != "watched":
|
|
_push_watch(background, db, user.id, it.id, rating_key, "watched")
|
|
return {"status": status}
|
|
|
|
|
|
def _bulk_state(
|
|
background: BackgroundTasks,
|
|
db: Session,
|
|
user_id: int,
|
|
eps: list[PlexItem],
|
|
watched: bool,
|
|
) -> int:
|
|
"""Mark every episode in `eps` watched (or unwatched) for a user, mirroring each change to a
|
|
linked Plex account in the background (best-effort). Returns how many rows actually changed."""
|
|
now = datetime.now(timezone.utc)
|
|
existing = {
|
|
s.item_id: s
|
|
for s in db.execute(
|
|
select(PlexState).where(
|
|
PlexState.user_id == user_id,
|
|
PlexState.item_id.in_([e.id for e in eps] or [0]),
|
|
)
|
|
).scalars()
|
|
}
|
|
pushable = plex_watch.link_for_push(db, user_id) is not None
|
|
to_push: list[tuple[int, str, str]] = []
|
|
for e in eps:
|
|
st = existing.get(e.id)
|
|
prev = st.status if st is not None else "new"
|
|
if watched:
|
|
if st is None:
|
|
st = PlexState(user_id=user_id, item_id=e.id)
|
|
db.add(st)
|
|
st.status = "watched"
|
|
st.watched_at = now
|
|
st.position_seconds = 0
|
|
st.synced_to_plex = False
|
|
if prev != "watched":
|
|
to_push.append((e.id, e.rating_key, "watched"))
|
|
else: # → new: drop any state (but keep the user's private "hidden" marks)
|
|
if st is not None and prev != "hidden":
|
|
db.delete(st)
|
|
if prev != "new":
|
|
to_push.append((e.id, e.rating_key, "unwatched"))
|
|
db.commit()
|
|
if pushable and to_push:
|
|
# One coalesced background task for the whole batch (reuses a single session + Plex client),
|
|
# not one task per episode — a 200-ep show mark is 1 task, not 200.
|
|
background.add_task(plex_watch.push_bulk_state_to_plex, user_id, to_push)
|
|
return len(to_push)
|
|
|
|
|
|
class WatchedIn(BaseModel):
|
|
watched: bool = False
|
|
|
|
|
|
@router.post("/show/{rating_key}/state")
|
|
def show_state(
|
|
rating_key: str,
|
|
body: WatchedIn,
|
|
background: BackgroundTasks,
|
|
user: User = Depends(current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
"""Mark a whole show watched/unwatched (all its episodes) for the current user + push to Plex."""
|
|
sh = by_rating_key(db, PlexShow, rating_key)
|
|
if sh is None:
|
|
raise HTTPException(status_code=404, detail="Unknown Plex show")
|
|
watched = body.watched
|
|
eps = (
|
|
db.execute(select(PlexItem).where(PlexItem.show_id == sh.id, PlexItem.kind == "episode"))
|
|
.scalars()
|
|
.all()
|
|
)
|
|
changed = _bulk_state(background, db, user.id, eps, watched)
|
|
return {"changed": changed, "watched": watched}
|
|
|
|
|
|
@router.post("/season/{rating_key}/state")
|
|
def season_state(
|
|
rating_key: str,
|
|
body: WatchedIn,
|
|
background: BackgroundTasks,
|
|
user: User = Depends(current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
"""Mark a whole season watched/unwatched (all its episodes) for the current user + push to Plex."""
|
|
se = by_rating_key(db, PlexSeason, rating_key)
|
|
if se is None:
|
|
raise HTTPException(status_code=404, detail="Unknown Plex season")
|
|
watched = body.watched
|
|
eps = (
|
|
db.execute(select(PlexItem).where(PlexItem.season_id == se.id, PlexItem.kind == "episode"))
|
|
.scalars()
|
|
.all()
|
|
)
|
|
changed = _bulk_state(background, db, user.id, eps, watched)
|
|
return {"changed": changed, "watched": watched}
|
|
|
|
|
|
@router.post("/stream/{rating_key}/session")
|
|
def stream_session(
|
|
rating_key: str,
|
|
# Finite: `start=inf` reached `int()` in the session builder and came back as a 500.
|
|
start: float = Query(0.0, allow_inf_nan=False),
|
|
audio: int | None = None,
|
|
aoff: float = Query(0.0, allow_inf_nan=False),
|
|
multi: bool = False,
|
|
user: User = Depends(current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
"""Prepare playback from `start` seconds. Direct-playable files are served raw; remux files
|
|
(re)start an HLS session at the offset (seek-restart, video stream-copied); transcode files (non-h264:
|
|
HEVC/VP9/AV1/…) take the same HLS path but with the video re-encoded to H.264 (VAAPI or libx264).
|
|
Selecting a non-default audio track (`audio` = stream ordinal) forces the HLS path so the chosen audio
|
|
can be muxed in. Subtitles are independent WebVTT tracks (GET /subtitle) — no session restart."""
|
|
it = item_or_404(db, rating_key)
|
|
transcode = it.playable == "transcode"
|
|
# `multi` (item has ≥2 audio tracks) forces the HLS path so ALL audio tracks ship as renditions and
|
|
# the client can switch audio without a restart — even for otherwise direct-playable files.
|
|
if it.playable == "direct" and audio is None and not aoff and not multi:
|
|
return {
|
|
"mode": "direct",
|
|
"url": f"/api/plex/stream/{it.rating_key}/file",
|
|
"start_s": 0.0,
|
|
}
|
|
if plex_paths.local_media_path(db, it.file_path) is None:
|
|
raise HTTPException(status_code=404, detail="Media file not found on disk")
|
|
s = plex_stream.start_session(db, it, user.id, start, audio, aoff, multi, transcode=transcode)
|
|
if s is None:
|
|
raise HTTPException(status_code=404, detail="Media file not found on disk")
|
|
# Report the REAL media start (the keyframe ffmpeg landed on, which hls.js zero-bases to), not the
|
|
# requested offset — the client's absolute clock and the subtitle cue-shift both key off this.
|
|
# `transcode`/`hw`/`tonemap` let the player show an honest cold-start ("Preparing stream…") and
|
|
# surface the active mode + encoder in the stats overlay (`tonemap` = the HW HDR→SDR path); both
|
|
# segment streams are otherwise identical.
|
|
return {
|
|
"mode": "hls",
|
|
"start_s": s.media_start_s,
|
|
# The session id is part of the path (not a query param) so the playlist's RELATIVE segment
|
|
# names resolve back into the same session — see `stream_hls_file` for why it has to.
|
|
"url": f"/api/plex/stream/{it.rating_key}/hls/{s.id}/{s.entry}",
|
|
"transcode": s.transcode,
|
|
"hw": s.hw,
|
|
"tonemap": s.tonemap,
|
|
}
|
|
|
|
|
|
@router.get("/stream/{rating_key}/file")
|
|
def stream_file(
|
|
rating_key: str,
|
|
download: bool = False,
|
|
user: User = Depends(current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> FileResponse:
|
|
"""Serve the raw local media file with HTTP range support. Used both for direct-play (inline)
|
|
and for the player's "download original" button (`?download=1` → attachment). The download is
|
|
the untouched physical file — no re-encode/repackage."""
|
|
it = item_or_404(db, rating_key)
|
|
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")
|
|
if download:
|
|
# A clean, safe download name from the title + the real file extension.
|
|
ext = src.suffix or (f".{it.container}" if it.container else "")
|
|
base = (
|
|
"".join(c for c in (it.title or "video") if c.isalnum() or c in " -_.").strip()
|
|
or "video"
|
|
)
|
|
return FileResponse(str(src), filename=f"{base}{ext}")
|
|
return FileResponse(str(src)) # Starlette honours the Range header (206)
|
|
|
|
|
|
# What a session directory can contain, and nothing else. No `.vtt`: ffmpeg runs with `-sn` and
|
|
# subtitles are independent tracks served by GET /subtitle, so a WebVTT name here could only ever be
|
|
# a name that will not arrive — and accepting it cost a 20 s parked worker per request.
|
|
_HLS_CT = {
|
|
".m3u8": "application/vnd.apple.mpegurl",
|
|
".ts": "video/mp2t",
|
|
}
|
|
# A real artifact name is ~20 characters. The bound is what keeps an absurd one out of the parser:
|
|
# `int()` on an unbounded digit run raises ValueError past 4300 digits (CPython's
|
|
# `sys.get_int_max_str_digits`), which reached the client as a 500 — the same shape of bug the
|
|
# subtitle routes were bounded for (`_RatingKey`, above).
|
|
_HLS_NAME_MAX = 128
|
|
|
|
|
|
@router.get("/stream/{rating_key}/hls/{session_id}/{filename}")
|
|
def stream_hls_file(
|
|
rating_key: str,
|
|
session_id: str,
|
|
filename: str,
|
|
user: User = Depends(current_user),
|
|
) -> Response:
|
|
"""Serve a file (playlist / segment / subtitle) from ONE HLS session's directory. The playlist
|
|
references its segments/subtitles by relative name, all resolving back here.
|
|
|
|
The session id in the path is what makes "back here" mean the same session. Sessions are keyed by
|
|
(viewer, rating_key) and a seek-restart replaces the viewer's registered one, while the DIRECTORY
|
|
is per (key, viewer, offset, tag) — so resolving by the key alone handed the old player the new
|
|
session's `seg_12.ts`, produced from a different offset under `-copyts`: wrong
|
|
bytes with a 200, which the player has no way to notice (deep-review round 5). A request for a
|
|
session that is no longer the registered one is a 404, and the player reclaims at the playhead.
|
|
|
|
That reclaim is also why a request with no live session is no longer auto-started here: the
|
|
auto-start could only ever begin at 0, silently throwing away the viewer's position."""
|
|
# Filename must be a plain hls artifact — no path separators, only the expected extensions.
|
|
ext = "." + filename.rsplit(".", 1)[-1] if "." in filename else ""
|
|
if (
|
|
len(filename) > _HLS_NAME_MAX
|
|
or ext not in _HLS_CT
|
|
or "/" in filename
|
|
or "\\" in filename
|
|
or ".." in filename
|
|
):
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
s = plex_stream.current_session(user.id, str(rating_key))
|
|
if s is None or s.id != session_id:
|
|
raise HTTPException(status_code=404, detail="No active playback session")
|
|
path = s.dir / filename
|
|
# Only a name THIS session can actually produce is worth waiting for: its own playlists, and a
|
|
# segment of one of its own renditions whose predecessor is already on disk (ffmpeg writes them
|
|
# in order, so if it is not there this one is not next). The session is passed, not just its
|
|
# directory, because the single- and multi-audio argv branches spell their names differently and
|
|
# neither can produce the other's. Answering now instead of waiting it out is what keeps a seek
|
|
# past the produced region — or a deliberate flood of hopeless names — from parking threadpool
|
|
# workers and their DB connections (see `artifact_may_arrive`).
|
|
if not plex_stream.artifact_may_arrive(s, filename):
|
|
raise HTTPException(status_code=404, detail="Not available yet")
|
|
timeout = 30 if ext == ".m3u8" else 20
|
|
# `proc` so a session whose ffmpeg has already exited answers at once instead of polling out the
|
|
# whole timeout: this is a SYNC route on the ~40-slot threadpool, holding a pooled DB connection
|
|
# with it, so every such wait is a worker (and a connection) parked for 20 s — an ordinary seek
|
|
# past the produced region was enough to do it (round 5).
|
|
if not plex_stream.wait_for(path, timeout, proc=s.proc):
|
|
raise HTTPException(status_code=404, detail="Not available yet")
|
|
try:
|
|
content = path.read_bytes()
|
|
except OSError:
|
|
# The session was torn down between the wait and the read — `_kill` rmtrees the whole
|
|
# directory, and nothing holds `_lock` across those two steps. It has to present as the same
|
|
# 404 a gone session does: the player's recovery is keyed on the STATUS (`code === 404`), so
|
|
# a 500 here skipped the reclaim and ended playback with "couldn't play" instead of
|
|
# restarting at the playhead. Ordinary trigger: `loadSession` awaits the new session — which
|
|
# kills and rmtrees the old one — before destroying the old hls instance, so the outgoing
|
|
# instance's in-flight segment request races the rmtree on every seek-restart.
|
|
raise HTTPException(status_code=404, detail="No active playback session") from None
|
|
return Response(
|
|
content=content,
|
|
media_type=_HLS_CT[ext],
|
|
headers={"Cache-Control": "no-store"},
|
|
)
|