Files
siftlode/backend/app/plex/sync.py
T
peter df4a79a1f1 fix(sync): an empty answer the remote's own count contradicts is not believed, one level down
E32 review round 1 (B1, B2, B3, B8).

- Plex watch reconcile: a section the server SAYS is empty (totalSize 0, nothing listed) is a
  complete listing. It read as "cut short" forever, so one empty library stopped un-watches from
  ever syncing. A listing with no total at all stays incomplete (the client now returns None for
  it instead of 0), and a whole scan that listed nothing still clears nothing. Tests now cover
  completeness across two sections, both orders.
- Plex collections: no members from a collection whose listing counts some (childCount) keeps
  the membership we hold, in the full sync and in the targeted resync after a write.
- YouTube playlists: no items for a playlist whose listing counts some (itemCount) keeps the
  mirror in the bulk pull, and is refused as retryable in "reset to YouTube" and in the push
  plan/merge, where it read as "YouTube deleted them all". The rule sits beside vanished().
- A playlist deleted on YouTube answered "reconnect your account"; it now says it is gone
  (errors.youtube.gone, HU+EN).
- The pure 3-way merge moved to sync/playlist_merge.py; playlists.py is back under 800 lines and
  off the size list.
2026-10-04 02:13:53 +02:00

776 lines
36 KiB
Python

