perf(plex): batch bulk pushes, cache show enrichment; remove dead code + DRY (F6/F8/F10)
F6: coalesce whole-show/season Plex watch-pushes into ONE background task (push_bulk_state_to_plex) that reuses a single DB session + keep-alive Plex client, instead of scheduling one task (own session + HTTP client) per episode. F8: cache a show's live Plex enrichment (metadata + related) per rating_key for a short TTL and fetch the two calls in parallel; repeat opens skip the network. Raw payloads are cached; per-user shaping stays out. An empty related list is not cached (plex.related swallows a transient failure as [] — don't pin it for the TTL). F10: remove dead code — the pre-unified /browse route + browse(), the /people route + people() + _person_photo(), api.plexPeople, interface PlexPerson, and the orphaned plex.people i18n block. DRY: appendPlexFilters() shared by plexLibrary + plexFacets; one exported plexDetailUi.Filterable (was Fil + Filterable); PlexInfo migrated onto the shared plexDetailUi hooks/menu (DetailCustomizeMenu gained overlay + extra props; useDetailPrefs exposes savePref; PrefToggle exported). Reviewed (high) → clean. F7 (facet aggregate collapse) deferred: self-exclusion gives each sub-aggregate a distinct WHERE, and no measurement shows /facets slow.
This commit is contained in:
npeter83
parent
9f8e410033
commit
b3af04a997
9 changed files with 180 additions and 450 deletions
|
|
@ -206,6 +206,51 @@ def push_state_to_plex(
|
|||
db.commit()
|
||||
|
||||
|
||||
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.query(PlexState).filter(
|
||||
PlexState.user_id == user_id, PlexState.item_id.in_(synced_ids)
|
||||
).update({PlexState.synced_to_plex: True}, synchronize_session=False)
|
||||
db.commit()
|
||||
|
||||
|
||||
# --- Phase C: incremental + full Plex ↔ Siftlode reconcile (scheduler jobs) ------------------------
|
||||
|
||||
# Clock-skew / round-trip slack when comparing a Plex timestamp to a Siftlode one. A state we just
|
||||
|
|
|
|||
|
|
@ -11,6 +11,9 @@ import os
|
|||
import re
|
||||
import subprocess
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from urllib.parse import quote
|
||||
|
|
@ -18,7 +21,7 @@ from urllib.parse import quote
|
|||
import httpx
|
||||
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, Request, Response
|
||||
from fastapi.responses import FileResponse
|
||||
from sqlalchemy import Integer, String, and_, case, cast, func, literal, null, or_, text
|
||||
from sqlalchemy import Integer, String, and_, case, cast, func, literal, null, or_
|
||||
from sqlalchemy.orm import Session, aliased
|
||||
|
||||
from app import sysconfig
|
||||
|
|
@ -389,171 +392,6 @@ def _apply_meta_filters(q, model, p: dict):
|
|||
return q.filter(*conds) if conds else q
|
||||
|
||||
|
||||
@router.get("/browse")
|
||||
def browse(
|
||||
library: str,
|
||||
q: str | None = None,
|
||||
sort: str = "added",
|
||||
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,
|
||||
sort_dir: str = "desc",
|
||||
user: User = Depends(current_user),
|
||||
db: Session = Depends(get_db),
|
||||
) -> dict:
|
||||
"""List a library's movies (leaves) or shows, with optional FTS search + sorting + a per-user
|
||||
watch-state filter (`show`: all|unwatched|in_progress|watched — movie libraries only). Movie
|
||||
libraries return playable movie cards; show libraries return show cards (drill in via /show)."""
|
||||
lib = db.query(PlexLibrary).filter_by(plex_key=str(library)).first()
|
||||
if lib is None:
|
||||
raise HTTPException(status_code=404, detail="Unknown Plex library")
|
||||
offset = max(0, offset)
|
||||
model = PlexItem if lib.kind == "movie" else PlexShow
|
||||
|
||||
query = db.query(model)
|
||||
if lib.kind == "movie":
|
||||
query = query.filter(PlexItem.library_id == lib.id, PlexItem.kind == "movie")
|
||||
# Per-user watch-state filter (movies only; a show isn't a single watch unit).
|
||||
st = aliased(PlexState)
|
||||
query = query.outerjoin(st, and_(st.item_id == PlexItem.id, st.user_id == user.id))
|
||||
if show == "watched":
|
||||
query = query.filter(st.status == "watched")
|
||||
elif show == "in_progress":
|
||||
query = query.filter(st.position_seconds > 0, or_(st.status.is_(None), st.status != "watched"))
|
||||
elif show == "unwatched":
|
||||
query = query.filter(or_(st.status.is_(None), st.status.notin_(["watched", "hidden"])))
|
||||
else: # all — hide only the explicitly hidden
|
||||
query = query.filter(or_(st.status.is_(None), st.status != "hidden"))
|
||||
# --- Metadata filters (movie libraries; @> containment is GIN-indexed) ---
|
||||
gsel = _csv(genres)
|
||||
if gsel:
|
||||
conds = [PlexItem.genres.contains([g]) for g in gsel]
|
||||
query = query.filter(and_(*conds) if genre_mode == "all" else or_(*conds))
|
||||
crs = _csv(content_ratings)
|
||||
if crs:
|
||||
query = query.filter(PlexItem.content_rating.in_(crs))
|
||||
if year_min is not None:
|
||||
query = query.filter(PlexItem.year >= year_min)
|
||||
if year_max is not None:
|
||||
query = query.filter(PlexItem.year <= year_max)
|
||||
if rating_min is not None:
|
||||
query = query.filter(PlexItem.rating >= rating_min)
|
||||
if duration_min is not None:
|
||||
query = query.filter(PlexItem.duration_s >= duration_min)
|
||||
if duration_max is not None:
|
||||
query = query.filter(PlexItem.duration_s <= duration_max)
|
||||
cutoff = _added_cutoff(added_within)
|
||||
if cutoff is not None:
|
||||
query = query.filter(PlexItem.added_at >= cutoff)
|
||||
for d in _csv(directors): # AND — titles this whole set directed
|
||||
query = query.filter(PlexItem.directors.contains([d]))
|
||||
for a in _csv(actors): # AND — titles featuring all these people
|
||||
query = query.filter(PlexItem.cast_names.contains([a]))
|
||||
stds = _csv(studios)
|
||||
if stds: # OR — a title has one studio
|
||||
query = query.filter(PlexItem.studio.in_(stds))
|
||||
else:
|
||||
query = query.filter(PlexShow.library_id == lib.id)
|
||||
# Aggregate per-user watch-state across the show's episodes (a show isn't one row of state).
|
||||
if show in ("watched", "in_progress", "unwatched"):
|
||||
agg = _show_agg_subq(db, user.id).subquery()
|
||||
query = query.outerjoin(agg, agg.c.show_id == PlexShow.id)
|
||||
tot = func.coalesce(agg.c.total, 0)
|
||||
wat = func.coalesce(agg.c.watched, 0)
|
||||
inp = func.coalesce(agg.c.inprog, 0)
|
||||
if show == "watched":
|
||||
query = query.filter(tot > 0, wat >= tot)
|
||||
elif show == "in_progress":
|
||||
query = query.filter(or_(wat > 0, inp > 0), wat < tot)
|
||||
else: # unwatched — no episode started
|
||||
query = query.filter(wat == 0, inp == 0)
|
||||
# --- Metadata filters (same as movies, minus duration which shows don't have) ---
|
||||
gsel = _csv(genres)
|
||||
if gsel:
|
||||
conds = [PlexShow.genres.contains([g]) for g in gsel]
|
||||
query = query.filter(and_(*conds) if genre_mode == "all" else or_(*conds))
|
||||
crs = _csv(content_ratings)
|
||||
if crs:
|
||||
query = query.filter(PlexShow.content_rating.in_(crs))
|
||||
if year_min is not None:
|
||||
query = query.filter(PlexShow.year >= year_min)
|
||||
if year_max is not None:
|
||||
query = query.filter(PlexShow.year <= year_max)
|
||||
if rating_min is not None:
|
||||
query = query.filter(PlexShow.rating >= rating_min)
|
||||
cutoff = _added_cutoff(added_within)
|
||||
if cutoff is not None:
|
||||
query = query.filter(PlexShow.added_at >= cutoff)
|
||||
for d in _csv(directors): # AND
|
||||
query = query.filter(PlexShow.directors.contains([d]))
|
||||
for a in _csv(actors): # AND
|
||||
query = query.filter(PlexShow.cast_names.contains([a]))
|
||||
stds = _csv(studios)
|
||||
if stds: # OR
|
||||
query = query.filter(PlexShow.studio.in_(stds))
|
||||
|
||||
# Collection filter applies to both movies and shows (both carry collection_keys).
|
||||
if collection:
|
||||
query = query.filter(model.collection_keys.contains([collection]))
|
||||
|
||||
ts = _to_tsquery_str(q) if q else None
|
||||
tsq = None
|
||||
if ts:
|
||||
tsq = func.to_tsquery(_TS_CONFIG, ts)
|
||||
query = query.filter(model.search_vector.op("@@")(tsq))
|
||||
|
||||
total = query.count()
|
||||
|
||||
if tsq is not None:
|
||||
order = [func.ts_rank(model.search_vector, tsq).desc()]
|
||||
else:
|
||||
if sort == "title":
|
||||
col = func.lower(model.title)
|
||||
elif sort == "year": # both movies and shows have year
|
||||
col = model.year
|
||||
elif sort == "rating": # both have rating (audienceRating)
|
||||
col = model.rating
|
||||
elif sort == "duration" and model is PlexItem:
|
||||
col = PlexItem.duration_s
|
||||
elif sort == "release": # both have originally_available_at
|
||||
col = model.originally_available_at
|
||||
else: # added
|
||||
col = model.added_at
|
||||
asc = sort_dir == "asc"
|
||||
order = [(col.asc() if asc else col.desc()).nullslast()]
|
||||
order.append(model.id.desc())
|
||||
|
||||
rows = query.order_by(*order).offset(offset).limit(limit).all()
|
||||
|
||||
if lib.kind == "movie":
|
||||
by_item = {r.id: r for r in rows}
|
||||
states = {
|
||||
s.item_id: s
|
||||
for s in db.query(PlexState).filter(
|
||||
PlexState.user_id == user.id, PlexState.item_id.in_(list(by_item.keys()) or [0])
|
||||
)
|
||||
}
|
||||
items = [_movie_card(r, states.get(r.id)) for r in rows]
|
||||
else:
|
||||
statuses = _show_state_map(db, user.id, [r.id for r in rows])
|
||||
items = [_show_card(r, statuses.get(r.id, "new")) for r in rows]
|
||||
|
||||
return {"kind": lib.kind, "total": total, "offset": offset, "limit": limit, "items": items}
|
||||
|
||||
|
||||
@router.get("/library")
|
||||
def unified_library(
|
||||
scope: str = "both",
|
||||
|
|
@ -888,69 +726,6 @@ def _like_escape(s: str) -> str:
|
|||
return s.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
||||
|
||||
|
||||
@router.get("/people")
|
||||
def people(
|
||||
q: str,
|
||||
library: str,
|
||||
user: User = Depends(current_user),
|
||||
db: Session = Depends(get_db),
|
||||
) -> dict:
|
||||
"""Cast/crew whose name matches the search term — drives the virtual person cards shown above the
|
||||
grid. Returns up to a few matches (most films first) with a count + a photo."""
|
||||
term = (q or "").strip()
|
||||
lib = db.query(PlexLibrary).filter_by(plex_key=str(library)).first()
|
||||
if lib is None or lib.kind != "movie" or len(term) < 2:
|
||||
return {"people": []}
|
||||
esc = _like_escape(term)
|
||||
pfx, word = f"{esc}%", f"% {esc}%" # name starts with the term, or any of its words does
|
||||
rows = db.execute(
|
||||
text(
|
||||
"SELECT name, kind, count(*) c FROM ("
|
||||
" SELECT jsonb_array_elements_text(cast_names) AS name, 'actor' AS kind FROM plex_items"
|
||||
" WHERE library_id = :lib AND kind = 'movie'"
|
||||
" UNION ALL"
|
||||
" SELECT jsonb_array_elements_text(directors) AS name, 'director' AS kind FROM plex_items"
|
||||
" WHERE library_id = :lib AND kind = 'movie'"
|
||||
") x WHERE name ILIKE :pfx OR name ILIKE :word "
|
||||
"GROUP BY name, kind ORDER BY c DESC, name LIMIT 5"
|
||||
),
|
||||
{"lib": lib.id, "pfx": pfx, "word": word},
|
||||
).all()
|
||||
return {
|
||||
"people": [
|
||||
{"name": name, "kind": kind, "count": c, "photo": _person_photo(db, lib.id, name, kind)}
|
||||
for name, kind, c in rows
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
def _person_photo(db: Session, lib_id: int, name: str, kind: str) -> str | None:
|
||||
"""A person's headshot from one representative film's live metadata (image bytes are disk-cached)."""
|
||||
col = PlexItem.directors if kind == "director" else PlexItem.cast_names
|
||||
rk = (
|
||||
db.query(PlexItem.rating_key)
|
||||
.filter(PlexItem.library_id == lib_id, col.contains([name]))
|
||||
.limit(1)
|
||||
.scalar()
|
||||
)
|
||||
if not rk:
|
||||
return None
|
||||
try:
|
||||
with PlexClient(db) as plex:
|
||||
meta = plex.metadata(rk) or {}
|
||||
except (PlexError, PlexNotConfigured):
|
||||
return None
|
||||
for r in meta.get("Director" if kind == "director" else "Role") or []:
|
||||
if r.get("tag") == name:
|
||||
thumb = str(r.get("thumb") or "")
|
||||
return (
|
||||
f"/api/plex/person-image?u={quote(thumb, safe='')}"
|
||||
if thumb.startswith(_PERSON_IMG_HOST)
|
||||
else None
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def _collection_editable(c: PlexCollection) -> bool:
|
||||
"""Effective editability: the admin marked it editable AND it's a plain manual collection — never
|
||||
smart, never an external auto-list (IMDb/TMDb/…), which are managed outside Plex."""
|
||||
|
|
@ -1351,6 +1126,42 @@ def reorder_playlist(pid: int, payload: dict, user: User = Depends(current_user)
|
|||
return {"ok": True}
|
||||
|
||||
|
||||
# Live-Plex enrichment for a show (cast/IMDb + related) is user-independent raw payload, so it's cached
|
||||
# per rating_key for a short TTL and shared across users; per-user shaping happens in the caller. Bounds
|
||||
# staleness to the TTL while turning repeat opens of the same show into zero network calls.
|
||||
_SHOW_ENRICH_TTL_S = 300
|
||||
_show_enrich_cache: dict[str, tuple[float, dict, list]] = {}
|
||||
_show_enrich_lock = threading.Lock()
|
||||
|
||||
|
||||
def _show_enrichment(db: Session, rating_key: str) -> tuple[dict, list]:
|
||||
"""(meta, related_raw) for a show's live Plex enrichment — served from the per-show TTL cache when
|
||||
warm, otherwise fetched with the two Plex calls (metadata + related) run IN PARALLEL. Returns
|
||||
({}, []) when Plex is unavailable/unconfigured (not cached, so it retries on the next open)."""
|
||||
now = time.monotonic()
|
||||
with _show_enrich_lock:
|
||||
hit = _show_enrich_cache.get(rating_key)
|
||||
if hit is not None and hit[0] > now:
|
||||
return hit[1], hit[2]
|
||||
try:
|
||||
with PlexClient(db) as plex:
|
||||
with ThreadPoolExecutor(max_workers=2) as ex:
|
||||
meta_f = ex.submit(plex.metadata, rating_key)
|
||||
related_f = ex.submit(plex.related, rating_key) # swallows its own errors → []
|
||||
meta = meta_f.result() or {}
|
||||
related_raw = related_f.result()
|
||||
except (PlexError, PlexNotConfigured):
|
||||
return {}, []
|
||||
# Don't cache an EMPTY related list: plex.related() swallows a transient failure as [] (same as a
|
||||
# genuinely no-related show), so caching it would serve an empty "related" section for the whole TTL
|
||||
# even after Plex recovers. Skip the cache in that case (retry next open, as before); shows that DO
|
||||
# have related items — the payload worth caching — still get cached.
|
||||
if related_raw:
|
||||
with _show_enrich_lock:
|
||||
_show_enrich_cache[rating_key] = (now + _SHOW_ENRICH_TTL_S, meta, related_raw)
|
||||
return meta, related_raw
|
||||
|
||||
|
||||
@router.get("/show/{rating_key}")
|
||||
def show_detail(
|
||||
rating_key: str,
|
||||
|
|
@ -1384,14 +1195,15 @@ def show_detail(
|
|||
all_cards.append(card)
|
||||
show_roll = _rollup(all_cards)
|
||||
|
||||
# Cast + IMDb (best-effort live metadata; same shape as the movie/episode info page).
|
||||
# Cast + IMDb (best-effort live metadata; same shape as the movie/episode info page). The two live
|
||||
# Plex calls are cached per-show + fetched in parallel by _show_enrichment; the per-user shaping
|
||||
# (_related_show_cards adds this user's watch-state) stays out of the shared cache.
|
||||
rich: dict = {}
|
||||
related: list[dict] = []
|
||||
try:
|
||||
with PlexClient(db) as plex:
|
||||
meta = plex.metadata(sh.rating_key) or {}
|
||||
rich = _rich_meta(meta)
|
||||
related = _related_show_cards(db, user.id, plex.related(sh.rating_key), exclude=sh.id)
|
||||
meta, related_raw = _show_enrichment(db, sh.rating_key)
|
||||
rich = _rich_meta(meta)
|
||||
related = _related_show_cards(db, user.id, related_raw, exclude=sh.id)
|
||||
except Exception:
|
||||
# Best-effort enrichment only — the local episode list is already assembled, so any live-Plex
|
||||
# failure (not configured, HTTP/JSON shape, _rich_meta parse error) must degrade, not 500.
|
||||
|
|
@ -2143,9 +1955,10 @@ def _bulk_state(
|
|||
if prev != "new":
|
||||
to_push.append((e.id, e.rating_key, "unwatched"))
|
||||
db.commit()
|
||||
if pushable:
|
||||
for item_id, rk, action in to_push:
|
||||
background.add_task(plex_watch.push_state_to_plex, user_id, item_id, rk, action, 0, 0)
|
||||
if pushable and to_push:
|
||||
# One coalesced background task for the whole batch (reuses a single session + Plex client),
|
||||
# not one task per episode — a 200-ep show mark is 1 task, not 200.
|
||||
background.add_task(plex_watch.push_bulk_state_to_plex, user_id, to_push)
|
||||
return len(to_push)
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue