peter
b8bff504d8
The button deletes the local row and mirrors the un-mark to Plex, but it decided whether to push from the previous STATUS alone (prev_status != "new") — and an in-progress item's row is status="new" carrying a position, which is exactly how the importer maps a bare viewOffset. So for the single case the button exists for, nothing was ever sent to Plex: the offset stayed, and the next pull imported it straight back. The user cleared Protector twice and it returned to Continue watching at 92s both times. Push now when Plex holds anything to clear — a view count, or a resume point. "hidden" stays Siftlode-only and pushes nothing, which is what this endpoint's docstring always claimed and the status-only guard contradicted. One unscrobble clears BOTH viewCount and viewOffset — measured against the live server from a watched+offset state and an offset-only one, so the push itself needed no change. The client docstring that claimed "clears viewCount" is what sent the first diagnosis down a wrong path (a timeline write with time=0, which does not clear an offset at all); it now records what the server actually does. Verified end to end on the real Plex server: the button leaves viewCount=None viewOffset=None, and a full 44,885-item re-import brings back no row for it.
647 lines
28 KiB
Python
647 lines
28 KiB
Python
"""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
|
|
|
|
import httpx
|
|
from fastapi import (
|
|
APIRouter,
|
|
BackgroundTasks,
|
|
Depends,
|
|
HTTPException,
|
|
Request,
|
|
Response,
|
|
)
|
|
from fastapi.responses import FileResponse
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import select
|
|
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 (
|
|
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)
|
|
|
|
|
|
@router.get("/subtitle/{rating_key}/{sub_ord}")
|
|
def subtitle(rating_key: str, sub_ord: int, request: Request, offset: float = 0.0) -> 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."""
|
|
_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")
|
|
|
|
|
|
# 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,
|
|
)
|
|
|
|
|
|
class ItemProgressIn(BaseModel):
|
|
# 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).
|
|
position_seconds: float = 0
|
|
duration_seconds: float = 0
|
|
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)
|
|
position = max(0, int(body.position_seconds))
|
|
duration = int(body.duration_seconds)
|
|
final = body.final
|
|
now = datetime.now(timezone.utc)
|
|
st = (
|
|
db.execute(
|
|
select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == it.id)
|
|
)
|
|
.scalars()
|
|
.first()
|
|
)
|
|
was_watched = st is not None and st.status == "watched"
|
|
|
|
near_end = duration > 0 and position > duration - _FINISH_MARGIN_S
|
|
if near_end:
|
|
if st is None:
|
|
st = PlexState(user_id=user.id, item_id=it.id)
|
|
db.add(st)
|
|
st.status = "watched"
|
|
st.watched_at = now
|
|
st.position_seconds = 0
|
|
st.progress_updated_at = now
|
|
st.synced_to_plex = False
|
|
# Terminal state for resume write-back: drop the throttle/ordering entry so the map doesn't
|
|
# retain a dead key for every finished item.
|
|
with _push_throttle_lock:
|
|
_last_push_at.pop((user.id, it.id), None)
|
|
db.commit()
|
|
# Immediate on the watched transition; skip a repeat scrobble if it was already watched
|
|
# (a later checkpoint of the same finished item would otherwise bump Plex's viewCount again).
|
|
if not was_watched:
|
|
_push_watch(background, db, user.id, it.id, rating_key, "watched")
|
|
return {"status": "watched", "position_seconds": 0}
|
|
|
|
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,
|
|
start: float = 0.0,
|
|
audio: int | None = None,
|
|
aoff: float = 0.0,
|
|
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."""
|
|
plex_stream.reap_idle() # opportunistic cleanup of idle sessions on each new playback
|
|
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, 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,
|
|
"url": f"/api/plex/stream/{it.rating_key}/hls/{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)
|
|
|
|
|
|
_HLS_CT = {
|
|
".m3u8": "application/vnd.apple.mpegurl",
|
|
".ts": "video/mp2t",
|
|
".vtt": "text/vtt",
|
|
}
|
|
|
|
|
|
@router.get("/stream/{rating_key}/hls/{filename}")
|
|
def stream_hls_file(
|
|
rating_key: str,
|
|
filename: str,
|
|
user: User = Depends(current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> Response:
|
|
"""Serve a file (playlist / segment / subtitle) from the current HLS session directory. The
|
|
playlist references its segments/subtitles by relative name, all resolving back here."""
|
|
# Filename must be a plain hls artifact — no path separators, only the expected extensions.
|
|
ext = "." + filename.rsplit(".", 1)[-1] if "." in filename else ""
|
|
if 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(str(rating_key))
|
|
if s is None:
|
|
# No session yet (e.g. reaped): the ENTRY playlist may auto-start one at 0 — master.m3u8 for a
|
|
# multi-audio item, index.m3u8 for a single-audio remux item.
|
|
it = item_or_404(db, rating_key)
|
|
if filename == "master.m3u8":
|
|
s = plex_stream.start_session(db, it, 0.0, multi=True)
|
|
elif filename == "index.m3u8" and it.playable == "remux":
|
|
s = plex_stream.start_session(db, it, 0.0)
|
|
else:
|
|
raise HTTPException(status_code=404, detail="No active playback session")
|
|
if s is None:
|
|
raise HTTPException(status_code=404, detail="Media file not found on disk")
|
|
path = s.dir / filename
|
|
timeout = 30 if ext == ".m3u8" else 20
|
|
if not plex_stream.wait_for(path, timeout):
|
|
raise HTTPException(status_code=404, detail="Not available yet")
|
|
return Response(
|
|
content=path.read_bytes(),
|
|
media_type=_HLS_CT[ext],
|
|
headers={"Cache-Control": "no-store"},
|
|
)
|