"""Mirror a locally-reachable Plex library into Siftlode's own catalog (plex_* tables).
Playback is from the local physical file, so this only mirrors METADATA + the physical-media facts
(codecs/container/path) needed to browse, search, and later play. It reuses Plex's own codec info
(no ffprobe needed) to classify each item's browser playability. Movies + episodes become playable
LEAF rows (plex_items); shows/seasons are mirrored for the drill-down + episode ordering.
The sync is a full reconcile of the enabled sections (upsert by rating_key) AND prunes rows for
titles that disappeared from Plex: each section diffs the rating_keys it just listed against the ones
stored, and deletes the stale ones (their watch-state / playlist links cascade). The prune is guarded
on a NON-empty listing — a section that returns zero items is left untouched, so a transient Plex
glitch that yields an empty page can't wipe a library (the metadata is cheap to re-sync; an accidental
mass-delete is the only real risk). Library-level reconcile (dropping a whole removed section, the
dead `enabled` column) is a separate follow-up. Cast + intro/credit markers are fetched lazily on the
item-detail/player path (P2), not here.
"""
from datetime import datetime, timezone
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app import logs, sysconfig
from app.models import PlexCollection, PlexItem, PlexLibrary, PlexSeason, PlexShow
from app.plex import detail, overlay, people
from app.plex.client import PlexClient, PlexError, PlexNotConfigured
from app.plex.genres import hu_genres
from app.plex.lookups import by_rating_key
log = logs.get_logger(__name__)
# Codecs/containers a browser can play in a native <video> without any transcoding.
_BROWSER_VIDEO = {"h264"}
_BROWSER_AUDIO = {"aac", "mp3", "opus", "vorbis"}
_BROWSER_CONTAINER = {"mp4", "mov", "m4v"}
# Plex section item types.
_TYPE_MOVIE = 1
_TYPE_SHOW = 2
_TYPE_SEASON = 3
_TYPE_EPISODE = 4
_PAGE = 200
def classify_playable(vcodec: str | None, acodec: str | None, container: str | None) -> str:
"""How the browser can play this file: 'direct' (native <video>), 'remux' (copy video, fix
container / transcode audio — cheap), or 'transcode' (re-encode video — expensive)."""
v = (vcodec or "").lower()
a = (acodec or "").lower()
c = (container or "").lower()
if not v or v not in _BROWSER_VIDEO:
return "transcode"
if a and a not in _BROWSER_AUDIO:
return "remux"
if c and c not in _BROWSER_CONTAINER:
return "remux"
return "direct"
# Video codecs the prod iGPU (UHD 630 / Quick Sync Gen 9.5) can hardware-DECODE — the precondition for
# the full-GPU VAAPI transcode pipeline. THE single source of this set: plex.stream imports it here so
# the runtime encoder choice (`_want_hw`) and this catalog facet can never disagree by codec. A source
# outside it (AV1, MPEG-4 ASP/XviD…) is software-only even on the iGPU (it can't decode it to feed its
# own encoder).
HW_DECODABLE_CODECS = frozenset({"h264", "hevc", "h265", "vp9", "vc1", "mpeg2video"})
# Color-transfer values marking an HDR source (PQ / HLG). A 10-bit/HDR source breaks the VAAPI
# scale=nv12 path on Gen 9.5 (see plex.stream) → software transcode, so it counts as NOT HW-capable.
_HDR_TRANSFERS = frozenset({"smpte2084", "arib-std-b67"})
def transcode_hw_capable(
playable: str | None,
vcodec: str | None,
bit_depth: int | None = None,
hdr: bool = False,
) -> bool | None:
"""The stored `PlexItem.transcode_hw` facet: for a `playable="transcode"` item, whether the prod
iGPU can HARDWARE-transcode it (else software libx264 only); None for a non-transcode item. HW needs
a HW-decodable codec AND an 8-bit SDR source — a 10-bit/HDR source (bit_depth>8 or an HDR transfer)
breaks the VAAPI nv12 pipeline and falls back to software. `bit_depth`/`hdr` are unknown at sync
time (the cheap listing carries no Stream), so they default to 8-bit SDR (codec-only) and the
bit-depth backfill refines a 10-bit/HDR HEVC to False later."""
if playable != "transcode":
return None
if (vcodec or "").lower() not in HW_DECODABLE_CODECS:
return False
if bit_depth is not None and bit_depth > 8:
return False
return not hdr
def _video_stream_props(meta: dict) -> tuple[int | None, bool]:
"""(bit_depth, is_hdr) from a FULL Plex metadata dict's video Stream (Media→Part→Stream, streamType
1). Only the rich metadata call carries Stream; the cheap listing does not — so this is read by the
bit-depth backfill, not the listing sync. (None, False) when absent/unparseable."""
for m in meta.get("Media") or []:
for part in m.get("Part") or []:
for s in part.get("Stream") or []:
if s.get("streamType") != 1: # 1 = video
continue
bd = s.get("bitDepth")
try:
depth = int(bd) if bd else None
# silent-ok: unknown bit depth, which is also what most listings report. The
# playback class then falls back to the codec alone, and per-stream noise over a
# whole library would drown the sync's own lines.
except (TypeError, ValueError):
depth = None
hdr = str(s.get("colorTrc") or "").lower() in _HDR_TRANSFERS
return depth, hdr
return None, False
def _epoch(v) -> datetime | None:
try:
return datetime.fromtimestamp(int(v), tz=timezone.utc) if v else None
# silent-ok: a field Plex leaves out of most listing rows. None is a legitimate stored value,
# and every caller treats it as "not known" rather than as an error.
except (TypeError, ValueError, OSError):
return None
def _date(v):
"""Parse a Plex 'YYYY-MM-DD' string (originallyAvailableAt) → date, or None."""
try:
return datetime.strptime(v, "%Y-%m-%d").date() if v else None
# silent-ok: as `_epoch` above — an absent or unparsable air date is stored as no air date.
except (TypeError, ValueError):
return None
def _tags(meta: dict, key: str, limit: int | None = None) -> list | None:
"""The .tag values of a Plex child-tag array (Genre/Director/Role) present in the listing."""
vals = [t.get("tag") for t in (meta.get(key) or []) if t.get("tag")]
if limit:
vals = vals[:limit]
return vals or None
def _rating(meta: dict) -> float | None:
try:
return float(meta.get("audienceRating") or meta.get("rating"))
# silent-ok: TypeError on `float(None)` is the ordinary path — most items carry no rating.
except (TypeError, ValueError):
return None
# Provider preference for the canonical guid (a stable id shared across a title's language twins).
_GUID_PREFERENCE = ("tmdb", "imdb", "tvdb")
def _canonical_guid(meta: dict) -> str | None:
"""The canonical external id used to pair an EN title with its HU-overlay twin (E5 S5). Reads the
`Guid` array (present only with includeGuids=1 in the listing), preferring tmdb > imdb > tvdb,
normalised to '<provider>:<id>' (e.g. 'tmdb:12345'). None when no known provider id is present."""
found: dict[str, str] = {}
for g in meta.get("Guid") or []:
raw = str(g.get("id") or "")
for prov in _GUID_PREFERENCE:
if raw.startswith(f"{prov}://"):
found.setdefault(prov, f"{prov}:{raw.split('://', 1)[1]}")
for prov in _GUID_PREFERENCE:
if prov in found:
return found[prov]
return None
def _media_facts(meta: dict) -> dict:
media = meta.get("Media") or []
if not media:
return {}
m = media[0]
part = (m.get("Part") or [{}])[0]
return {
"container": m.get("container") or part.get("container"),
"codec_video": m.get("videoCodec"),
"codec_audio": m.get("audioCodec"),
"size_bytes": part.get("size"),
"file_path": part.get("file"),
"part_key": part.get("key"),
}
def _dur_s(meta: dict) -> int | None:
d = meta.get("duration")
return int(d / 1000) if d else None
def _enabled_section_keys(db: Session) -> set[str] | None:
"""The configured allowlist of section keys, or None to mean 'all movie/show sections'."""
raw = sysconfig.get_str(db, "plex_libraries")
keys = {k.strip() for k in raw.split(",") if k.strip()}
return keys or None
def sync(db: Session) -> dict:
"""Full reconcile of the enabled movie/show sections. Returns a small stats dict."""
if not sysconfig.get_bool(db, "plex_enabled"):
return {"skipped": "disabled"}
stats = {
"libraries": 0,
"movies": 0,
"shows": 0,
"seasons": 0,
"episodes": 0,
"collections": 0,
"overlaid": 0,
"pruned": 0,
}
wanted = _enabled_section_keys(db)
overlay_map = overlay.overlay_map(db) # {huKey: enKey}
overlay_keys = set(overlay_map)
try:
with PlexClient(db) as plex:
sections = {
str(s.get("key")): s for s in plex.sections() if s.get("type") in ("movie", "show")
}
libs_by_key: dict[str, PlexLibrary] = {}
# Pass 1 — browsable (non-overlay) sections. HU-overlay sections are handled in pass 2 and
# are NEVER mirrored as browsable libraries of their own.
for key, s in sections.items():
if key in overlay_keys:
continue
if wanted is not None and key not in wanted:
continue
lib = _upsert_library(db, key, s.get("title") or key, s.get("type"))
libs_by_key[key] = lib
stats["libraries"] += 1
if s.get("type") == "movie":
_sync_movies(db, plex, lib, stats)
else:
_sync_shows(db, plex, lib, stats)
_sync_collections(db, plex, lib, stats)
# Commit each section as it completes: a PlexError on a LATER section then keeps the
# libraries already mirrored (idempotent — the next run reconciles the rest), instead
# of the old single end-of-loop commit that a late failure rolled back entirely.
db.commit()
# Pass 2 — HU overlays. The EN twins are now synced (with guids), so write each overlay
# section's Hungarian metadata onto the matched EN rows. Requires the EN section to have been
# synced in pass 1 (in the allowlist / all) and to share the overlay's kind.
for hu_key, en_key in overlay_map.items():
s = sections.get(hu_key)
en_lib = libs_by_key.get(en_key)
if s is None or en_lib is None or s.get("type") != en_lib.kind:
log.warning(
"Plex overlay skipped: hu=%s en=%s (missing section/EN lib or kind mismatch)",
hu_key,
en_key,
)
continue
lib = _upsert_library(db, hu_key, s.get("title") or hu_key, s.get("type"))
lib.metadata_lang = "hu"
lib.overlay_of = en_lib.id
db.flush()
overlay.sync_overlay(db, plex, lib, en_lib.id, stats)
db.commit()
# Rich-fetch a bounded batch of not-yet-enriched titles' FULL cast, so the whole library
# becomes filterable over successive runs (the cheap listing above only stored the top ~3).
try:
stats["people_enriched"] = _enrich_people_batch(db, plex)
except Exception: # best-effort; never fail the sync over a backfill
# With the traceback: a blind catch over a batch that fetches AND writes can be
# a Plex shape change, a DB error or our own bug, and `%s` told them apart badly.
log.exception("the Plex people-enrich batch failed")
# Probe a bounded batch of HW-decodable transcode items for bit-depth/HDR, so the
# playback-class filter's HW/SW split is precise (a 10-bit/HDR HEVC → software) over
# successive runs. Only HW-decodable-codec transcodes are probed — the rest are SW by codec.
try:
stats["media_probed"] = _probe_media_batch(db, plex)
except Exception: # best-effort; never fail the sync over a backfill
log.exception("the Plex media-probe batch failed")
# silent-ok: the return value IS the report. `{"skipped": …}` goes to the scheduler, which
# logs the job result, and to the admin route, which answers it to the caller.
except PlexNotConfigured as e:
return {"skipped": str(e)}
except PlexError as e:
db.rollback()
log.warning("Plex sync failed: %s", e)
return {"error": str(e)}
# Prune the headshot cache: a reconcile may have removed titles, orphaning people who no longer
# appear in ANY item's cast/directors — drop those rows, mirroring how Plex drops orphaned metadata.
try:
stats["people_pruned"] = people.prune_orphan_people(db)
except Exception: # best-effort housekeeping; never fail the sync over it
log.exception("the Plex people prune failed")
log.info("Plex sync done: %s", stats)
return stats
# How many titles to rich-fetch per sync run for the cast backfill — bounds the extra Plex load; the
# whole library is covered over successive runs (each fetched item is stamped people_enriched_at once).
_ENRICH_BATCH = 60
# Same bounded-per-run pattern for the bit-depth probe. Only HW-decodable-codec transcode items are
# probed (typically a small minority — the bulk of transcodes are old codecs that are SW by codec), so
# a modest batch covers the work-list in a run or two; a stuck item retries next run (watermark unset).
_PROBE_BATCH = 60
def _probe_media_batch(db: Session, plex: PlexClient, limit: int = _PROBE_BATCH) -> int:
"""Rich-fetch up to `limit` not-yet-probed HW-decodable transcode items, read the video Stream's
bit-depth + HDR, and store `video_bit_depth` + the refined `transcode_hw` (a 10-bit/HDR HEVC →
False/software). Stamps `media_probed_at` on every attempt (success OR a Plex miss with no Stream),
so an item is probed once and not re-fetched every sync; a transient fetch FAILURE leaves the
watermark unset to retry. Oldest-id first. Returns how many items were probed this run."""
rows = (
db.execute(
select(PlexItem)
.where(
PlexItem.playable == "transcode",
func.lower(PlexItem.codec_video).in_(HW_DECODABLE_CODECS),
PlexItem.media_probed_at.is_(None),
)
.order_by(PlexItem.id)
.limit(limit)
)
.scalars()
.all()
)
now = datetime.now(timezone.utc)
n = 0
for row in rows:
try:
meta = plex.metadata(row.rating_key) or {}
# No traceback and only debug: this is per item over a whole library, and the stack is
# the same `plex.metadata` call every time. The retry next run is the handling.
except Exception as e: # noqa: BLE001 — transient; leave media_probed_at NULL to retry
log.debug("Plex media probe fetch failed for %s: %s", row.rating_key, e)
continue
bit_depth, hdr = _video_stream_props(meta)
row.video_bit_depth = bit_depth
row.transcode_hw = transcode_hw_capable("transcode", row.codec_video, bit_depth, hdr)
row.media_probed_at = (
now # probed (even if Plex carried no Stream — don't re-fetch forever)
)
n += 1
db.commit()
return n
def _enrich_people_batch(db: Session, plex: PlexClient, limit: int = _ENRICH_BATCH) -> int:
"""Rich-fetch the FULL cast of up to `limit` not-yet-enriched movies/shows and persist it (via the
shared enrich_row_people). Oldest-id first, movies before shows. Stamps people_enriched_at only on
a SUCCESSFUL fetch, so a transient per-item failure retries next run rather than being lost.
Returns how many items were enriched this run."""
movies = (
db.execute(
select(PlexItem)
.where(PlexItem.kind == "movie", PlexItem.people_enriched_at.is_(None))
.order_by(PlexItem.id)
.limit(limit)
)
.scalars()
.all()
)
remaining = limit - len(movies)
shows = (
db.execute(
select(PlexShow)
.where(PlexShow.people_enriched_at.is_(None))
.order_by(PlexShow.id)
.limit(remaining)
)
.scalars()
.all()
if remaining > 0
else []
)
now = datetime.now(timezone.utc)
n = 0
for row in [*movies, *shows]:
try:
meta = plex.metadata(row.rating_key) or {}
# As in `_probe_media_batch` above: per item, so the message without the stack.
except Exception as e: # noqa: BLE001 — transient; leave people_enriched_at NULL
log.debug("Plex people enrich fetch failed for %s: %s", row.rating_key, e)
continue # leave people_enriched_at NULL so it's retried on a later run
people.enrich_row_people(db, row, people.cast_from_meta(meta))
detail.apply_rich_detail(
row, meta
) # imdb/tagline/full_cast → served from the row, not live
row.people_enriched_at = now
n += 1
db.commit()
return n
def _upsert_library(db: Session, key: str, title: str, kind: str) -> PlexLibrary:
lib = db.execute(select(PlexLibrary).where(PlexLibrary.plex_key == key)).scalars().first()
if lib is None:
lib = PlexLibrary(plex_key=key)
db.add(lib)
lib.title = title
lib.kind = kind
lib.synced_at = datetime.now(timezone.utc)
db.flush()
return lib
def _paginate(plex: PlexClient, key: str, item_type: int, progress: dict | None = None):
"""Yield a section's items page by page. If `progress` is given, set `progress["complete"]` True
ONLY when the walk reached the server's reported `total` — a truncated fetch (an empty/short page
mid-sequence, e.g. a transient Plex hiccup, before `total` is reached) leaves it False so the
caller can skip pruning. Pruning on a partial listing would delete rows the listing merely failed
to mention (the metadata is re-syncable, but that still cascades watch-state)."""
start = 0
total = None
while True:
items, total = plex.section_items(key, item_type, start, _PAGE)
if not items:
break
for it in items:
yield it
start += len(items)
if total and start >= total:
break
if len(items) < _PAGE:
break
if progress is not None:
# A truthy `total` we reached = a trustworthy complete listing. `total` unknown/zero-with-rows
# (a Plex that doesn't report totalSize) is treated as NOT complete, so prune stays conservative.
# `total` itself is kept for a caller to whom a server that SAID 0 is news it can trust (the
# watch reconcile); an empty mirror or overlay must not read it so (E32 S2).
progress["complete"] = bool(total) and start >= total
progress["total"] = total
# The section listing exposes at most the top ~20 cast (Plex's own cap); a stored cast longer than this
# can only have come from the rich enrich, never the listing.
_LISTING_CAST_CAP = 20
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")
# Every row has Hungarian genres, not only one with an HU overlay twin (the overlay pass, which runs
# after this, puts the twin's own list over it): the HU genre facets and filter read this column,
# and an English word there could never match a filter read in Hungarian.
row.genres_hu = hu_genres(row.genres)
row.directors = _tags(meta, "Director")
# The listing carries only the top ~20 cast, so re-applying it would DOWNGRADE an already-enriched
# full cast (up to 24, grown by the rich backfill/info-page enrich), silently breaking the
# person-page cast filter for names past the top ~20. The watermark gate below (`people_enriched_at
# is None`) normally protects this, but it is NOT enough on its own: a migration that resets the
# watermark re-opens the regression (0065 nulled it, which is exactly how the S5b fix got undone in
# prod). So guard STRUCTURALLY too, by LENGTH against the listing cap: a stored cast LONGER than the
# cap can only be an enriched fuller list, so keep it (robust to any watermark reset). A stored cast
# at or below the cap is itself listing-sourced, so take the fresh listing — which tracks Plex cast
# EDITS in EITHER direction (comparing against this listing's length instead would silently keep a
# stale longer cast when Plex TRIMS one).
if row.people_enriched_at is None:
listing_cast = _tags(meta, "Role", limit=_LISTING_CAST_CAP)
if len(row.cast_names or []) <= _LISTING_CAST_CAP:
row.cast_names = listing_cast
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
# EN<->HU pairing key (movies; harmless/unused on episodes). Keep the prior guid if this listing
# carries none (a server ignoring includeGuids mustn't wipe the overlay's only match key).
row.guid = _canonical_guid(meta) or row.guid
row.title = (meta.get("title") or "").strip() or "Untitled"
row.summary = meta.get("summary")
row.year = meta.get("year")
row.duration_s = _dur_s(meta)
row.thumb_key = meta.get("thumb")
row.art_key = meta.get("art") or meta.get("grandparentArt")
row.added_at = _epoch(meta.get("addedAt"))
_apply_facet_fields(row, meta) # filterable/orderable extras (shared with shows)
mf = _media_facts(meta)
new_vcodec = mf.get("codec_video")
new_part_key = mf.get("part_key")
# The physical file was replaced when its video codec changes OR — for a SINGLE-version item — when its
# Plex Part key changes (a re-encode / quality upgrade gets a new part id). Either way any probed
# bit-depth is stale, so reset the probe watermark and re-derive transcode_hw from codec alone until the
# backfill re-probes. Codec alone is NOT enough: a same-codec bit-depth re-encode (8-bit→10-bit HEVC)
# keeps the codec `hevc` but flips the HW class, and only the part-key move catches it.
#
# The part-key check is gated on single-version because `_media_facts` reads media[0]/part[0], and for a
# MULTI-version item (an original + a Plex "Optimized" copy) that order isn't guaranteed stable across
# listings — an unguarded part-key compare would then reset+re-probe every sync and oscillate the
# transcode_hw bucket between the two versions. A multi-version same-codec re-encode not self-healing is
# an acceptable rare edge; a per-sync re-probe loop is not.
single_version = len(meta.get("Media") or []) <= 1
codec_changed = (row.codec_video or "").lower() != (new_vcodec or "").lower()
if codec_changed or (single_version and row.part_key != new_part_key):
row.video_bit_depth = None
row.media_probed_at = None
row.file_path = mf.get("file_path")
row.part_key = new_part_key
row.container = mf.get("container")
row.codec_video = new_vcodec
row.codec_audio = mf.get("codec_audio")
row.size_bytes = mf.get("size_bytes")
row.playable = classify_playable(
mf.get("codec_video"), mf.get("codec_audio"), mf.get("container")
)
# transcode_hw: keep the bit-depth backfill's authoritative value (it folds in bit-depth + HDR) once
# probed; otherwise (re)derive the provisional codec-only class. A non-transcode item is always None,
# even if it had been probed as a transcode before (an audio/container flip to remux without a codec
# change wouldn't reset the watermark, so pin it here).
if row.playable != "transcode":
row.transcode_hw = None
elif row.media_probed_at is None:
row.transcode_hw = transcode_hw_capable("transcode", row.codec_video, row.video_bit_depth)
def _prune_stale(
db: Session, existing: dict, seen: set[str], stats: dict, *, complete: bool
) -> None:
"""Delete stored rows whose rating_key is no longer in the freshly-listed `seen` set — the titles
that disappeared from Plex (their watch-state / playlist links cascade via the FKs). GUARDED on a
COMPLETE, non-empty listing: a truncated fetch (`complete` False — a mid-sequence empty/short page
before the server's total) or a zero-item listing prunes NOTHING, so a transient Plex hiccup can
never wipe a library over rows it merely failed to list. Callers pass the rating_key -> row map
they already built for the upsert."""
if not complete or not seen:
return
for rk, row in existing.items():
if rk not in seen:
db.delete(row)
stats["pruned"] = stats.get("pruned", 0) + 1
db.flush()
def _sync_movies(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> None:
existing = {
r.rating_key: r
for r in db.execute(
select(PlexItem).where(PlexItem.library_id == lib.id, PlexItem.kind == "movie")
).scalars()
}
seen: set[str] = set()
progress: dict = {}
for meta in _paginate(plex, lib.plex_key, _TYPE_MOVIE, progress):
rk = str(meta.get("ratingKey") or "")
if not rk:
continue
row = existing.get(rk) or PlexItem(rating_key=rk)
if row.id is None:
db.add(row)
_apply_item(row, lib.id, "movie", meta)
seen.add(rk)
stats["movies"] += 1
db.flush()
_prune_stale(db, existing, seen, stats, complete=progress.get("complete", False))
def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> None:
# 1) shows
shows = {
r.rating_key: r
for r in db.execute(select(PlexShow).where(PlexShow.library_id == lib.id)).scalars()
}
shows_seen: set[str] = set()
shows_progress: dict = {}
for meta in _paginate(plex, lib.plex_key, _TYPE_SHOW, shows_progress):
rk = str(meta.get("ratingKey") or "")
if not rk:
continue
shows_seen.add(rk)
sh = shows.get(rk) or PlexShow(rating_key=rk)
if sh.id is None:
db.add(sh)
shows[rk] = sh
sh.library_id = lib.id
sh.guid = (
_canonical_guid(meta) or sh.guid
) # EN<->HU pairing key; keep prior if listing has none
sh.title = (meta.get("title") or "").strip() or "Untitled"
sh.summary = meta.get("summary")
sh.year = meta.get("year")
sh.thumb_key = meta.get("thumb")
sh.art_key = meta.get("art")
sh.child_count = meta.get("childCount")
sh.added_at = _epoch(meta.get("addedAt"))
_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()}
# 2) seasons
seasons = {
r.rating_key: r
for r in db.execute(
select(PlexSeason).join(PlexShow).where(PlexShow.library_id == lib.id)
).scalars()
}
seasons_seen: set[str] = set()
seasons_progress: dict = {}
for meta in _paginate(plex, lib.plex_key, _TYPE_SEASON, seasons_progress):
rk = str(meta.get("ratingKey") or "")
sid = show_id.get(str(meta.get("parentRatingKey") or ""))
if not rk or sid is None:
continue
seasons_seen.add(rk)
se = seasons.get(rk) or PlexSeason(rating_key=rk)
if se.id is None:
db.add(se)
seasons[rk] = se
se.show_id = sid
se.title = meta.get("title")
se.season_number = meta.get("index")
se.thumb_key = meta.get("thumb")
stats["seasons"] += 1
db.flush()
season_id = {rk: se.id for rk, se in seasons.items()}
# 3) episodes (playable leaves)
episodes = {
r.rating_key: r
for r in db.execute(
select(PlexItem).where(PlexItem.library_id == lib.id, PlexItem.kind == "episode")
).scalars()
}
episodes_seen: set[str] = set()
episodes_progress: dict = {}
for meta in _paginate(plex, lib.plex_key, _TYPE_EPISODE, episodes_progress):
rk = str(meta.get("ratingKey") or "")
if not rk:
continue
episodes_seen.add(rk)
row = episodes.get(rk) or PlexItem(rating_key=rk)
if row.id is None:
db.add(row)
_apply_item(row, lib.id, "episode", meta)
row.show_id = show_id.get(str(meta.get("grandparentRatingKey") or ""))
row.season_id = season_id.get(str(meta.get("parentRatingKey") or ""))
row.season_number = meta.get("parentIndex")
row.episode_number = meta.get("index")
stats["episodes"] += 1
db.flush()
# Prune leaf-first (episodes -> seasons -> shows) so a stale row is deleted directly, before its
# parent's cascade would reach it — avoids deleting the same ORM object twice. Each level is
# independently guarded on a non-empty listing by _prune_stale.
_prune_stale(
db, episodes, episodes_seen, stats, complete=episodes_progress.get("complete", False)
)
_prune_stale(db, seasons, seasons_seen, stats, complete=seasons_progress.get("complete", False))
_prune_stale(db, shows, shows_seen, stats, complete=shows_progress.get("complete", False))
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.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, by_rating_key(db, PlexCollection, rk), rk, lib.id, meta)
Model = PlexItem if lib.kind == "movie" else PlexShow
children = _believed_children(plex, rk, meta)
if children is None:
db.flush()
return col
new_members = {str(c.get("ratingKey")) for c in children if c.get("ratingKey")}
current = db.execute(select(Model).where(Model.collection_keys.contains([rk]))).scalars().all()
for row in current: # drop rk from rows that are no longer members
if row.rating_key not in new_members:
row.collection_keys = [k for k in (row.collection_keys or []) if k != rk] or None
to_add = new_members - {r.rating_key for r in current}
if to_add:
for row in db.execute(select(Model).where(Model.rating_key.in_(to_add))).scalars():
row.collection_keys = list(dict.fromkeys([*(row.collection_keys or []), rk]))
db.flush()
return col
def _believed_children(plex: PlexClient, rk: str, meta: dict) -> list[dict] | None:
"""A collection's members, or None when the answer is not believed: no members from a collection
whose own listing counts some (`childCount`) is a glitch, not a collection emptied — the same
empty-answer rule as the collection listing's (E32 review). Rebuilding on it showed the
collection empty until the next sync."""
children = plex.collection_children(rk)
count = str(meta.get("childCount") or "0")
if children or not count.isdigit() or int(count) == 0:
return children
log.warning(
"Plex listed no members for a non-empty collection; keeping its membership",
extra=logs.event("plex.sync.collection_children_empty", collection=rk),
)
return None
def delete_collection_local(db: Session, lib: PlexLibrary, rk: str) -> None:
"""Local cleanup after a collection is deleted from Plex: strip rk from members + drop the row."""
Model = PlexItem if lib.kind == "movie" else PlexShow
for row in db.execute(select(Model).where(Model.collection_keys.contains([rk]))).scalars():
row.collection_keys = [k for k in (row.collection_keys or []) if k != rk] or None
col = by_rating_key(db, PlexCollection, rk)
if col is not None:
db.delete(col)
db.flush()
def _rebuild_membership(
db: Session, lib: PlexLibrary, membership: dict[str, list[str]], kept: set[str]
) -> None:
"""Rebuild collection_keys on every member row of this library (clear + set), carrying over the
keys of a collection whose empty answer was not believed (`_believed_children`)."""
for Model in (PlexItem, PlexShow):
for row in db.execute(select(Model).where(Model.library_id == lib.id)).scalars():
held = [k for k in (row.collection_keys or []) if k in kept]
keys = [*membership.get(row.rating_key, []), *held]
row.collection_keys = list(dict.fromkeys(keys)) or None
db.flush()
def _sync_collections(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> None:
"""Mirror a library's collections + membership. Reads never touch Plex afterwards. Membership is
written as `collection_keys` on member movies (plex_items) / shows (plex_shows). A full reconcile:
collections gone from Plex are pruned, and each member row's keys are rebuilt from scratch."""
cols = plex.collections(lib.plex_key)
existing = {
c.rating_key: c
for c in db.execute(
select(PlexCollection).where(PlexCollection.library_id == lib.id)
).scalars()
}
if not cols and existing:
# The catalog's own rule (see the module doc): an empty listing while we hold rows is a Plex
# glitch, not a library emptied of collections. Reconciling on it dropped every collection
# (its user-set `editable` flag with it) and every member's keys (E32 S2).
log.warning(
"Plex listed no collections for a library that has some; keeping them this run",
extra=logs.event("plex.sync.collections_empty", library=lib.plex_key),
)
return
seen: set[str] = set()
kept: set[str] = set() # collections whose empty children answer is not believed
membership: dict[str, list[str]] = {} # member rating_key -> [collection rating_key, ...]
for meta in cols:
rk = str(meta.get("ratingKey") or "")
if not rk:
continue
_upsert_collection(db, existing.get(rk), rk, lib.id, meta)
seen.add(rk)
children = _believed_children(plex, rk, meta)
if children is None:
kept.add(rk)
for child in children or []:
crk = str(child.get("ratingKey") or "")
if crk:
membership.setdefault(crk, []).append(rk)
stats["collections"] += 1
db.flush()
for rk, col in existing.items():
if rk not in seen:
db.delete(col)
_rebuild_membership(db, lib, membership, kept)