692 lines
33 KiB
Python
692 lines
33 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) build on this and land in later ships. As of S2 Plex is the source of truth: the
|
|
player pushes the resume position back to Plex on the periodic checkpoint (not just the final flush),
|
|
and the incremental pull resolves a genuine divergence to Plex's state (see `_pull_apply`) rather than
|
|
running a last-write-wins comparison.
|
|
"""
|
|
|
|
# On the PlexError lines in this module: none of them carries a traceback, deliberately. A
|
|
# PlexError is httpx's own message — the method, the URL and the status of one call — and the stack
|
|
# beneath it is always this `except` plus `PlexClient._get`. The attrs beside the line say which
|
|
# user and which phase, which is the part the message cannot know.
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import func, select, update
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import logs, sysconfig
|
|
from app.db import SessionLocal
|
|
from app.models import PlexItem, PlexLibrary, 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 = logs.get_logger(__name__)
|
|
# 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 _rk_to_id(db: Session) -> dict[str, int]:
|
|
"""The rating_key → PlexItem.id lookup used to resolve a Plex leaf back to its mirrored row."""
|
|
return {rk: iid for rk, iid in db.execute(select(PlexItem.rating_key, PlexItem.id)).all()}
|
|
|
|
|
|
def _scan_plex_states(
|
|
db: Session, plex: PlexClient
|
|
) -> tuple[dict[int, tuple[str, int, datetime | None, datetime | None]], int, set[str], set[str]]:
|
|
"""Scan the enabled movie/show sections and return ``{item_id: (status, pos, watched_at,
|
|
prog_at)}`` for every mirrored leaf Plex holds a watch signal for, the scanned-leaf count, the
|
|
keys of the sections SETTLED — walked to the server's total (or said empty by a server we mirror
|
|
nothing from) — and the keys of every section walked.
|
|
Shared by the one-time import and the full reconcile (both need Plex's whole current picture).
|
|
The settled keys matter to the reconcile alone: it reads "absent from the scan" as "Plex
|
|
un-watched it", which a walk cut short by a Plex hiccup does not mean (see `_paginate`). Settled
|
|
is decided PER SECTION: one section that never settles (a library emptied on Plex while its
|
|
mirrored items are kept) must not stop un-watches syncing in all the others (E32 review)."""
|
|
item_id_by_rk = _rk_to_id(db)
|
|
wanted = _enabled_section_keys(db)
|
|
# The sections we mirror something from: a total of 0 from one of those contradicts the mirror.
|
|
mirrored = set(
|
|
db.execute(
|
|
select(PlexLibrary.plex_key).join(PlexItem, PlexItem.library_id == PlexLibrary.id)
|
|
).scalars()
|
|
)
|
|
scanned = 0
|
|
settled: set[str] = set()
|
|
walked: set[str] = set()
|
|
out: dict[int, tuple[str, int, datetime | None, datetime | None]] = {}
|
|
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
|
|
walked.add(key)
|
|
progress: dict = {}
|
|
before = scanned
|
|
for meta in _paginate(plex, key, item_type, progress):
|
|
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 not None:
|
|
out[item_id] = res
|
|
# A section the server SAYS is empty (total 0, nothing listed) settles too. Believed only for
|
|
# a section we mirror nothing from: one whose items we hold answering 0 is a Plex that lost
|
|
# the library for a moment, not one that un-watched all of it (E32 review) — so it is left
|
|
# unsettled, and its rows are not cleared.
|
|
listed = scanned > before
|
|
said_empty = progress.get("total") == 0 and not listed and key not in mirrored
|
|
if (progress.get("complete") and listed) or said_empty:
|
|
settled.add(key)
|
|
return out, scanned, settled, walked
|
|
|
|
|
|
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"}
|
|
|
|
stats = {"watched": 0, "in_progress": 0, "scanned": 0}
|
|
try:
|
|
with PlexClient(db) as plex:
|
|
# A partial scan is still safe here: the import only ever ADDS what Plex reports.
|
|
scanned_states, stats["scanned"], _settled, _walked = _scan_plex_states(db, plex)
|
|
# 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 — it re-updates.
|
|
rows = []
|
|
for item_id, (status, pos, watched_at, prog_at) in scanned_states.items():
|
|
rows.append(
|
|
{
|
|
"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
|
|
# scanned) are never referenced, so they're preserved untouched (union).
|
|
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()
|
|
# silent-ok: the `{"skipped": …}` return is the report — the admin route answers it and the
|
|
# scheduler logs it on the job line.
|
|
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.execute(select(PlexLink).where(PlexLink.user_id == user_id)).scalars().first()
|
|
if link and link.uses_admin and link.sync_enabled and link.initial_import_done:
|
|
return link
|
|
return None
|
|
|
|
|
|
def _state_marker(st: PlexState | None) -> datetime | None:
|
|
"""The freshest local-edit timestamp on a state row (watched_at / progress_updated_at). A push
|
|
captures it at schedule time so it can tell, when it later settles the `synced_to_plex` flag,
|
|
whether the row has been re-edited in the meantime (see `push_state_to_plex`)."""
|
|
if st is None:
|
|
return None
|
|
return max([t for t in (st.watched_at, st.progress_updated_at) if t], default=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,
|
|
expect_ts: datetime | None = None,
|
|
) -> 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).
|
|
`expect_ts` is the row's freshest timestamp captured when this push was scheduled: if the row has
|
|
since been re-edited (a newer local change with its own pending push), we must NOT mark it clean —
|
|
doing so would strand that newer value, never pushing it to Plex (lost update)."""
|
|
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). Skip the flag if the row was
|
|
# re-edited since we captured `expect_ts` (leave it dirty for its own push / the reconcile).
|
|
st = (
|
|
db.execute(
|
|
select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id)
|
|
)
|
|
.scalars()
|
|
.first()
|
|
)
|
|
if st is not None and _row_matches(st, expect_ts):
|
|
st.synced_to_plex = True
|
|
db.commit()
|
|
|
|
|
|
def _row_matches(st: PlexState, expect_ts: datetime | None) -> bool:
|
|
"""Whether a state row is still the one a push was scheduled for — true when no marker was given
|
|
(legacy/edge), the row carries no timestamp, or its freshest timestamp is within a second of the
|
|
captured marker (a genuine newer edit is always many seconds later)."""
|
|
if expect_ts is None:
|
|
return True
|
|
cur = _state_marker(st)
|
|
return cur is None or abs((cur - expect_ts).total_seconds()) < 1.0
|
|
|
|
|
|
def push_bulk_state_to_plex(user_id: int, changes: list[tuple[int, str, str]]) -> None:
|
|
"""Coalesced Siftlode→Plex push for a whole-show / whole-season mark (see `_bulk_state`). Same
|
|
contract as `push_state_to_plex` (own DB session, best-effort, never raises) but ONE background
|
|
task for the entire batch: it reuses a single DB session and a single keep-alive PlexClient across
|
|
all `changes` instead of scheduling N tasks that each rebuild a session + HTTP client. Plex has no
|
|
batch-scrobble endpoint, so the per-item HTTP calls remain (looped over the shared connection), but
|
|
the O(N) task/session/client setup is gone. `changes` is a list of (item_id, rating_key, action)
|
|
with action ``watched`` (scrobble) or ``unwatched`` (unscrobble)."""
|
|
if not changes:
|
|
return
|
|
with SessionLocal() as db:
|
|
if link_for_push(db, user_id) is None:
|
|
return
|
|
synced_ids: list[int] = []
|
|
try:
|
|
with PlexClient(db) as plex:
|
|
for item_id, rating_key, action in changes:
|
|
try:
|
|
if action == "watched":
|
|
plex.scrobble(rating_key)
|
|
elif action == "unwatched":
|
|
plex.unscrobble(rating_key)
|
|
else:
|
|
continue
|
|
except PlexError as e:
|
|
# One bad item mustn't abort the rest of the batch; leave its row dirty
|
|
# (synced_to_plex=False) for a later reconcile.
|
|
log.warning(
|
|
"Plex bulk push failed (user %s, item %s, %s): %s",
|
|
user_id,
|
|
item_id,
|
|
action,
|
|
e,
|
|
)
|
|
continue
|
|
synced_ids.append(item_id)
|
|
except PlexError as e:
|
|
# Couldn't even open the client — nothing pushed; everything stays dirty for reconcile.
|
|
log.warning("Plex bulk push could not connect (user %s): %s", user_id, e)
|
|
return
|
|
# Flag every successfully-pushed row that still exists (an "unwatched" deletes the row, so
|
|
# only the "watched" changes have a row to flag).
|
|
if synced_ids:
|
|
db.execute(
|
|
update(PlexState)
|
|
.where(PlexState.user_id == user_id, PlexState.item_id.in_(synced_ids))
|
|
.values(synced_to_plex=True)
|
|
)
|
|
db.commit()
|
|
|
|
|
|
# --- Phase C: incremental + full Plex ↔ Siftlode reconcile (scheduler jobs) ------------------------
|
|
|
|
|
|
def _active_sync_links(db: Session) -> list[PlexLink]:
|
|
"""Every link with two-way sync actually running (owner MVP = one row). Iterated uniformly so
|
|
P5b friend links slot in later without touching this loop."""
|
|
return (
|
|
db.execute(
|
|
select(PlexLink).where(
|
|
PlexLink.uses_admin.is_(True),
|
|
PlexLink.sync_enabled.is_(True),
|
|
PlexLink.initial_import_done.is_(True),
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
|
|
|
|
def _resolve_account_id(db: Session, plex: PlexClient, link: PlexLink) -> int:
|
|
"""The owner's Plex accountID, cached on the link. On a Plex Media Server, accountID 1 is reserved
|
|
for the server owner/admin (managed + shared users get id > 1), and an owner link (`uses_admin`)
|
|
always holds the admin token — so 1 is correct here, not an assumption to second-guess. We still
|
|
confirm + grab the owner's username from /accounts, falling back to 1 if that call is unavailable."""
|
|
if link.plex_account_id:
|
|
return link.plex_account_id
|
|
acct_id, username = 1, None
|
|
try:
|
|
for a in plex.accounts():
|
|
if int(a.get("id") or -1) == 1:
|
|
username = a.get("name")
|
|
break
|
|
except PlexError as exc:
|
|
# The fallback account id 1 is right for the admin-token setup, so sync keeps working —
|
|
# but the username is never stored, and the UI then has no name to show for the link,
|
|
# exactly as if the server had not offered one.
|
|
log.info(
|
|
"the Plex account list could not be read; the link keeps the default account",
|
|
extra=logs.event(
|
|
"plex.watch.accounts_probe_failed",
|
|
user_id=link.user_id,
|
|
reason=type(exc).__name__,
|
|
),
|
|
)
|
|
link.plex_account_id = acct_id
|
|
if username:
|
|
link.plex_username = username
|
|
db.commit()
|
|
return acct_id
|
|
|
|
|
|
def _same_state(st: PlexState | None, status: str, pos: int) -> bool:
|
|
"""Whether a row already matches a target (status, pos) — the core ping-pong guard: when nothing
|
|
would change we skip the write entirely (no row churn, no timestamp bump, no re-pull)."""
|
|
if st is None:
|
|
return False
|
|
if status == "watched":
|
|
return st.status == "watched"
|
|
return st.status != "watched" and abs((st.position_seconds or 0) - pos) <= _PROGRESS_MIN_S
|
|
|
|
|
|
def _pull_apply(
|
|
db: Session, user_id: int, item_id: int, status: str, pos: int, plex_ts: datetime | None
|
|
) -> str | None:
|
|
"""Bring one Plex-side positive signal into Siftlode under last-write-wins. Returns
|
|
"watched"/"in_progress" if it changed the row, else None. Never clears a row (un-watch is the
|
|
full reconcile's job) — this only pulls in watched/resume."""
|
|
st = (
|
|
db.execute(
|
|
select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id)
|
|
)
|
|
.scalars()
|
|
.first()
|
|
)
|
|
if _same_state(st, status, pos):
|
|
if st is not None and not st.synced_to_plex:
|
|
st.synced_to_plex = True # already matches Plex → just settle the flag, no churn
|
|
return None
|
|
# Plex-authoritative (S2): on a genuine divergence for an item Plex knows about, Plex wins — no
|
|
# last-write-wins tug-of-war. This is safe because prompt write-back (playback.py pushes the resume
|
|
# position to Plex on the ~10s checkpoint, not just on the final flush) keeps Plex current, so a
|
|
# local edit that's "newer" than Plex has almost always already been pushed and matched above by
|
|
# `_same_state`; a row that reaches here really is out of step with the source of truth. The
|
|
# `_same_state` + `synced_to_plex` short-circuit above still absorbs the freshly-pushed case (Plex
|
|
# echoes the same shape a beat later) so a just-pushed row never bounces.
|
|
if st is not None and not st.synced_to_plex:
|
|
# Durability guard: this row holds a local change that never reached Plex — its immediate push
|
|
# failed, so it's still dirty. Do NOT let the Plex-authoritative pull overwrite it and clear the
|
|
# flag: that would strand the local value forever, since `_repush_dirty` only retries dirty rows.
|
|
# Leave it dirty; `_repush_dirty` (later in this same run) pushes the local value up to Plex. This
|
|
# closes the "Plex flapped mid-session" lost-update — the one case the Plex-wins comment's
|
|
# "almost always already pushed" assumption above excludes.
|
|
return None
|
|
if st is None:
|
|
st = PlexState(user_id=user_id, item_id=item_id)
|
|
db.add(st)
|
|
if status == "watched":
|
|
st.status, st.position_seconds, st.watched_at, st.progress_updated_at = (
|
|
"watched",
|
|
0,
|
|
plex_ts,
|
|
plex_ts,
|
|
)
|
|
else:
|
|
st.status, st.position_seconds, st.progress_updated_at = "new", pos, plex_ts
|
|
st.synced_to_plex = True
|
|
return "watched" if status == "watched" else "in_progress"
|
|
|
|
|
|
def _repush_dirty(db: Session, plex: PlexClient, link: PlexLink) -> int:
|
|
"""Belt-and-suspenders: push local states that never reached Plex (synced_to_plex=False) — e.g.
|
|
an immediate Phase B push that failed while Plex was down. `hidden` is excluded (Siftlode-only,
|
|
never goes to Plex). Returns the count re-pushed."""
|
|
# id → (rating_key, duration_s): duration comes straight from the mirrored row, so a resume
|
|
# re-push doesn't need an extra per-item plex.metadata() round-trip just to fill set_timeline's
|
|
# duration argument.
|
|
info_by_id: dict[int, tuple[str, int | None]] = {
|
|
iid: (rk, dur)
|
|
for iid, rk, dur in db.execute(
|
|
select(PlexItem.id, PlexItem.rating_key, PlexItem.duration_s)
|
|
).all()
|
|
}
|
|
dirty = (
|
|
db.execute(
|
|
select(PlexState).where(
|
|
PlexState.user_id == link.user_id,
|
|
PlexState.synced_to_plex.is_(False),
|
|
PlexState.status != "hidden",
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
n = 0
|
|
for st in dirty:
|
|
info = info_by_id.get(st.item_id)
|
|
if info is None:
|
|
continue
|
|
rk, dur_s = info
|
|
try:
|
|
if st.status == "watched":
|
|
plex.scrobble(rk)
|
|
elif st.position_seconds and st.position_seconds >= _PROGRESS_MIN_S:
|
|
plex.set_timeline(rk, st.position_seconds * 1000, int(dur_s or 0) * 1000)
|
|
else:
|
|
plex.unscrobble(rk)
|
|
st.synced_to_plex = True
|
|
n += 1
|
|
except PlexError as e:
|
|
log.warning(
|
|
"Plex dirty re-push failed (user %s, item %s): %s", link.user_id, st.item_id, e
|
|
)
|
|
return n
|
|
|
|
|
|
# How many stale in-progress movies we probe per incremental run (one metadata() call each). The set
|
|
# is normally tiny (movies started but no longer on Plex's Continue-Watching); the daily full reconcile
|
|
# is the backstop for anything beyond the cap. Most-recently-touched are probed first.
|
|
_MOVIE_PROBE_CAP = 100
|
|
|
|
|
|
def _reconcile_inprogress_movies(
|
|
db: Session, plex: PlexClient, link: PlexLink, on_deck_ids: set[int]
|
|
) -> dict[str, int]:
|
|
"""Close the movie "stuck in progress" gap that the history + onDeck feeds structurally miss.
|
|
|
|
When a movie is finished on Plex (often by scrubbing through the end credits) Plex sets
|
|
``viewCount>0``, clears ``viewOffset`` and DROPS the movie from Continue-Watching. So onDeck no
|
|
longer lists it (and onDeck can only ever ADD an in-progress row, never resolve one to watched —
|
|
see `_pull_apply`), and a scrub-to-end frequently emits no `/status/sessions/history` session row,
|
|
so the history feed misses it too. The stale local ``status='new', position>0`` row then survives
|
|
until the *daily* full reconcile. Movies are hit hardest: a movie is a single leaf with no
|
|
episode-rollup to mask the stuck row, and once watched it leaves onDeck entirely.
|
|
|
|
Fix: for each local in-progress MOVIE that Plex's onDeck no longer lists, read the item's current
|
|
per-(token-)account ``viewCount``/``viewOffset`` with one cheap `metadata()` call and apply it via
|
|
the shared `_pull_apply` — flipping to watched when Plex says so. Returns ``{"watched", "in_progress",
|
|
"probed"}`` counts. Episodes are intentionally left to onDeck/history (they promote the next episode
|
|
and aren't affected). Note: like the rest of this job, `metadata()` reads the TOKEN account (the
|
|
owner) — a HTPC playing under a managed Plex user (accountID>1) is out of scope here as elsewhere."""
|
|
conds = [
|
|
PlexState.user_id == link.user_id,
|
|
PlexState.status.notin_(("watched", "hidden")),
|
|
PlexState.position_seconds > _PROGRESS_MIN_S,
|
|
PlexItem.kind == "movie",
|
|
]
|
|
if on_deck_ids:
|
|
conds.append(PlexState.item_id.notin_(on_deck_ids))
|
|
rows = db.execute(
|
|
select(PlexState, PlexItem.rating_key)
|
|
.join(PlexItem, PlexItem.id == PlexState.item_id)
|
|
.where(*conds)
|
|
.order_by(PlexState.progress_updated_at.desc().nullslast())
|
|
.limit(_MOVIE_PROBE_CAP + 1)
|
|
).all()
|
|
if len(rows) > _MOVIE_PROBE_CAP:
|
|
rows = rows[:_MOVIE_PROBE_CAP]
|
|
log.warning(
|
|
"in-progress movie probe hit the %d cap (user %s); reconcile catches the tail",
|
|
_MOVIE_PROBE_CAP,
|
|
link.user_id,
|
|
)
|
|
out = {"watched": 0, "in_progress": 0, "probed": 0}
|
|
for st, rk in rows:
|
|
try:
|
|
meta = plex.metadata(rk)
|
|
except PlexError as e:
|
|
log.warning(
|
|
"movie watch-probe failed (user %s, item %s): %s", link.user_id, st.item_id, e
|
|
)
|
|
continue
|
|
out["probed"] += 1
|
|
if meta is None:
|
|
continue
|
|
res = _plex_watch_to_state(
|
|
meta.get("viewCount"), meta.get("viewOffset"), meta.get("lastViewedAt")
|
|
)
|
|
if res is None:
|
|
# Plex holds no watch signal now (viewCount 0, no offset) — don't clear it here; an un-watch
|
|
# is the daily reconcile's job, and a Siftlode-only in-progress row must be preserved.
|
|
continue
|
|
status, pos, watched_at, prog_at = res
|
|
ch = _pull_apply(db, link.user_id, st.item_id, status, pos, watched_at or prog_at)
|
|
if ch:
|
|
out[ch] += 1
|
|
return out
|
|
|
|
|
|
def run_plex_watch_sync(db: Session) -> dict:
|
|
"""Scheduler job (default 15 min) — the cheap incremental pass. Pulls Plex-side changes since the
|
|
high-water mark (watch history + on-deck, filtered to the owner account) into Siftlode, resolving a
|
|
divergence to Plex (the source of truth — see `_pull_apply`), reconciles stale in-progress movies
|
|
Plex has since finished (see `_reconcile_inprogress_movies`), then re-pushes any still-unsynced
|
|
local states. Does NOT propagate un-watches (the history feed only carries positive views) — that's
|
|
`run_plex_watch_reconcile`'s job."""
|
|
if not sysconfig.get_bool(db, "plex_enabled"):
|
|
return {"skipped": "disabled"}
|
|
total = {"links": 0, "watched": 0, "in_progress": 0, "repushed": 0, "movie_probes": 0}
|
|
for link in _active_sync_links(db):
|
|
total["links"] += 1
|
|
try:
|
|
with PlexClient(db) as plex:
|
|
acct = _resolve_account_id(db, plex, link)
|
|
since = int(link.last_watch_sync_at.timestamp()) if link.last_watch_sync_at else 0
|
|
now = datetime.now(timezone.utc)
|
|
rk_to_id = _rk_to_id(db)
|
|
|
|
for row in plex.watch_history(min_viewed_at=since, account_id=acct):
|
|
iid = rk_to_id.get(str(row.get("ratingKey") or ""))
|
|
if iid is None:
|
|
continue
|
|
ch = _pull_apply(
|
|
db, link.user_id, iid, "watched", 0, _epoch(row.get("viewedAt"))
|
|
)
|
|
if ch:
|
|
total[ch] += 1
|
|
|
|
# onDeck resolves resume-position changes AND tells us which items Plex still considers
|
|
# in-progress (needed by the movie reconcile below to know which local rows are stale).
|
|
on_deck_ids: set[int] = set()
|
|
for row in plex.on_deck():
|
|
iid = rk_to_id.get(str(row.get("ratingKey") or ""))
|
|
if iid is None:
|
|
continue
|
|
on_deck_ids.add(iid)
|
|
off = int(row.get("viewOffset") or 0) // 1000
|
|
if off < _PROGRESS_MIN_S:
|
|
continue
|
|
ch = _pull_apply(
|
|
db, link.user_id, iid, "new", off, _epoch(row.get("lastViewedAt"))
|
|
)
|
|
if ch:
|
|
total[ch] += 1
|
|
|
|
mv = _reconcile_inprogress_movies(db, plex, link, on_deck_ids)
|
|
total["watched"] += mv["watched"]
|
|
total["in_progress"] += mv["in_progress"]
|
|
total["movie_probes"] += mv["probed"]
|
|
|
|
total["repushed"] += _repush_dirty(db, plex, link)
|
|
link.last_watch_sync_at = now
|
|
db.commit()
|
|
# silent-ok: the `{"skipped": …}` return is the report, as above.
|
|
except PlexNotConfigured as e:
|
|
return {"skipped": str(e)}
|
|
except PlexError as e:
|
|
db.rollback()
|
|
log.warning("Plex watch sync failed (user %s): %s", link.user_id, e)
|
|
log.info("Plex watch sync: %s", total)
|
|
return total
|
|
|
|
|
|
def _clear_unwatched(db: Session, user_id: int, plex_states: dict, settled: set[str]) -> int:
|
|
"""Clear the user's rows Plex no longer has a record for — the mirrored ones (Plex un-watched
|
|
them); hidden and dirty rows stay (a dirty one is re-pushed by `_repush_dirty`). Only rows of the
|
|
sections the scan SETTLED: a walk a Plex hiccup cut short leaves out rows it merely failed to
|
|
reach, and clearing those un-watched them here until the next complete run (E32 S2); an item of
|
|
a library the allowlist leaves out is absent from every scan, and was cleared on every run (E32
|
|
review)."""
|
|
n = 0
|
|
rows = db.execute(
|
|
select(PlexState)
|
|
.join(PlexItem, PlexItem.id == PlexState.item_id)
|
|
.join(PlexLibrary, PlexLibrary.id == PlexItem.library_id)
|
|
.where(PlexState.user_id == user_id, PlexLibrary.plex_key.in_(settled))
|
|
)
|
|
for st in rows.scalars().all():
|
|
if st.item_id in plex_states or st.status == "hidden" or not st.synced_to_plex:
|
|
continue
|
|
db.delete(st)
|
|
n += 1
|
|
return n
|
|
|
|
|
|
def run_plex_watch_reconcile(db: Session) -> dict:
|
|
"""Scheduler job (default daily) — the authoritative full reconcile. Rescans every section and,
|
|
per item, uses `synced_to_plex` to settle what the incremental feed can't:
|
|
- Plex has a state, Siftlode differs → Plex wins (the source of truth — see `_pull_apply`).
|
|
- Siftlode has a state, Plex has NO record:
|
|
· synced_to_plex=True → Plex un-watched it since our last sync → clear it (un-watch).
|
|
· synced_to_plex=False → a local change that never reached Plex → re-push it (via
|
|
`_repush_dirty`), never dropped.
|
|
- `hidden` rows are Siftlode-only and never cleared.
|
|
Union-preserving: a never-synced Siftlode-only state is pushed up, not lost."""
|
|
if not sysconfig.get_bool(db, "plex_enabled"):
|
|
return {"skipped": "disabled"}
|
|
total = {"links": 0, "applied": 0, "cleared": 0, "repushed": 0, "scanned": 0}
|
|
for link in _active_sync_links(db):
|
|
total["links"] += 1
|
|
try:
|
|
with PlexClient(db) as plex:
|
|
_resolve_account_id(db, plex, link)
|
|
plex_states, scanned, settled, walked = _scan_plex_states(db, plex)
|
|
total["scanned"] += scanned
|
|
|
|
for item_id, (status, pos, watched_at, prog_at) in plex_states.items():
|
|
ch = _pull_apply(db, link.user_id, item_id, status, pos, watched_at or prog_at)
|
|
if ch:
|
|
total["applied"] += 1
|
|
|
|
total["cleared"] += _clear_unwatched(db, link.user_id, plex_states, settled)
|
|
if unsettled := sorted(walked - settled):
|
|
total["incomplete"] = total.get("incomplete", 0) + 1
|
|
log.warning(
|
|
"Plex sections %s were cut short or answered empty; their rows are not cleared"
|
|
" this run",
|
|
", ".join(unsettled),
|
|
extra=logs.event(
|
|
"plex.watch.reconcile_incomplete",
|
|
user_id=link.user_id,
|
|
sections=unsettled,
|
|
),
|
|
)
|
|
|
|
total["repushed"] += _repush_dirty(db, plex, link)
|
|
link.last_watch_sync_at = datetime.now(timezone.utc)
|
|
db.commit()
|
|
# silent-ok: the `{"skipped": …}` return is the report, as above.
|
|
except PlexNotConfigured as e:
|
|
return {"skipped": str(e)}
|
|
except PlexError as e:
|
|
db.rollback()
|
|
log.warning("Plex watch reconcile failed (user %s): %s", link.user_id, e)
|
|
log.info("Plex watch reconcile: %s", total)
|
|
return total
|