peter
c5f121139b
A season marked watched, a pull from Plex and a YouTube resume point made their first row by a read then an add, and failed on the unique key when a first checkpoint or the watched mark made it in between. They go through the same insert-then-read as the checkpoint (the resume point is one upsert); a pull that meets a local change still waiting for its push leaves it. Folding duplicate Watch later lists moved their videos but not a Home pin, which the cascade then deleted; the first pin of a dropped copy moves onto the kept list. Marking one notice read trimmed before its own flag was flushed, so the trim left one past the cap. The route test now crosses the cap and reads the result from a session of its own.
683 lines
33 KiB
Python
683 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. Plex is now 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.userscope import owned_by
|
|
from app.plex.models import PlexItem, PlexLibrary, PlexLink, PlexState
|
|
from app.plex.client import PlexClient, PlexError, PlexNotConfigured
|
|
from app.plex import queries
|
|
from app.plex.service import state_row
|
|
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."""
|
|
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 — 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(owned_by(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 = queries.state(db, user_id, item_id)
|
|
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(owned_by(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 = queries.state(db, user_id, item_id)
|
|
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: 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 = state_row(db, user_id, item_id, synced=True)
|
|
if not st.synced_to_plex: # a local change made it first, beside this pull: the guard above
|
|
return None
|
|
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(
|
|
owned_by(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 = [
|
|
owned_by(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; an item of
|
|
a library the allowlist leaves out is absent from every scan, and was cleared on every run."""
|
|
n = 0
|
|
rows = db.execute(
|
|
select(PlexState)
|
|
.join(PlexItem, PlexItem.id == PlexState.item_id)
|
|
.join(PlexLibrary, PlexLibrary.id == PlexItem.library_id)
|
|
.where(owned_by(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
|