3a3ba17fb8
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).
197 lines
8.9 KiB
Python
197 lines
8.9 KiB
Python
"""Plex ↔ Siftlode watch-state sync.
|
|
|
|
Plex records watch state PER PLEX ACCOUNT; Siftlode records it per Siftlode user in `plex_states`.
|
|
A `PlexLink` row maps one Siftlode user to a Plex account. For the owner (MVP) that account is the
|
|
server admin (`uses_admin=True`), so the existing admin token already reads/writes the owner's state
|
|
— no separate Plex login.
|
|
|
|
Phase A (this module today): the one-time **"Plex is master" import** — read the owner account's
|
|
viewCount / viewOffset / lastViewedAt (present in the same cheap section listing the catalog mirror
|
|
already pages) and upsert them into the owner's `plex_states`. Plex wins on overlap; states that
|
|
exist only in Siftlode are left untouched (union, Plex-authoritative on the intersection).
|
|
|
|
Phases B (Siftlode→Plex push via scrobble/timeline) and C (incremental Plex→Siftlode via the watch
|
|
history + onDeck, last-write-wins) build on this and land in later ships.
|
|
"""
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import func
|
|
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 (
|
|
_TYPE_EPISODE,
|
|
_TYPE_MOVIE,
|
|
_enabled_section_keys,
|
|
_epoch,
|
|
_paginate,
|
|
)
|
|
|
|
log = logging.getLogger("siftlode.plex.watch")
|
|
|
|
# A resume position below this many seconds isn't worth mirroring — matches the player's own
|
|
# _PROGRESS_MIN_S (a stray few-second seek is not "in progress").
|
|
_PROGRESS_MIN_S = 5
|
|
|
|
|
|
def _plex_watch_to_state(
|
|
view_count, view_offset_ms, last_viewed_at
|
|
) -> tuple[str, int, datetime | None, datetime | None] | None:
|
|
"""Translate a Plex leaf's per-account watch fields into a Siftlode state tuple
|
|
``(status, position_seconds, watched_at, progress_updated_at)``, or None when Plex holds no
|
|
meaningful watch signal for it (so the import skips it and never clears a Siftlode-only state).
|
|
|
|
In Siftlode's model (mirrors item_progress/item_state): a *watched* item is ``status="watched"``
|
|
with position 0; an *in-progress* item keeps the default status and carries ``position_seconds``
|
|
plus ``progress_updated_at``."""
|
|
ts = _epoch(last_viewed_at)
|
|
if view_count and int(view_count) > 0:
|
|
return ("watched", 0, ts, ts)
|
|
if view_offset_ms:
|
|
pos = int(view_offset_ms) // 1000
|
|
if pos >= _PROGRESS_MIN_S:
|
|
return ("new", pos, None, ts)
|
|
return None
|
|
|
|
|
|
def import_owner_watch_state(db: Session, link: PlexLink) -> dict:
|
|
"""One-time "Plex is master" import of the linked user's Plex watch state into their
|
|
`plex_states`. Idempotent (safe to re-run — Plex simply re-wins on the intersection). Only
|
|
touches items Plex has a watch record for; Siftlode-only states are preserved."""
|
|
if not sysconfig.get_bool(db, "plex_enabled"):
|
|
return {"skipped": "disabled"}
|
|
|
|
# rating_key -> plex_items.id for every mirrored playable leaf (movie/episode).
|
|
item_id_by_rk: dict[str, int] = {
|
|
rk: iid for rk, iid in db.query(PlexItem.rating_key, PlexItem.id)
|
|
}
|
|
stats = {"watched": 0, "in_progress": 0, "scanned": 0}
|
|
wanted = _enabled_section_keys(db)
|
|
# Accumulate one row per item (keyed by item_id → inherently de-duped) then bulk-UPSERT. An upsert
|
|
# is idempotent AND concurrency-safe: a repeated/overlapping enable (the toggle firing twice while
|
|
# the multi-second import runs) can't raise a duplicate-key error — the conflict just re-updates.
|
|
rows_by_item: dict[int, dict] = {}
|
|
|
|
try:
|
|
with PlexClient(db) as plex:
|
|
for s in plex.sections():
|
|
stype = s.get("type")
|
|
if stype not in ("movie", "show"):
|
|
continue
|
|
key = str(s.get("key"))
|
|
if wanted is not None and key not in wanted:
|
|
continue
|
|
item_type = _TYPE_MOVIE if stype == "movie" else _TYPE_EPISODE
|
|
for meta in _paginate(plex, key, item_type):
|
|
stats["scanned"] += 1
|
|
item_id = item_id_by_rk.get(str(meta.get("ratingKey") or ""))
|
|
if item_id is None:
|
|
continue
|
|
res = _plex_watch_to_state(
|
|
meta.get("viewCount"),
|
|
meta.get("viewOffset"),
|
|
meta.get("lastViewedAt"),
|
|
)
|
|
if res is None:
|
|
continue
|
|
status, pos, watched_at, prog_at = res
|
|
rows_by_item[item_id] = {
|
|
"user_id": link.user_id,
|
|
"item_id": item_id,
|
|
"status": status,
|
|
"position_seconds": pos,
|
|
"watched_at": watched_at,
|
|
"progress_updated_at": prog_at,
|
|
# Mirrors Plex now — so Phase B's push doesn't bounce it straight back.
|
|
"synced_to_plex": True,
|
|
}
|
|
stats["watched" if status == "watched" else "in_progress"] += 1
|
|
|
|
# Bulk upsert in chunks. Plex wins on the intersection; Siftlode-only states (item_ids not
|
|
# in rows_by_item) are never referenced, so they're preserved untouched (union).
|
|
rows = list(rows_by_item.values())
|
|
for i in range(0, len(rows), 1000):
|
|
stmt = pg_insert(PlexState).values(rows[i : i + 1000])
|
|
stmt = stmt.on_conflict_do_update(
|
|
index_elements=[PlexState.user_id, PlexState.item_id],
|
|
set_={
|
|
"status": stmt.excluded.status,
|
|
"position_seconds": stmt.excluded.position_seconds,
|
|
"watched_at": stmt.excluded.watched_at,
|
|
"progress_updated_at": stmt.excluded.progress_updated_at,
|
|
"synced_to_plex": stmt.excluded.synced_to_plex,
|
|
"updated_at": func.now(),
|
|
},
|
|
)
|
|
db.execute(stmt)
|
|
link.initial_import_done = True
|
|
link.last_watch_sync_at = datetime.now(timezone.utc)
|
|
db.commit()
|
|
except PlexNotConfigured as e:
|
|
return {"skipped": str(e)}
|
|
except PlexError as e:
|
|
db.rollback()
|
|
log.warning("Plex watch import failed (user %s): %s", link.user_id, e)
|
|
return {"error": str(e)}
|
|
|
|
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()
|