From 28061353eca15f3e637fe3020d95cc52d187183a Mon Sep 17 00:00:00 2001 From: npeter83 Date: Sat, 11 Jul 2026 23:48:35 +0200 Subject: [PATCH] Plex backend pass (#9): watch_sync bugs + backend dedup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the Plex #9 backend backlog (frontend was the prior follow-up). Behavior- neutral cleanup + two two-way-sync bug fixes; PW2 verified a non-bug. fix(plex): watch-sync PW1 pagination + PW3 lost-update guard - PW1 watch_history: the history is a GLOBAL cross-account feed read as a single 500-row page, then filtered to the owner — on a busy family server the owner's recent views can sit past page 1 and be missed. Now paginate (desc) until the since-cutoff, bounded by max_pages (daily reconcile is the backstop; logs if hit). - PW3 push_state_to_plex: it flagged synced_to_plex=True unconditionally after a push, so a local edit landing between the push and the flag-set was marked clean and never pushed (lost update). Capture the row's freshest timestamp when the push is scheduled (_push_watch) and only settle the flag if the row is unchanged. - PW2 was flagged as "accountID hardcoded to 1" but is NOT a bug: on a PMS account 1 is reserved for the owner/admin and an owner link always holds the admin token — added a clarifying comment so it isn't re-flagged. chore(plex): backend dedup - PC2 unified_library + facets shared a ~13-field filter Query signature + p-dict → one LibraryFilters dataclass injected via Depends(); as_dict() feeds the builders. - PS-C1 sync.py collection upsert dup (resync_collection ≈ _sync_collections) → _upsert_collection(). - PS-C2 sync.py 8-field filterable-metadata block dup (_apply_item ≈ _sync_shows) → _apply_facet_fields(). - PW-C1 _repush_dirty read duration from the mirrored PlexItem.duration_s instead of a per-row plex.metadata() round-trip. - PW-C2 rk→id map built inline in two places → _rk_to_id() helper. --- backend/app/plex/client.py | 53 +++++++++++++------- backend/app/plex/sync.py | 74 ++++++++++++++-------------- backend/app/plex/watch_sync.py | 62 +++++++++++++++++------ backend/app/routes/plex.py | 90 +++++++++++++++++----------------- 4 files changed, 164 insertions(+), 115 deletions(-) diff --git a/backend/app/plex/client.py b/backend/app/plex/client.py index 54fbb28..e0341a3 100644 --- a/backend/app/plex/client.py +++ b/backend/app/plex/client.py @@ -217,27 +217,42 @@ class PlexClient: return self._get("/accounts").get("Account", []) or [] def watch_history( - self, min_viewed_at: int = 0, account_id: int | None = None, size: int = 500 + self, min_viewed_at: int = 0, account_id: int | None = None, size: int = 500, max_pages: int = 40 ) -> list[dict]: - """The most recent watch-history rows (ratingKey / viewedAt / accountID / type), newest - first — works without Plex Pass. Filtered client-side to `account_id` and to rows at/after - `min_viewed_at` (epoch seconds): the efficient "what changed since T" feed. Sorted desc, so - we stop at the first row older than the cutoff.""" - mc = self._get( - "/status/sessions/history/all", - params={ - "sort": "viewedAt:desc", - "X-Plex-Container-Start": 0, - "X-Plex-Container-Size": size, - }, - ) - out = [] - for r in mc.get("Metadata", []) or []: - if int(r.get("viewedAt") or 0) < min_viewed_at: + """The watch-history rows (ratingKey / viewedAt / accountID / type) newer than `min_viewed_at` + (epoch seconds), newest first — works without Plex Pass. The efficient "what changed since T" + feed, filtered client-side to `account_id`. + + The history is a GLOBAL feed across all server accounts, so paginate (sorted desc) until we + reach a row at/older than the cutoff: on a busy family server the owner's recent views can sit + past the first page behind other accounts' rows. `max_pages` bounds the scan (the daily full + reconcile is the backstop for anything beyond it).""" + out: list[dict] = [] + for page in range(max_pages): + mc = self._get( + "/status/sessions/history/all", + params={ + "sort": "viewedAt:desc", + "X-Plex-Container-Start": page * size, + "X-Plex-Container-Size": size, + }, + ) + rows = mc.get("Metadata", []) or [] + reached_cutoff = False + for r in rows: + if int(r.get("viewedAt") or 0) < min_viewed_at: + reached_cutoff = True + break + if account_id is not None and int(r.get("accountID") or -1) != account_id: + continue + out.append(r) + if reached_cutoff or len(rows) < size: break - if account_id is not None and int(r.get("accountID") or -1) != account_id: - continue - out.append(r) + else: + log.warning( + "watch_history hit the %d-page scan cap (since=%s); reconcile will catch the tail", + max_pages, min_viewed_at, + ) return out def on_deck(self) -> list[dict]: diff --git a/backend/app/plex/sync.py b/backend/app/plex/sync.py index 975369c..cd4201f 100644 --- a/backend/app/plex/sync.py +++ b/backend/app/plex/sync.py @@ -166,6 +166,21 @@ def _paginate(plex: PlexClient, key: str, item_type: int): break +def _apply_facet_fields(row, meta: dict) -> None: + """Set the filterable/orderable metadata columns shared by plex_items + plex_shows (identical + column names) from the cheap section listing: rating, content_rating, studio, release date, + genres/directors/cast, and the searchable people blob folded into search_vector at sync time.""" + row.rating = _rating(meta) + row.content_rating = (meta.get("contentRating") or None) + row.studio = (meta.get("studio") or None) and str(meta.get("studio"))[:255] + row.originally_available_at = _date(meta.get("originallyAvailableAt")) + row.genres = _tags(meta, "Genre") + row.directors = _tags(meta, "Director") + row.cast_names = _tags(meta, "Role", limit=20) + people = list(dict.fromkeys((row.directors or []) + (row.cast_names or []))) + row.people_text = " ".join(people) or None + + def _apply_item(row: PlexItem, lib_id: int, kind: str, meta: dict) -> None: row.library_id = lib_id row.kind = kind @@ -176,17 +191,7 @@ def _apply_item(row: PlexItem, lib_id: int, kind: str, meta: dict) -> None: row.thumb_key = meta.get("thumb") row.art_key = meta.get("art") or meta.get("grandparentArt") row.added_at = _epoch(meta.get("addedAt")) - # Filterable / orderable extras — all present in the cheap section listing. - row.rating = _rating(meta) - row.content_rating = (meta.get("contentRating") or None) - row.studio = (meta.get("studio") or None) and str(meta.get("studio"))[:255] - row.originally_available_at = _date(meta.get("originallyAvailableAt")) - row.genres = _tags(meta, "Genre") - row.directors = _tags(meta, "Director") - row.cast_names = _tags(meta, "Role", limit=20) - # Searchable people blob (cast + directors, de-duped) → folded into search_vector at sync time. - people = list(dict.fromkeys((row.directors or []) + (row.cast_names or []))) - row.people_text = " ".join(people) or None + _apply_facet_fields(row, meta) # filterable/orderable extras (shared with shows) mf = _media_facts(meta) row.file_path = mf.get("file_path") row.part_key = mf.get("part_key") @@ -230,16 +235,7 @@ def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> sh.art_key = meta.get("art") sh.child_count = meta.get("childCount") sh.added_at = _epoch(meta.get("addedAt")) - # Filterable / orderable metadata — same cheap section-listing tags as movies. - sh.rating = _rating(meta) - sh.content_rating = (meta.get("contentRating") or None) - sh.studio = (meta.get("studio") or None) and str(meta.get("studio"))[:255] - sh.originally_available_at = _date(meta.get("originallyAvailableAt")) - sh.genres = _tags(meta, "Genre") - sh.directors = _tags(meta, "Director") - sh.cast_names = _tags(meta, "Role", limit=20) - people = list(dict.fromkeys((sh.directors or []) + (sh.cast_names or []))) - sh.people_text = " ".join(people) or None + _apply_facet_fields(sh, meta) # same filterable/orderable tags as movies stats["shows"] += 1 db.flush() show_id = {rk: sh.id for rk, sh in shows.items()} @@ -281,21 +277,32 @@ def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> db.flush() -def resync_collection(db: Session, plex: PlexClient, lib: PlexLibrary, rk: str) -> PlexCollection: - """Targeted re-sync of ONE collection after a write-back (create/add/remove/rename) — avoids the - full ~4-min library sync. Upserts the row (preserving `editable`) and reconciles ONLY this - collection's membership on the affected member rows.""" - meta = plex.metadata(rk) or {} - col = db.query(PlexCollection).filter_by(rating_key=rk).first() or PlexCollection(rating_key=rk) +def _upsert_collection( + db: Session, existing: PlexCollection | None, rk: str, lib_id: int, meta: dict +) -> PlexCollection: + """Get-or-create a PlexCollection row and apply its metadata from a Plex `meta` dict. `editable` + is user-managed and preserved (never set here). Shared by the full sync + the targeted re-sync.""" + col = existing or PlexCollection(rating_key=rk) if col.id is None: db.add(col) - col.library_id = lib.id + col.library_id = lib_id col.title = (meta.get("title") or "").strip() or "Untitled" col.summary = meta.get("summary") col.thumb_key = meta.get("thumb") col.child_count = meta.get("childCount") col.smart = str(meta.get("smart")) == "1" col.synced_at = datetime.now(timezone.utc) + return col + + +def resync_collection(db: Session, plex: PlexClient, lib: PlexLibrary, rk: str) -> PlexCollection: + """Targeted re-sync of ONE collection after a write-back (create/add/remove/rename) — avoids the + full ~4-min library sync. Upserts the row (preserving `editable`) and reconciles ONLY this + collection's membership on the affected member rows.""" + meta = plex.metadata(rk) or {} + col = _upsert_collection( + db, db.query(PlexCollection).filter_by(rating_key=rk).first(), rk, lib.id, meta + ) Model = PlexItem if lib.kind == "movie" else PlexShow new_members = {str(c.get("ratingKey")) for c in plex.collection_children(rk) if c.get("ratingKey")} current = db.query(Model).filter(Model.collection_keys.contains([rk])).all() @@ -333,16 +340,7 @@ def _sync_collections(db: Session, plex: PlexClient, lib: PlexLibrary, stats: di rk = str(meta.get("ratingKey") or "") if not rk: continue - col = existing.get(rk) or PlexCollection(rating_key=rk) - if col.id is None: - db.add(col) - col.library_id = lib.id - col.title = (meta.get("title") or "").strip() or "Untitled" - col.summary = meta.get("summary") - col.thumb_key = meta.get("thumb") - col.child_count = meta.get("childCount") - col.smart = str(meta.get("smart")) == "1" # `editable` is preserved (user-managed, not here) - col.synced_at = datetime.now(timezone.utc) + _upsert_collection(db, existing.get(rk), rk, lib.id, meta) seen.add(rk) for child in plex.collection_children(rk): crk = str(child.get("ratingKey") or "") diff --git a/backend/app/plex/watch_sync.py b/backend/app/plex/watch_sync.py index 9799972..6e1c3d7 100644 --- a/backend/app/plex/watch_sync.py +++ b/backend/app/plex/watch_sync.py @@ -59,15 +59,18 @@ def _plex_watch_to_state( 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.query(PlexItem.rating_key, PlexItem.id)} + + def _scan_plex_states( db: Session, plex: PlexClient ) -> tuple[dict[int, tuple[str, int, datetime | None, datetime | None]], int]: """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, plus the scanned-leaf count. Shared by the one-time import and the full reconcile (both need Plex's whole current picture).""" - item_id_by_rk: dict[str, int] = { - rk: iid for rk, iid in db.query(PlexItem.rating_key, PlexItem.id) - } + item_id_by_rk = _rk_to_id(db) wanted = _enabled_section_keys(db) scanned = 0 out: dict[int, tuple[str, int, datetime | None, datetime | None]] = {} @@ -165,6 +168,15 @@ def link_for_push(db: Session, user_id: int) -> PlexLink | None: 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, @@ -172,6 +184,7 @@ def push_state_to_plex( 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 @@ -179,7 +192,10 @@ def push_state_to_plex( 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).""" + `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 @@ -199,13 +215,24 @@ def push_state_to_plex( ) 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). + # 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.query(PlexState).filter_by(user_id=user_id, item_id=item_id).first() - if st is not None: + 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 @@ -271,8 +298,10 @@ def _active_sync_links(db: Session) -> list[PlexLink]: def _resolve_account_id(db: Session, plex: PlexClient, link: PlexLink) -> int: - """The owner's Plex accountID, cached on the link. The server owner is accountID 1 on their own - server; we confirm + grab their username from /accounts, falling back to 1 if unavailable.""" + """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 @@ -335,7 +364,12 @@ 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.""" - rk_by_id = dict(db.query(PlexItem.id, PlexItem.rating_key)) + # 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.query(PlexItem.id, PlexItem.rating_key, PlexItem.duration_s) + } dirty = ( db.query(PlexState) .filter( @@ -347,15 +381,15 @@ def _repush_dirty(db: Session, plex: PlexClient, link: PlexLink) -> int: ) n = 0 for st in dirty: - rk = rk_by_id.get(st.item_id) - if rk is None: + 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: - dur = int((plex.metadata(rk) or {}).get("duration") or 0) - plex.set_timeline(rk, st.position_seconds * 1000, dur) + plex.set_timeline(rk, st.position_seconds * 1000, int(dur_s or 0) * 1000) else: plex.unscrobble(rk) st.synced_to_plex = True @@ -382,7 +416,7 @@ def run_plex_watch_sync(db: Session) -> dict: 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: iid for rk, iid in db.query(PlexItem.rating_key, PlexItem.id)} + 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 "")) diff --git a/backend/app/routes/plex.py b/backend/app/routes/plex.py index 3b8b408..9c14e95 100644 --- a/backend/app/routes/plex.py +++ b/backend/app/routes/plex.py @@ -14,6 +14,7 @@ import tempfile import threading import time from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path from urllib.parse import quote @@ -386,6 +387,37 @@ def _apply_meta_filters(q, model, p: dict): return q.filter(*conds) if conds else q +@dataclass +class LibraryFilters: + """The shared sidebar filter query-params for /library + /facets, declared once and injected via + ``Depends()``. `as_dict()` yields the `p` dict the query builders read; `duration_*` is included + for /facets' wsq but ignored by `_meta_filter_conds` (both callers apply duration separately).""" + + genres: str | None = None + genre_mode: str = "any" + content_ratings: str | None = None + year_min: int | None = None + year_max: int | None = None + rating_min: float | None = None + duration_min: int | None = None + duration_max: int | None = None + added_within: str | None = None + directors: str | None = None + actors: str | None = None + studios: str | None = None + collection: str | None = None + + def as_dict(self) -> dict: + return { + "genres": self.genres, "genre_mode": self.genre_mode, + "content_ratings": self.content_ratings, "year_min": self.year_min, + "year_max": self.year_max, "rating_min": self.rating_min, + "duration_min": self.duration_min, "duration_max": self.duration_max, + "added_within": self.added_within, "directors": self.directors, + "actors": self.actors, "studios": self.studios, "collection": self.collection, + } + + @router.get("/library") def unified_library( scope: str = "both", @@ -395,19 +427,7 @@ def unified_library( show: str = "all", offset: int = 0, limit: int = Query(default=40, ge=1, le=100), - genres: str | None = None, - genre_mode: str = "any", - content_ratings: str | None = None, - year_min: int | None = None, - year_max: int | None = None, - rating_min: float | None = None, - duration_min: int | None = None, - duration_max: int | None = None, - added_within: str | None = None, - directors: str | None = None, - actors: str | None = None, - studios: str | None = None, - collection: str | None = None, + f: LibraryFilters = Depends(), user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: @@ -416,12 +436,7 @@ def unified_library( its episodes). On a search that also matches episodes (and shows are in scope), the matching episodes come back in a separate `episodes` list (grouped result — the 'Episodes' section).""" offset = max(0, offset) - p = { - "genres": genres, "genre_mode": genre_mode, "content_ratings": content_ratings, - "year_min": year_min, "year_max": year_max, "rating_min": rating_min, - "added_within": added_within, "directors": directors, "actors": actors, - "studios": studios, "collection": collection, - } + p = f.as_dict() libs = db.query(PlexLibrary).filter_by(enabled=True).all() movie_lib_ids = [lb.id for lb in libs if lb.kind == "movie"] show_lib_ids = [lb.id for lb in libs if lb.kind == "show"] @@ -467,10 +482,10 @@ def unified_library( else: mq = mq.filter(or_(st.status.is_(None), st.status != "hidden")) mq = _apply_meta_filters(mq, PlexItem, p) - if duration_min is not None: - mq = mq.filter(PlexItem.duration_s >= duration_min) - if duration_max is not None: - mq = mq.filter(PlexItem.duration_s <= duration_max) + if f.duration_min is not None: + mq = mq.filter(PlexItem.duration_s >= f.duration_min) + if f.duration_max is not None: + mq = mq.filter(PlexItem.duration_s <= f.duration_max) if tsq is not None: mq = mq.filter(PlexItem.search_vector.op("@@")(tsq)) selects.append(mq) @@ -576,19 +591,7 @@ def unified_library( def facets( scope: str = "both", show: str = "all", - genres: str | None = None, - genre_mode: str = "any", - content_ratings: str | None = None, - year_min: int | None = None, - year_max: int | None = None, - rating_min: float | None = None, - duration_min: int | None = None, - duration_max: int | None = None, - added_within: str | None = None, - directors: str | None = None, - actors: str | None = None, - studios: str | None = None, - collection: str | None = None, + f: LibraryFilters = Depends(), user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: @@ -611,12 +614,7 @@ def facets( show_ids = [lb.id for lb in libs if lb.kind == "show"] if scope in ("show", "both") else [] if not movie_ids and not show_ids: return empty - p = { - "genres": genres, "genre_mode": genre_mode, "content_ratings": content_ratings, - "year_min": year_min, "year_max": year_max, "rating_min": rating_min, - "duration_min": duration_min, "duration_max": duration_max, "added_within": added_within, - "directors": directors, "actors": actors, "studios": studios, "collection": collection, - } + p = f.as_dict() uid = user.id @@ -714,7 +712,7 @@ def facets( # be absent): the sidebar renders genres solely from this list and hides the whole genre section — # chips AND the Any/All toggle — when it's empty, which would trap the user with no per-chip way to # undo a zero-result selection. Count 0 is fine (genre chips don't display counts). - for g in _csv(genres): + for g in _csv(f.genres): genre_counts.setdefault(g, 0) return { "genres": [{"value": g, "count": c} for g, c in sorted(genre_counts.items(), key=lambda kv: kv[0])], @@ -1828,8 +1826,12 @@ def _push_watch( 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 + # Capture the row's freshest timestamp NOW so the background push can tell if a newer local edit + # lands before it settles synced_to_plex (which would otherwise strand that newer value). + st = db.query(PlexState).filter_by(user_id=user_id, item_id=item_id).first() background.add_task( - plex_watch.push_state_to_plex, user_id, item_id, rating_key, action, time_ms, duration_ms + plex_watch.push_state_to_plex, user_id, item_id, rating_key, action, time_ms, duration_ms, + plex_watch._state_marker(st), )