5592f60e85
The Plex watch import inserted plex_states via per-row ORM add against a pre-read 'existing' set, which is not safe when the enable call fires more than once while the multi-second import runs (the toggle wasn't disabled during it): overlapping imports raced to INSERT the same (user_id, item_id) rows and all but the first hit a UniqueViolation -> 500 (seen on prod). Rewrite the write path as a chunked PostgreSQL UPSERT (on_conflict_do_update, rows de-duped by item_id) — idempotent and concurrency-safe, Plex still wins on the intersection and Siftlode-only states are untouched. Also disable the Settings toggle (and ignore its onChange) while a sync mutation is pending, so it can't be fired repeatedly; adds a reusable 'disabled' prop to the Switch primitive.
142 lines
6.5 KiB
Python
142 lines
6.5 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.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
|