peter
24190e4c16
The 10 s tick in flight when the player finishes (or a refusal rollback whose state write in fact committed) wrote its mid-item position onto the freshly watched row and scheduled a resume push after the scrobble, so Plex showed the item in progress again. The checkpoint is now a conditional write that skips a row marked watched in the last 30 s and answers like a finish; a rewatch past that is checkpointed as before. Marking an item (or a season/show) watched also supersedes resume pushes still in flight, as the near-end finish already did.
435 lines
16 KiB
Python
435 lines
16 KiB
Python
"""Plex's writes for the HTTP layer, one function per use case. None of them commits: the handler
|
|
does, so a page that fails to save an enrichment still renders, and a Plex write and its local
|
|
mirror land together.
|
|
|
|
The rest of the mirror's writes stay with the sync (`plex.sync`): a collection's targeted re-sync
|
|
after a write-back, its local cleanup after a delete.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import CursorResult, delete, func, or_, select, update
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import logs
|
|
from app.db import SessionLocal
|
|
from app.plex import detail as plex_detail
|
|
from app.plex import people as plex_people
|
|
from app.plex import tmdb as plex_tmdb
|
|
from app.plex.lookups import by_rating_key
|
|
from app.models import User
|
|
from app.plex import queries, rollup
|
|
from app.plex.models import (
|
|
PlexCollection,
|
|
PlexItem,
|
|
PlexLibrary,
|
|
PlexLink,
|
|
PlexPerson,
|
|
PlexPlaylist,
|
|
PlexPlaylistItem,
|
|
PlexShow,
|
|
PlexState,
|
|
)
|
|
from app.userscope import owned_by
|
|
|
|
log = logs.get_logger(__name__)
|
|
|
|
|
|
def enrich_show(db: Session, sh: PlexShow, meta: dict) -> None:
|
|
"""Fill a not-yet-enriched show from its live metadata: the full cast (filterable), IMDb and
|
|
tagline, stamped so later views serve from the row."""
|
|
plex_people.enrich_row_people(db, sh, plex_people.cast_from_meta(meta))
|
|
plex_detail.apply_rich_detail(sh, meta)
|
|
sh.people_enriched_at = datetime.now(timezone.utc)
|
|
|
|
|
|
def enrich_movie(db: Session, it: PlexItem, meta: dict) -> None:
|
|
"""The same for a movie. The listing sync stored only its top ~3 cast; the live metadata
|
|
carries all of them."""
|
|
cast = plex_people.cast_from_meta(meta)
|
|
if cast:
|
|
plex_people.enrich_row_people(db, it, cast)
|
|
plex_detail.apply_rich_detail(it, meta)
|
|
it.people_enriched_at = datetime.now(timezone.utc)
|
|
|
|
|
|
def set_collection_editable(col: PlexCollection, editable: bool) -> None:
|
|
col.editable = editable
|
|
|
|
|
|
def keep_created_collection(
|
|
db: Session, lib: PlexLibrary, rating_key: str, title: str, seed: str
|
|
) -> None:
|
|
"""A collection Plex made, whose follow-up read failed: its row is kept, editable. The full sync
|
|
preserves `editable` but creates a missing row read-only, so the collection would drop out of
|
|
every "add to" list; and the duplicate-name check reads these rows, so a retry of the same name
|
|
would make a second one on Plex. Plex made it with exactly the seed in it, so the seed's row says
|
|
so too — otherwise the title showed unticked in the collection it was just added to, and the
|
|
collection empty, until that sync."""
|
|
db.add(
|
|
PlexCollection(
|
|
rating_key=rating_key, library_id=lib.id, title=title, editable=True, child_count=1
|
|
)
|
|
)
|
|
Model = PlexItem if lib.kind == "movie" else PlexShow
|
|
if row := by_rating_key(db, Model, seed):
|
|
row.collection_keys = list(dict.fromkeys([*(row.collection_keys or []), rating_key]))
|
|
db.flush()
|
|
|
|
|
|
def enrich_person_bg(person_id: int) -> None:
|
|
"""Fetch + persist one person's TMDB bio OFF the request path — a FastAPI background task that runs
|
|
AFTER the response is sent, so the outbound TMDB call never sits on the serve latency (the module's
|
|
"no live TMDB call at serve time" invariant). The request's session is already closed by then, so
|
|
open a fresh one. Best-effort: a TMDB failure just leaves `tmdb_enriched_at` NULL to retry on the
|
|
next open, and the bio lands from the row by then."""
|
|
db = SessionLocal()
|
|
try:
|
|
person = db.get(PlexPerson, person_id)
|
|
if person is not None:
|
|
plex_tmdb.enrich_person(db, person)
|
|
except Exception: # noqa: BLE001 — background best-effort; never surfaces to a user
|
|
db.rollback()
|
|
log.warning("background person bio enrich failed for id=%s", person_id, exc_info=True)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def set_watch_link(db: Session, user: User, enabled: bool) -> PlexLink:
|
|
"""Turn the user's two-way watch sync on or off, creating their link on first use. The link uses
|
|
the server admin token (`uses_admin`), so it needs no Plex login of its own."""
|
|
link = queries.watch_link(db, user)
|
|
if link is None:
|
|
link = PlexLink(user_id=user.id, uses_admin=True, linked_at=datetime.now(timezone.utc))
|
|
db.add(link)
|
|
link.sync_enabled = enabled
|
|
db.flush()
|
|
return link
|
|
|
|
|
|
# --- a user's playlists ----------------------------------------------------------------------
|
|
|
|
|
|
def create_playlist(db: Session, user: User, title: str, seed: PlexItem | None) -> PlexPlaylist:
|
|
"""A new playlist of the user's, holding `seed` when one is given."""
|
|
pl = PlexPlaylist(user_id=user.id, title=title)
|
|
db.add(pl)
|
|
db.flush()
|
|
if seed is not None:
|
|
db.add(PlexPlaylistItem(playlist_id=pl.id, item_id=seed.id, position=0))
|
|
db.flush()
|
|
return pl
|
|
|
|
|
|
def rename_playlist(pl: PlexPlaylist, title: str) -> None:
|
|
pl.title = title
|
|
|
|
|
|
def delete_playlist(db: Session, pl: PlexPlaylist) -> None:
|
|
db.delete(pl)
|
|
db.flush()
|
|
|
|
|
|
def _touch(pl: PlexPlaylist) -> None:
|
|
pl.updated_at = datetime.now(timezone.utc)
|
|
|
|
|
|
def _last_position(db: Session, playlist_id: int) -> int:
|
|
return db.execute(
|
|
select(func.coalesce(func.max(PlexPlaylistItem.position), -1)).where(
|
|
PlexPlaylistItem.playlist_id == playlist_id
|
|
)
|
|
).scalar_one()
|
|
|
|
|
|
def add_item(db: Session, pl: PlexPlaylist, it: PlexItem) -> None:
|
|
"""Append `it` to the (locked) playlist; an item already in it stays where it is."""
|
|
exists = db.execute(
|
|
select(PlexPlaylistItem.id).where(
|
|
PlexPlaylistItem.playlist_id == pl.id, PlexPlaylistItem.item_id == it.id
|
|
)
|
|
).first()
|
|
if exists is None:
|
|
pos = _last_position(db, pl.id) + 1
|
|
db.add(PlexPlaylistItem(playlist_id=pl.id, item_id=it.id, position=pos))
|
|
_touch(pl)
|
|
db.flush()
|
|
|
|
|
|
def remove_item(db: Session, pl: PlexPlaylist, it: PlexItem) -> None:
|
|
db.execute(
|
|
delete(PlexPlaylistItem).where(
|
|
PlexPlaylistItem.playlist_id == pl.id, PlexPlaylistItem.item_id == it.id
|
|
)
|
|
)
|
|
_touch(pl)
|
|
db.flush()
|
|
|
|
|
|
def add_items(db: Session, pl: PlexPlaylist, items: list[PlexItem]) -> int:
|
|
"""Append many items in their given order (a whole season or show); those already in the list
|
|
are skipped. Says how many were added."""
|
|
existing = set(
|
|
db.execute(
|
|
select(PlexPlaylistItem.item_id).where(PlexPlaylistItem.playlist_id == pl.id)
|
|
).scalars()
|
|
)
|
|
pos = _last_position(db, pl.id)
|
|
added = 0
|
|
for it in items:
|
|
if it.id in existing:
|
|
continue
|
|
pos += 1
|
|
db.add(PlexPlaylistItem(playlist_id=pl.id, item_id=it.id, position=pos))
|
|
existing.add(it.id)
|
|
added += 1
|
|
if added:
|
|
_touch(pl)
|
|
db.flush()
|
|
return added
|
|
|
|
|
|
def insert_items_at(db: Session, pl: PlexPlaylist, placed: list[tuple[int, PlexItem]]) -> int:
|
|
"""Insert new rows at their indices (ascending, so each index is where the row stood when the
|
|
rows before it were back), renumber the whole list 0..n-1, and say how many were added. An item
|
|
already in the list does not move: the add stays idempotent."""
|
|
rows = list(
|
|
db.execute(
|
|
select(PlexPlaylistItem)
|
|
.where(PlexPlaylistItem.playlist_id == pl.id)
|
|
.order_by(PlexPlaylistItem.position, PlexPlaylistItem.id)
|
|
).scalars()
|
|
)
|
|
present = {r.item_id for r in rows}
|
|
added = 0
|
|
for at, it in sorted(placed, key=lambda p: p[0]):
|
|
if it.id in present:
|
|
continue
|
|
row = PlexPlaylistItem(playlist_id=pl.id, item_id=it.id, position=0)
|
|
db.add(row)
|
|
rows.insert(min(at, len(rows)), row)
|
|
present.add(it.id)
|
|
added += 1
|
|
for i, r in enumerate(rows):
|
|
r.position = i
|
|
if added:
|
|
_touch(pl)
|
|
db.flush()
|
|
return added
|
|
|
|
|
|
def remove_items(db: Session, pl: PlexPlaylist, item_ids: list[int]) -> int:
|
|
"""Remove these items from the playlist; says how many were there."""
|
|
if not item_ids:
|
|
return 0
|
|
result = db.execute(
|
|
delete(PlexPlaylistItem).where(
|
|
PlexPlaylistItem.playlist_id == pl.id, PlexPlaylistItem.item_id.in_(item_ids)
|
|
),
|
|
execution_options={"synchronize_session": False},
|
|
)
|
|
removed = result.rowcount if isinstance(result, CursorResult) else 0
|
|
if removed:
|
|
_touch(pl)
|
|
db.flush()
|
|
return removed
|
|
|
|
|
|
def reorder_playlist(db: Session, pl: PlexPlaylist, rating_keys: list[str]) -> None:
|
|
"""Set the play order from an explicit list of item keys (0-based positions)."""
|
|
id_by_rk = {
|
|
it.rating_key: it.id
|
|
for it in db.execute(
|
|
select(PlexItem).where(PlexItem.rating_key.in_(rating_keys or [""]))
|
|
).scalars()
|
|
}
|
|
rows = {
|
|
r.item_id: r
|
|
for r in db.execute(
|
|
select(PlexPlaylistItem).where(PlexPlaylistItem.playlist_id == pl.id)
|
|
).scalars()
|
|
}
|
|
pos = 0
|
|
seen: set[int] = set()
|
|
for rk in rating_keys:
|
|
row = rows.get(id_by_rk.get(rk, -1))
|
|
if row is not None and row.item_id not in seen:
|
|
row.position = pos
|
|
pos += 1
|
|
seen.add(row.item_id)
|
|
# Rows the payload omitted (added in another tab meanwhile) keep their relative order after
|
|
# the given ones, so no two rows share a position (the YouTube reorder does the same).
|
|
for row in sorted(
|
|
(r for iid, r in rows.items() if iid not in seen), key=lambda r: (r.position, r.id)
|
|
):
|
|
row.position = pos
|
|
pos += 1
|
|
_touch(pl)
|
|
db.flush()
|
|
|
|
|
|
# --- a user's watch state --------------------------------------------------------------------
|
|
|
|
# How long after a watched mark a mid-item position is read as the finishing player's late tick
|
|
# (`checkpoint`): the tick is every 10 s, plus the request's own latency and a retry.
|
|
_LATE_CHECKPOINT_S = 30
|
|
|
|
|
|
def mark_finished(db: Session, user_id: int, item_id: int, now: datetime) -> bool:
|
|
"""Mark the item watched (position cleared) and say whether THIS call made it so. Two near-end
|
|
checkpoints arrive together routinely (the 10 s tick and the debounced `seeked` flush), and a
|
|
read-then-write let both see "not watched" and both scrobble. The transition is decided by the
|
|
write itself: an insert that did not conflict, or an update conditional on the old status —
|
|
under READ COMMITTED the second writer re-reads the committed row and matches nothing."""
|
|
values = {
|
|
"status": "watched",
|
|
"watched_at": now,
|
|
"position_seconds": 0,
|
|
"progress_updated_at": now,
|
|
"synced_to_plex": False,
|
|
}
|
|
inserted = db.execute(
|
|
pg_insert(PlexState)
|
|
.values(user_id=user_id, item_id=item_id, **values)
|
|
.on_conflict_do_nothing(constraint="uq_plex_user_item")
|
|
.returning(PlexState.id)
|
|
).first()
|
|
if inserted is not None:
|
|
return True
|
|
mine = owned_by(PlexState.user_id, user_id) & (PlexState.item_id == item_id)
|
|
result = db.execute(update(PlexState).where(mine, PlexState.status != "watched").values(values))
|
|
if isinstance(result, CursorResult) and result.rowcount:
|
|
return True
|
|
# Already watched: a re-finish still restamps the row, as it always did — just no scrobble.
|
|
db.execute(update(PlexState).where(mine).values(values))
|
|
return False
|
|
|
|
|
|
def state_row(
|
|
db: Session,
|
|
user: User | int,
|
|
item_id: int,
|
|
st: PlexState | None = None,
|
|
*,
|
|
synced: bool = False,
|
|
) -> PlexState:
|
|
"""The user's state row on this item, made if the read before found none. Made by the insert
|
|
itself (ON CONFLICT DO NOTHING) and then read: the first checkpoint of an item and the `final`
|
|
flush beside it routinely both read "none", and a plain add let the second one fail on
|
|
`uq_plex_user_item` — a 500, and the push it was carrying lost. Every first row goes through
|
|
here: a checkpoint, a status, a season marked at once, a pull from Plex. `synced` is what a row
|
|
made now says about Plex; a row another writer made first keeps its own."""
|
|
if st is not None:
|
|
return st
|
|
db.execute(
|
|
pg_insert(PlexState)
|
|
.values(
|
|
user_id=user if isinstance(user, int) else user.id,
|
|
item_id=item_id,
|
|
synced_to_plex=synced,
|
|
)
|
|
.on_conflict_do_nothing(constraint="uq_plex_user_item")
|
|
)
|
|
made = queries.state(db, user, item_id)
|
|
assert made is not None # inserted now, or committed by the request that won
|
|
return made
|
|
|
|
|
|
def clear_progress(db: Session, st: PlexState | None, now: datetime) -> None:
|
|
"""A checkpoint too early to keep: a state that held nothing else goes, any other forgets its
|
|
position."""
|
|
if st is not None and st.status == "new":
|
|
db.delete(st)
|
|
elif st is not None:
|
|
st.position_seconds = 0
|
|
st.progress_updated_at = now
|
|
db.flush()
|
|
|
|
|
|
def checkpoint(
|
|
db: Session,
|
|
user: User,
|
|
item_id: int,
|
|
st: PlexState | None,
|
|
position: int,
|
|
now: datetime,
|
|
*,
|
|
dirty: bool,
|
|
) -> bool:
|
|
"""Record a resume position, and say whether it was recorded. `dirty` marks it not yet mirrored
|
|
to Plex — set before a push, so the reconcile re-pushes it if the push fails and the push can
|
|
settle it once Plex confirms.
|
|
|
|
Not onto a row marked watched in the last `_LATE_CHECKPOINT_S`: that position is the player's
|
|
tick sent just before its own finish (or a mark from another tab), and writing it made a watched
|
|
item in progress again — on Plex too, through the resume push it carried. Decided by the write
|
|
itself, since the mark can commit between any read here and the write. A rewatch is checkpointed
|
|
as before once the mark is that far behind it."""
|
|
st = state_row(db, user, item_id, st)
|
|
values: dict[str, object] = {"position_seconds": position, "progress_updated_at": now}
|
|
if dirty:
|
|
values["synced_to_plex"] = False
|
|
not_just_finished = or_(
|
|
PlexState.status != "watched",
|
|
PlexState.watched_at.is_(None),
|
|
PlexState.watched_at <= now - timedelta(seconds=_LATE_CHECKPOINT_S),
|
|
)
|
|
result = db.execute(
|
|
update(PlexState)
|
|
.where(PlexState.id == st.id, not_just_finished)
|
|
.values(values)
|
|
.execution_options(synchronize_session="fetch")
|
|
)
|
|
db.flush()
|
|
return isinstance(result, CursorResult) and result.rowcount > 0
|
|
|
|
|
|
def forget_state(db: Session, st: PlexState) -> None:
|
|
db.delete(st)
|
|
db.flush()
|
|
|
|
|
|
def set_status(
|
|
db: Session, user: User, item_id: int, st: PlexState | None, status: str
|
|
) -> PlexState:
|
|
"""Set a watched or hidden status; watched also clears the resume point."""
|
|
st = state_row(db, user, item_id, st)
|
|
st.status = status
|
|
if status == "watched":
|
|
st.watched_at = datetime.now(timezone.utc)
|
|
st.position_seconds = 0
|
|
st.synced_to_plex = False
|
|
db.flush()
|
|
return st
|
|
|
|
|
|
def bulk_state(
|
|
db: Session, user_id: int, eps: list[PlexItem], watched: bool
|
|
) -> list[tuple[int, str, str]]:
|
|
"""Mark every episode in `eps` watched (or unwatched) for a user and say what changed, as the
|
|
(item id, rating key, action) pushes a linked Plex account needs. Unwatching keeps the user's
|
|
private "hidden" marks."""
|
|
now = datetime.now(timezone.utc)
|
|
existing = rollup.item_state_map(db, user_id, [e.id for e in eps])
|
|
to_push: list[tuple[int, str, str]] = []
|
|
for e in eps:
|
|
st = existing.get(e.id)
|
|
prev = st.status if st is not None else "new"
|
|
if watched:
|
|
st = state_row(db, user_id, e.id, st)
|
|
prev = st.status # a first checkpoint beside this may have made it meanwhile
|
|
st.status = "watched"
|
|
st.watched_at = now
|
|
st.position_seconds = 0
|
|
st.synced_to_plex = False
|
|
if prev != "watched":
|
|
to_push.append((e.id, e.rating_key, "watched"))
|
|
else: # → new: drop any state (but keep the user's private "hidden" marks)
|
|
if st is not None and prev != "hidden":
|
|
db.delete(st)
|
|
if prev != "new":
|
|
to_push.append((e.id, e.rating_key, "unwatched"))
|
|
db.flush()
|
|
return to_push
|