peter
c5f121139b
A season marked watched, a pull from Plex and a YouTube resume point made their first row by a read then an add, and failed on the unique key when a first checkpoint or the watched mark made it in between. They go through the same insert-then-read as the checkpoint (the resume point is one upsert); a pull that meets a local change still waiting for its push leaves it. Folding duplicate Watch later lists moved their videos but not a Home pin, which the cascade then deleted; the first pin of a dropped copy moves onto the kept list. Marking one notice read trimmed before its own flag was flushed, so the trim left one past the cap. The route test now crosses the cap and reads the result from a session of its own.
414 lines
15 KiB
Python
414 lines
15 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, timezone
|
|
|
|
from sqlalchemy import CursorResult, delete, func, 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 --------------------------------------------------------------------
|
|
|
|
|
|
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,
|
|
) -> PlexState:
|
|
"""Record a resume position. `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."""
|
|
st = state_row(db, user, item_id, st)
|
|
st.position_seconds = position
|
|
st.progress_updated_at = now
|
|
if dirty:
|
|
st.synced_to_plex = False
|
|
db.flush()
|
|
return st
|
|
|
|
|
|
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
|