From 3a3ba17fb8c90f8cbf17a4e3d92642eb8f29b2d0 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Fri, 10 Jul 2026 00:17:56 +0200 Subject: [PATCH 1/2] =?UTF-8?q?feat(plex):=20Phase=20B=20=E2=80=94=20Siftl?= =?UTF-8?q?ode=E2=86=92Plex=20watch-state=20push?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mirror watched/unwatched/resume from Siftlode to a linked (owner) Plex account. - PlexClient: scrobble / unscrobble / set_timeline (/:/scrobble, /:/unscrobble, /:/timeline). - watch_sync: link_for_push() gates on owner + sync_enabled + initial_import_done; best-effort push_state_to_plex() runs as a background task with its own session, never raises (Plex being down must not break the user's action), and flips synced_to_plex on success so Phase C's pull won't bounce it back. - item_state / item_progress schedule the push: watched/unwatched immediately, resume only on a "final" checkpoint (pause / pagehide / unmount) — not every 10s tick — so one watch doesn't spray /:/timeline at the server. `hidden` is Siftlode-only and never touches Plex. - Frontend: plexProgress / plexProgressBeacon carry a `final` flag; the periodic checkpoint is non-final, pause/pagehide/leaving the player are final. Verified live against the real Plex server (scrobble/unscrobble/timeline round-trip + restore; link gating; synced_to_plex flip). --- backend/app/plex/client.py | 32 +++++++++++++++ backend/app/plex/watch_sync.py | 55 ++++++++++++++++++++++++++ backend/app/routes/plex.py | 53 +++++++++++++++++++++++-- frontend/src/components/PlexPlayer.tsx | 18 ++++++--- frontend/src/lib/api.ts | 21 ++++++++-- 5 files changed, 167 insertions(+), 12 deletions(-) diff --git a/backend/app/plex/client.py b/backend/app/plex/client.py index 2dd0af6..721bb18 100644 --- a/backend/app/plex/client.py +++ b/backend/app/plex/client.py @@ -166,6 +166,38 @@ class PlexClient: items = mc.get("Metadata", []) or [] return items[0] if items else None + # --- Watch-state write-back (P5 Phase B — Siftlode→Plex push; acts on the token's account) --- + _LIBRARY_IDENTIFIER = "com.plexapp.plugins.library" + + def scrobble(self, rating_key: str) -> None: + """Mark the item *watched* on the token's Plex account (viewCount++, resume cleared).""" + self._write( + "GET", + f"/:/scrobble?identifier={self._LIBRARY_IDENTIFIER}" + f"&key={quote(str(rating_key), safe='')}", + ) + + def unscrobble(self, rating_key: str) -> None: + """Mark the item *unwatched* on the token's Plex account (clears viewCount).""" + self._write( + "GET", + f"/:/unscrobble?identifier={self._LIBRARY_IDENTIFIER}" + f"&key={quote(str(rating_key), safe='')}", + ) + + def set_timeline( + self, rating_key: str, time_ms: int, duration_ms: int, state: str = "stopped" + ) -> None: + """Report a resume position (viewOffset) for the item on the token's account. `state` is a + Plex timeline state (playing|paused|stopped); we push the settled position as ``stopped``.""" + key = quote(f"/library/metadata/{rating_key}", safe="") + self._write( + "GET", + f"/:/timeline?identifier={self._LIBRARY_IDENTIFIER}" + f"&ratingKey={quote(str(rating_key), safe='')}&key={key}" + f"&state={state}&time={int(time_ms)}&duration={int(duration_ms)}", + ) + def raw_get(self, path: str) -> bytes: """Raw bytes of an arbitrary Plex resource path (e.g. an external subtitle stream `key` like ``/library/streams/553184``). Keeps the admin token server-side.""" diff --git a/backend/app/plex/watch_sync.py b/backend/app/plex/watch_sync.py index 8763785..0a41889 100644 --- a/backend/app/plex/watch_sync.py +++ b/backend/app/plex/watch_sync.py @@ -21,6 +21,7 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.orm import Session from app import sysconfig +from app.db import SessionLocal from app.models import PlexItem, PlexLink, PlexState from app.plex.client import PlexClient, PlexError, PlexNotConfigured from app.plex.sync import ( @@ -140,3 +141,57 @@ def import_owner_watch_state(db: Session, link: PlexLink) -> dict: log.info("Plex watch import (user %s): %s", link.user_id, stats) return stats + + +# --- Phase B: Siftlode → Plex push ----------------------------------------------------------------- + + +def link_for_push(db: Session, user_id: int) -> PlexLink | None: + """The active owner watch-push link for this user, or None when push is off / not the owner. + Push is gated on `initial_import_done` so we never push a half-known local state over Plex's + authoritative one before the one-time "Plex is master" import has reconciled them.""" + link = db.query(PlexLink).filter_by(user_id=user_id).first() + if link and link.uses_admin and link.sync_enabled and link.initial_import_done: + return link + return None + + +def push_state_to_plex( + user_id: int, + item_id: int, + rating_key: str, + action: str, + time_ms: int = 0, + duration_ms: int = 0, +) -> None: + """Best-effort Siftlode→Plex push, run in a FastAPI BackgroundTask (its OWN DB session — the + request's session is already closed by the time this runs). Never raises: Plex being down or slow + must not break the user's action. On success it flags `synced_to_plex` so Phase C's incremental + pull won't bounce the change straight back; on failure the row stays dirty (synced_to_plex=False) + for a later retry/reconcile. + + `action` is one of ``watched`` (scrobble), ``unwatched`` (unscrobble), ``resume`` (timeline).""" + with SessionLocal() as db: + if link_for_push(db, user_id) is None: + return + try: + with PlexClient(db) as plex: + if action == "watched": + plex.scrobble(rating_key) + elif action == "unwatched": + plex.unscrobble(rating_key) + elif action == "resume": + plex.set_timeline(rating_key, time_ms, duration_ms) + else: + return + except PlexError as e: + log.warning( + "Plex watch push failed (user %s, item %s, %s): %s", user_id, item_id, action, e + ) + return + # Flag the row synced — but an "unwatched" via status=new deletes the row, so there may be + # nothing to flag; that's fine (unscrobble still happened). + st = db.query(PlexState).filter_by(user_id=user_id, item_id=item_id).first() + if st is not None: + st.synced_to_plex = True + db.commit() diff --git a/backend/app/routes/plex.py b/backend/app/routes/plex.py index 19d390f..f44b5ec 100644 --- a/backend/app/routes/plex.py +++ b/backend/app/routes/plex.py @@ -16,7 +16,7 @@ from pathlib import Path from urllib.parse import quote import httpx -from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response +from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, Request, Response from fastapi.responses import FileResponse from sqlalchemy import and_, func, or_, text from sqlalchemy.orm import Session, aliased @@ -1486,22 +1486,49 @@ def item_detail( } +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, +) -> 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.""" + if plex_watch.link_for_push(db, user_id) is None: + return + background.add_task( + plex_watch.push_state_to_plex, user_id, item_id, rating_key, action, time_ms, duration_ms + ) + + @router.post("/item/{rating_key}/progress") def item_progress( rating_key: str, payload: dict, + 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).""" + """Checkpoint the resume position while playing (near the end → mark watched + clear it). + + `final` (sent on the pause/pagehide/unmount flush, not the periodic 10s checkpoint) is what + debounces the Plex resume push: we mirror the settled position to Plex only on a final flush, so + a single watch doesn't spray `/:/timeline` calls at the server every 10 seconds.""" it = _item_or_404(db, rating_key) try: position = max(0, int(payload.get("position_seconds") or 0)) duration = int(payload.get("duration_seconds") or 0) except (TypeError, ValueError): raise HTTPException(status_code=400, detail="position_seconds must be an integer") + final = bool(payload.get("final")) now = datetime.now(timezone.utc) st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).first() + was_watched = st is not None and st.status == "watched" near_end = duration > 0 and position > duration - _FINISH_MARGIN_S if near_end: @@ -1512,7 +1539,12 @@ def item_progress( st.watched_at = now st.position_seconds = 0 st.progress_updated_at = now + st.synced_to_plex = False 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: @@ -1529,7 +1561,13 @@ def item_progress( db.add(st) st.position_seconds = position st.progress_updated_at = now + if final: + st.synced_to_plex = False db.commit() + if final: + _push_watch( + background, db, user.id, it.id, rating_key, "resume", position * 1000, duration * 1000 + ) return {"position_seconds": position} @@ -1537,19 +1575,25 @@ def item_progress( def item_state( rating_key: str, payload: dict, + background: BackgroundTasks, user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - """Set the per-user watch status (new|watched|hidden).""" + """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 = payload.get("status") if status not in ("new", "watched", "hidden"): raise HTTPException(status_code=400, detail="Invalid status") st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).first() + prev_status = st.status if st is not None else "new" if status == "new": if st is not None: db.delete(st) db.commit() + if prev_status != "new": # something → unwatched: mirror the un-mark to 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) @@ -1558,7 +1602,10 @@ def item_state( 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} diff --git a/frontend/src/components/PlexPlayer.tsx b/frontend/src/components/PlexPlayer.tsx index 792b2dd..bb834c4 100644 --- a/frontend/src/components/PlexPlayer.tsx +++ b/frontend/src/components/PlexPlayer.tsx @@ -452,23 +452,31 @@ export default function PlexPlayer({ itemId, onClose, queue }: Props) { const absRef = useRef(0); absRef.current = abs; useEffect(() => { - const flush = () => { + // `final` = a settled checkpoint (pause / page-hide / leaving the player) vs. the periodic tick. + // Only final checkpoints get mirrored to a linked Plex account (backend-side debounce), so Plex + // sees the resume position when the user actually stops — not a timeline call every 10 seconds. + const flush = (final = false) => { if (absRef.current > 0 && durationRef.current > 0) - api.plexProgress(id, Math.floor(absRef.current), durationRef.current).catch(() => {}); + api.plexProgress(id, Math.floor(absRef.current), durationRef.current, final).catch(() => {}); }; const iv = window.setInterval(flush, 10000); + // Pausing is a settled position — push it to Plex now rather than waiting for unmount/unload. + const onPauseFlush = () => flush(true); + const video = videoRef.current; + video?.addEventListener("pause", onPauseFlush); // Save on page unload too (F5/close/navigate): React effect cleanup does NOT run on a full reload, // so without this the resume point lags up to 10s — e.g. a seek right before F5 was lost. Uses a - // keepalive beacon so the POST survives the unload. + // keepalive beacon so the POST survives the unload (a final checkpoint — the user is leaving). const onHide = () => { if (absRef.current > 0 && durationRef.current > 0) - api.plexProgressBeacon(id, Math.floor(absRef.current), durationRef.current); + api.plexProgressBeacon(id, Math.floor(absRef.current), durationRef.current, true); }; window.addEventListener("pagehide", onHide); return () => { window.clearInterval(iv); window.removeEventListener("pagehide", onHide); - flush(); + video?.removeEventListener("pause", onPauseFlush); + flush(true); // Back (unmount) persists the position server-side via flush(), but reopening the same item in // the SAME session read react-query's CACHED detail — whose position_seconds predated this // watch — so the resume landed at the old spot (the progress GET could also race the POST). diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index b8f889a..ab645b3 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -1244,16 +1244,29 @@ export const api = { // cue times onto the HLS session's zero-based clock so subtitles line up with speech after a seek. plexSubtitleUrl: (id: string, ord: number, offset = 0): string => `/api/plex/subtitle/${encodeURIComponent(id)}/${ord}${offset ? `?offset=${offset}` : ""}`, - plexProgress: (id: string, position_seconds: number, duration_seconds: number): Promise => + // `final` marks a settled checkpoint (pause / page-hide / leaving the player) as opposed to the + // periodic 10s tick — the backend mirrors the resume position to a linked Plex account only on a + // final checkpoint, so a single watch doesn't spray Plex with timeline updates. + plexProgress: ( + id: string, + position_seconds: number, + duration_seconds: number, + final = false, + ): Promise => req(`/api/plex/item/${encodeURIComponent(id)}/progress`, { method: "POST", - body: JSON.stringify({ position_seconds, duration_seconds }), + body: JSON.stringify({ position_seconds, duration_seconds, final }), }), // Fire-and-forget progress save that SURVIVES page unload (F5/close/navigate): a normal fetch is // cancelled on unload and React effect cleanup doesn't run on a full reload, so the last position // (esp. right after a seek) would be lost. `keepalive` lets the POST complete during unload; we // replicate req()'s credentials + account header (no retry — there's no page left to retry on). - plexProgressBeacon: (id: string, position_seconds: number, duration_seconds: number): void => { + plexProgressBeacon: ( + id: string, + position_seconds: number, + duration_seconds: number, + final = false, + ): void => { const active = getActiveAccount(); try { void fetch(`/api/plex/item/${encodeURIComponent(id)}/progress`, { @@ -1264,7 +1277,7 @@ export const api = { "Content-Type": "application/json", ...(active != null ? { [ACTIVE_ACCOUNT_HEADER]: String(active) } : {}), }, - body: JSON.stringify({ position_seconds, duration_seconds }), + body: JSON.stringify({ position_seconds, duration_seconds, final }), }).catch(() => {}); } catch { /* ignore */ From 363b4d17fcbe6e247b0ce7abd572912299042db7 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Fri, 10 Jul 2026 00:18:44 +0200 Subject: [PATCH 2/2] fix(plex): mark the finished episode on the correct id when auto-advancing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The player's watched-mark on completion keyed off the reactive `id`, which can run ahead of the media actually playing: when the next episode fails to load (e.g. no local file), loadSession returns without replacing the