diff --git a/VERSION b/VERSION index 0f1a7df..ca75280 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.37.0 +0.38.0 diff --git a/backend/app/auth.py b/backend/app/auth.py index 29eafa4..97bb25d 100644 --- a/backend/app/auth.py +++ b/backend/app/auth.py @@ -22,6 +22,10 @@ from app.utils import valid_email PASSWORD_MIN_LEN = 10 VERIFY_TTL = timedelta(hours=24) RESET_TTL = timedelta(hours=1) +# A throwaway argon2 hash to verify against when the email is unknown / has no password, so a +# login attempt pays the same KDF cost either way and its timing can't reveal whether the account +# exists (anti-enumeration). Computed once at startup; it never matches a real password. +_DECOY_PASSWORD_HASH = hash_password(secrets.token_urlsafe(16)) _register_limiter = RateLimiter(max_events=5, window_seconds=300) _login_limiter = RateLimiter(max_events=10, window_seconds=300) _reset_limiter = RateLimiter(max_events=5, window_seconds=300) @@ -276,6 +280,12 @@ async def callback( # hijack it. (Near-impossible with real Google accounts: email↔sub is stable.) log.warning("Google login email collision (different sub): %s", email) return RedirectResponse(url="/?access=denied") + if not userinfo.get("email_verified", False): + # Adopting (and activating) a pre-existing password account requires Google to + # actually attest the address — otherwise an unverified Google email that merely + # collides with a pending registration could seize that account. Defense-in-depth. + log.warning("Google login: unverified email, refusing to adopt account: %s", email) + return RedirectResponse(url="/?access=denied") user = existing user.google_sub = sub # Google verified the email and is_allowed granted access, so finish activating a @@ -291,7 +301,13 @@ async def callback( log.info("Suspended account blocked at Google login: %s", email) _notify_suspended(background, user.email) return RedirectResponse(url="/?login=suspended") - user.email = email + # Keep the email in sync with Google, but never overwrite it with one another account already + # owns — that would hit the unique constraint and 500, wedging this user's sign-in (rare: a + # Google-side email change to a value that collides with a different local account). + if user.email != email: + clash = db.query(User).filter(User.email == email, User.id != user.id).one_or_none() + if clash is None: + user.email = email user.display_name = userinfo.get("name") user.avatar_url = userinfo.get("picture") # ADMIN_EMAILS (env) is the bootstrap admin list and always wins, so the configured admin can't @@ -337,7 +353,9 @@ def _store_token(db: Session, user: User, token: dict) -> None: tok = user.token or OAuthToken(user=user) if token.get("refresh_token"): tok.refresh_token_enc = encrypt(token["refresh_token"]) - tok.access_token = token.get("access_token") + # Encrypt the access token at rest too (it's a live ~1h bearer credential), matching the + # refresh token — a DB/backup/log leak otherwise hands out working Google API tokens. + tok.access_token = encrypt(token.get("access_token")) expires_at = token.get("expires_at") tok.expiry = datetime.fromtimestamp(expires_at, tz=timezone.utc) if expires_at else None tok.scopes = token.get("scope") or BASE_SCOPES @@ -471,6 +489,9 @@ def demo_login(payload: dict, request: Request, db: Session = Depends(get_db)) - email = (payload.get("email") or "").strip().lower() if valid_email(email) and is_demo_allowed(db, email): demo = get_or_create_demo_user(db) + # Isolate the demo: drop any real-account wallet on this tab first, so the demo session + # can't switch to / act as a previously signed-in account via the multi-account header. + request.session.clear() request.session["user_id"] = demo.id log.info("Demo login (key=%s, demo_id=%s)", email, demo.id) return {"authenticated": True} @@ -603,7 +624,11 @@ def password_login( email = (payload.get("email") or "").strip().lower() password = payload.get("password") or "" user = db.execute(select(User).where(User.email == email)).scalar_one_or_none() - if user is None or user.is_demo or not verify_password(password, user.password_hash): + # Always run the KDF (against a decoy hash for an unknown / passwordless / demo account) so the + # response time is the same whether or not the email exists — no timing-based enumeration. + candidate_hash = user.password_hash if (user and not user.is_demo) else None + password_ok = verify_password(password, candidate_hash or _DECOY_PASSWORD_HASH) + if user is None or user.is_demo or not password_ok: raise HTTPException(status_code=401, detail="Invalid email or password.") if user.is_suspended: _notify_suspended(background, user.email) diff --git a/backend/app/config.py b/backend/app/config.py index c83f0f7..3f75965 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -211,7 +211,7 @@ class Settings(BaseSettings): plex_watch_reconcile_interval_min: int = 1440 # Max concurrent transcodes for the P3 fallback. Low by default — CPU-only transcode is # expensive; direct-serve (browser-compatible files) has no such limit. - plex_max_transcodes: int = 1 + plex_max_transcodes: int = 4 # Where on-the-fly HLS remux segments are written (transient, reaped). Empty → a `.plex-hls` # dir under download_root. On localdev the download_root is a slow Windows bind-mount, so point # this at a fast container-local path (e.g. /var/tmp/plex-hls); prod's download_root is fast diff --git a/backend/app/downloads/edit.py b/backend/app/downloads/edit.py index b52064f..cff449a 100644 --- a/backend/app/downloads/edit.py +++ b/backend/app/downloads/edit.py @@ -79,13 +79,17 @@ def normalize_edit_spec(spec: dict | None) -> dict: # Single TRIM (one output file; the frontend fans a "separate" split out into N of these). trim = spec.get("trim") or {} if trim: - start = max(0.0, float(trim.get("start_s") or 0.0)) - end_raw = trim.get("end_s") + # Guard the numeric coercion like the crop/segments branches do — a malformed value must + # drop the trim (→ empty spec → 400 EditError), not raise an unhandled 500. + try: + start = max(0.0, float(trim.get("start_s") or 0.0)) + end_raw = trim.get("end_s") + end = float(end_raw) if end_raw is not None else None + except (TypeError, ValueError): + start, end = 0.0, None t: dict = {"start_s": round(start, 3)} - if end_raw is not None: - end = float(end_raw) - if end > start: - t["end_s"] = round(end, 3) + if end is not None and end > start: + t["end_s"] = round(end, 3) # A trim with only a start (open-ended) is valid (cut to the end). if t.get("start_s") or "end_s" in t: out["trim"] = t diff --git a/backend/app/downloads/formats.py b/backend/app/downloads/formats.py index ef38ef9..28b7f45 100644 --- a/backend/app/downloads/formats.py +++ b/backend/app/downloads/formats.py @@ -62,7 +62,12 @@ def normalize(spec: dict) -> dict: if s["mode"] not in ("av", "v", "a"): s["mode"] = "av" if s["max_height"] is not None: - s["max_height"] = int(s["max_height"]) + # A malformed max_height (non-numeric inline spec / stored profile) falls back to "best" + # rather than raising an unhandled 500 at enqueue. + try: + s["max_height"] = int(s["max_height"]) + except (TypeError, ValueError): + s["max_height"] = None for b in ("embed_subs", "embed_chapters", "embed_thumbnail", "sponsorblock"): s[b] = bool(s[b]) # Audio-only files can't embed subtitles/thumbnails as video; keep the flags meaningful. @@ -80,14 +85,6 @@ def format_sig(spec: dict) -> str: return hashlib.sha256(payload.encode()).hexdigest()[:24] -def target_ext(spec: dict) -> str: - """Best-effort final extension for the produced file (also used to name the staging output).""" - s = normalize(spec) - if s["mode"] == "a": - return s["audio_format"] or "m4a" - return s["container"] or "mp4" - - def build_ydl_opts( spec: dict, outtmpl: str, diff --git a/backend/app/downloads/links.py b/backend/app/downloads/links.py index 08123c7..6b18cbd 100644 --- a/backend/app/downloads/links.py +++ b/backend/app/downloads/links.py @@ -75,13 +75,7 @@ def public_meta(link: DownloadLink, job, asset, file_url: str, poster_url: str | auto-extracted values, and surfaces the source + any extra reference links as clickable URLs.""" from app.downloads import service - source_url = None - if job is not None and job.job_kind != "edit": - source_url = asset.source_webpage_url or ( - service.source_url(job.source_kind, job.source_ref) - if job.source_kind in ("youtube", "url") - else None - ) + source_url = service.reference_url(job, asset) return { "title": (job and job.display_name) or asset.title, "uploader": (job and job.display_uploader) or asset.uploader, diff --git a/backend/app/downloads/service.py b/backend/app/downloads/service.py index b20304b..e2bf16f 100644 --- a/backend/app/downloads/service.py +++ b/backend/app/downloads/service.py @@ -32,6 +32,20 @@ def source_url(source_kind: str, source_ref: str) -> str: return source_ref +def reference_url(job, asset) -> str | None: + """The clean "downloaded from" URL for a job — shared by the private Downloads page and the public + watch page so both surfaces always agree. None for edit derivatives (a clip's source is the user's + own earlier download, not a web page); else the canonical page URL yt-dlp recorded; else one derived + from source_kind+source_ref (covers queued/older rows before the worker fills it).""" + if job is None or job.job_kind == "edit": + return None + if asset is not None and asset.source_webpage_url: + return asset.source_webpage_url + if job.source_kind in ("youtube", "url"): + return source_url(job.source_kind, job.source_ref) + return None + + def get_or_create_asset( db: Session, source_kind: str, source_ref: str, format_sig: str ) -> MediaAsset: @@ -46,6 +60,14 @@ def get_or_create_asset( ) asset = db.execute(stmt).scalar_one_or_none() if asset is not None: + # A previously-FAILED shared asset must be re-attempted for the new job about to hold it — + # reset it to pending (+clear the error) so the worker actually re-downloads instead of + # short-circuiting the new job with the asset's stale error. (Mirrors resume_download; without + # this a fresh enqueue of a once-failed (source,format) pair is permanently poisoned, since + # errored assets carry no expires_at and GC never clears them.) + if asset.status == "error": + asset.status = "pending" + asset.error = None return asset asset = MediaAsset( source_kind=source_kind, diff --git a/backend/app/downloads/storage.py b/backend/app/downloads/storage.py index d15bb3a..1aee7b8 100644 --- a/backend/app/downloads/storage.py +++ b/backend/app/downloads/storage.py @@ -96,10 +96,32 @@ def edit_rel_path(user_id: int, asset_id: int, ext: str) -> str: return f".edits/{user_id}/clip_{asset_id}.{ext}" +def download_filename(display_name: str, container: str | None, path: Path) -> str: + """The Content-Disposition filename for a served download: the cleaned display name carrying a single + correct extension — the asset container, else the file's own suffix — without doubling it when the + name already ends in that extension. Shared by the private download and the public watch endpoints.""" + ext = container or path.suffix.lstrip(".") + base = display_filename(display_name) + if base.lower().endswith(f".{ext.lower()}"): + base = base[: -(len(ext) + 1)] + return f"{base}.{ext}" + + def abs_path(download_root: str, rel: str) -> Path: return Path(download_root) / rel +def safe_abs_path(download_root: str, rel: str) -> Path | None: + """The resolved absolute path for `rel`, but ONLY if it stays inside DOWNLOAD_ROOT and exists on + disk — otherwise None. Centralizes the path-traversal + existence guard every file-serving endpoint + shares, so the containment check lives in exactly one place.""" + root = Path(download_root).resolve() + path = abs_path(download_root, rel).resolve() + if root not in path.parents or not path.exists(): + return None + return path + + def place_file(staging_file: Path, download_root: str, rel: str) -> None: """Move a completed staging file into its final location under DOWNLOAD_ROOT.""" dest = abs_path(download_root, rel) diff --git a/backend/app/models.py b/backend/app/models.py index 5e0c915..d65ee87 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -232,6 +232,11 @@ class Subscription(Base, TimestampMixin): channel: Mapped["Channel"] = relationship(back_populates="subscriptions") +# Video.live_status values that mean "currently a live/upcoming broadcast" — transient states +# hidden from feeds/search and revisited by the refresh job until they settle into a VOD. +LIVE_OR_UPCOMING = ("live", "upcoming") + + class Video(Base, TimestampMixin): """A YouTube video. Shared across users; per-user state lives elsewhere.""" 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/stream.py b/backend/app/plex/stream.py index 129e050..5d1f5c2 100644 --- a/backend/app/plex/stream.py +++ b/backend/app/plex/stream.py @@ -23,6 +23,7 @@ from pathlib import Path from sqlalchemy.orm import Session as DbSession +from app import sysconfig from app.config import settings from app.models import PlexItem from app.plex import paths @@ -31,7 +32,6 @@ log = logging.getLogger("siftlode.plex") _HLS_ROOT = Path(settings.plex_hls_dir) if settings.plex_hls_dir else Path(settings.download_root) / ".plex-hls" _SEG_SECONDS = 6 -_MAX_SESSIONS = 4 # concurrent remux sessions (safety valve for the CPU-only host) _SESSION_IDLE_S = 600 # reap a session with no access for this long _lock = threading.Lock() @@ -168,11 +168,11 @@ def _kill(s: HlsSession) -> None: shutil.rmtree(s.dir, ignore_errors=True) -def _enforce_cap() -> None: +def _enforce_cap(cap: int) -> None: # Caller holds _lock. Drop the least-recently-accessed sessions over the cap. - if len(_sessions) <= _MAX_SESSIONS: + if len(_sessions) <= cap: return - for key, s in sorted(_sessions.items(), key=lambda kv: kv[1].last_access)[: len(_sessions) - _MAX_SESSIONS]: + for key, s in sorted(_sessions.items(), key=lambda kv: kv[1].last_access)[: len(_sessions) - cap]: _kill(s) _sessions.pop(key, None) @@ -211,7 +211,9 @@ def start_session( ) s = HlsSession(key, directory, proc, start_s, "master.m3u8" if is_multi else "index.m3u8") _sessions[key] = s - _enforce_cap() + # Admin-configurable concurrency cap (Configuration → Plex → Max concurrent transcodes); + # at least 1 so playback can't be capped to zero. + _enforce_cap(max(1, sysconfig.get_int(db, "plex_max_transcodes"))) # Learn the REAL start offset K from the first VIDEO segment (outside the lock — this blocks on # ffmpeg producing it, and we must not hold up other sessions). The client reads `media_start_s`, so # its absolute clock + subtitle shift key off the true keyframe, not the requested offset. On 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/admin.py b/backend/app/routes/admin.py index 1d86ed0..3edd262 100644 --- a/backend/app/routes/admin.py +++ b/backend/app/routes/admin.py @@ -166,7 +166,12 @@ def set_user_role( raise HTTPException(status_code=400, detail="The demo account's role can't be changed.") if target.id == admin.id: raise HTTPException(status_code=400, detail="You can't change your own role.") - if target.role == "admin" and role == "user" and count_admins(db) <= 1: + if ( + target.role == "admin" + and role == "user" + and not target.is_suspended + and count_admins(db, active_only=True) <= 1 + ): raise HTTPException(status_code=400, detail="Can't remove the last admin.") changed = target.role != role target.role = role @@ -233,7 +238,7 @@ def admin_delete_user( raise HTTPException( status_code=400, detail="Delete your own account from Settings → Account." ) - if target.role == "admin" and count_admins(db) <= 1: + if target.role == "admin" and not target.is_suspended and count_admins(db, active_only=True) <= 1: raise HTTPException(status_code=400, detail="Can't delete the last admin.") purge_user(db, target, background) return {"deleted": user_id} diff --git a/backend/app/routes/channels.py b/backend/app/routes/channels.py index 4c47b71..bdb5230 100644 --- a/backend/app/routes/channels.py +++ b/backend/app/routes/channels.py @@ -5,13 +5,14 @@ import logging from datetime import datetime, timezone from fastapi import APIRouter, Depends, HTTPException -from sqlalchemy import and_, func, select, update +from sqlalchemy import func, select, update from sqlalchemy.orm import Session from app import quota, sysconfig from app.auth import current_user, has_write_scope, require_human from app.db import get_db from app.models import ( + LIVE_OR_UPCOMING, BlockedChannel, Channel, ChannelTag, @@ -59,7 +60,7 @@ def list_channels( func.max(Video.published_at), func.coalesce(func.sum(Video.duration_seconds), 0), func.count(Video.id).filter(Video.is_short.is_(True)), - func.count(Video.id).filter(Video.live_status.in_(("live", "upcoming"))), + func.count(Video.id).filter(Video.live_status.in_(LIVE_OR_UPCOMING)), ) .where(Video.channel_id.in_(channel_ids)) .group_by(Video.channel_id) @@ -100,12 +101,7 @@ def list_channels( return [ { - "id": ch.id, - "title": ch.title, - "handle": ch.handle, - "thumbnail_url": ch.thumbnail_url, - "subscriber_count": ch.subscriber_count, - "video_count": ch.video_count, + **_channel_summary(ch), "stored_videos": (agg.get(ch.id) or {}).get("stored", 0), "last_video_at": (agg.get(ch.id) or {}).get("last_video_at"), "total_duration_seconds": (agg.get(ch.id) or {}).get("total_duration_seconds", 0), @@ -194,12 +190,7 @@ def discover_channels( return [ { - "id": ch.id, - "title": ch.title, - "handle": ch.handle, - "thumbnail_url": ch.thumbnail_url, - "subscriber_count": ch.subscriber_count, - "video_count": ch.video_count, + **_channel_summary(ch), "playlist_video_count": int(vid_count), "playlist_count": int(pl_count), "details_synced": ch.details_synced_at is not None, @@ -208,6 +199,18 @@ def discover_channels( ] +def _channel_summary(ch: Channel) -> dict: + """The channel fields common to the list and discovery projections.""" + return { + "id": ch.id, + "title": ch.title, + "handle": ch.handle, + "thumbnail_url": ch.thumbnail_url, + "subscriber_count": ch.subscriber_count, + "video_count": ch.video_count, + } + + def _channel_detail_dict( channel: Channel, *, subscribed: bool, explored: bool, blocked: bool, stored: int ) -> dict: @@ -267,11 +270,7 @@ def channel_detail( ExploredChannel.user_id == user.id, ExploredChannel.channel_id == channel_id ) ).first() is not None - blocked = db.execute( - select(BlockedChannel.id).where( - BlockedChannel.user_id == user.id, BlockedChannel.channel_id == channel_id - ) - ).first() is not None + blocked = _is_blocked(db, user, channel_id) stored = db.scalar(select(func.count(Video.id)).where(Video.channel_id == channel_id)) or 0 return _channel_detail_dict( channel, subscribed=subscribed, explored=explored, blocked=blocked, stored=int(stored) @@ -294,11 +293,7 @@ def explore_channel( channel = db.get(Channel, channel_id) if channel is None: raise HTTPException(status_code=404, detail="Unknown channel") - if db.execute( - select(BlockedChannel.id).where( - BlockedChannel.user_id == user.id, BlockedChannel.channel_id == channel_id - ) - ).first() is not None: + if _is_blocked(db, user, channel_id): raise HTTPException(status_code=403, detail="You've blocked this channel.") if quota.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"): raise HTTPException( @@ -358,6 +353,19 @@ def _user_subscription(db: Session, user: User, channel_id: str) -> Subscription return sub +def _is_blocked(db: Session, user: User, channel_id: str) -> bool: + """Whether this user has blocked this channel.""" + return ( + db.execute( + select(BlockedChannel.id).where( + BlockedChannel.user_id == user.id, + BlockedChannel.channel_id == channel_id, + ) + ).first() + is not None + ) + + @router.patch("/{channel_id}") def update_channel( channel_id: str, @@ -545,12 +553,7 @@ def block_channel( un-kept search/explore videos are reclaimed by the discovery-cleanup job in due course.""" if db.get(Channel, channel_id) is None: raise HTTPException(status_code=404, detail="Unknown channel") - exists = db.execute( - select(BlockedChannel.id).where( - BlockedChannel.user_id == user.id, BlockedChannel.channel_id == channel_id - ) - ).first() - if exists is None: + if not _is_blocked(db, user, channel_id): db.add(BlockedChannel(user_id=user.id, channel_id=channel_id)) # Stop any active exploration of it so it can be cleaned up. db.execute( diff --git a/backend/app/routes/downloads.py b/backend/app/routes/downloads.py index ee5490e..71153d5 100644 --- a/backend/app/routes/downloads.py +++ b/backend/app/routes/downloads.py @@ -7,7 +7,6 @@ serves finished files (range-aware, with the user's custom display name), and sh """ import re from datetime import datetime, timedelta, timezone -from pathlib import Path from urllib.parse import parse_qs, urlparse from fastapi import APIRouter, Depends, HTTPException, Request @@ -15,6 +14,7 @@ from fastapi.responses import FileResponse from sqlalchemy import func, select from sqlalchemy.orm import Session +from app import sysconfig from app.auth import admin_user, require_human from app.config import settings from app.db import get_db @@ -68,20 +68,6 @@ def resolve_source(raw: str) -> tuple[str, str]: raise HTTPException(status_code=400, detail="That doesn't look like a valid video link.") -def _reference_url(job: DownloadJob, asset: MediaAsset | None) -> str | None: - """The clean "downloaded from" URL for the Downloads page: the canonical page URL yt-dlp - recorded, else one derived from source_kind+source_ref (covers queued/older rows before the - worker fills it). Returns None for edit derivatives — a clip's source is the user's own earlier - download, not a web page, so there's no meaningful external URL to show.""" - if job.job_kind == "edit": - return None - if asset is not None and asset.source_webpage_url: - return asset.source_webpage_url - if job.source_kind in ("youtube", "url"): - return service.source_url(job.source_kind, job.source_ref) - return None - - def _serialize(job: DownloadJob, asset: MediaAsset | None) -> dict: ready = asset is not None and asset.status == "ready" and bool(asset.rel_path) return { @@ -96,7 +82,7 @@ def _serialize(job: DownloadJob, asset: MediaAsset | None) -> dict: "created_at": job.created_at.isoformat() if job.created_at else None, "source_kind": job.source_kind, "source_ref": job.source_ref, - "source_url": _reference_url(job, asset), + "source_url": service.reference_url(job, asset), # A locally-generated poster to fall back to when the source had no thumbnail. "poster_url": f"/api/downloads/{job.id}/poster.jpg" if (asset is not None and asset.poster_path) @@ -140,6 +126,13 @@ def _assets_for(db: Session, jobs: list[DownloadJob]) -> dict[int, MediaAsset]: return {a.id: a for a in rows} +def _serialize_job(db: Session, job: DownloadJob) -> dict: + """Resolve a job's cache asset (if any) and serialize the pair — the resolve-then-serialize step + every single-job handler shares.""" + asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None + return _serialize(job, asset) + + def _own_job(db: Session, user: User, job_id: int) -> DownloadJob: job = db.get(DownloadJob, job_id) if job is None or job.user_id != user.id: @@ -273,8 +266,7 @@ def enqueue_download( ) except quota.QuotaExceeded as e: raise HTTPException(status_code=422, detail=_quota_message(e)) - asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None - return _serialize(job, asset) + return _serialize_job(db, job) @router.post("/edit") @@ -299,8 +291,7 @@ def enqueue_edit( raise HTTPException(status_code=422, detail=_quota_message(e)) except service.EditError as e: raise HTTPException(status_code=400, detail=_edit_message(e)) - asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None - return _serialize(job, asset) + return _serialize_job(db, job) def _edit_message(e: service.EditError) -> str: @@ -430,8 +421,7 @@ def update_download_meta( cleaned = [u for u in (_clean_url(x) for x in raw) if u][:_MAX_EXTRA_LINKS] job.extra_links = cleaned or None db.commit() - asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None - return _serialize(job, asset) + return _serialize_job(db, job) def _set_status(db: Session, job: DownloadJob, status: str) -> None: @@ -446,8 +436,7 @@ def pause_download( job = _own_job(db, user, job_id) if job.status in ("queued", "running"): _set_status(db, job, "paused") # a running worker aborts cooperatively via the hook - asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None - return _serialize(job, asset) + return _serialize_job(db, job) @router.post("/{job_id}/resume") @@ -467,8 +456,7 @@ def resume_download( asset.status = "pending" asset.error = None _set_status(db, job, "queued") - asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None - return _serialize(job, asset) + return _serialize_job(db, job) @router.post("/{job_id}/cancel") @@ -480,8 +468,7 @@ def cancel_download( _release_asset(db, job) # release while the job still counts as holding job.status = "canceled" db.commit() - asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None - return _serialize(job, asset) + return _serialize_job(db, job) @router.delete("/{job_id}") @@ -499,9 +486,18 @@ def _release_asset(db: Session, job: DownloadJob) -> None: """Drop this job's hold on its asset when it leaves a holding state. Once NO job holds the asset anymore, delete the file + row immediately — a deleted download should free its disk, and the shared cache only needs to span *overlapping* holders (a later re-add just downloads - again). Idempotent: a job that's already left holding (canceled/error) is a no-op, so delete - after cancel doesn't double-release.""" - if not job.asset_id or job.status not in ("queued", "running", "paused", "done"): + again). `error` counts as holding: enqueue always +1'd ref_count and the worker never + decrements on failure (ref_count is the route's job), so an errored job releases here on + delete — else its +1 leaks and a later resume→ready keeps the file pinned past its last holder. + Idempotent for the cancel path: cancel already released while the job was holding, then set it + `canceled` (∉ the set below), so delete-after-cancel is a no-op and never double-releases. + + We DON'T delete an errored asset row here even at ref==0: another request may be concurrently + re-enqueuing the same (source,format) — get_or_create_asset would reset that row to `pending` and + +1 it, and deleting it under that would FK-null the new job's asset (a lost update on ref_count). + A ready asset is still freed at ref==0 (its file is the point); an orphaned errored row is cheap + (no file) and gets reused+reset by the next enqueue, or reclaimed by a later GC pass.""" + if not job.asset_id or job.status not in ("queued", "running", "paused", "done", "error"): return asset = db.get(MediaAsset, job.asset_id) if asset is None: @@ -515,11 +511,6 @@ def _release_asset(db: Session, job: DownloadJob) -> None: # --- file download (range-aware, custom display name) -------------------------------------- -def _clean_basename(name: str) -> str: - # Emoji/symbol-free but keeps spaces + accents (a readable device filename). - return storage.display_filename(name) - - def _accessible_job(db: Session, user: User, job_id: int) -> DownloadJob: job = db.get(DownloadJob, job_id) if job is None: @@ -547,16 +538,13 @@ def download_file( asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None if asset is None or asset.status != "ready" or not asset.rel_path: raise HTTPException(status_code=409, detail="This download isn't ready.") - path = storage.abs_path(settings.download_root, asset.rel_path).resolve() - root = Path(settings.download_root).resolve() - if root not in path.parents or not path.exists(): + path = storage.safe_abs_path(settings.download_root, asset.rel_path) + if path is None: raise HTTPException(status_code=404, detail="File is no longer available.") - ext = asset.container or path.suffix.lstrip(".") - base = _clean_basename(job.display_name or asset.title or asset.source_ref) - if base.lower().endswith(f".{ext.lower()}"): - base = base[: -(len(ext) + 1)] - filename = f"{base}.{ext}" + filename = storage.download_filename( + job.display_name or asset.title or asset.source_ref, asset.container, path + ) asset.last_access_at = datetime.now(timezone.utc) db.commit() @@ -601,9 +589,8 @@ def poster_image( asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None if asset is None or not asset.poster_path: raise HTTPException(status_code=404, detail="No poster.") - path = storage.abs_path(settings.download_root, asset.poster_path).resolve() - root = Path(settings.download_root).resolve() - if root not in path.parents or not path.exists(): + path = storage.safe_abs_path(settings.download_root, asset.poster_path) + if path is None: raise HTTPException(status_code=404, detail="No poster.") return FileResponse(path, media_type="image/jpeg") @@ -616,9 +603,8 @@ def storyboard_image( asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None if asset is None or not asset.storyboard_path: raise HTTPException(status_code=404, detail="No filmstrip.") - path = storage.abs_path(settings.download_root, asset.storyboard_path).resolve() - root = Path(settings.download_root).resolve() - if root not in path.parents or not path.exists(): + path = storage.safe_abs_path(settings.download_root, asset.storyboard_path) + if path is None: raise HTTPException(status_code=404, detail="No filmstrip.") return FileResponse(path, media_type="image/jpeg") @@ -675,7 +661,7 @@ def unshare_download( user: User = Depends(require_human), db: Session = Depends(get_db), ) -> dict: - job = _own_job(db, user, job_id) + _own_job(db, user, job_id) # ownership guard (404s if not the caller's job) recipient = db.execute( select(User).where(func.lower(User.email) == email.strip().lower()) ).scalar_one_or_none() @@ -766,7 +752,7 @@ def list_links( .scalars() .all() ) - return [linksmod.owner_view(l) for l in rows] + return [linksmod.owner_view(row) for row in rows] @router.post("/{job_id}/links") @@ -876,7 +862,7 @@ def admin_storage( "ready_files": ready[0], "total_bytes": int(ready[1]), "total_assets": total_assets, - "total_cap_bytes": settings.download_total_max_bytes, + "total_cap_bytes": sysconfig.get_int(db, "download_total_max_bytes"), "per_user": [ {"user_id": uid, "email": emails.get(uid), "footprint_bytes": quota.footprint(db, uid)} for uid, _ in per_user diff --git a/backend/app/routes/feed.py b/backend/app/routes/feed.py index 7b76021..54f6b49 100644 --- a/backend/app/routes/feed.py +++ b/backend/app/routes/feed.py @@ -14,6 +14,7 @@ from app import quota from app.auth import current_user from app.db import get_db from app.models import ( + LIVE_OR_UPCOMING, BlockedChannel, Channel, ChannelTag, @@ -34,7 +35,6 @@ router = APIRouter(prefix="/api", tags=["feed"]) # "saved" used to be a status; it's now membership in the built-in Watch later playlist. VALID_STATES = {"new", "watched", "hidden"} -HIDDEN_LIVE = ("live", "upcoming") # Resume-position thresholds (mirror the client): positions below this are "didn't # really start" and within this of the end are "basically finished" — neither is worth @@ -132,7 +132,7 @@ def _filtered_query( exclude_tag_category: str | None = None, ) -> tuple[Select, object]: """Build the feed query (joins + all WHERE filters), shared by /feed and /feed/count. - Returns the column-bearing select plus the watch-status expression for sorting. + Returns the column-bearing select plus the priority/relevance rank expression for sorting. `scope="my"` (default) restricts the feed to the user's own non-hidden subscriptions. `scope="all"` shows every video in the shared catalog (any user's ingested channels); @@ -313,12 +313,12 @@ def _filtered_query( type_clauses = [] if show_normal: type_clauses.append( - and_(Video.is_short.is_(False), Video.live_status.notin_(HIDDEN_LIVE)) + and_(Video.is_short.is_(False), Video.live_status.notin_(LIVE_OR_UPCOMING)) ) if include_shorts: type_clauses.append(Video.is_short.is_(True)) if include_live: - type_clauses.append(Video.live_status.in_(HIDDEN_LIVE)) + type_clauses.append(Video.live_status.in_(LIVE_OR_UPCOMING)) query = query.where(or_(*type_clauses) if type_clauses else false()) if show == "unwatched": @@ -335,7 +335,7 @@ def _filtered_query( else: # all query = query.where(status_expr != "hidden") - return query, status_expr, rank_expr + return query, rank_expr def _to_tsquery_str(q: str) -> str | None: @@ -492,7 +492,7 @@ def get_feed( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - query, _status, rank_expr = _filtered_query(db, user, **params) + query, rank_expr = _filtered_query(db, user, **params) keys = _sort_keys(sort, seed, rank_expr) # Expose each key column on the row so we can build the next cursor from the last item. @@ -522,7 +522,7 @@ def get_feed_count( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - query, _status, _rank = _filtered_query(db, user, **params) + query, _rank = _filtered_query(db, user, **params) total = db.scalar(select(func.count()).select_from(query.subquery())) return {"count": total or 0} @@ -550,7 +550,7 @@ def get_facets( # keeps them applied, so each remaining chip narrows to channels that ALSO have all # already-selected topics — and tags that can't co-occur drop to zero (hidden). conjunctive = category == "topic" and params.get("tag_mode") == "and" - base, _status, _rank = _filtered_query( + base, _rank = _filtered_query( db, user, **{**params, "exclude_tag_category": None if conjunctive else category}, diff --git a/backend/app/routes/me.py b/backend/app/routes/me.py index caac1c5..d21e9db 100644 --- a/backend/app/routes/me.py +++ b/backend/app/routes/me.py @@ -63,6 +63,10 @@ def switch_account( raise HTTPException(status_code=404, detail="That account no longer exists.") if not is_allowed(db, u.email): raise HTTPException(status_code=403, detail="That account no longer has access.") + if u.is_suspended: + # Don't make a suspended account the active one — the next request's current_user would + # see the suspension and clear the WHOLE wallet session, logging out every account here. + raise HTTPException(status_code=403, detail="That account is suspended.") request.session["user_id"] = target return {"ok": True, "user_id": target} diff --git a/backend/app/routes/playlists.py b/backend/app/routes/playlists.py index a3ebf8a..e81a2c9 100644 --- a/backend/app/routes/playlists.py +++ b/backend/app/routes/playlists.py @@ -74,6 +74,23 @@ def _add_item(db: Session, pl: Playlist, video_id: str, *, mark_dirty: bool) -> return True +def _remove_item(db: Session, pl: Playlist, video_id: str, *, mark_dirty: bool) -> bool: + """Remove a video from a playlist if present (idempotent). Returns whether it was removed. + `mark_dirty` recomputes the YouTube-sync dirty flag (skip for Watch later).""" + item = db.scalar( + select(PlaylistItem).where( + PlaylistItem.playlist_id == pl.id, PlaylistItem.video_id == video_id + ) + ) + if item is None: + return False + db.delete(item) + if mark_dirty: + recompute_dirty(db, pl) + db.commit() + return True + + def _summary( db: Session, pl: Playlist, @@ -239,14 +256,7 @@ def remove_watch_later( ) ) if pl is not None: - item = db.scalar( - select(PlaylistItem).where( - PlaylistItem.playlist_id == pl.id, PlaylistItem.video_id == video_id - ) - ) - if item is not None: - db.delete(item) - db.commit() + _remove_item(db, pl, video_id, mark_dirty=False) return {"saved": False} @@ -452,15 +462,7 @@ def remove_item( db: Session = Depends(get_db), ) -> dict: pl = _own_playlist(db, user, playlist_id) - item = db.scalar( - select(PlaylistItem).where( - PlaylistItem.playlist_id == pl.id, PlaylistItem.video_id == video_id - ) - ) - if item is not None: - db.delete(item) - recompute_dirty(db, pl) - db.commit() + _remove_item(db, pl, video_id, mark_dirty=True) return {"playlist_id": pl.id, "video_id": video_id} @@ -480,11 +482,20 @@ def reorder_items( ).scalars() } pos = 0 + seen: set[str] = set() for vid in order: it = items.get(vid) if it is not None: it.position = pos pos += 1 + seen.add(vid) + # Any items the payload omitted (e.g. added concurrently in another tab) keep their relative + # order but get fresh trailing positions, so no two items collide on the same position. + for it in sorted( + (it for vid, it in items.items() if vid not in seen), key=lambda i: i.position + ): + it.position = pos + pos += 1 recompute_dirty(db, pl) db.commit() return {"playlist_id": pl.id, "count": pos} diff --git a/backend/app/routes/plex.py b/backend/app/routes/plex.py index af1a177..61cdd5a 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 @@ -197,12 +198,6 @@ def watch_import_now( # --- Read: libraries / browse / show / image -------------------------------------------------- -def _enabled() -> None: - """Guard read endpoints when the module is off (avoids leaking a stale mirror).""" - # Read endpoints stay usable as long as a mirror exists; the toggle only gates sync + UI. - return None - - @router.get("/libraries") def list_libraries(user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict: """The mirrored libraries the user can browse (drives the Plex scope selector).""" @@ -392,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", @@ -401,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: @@ -422,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"] @@ -473,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) @@ -582,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: @@ -617,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 @@ -720,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])], @@ -1354,11 +1346,15 @@ def _vtt_ts_to_s(ts: str) -> float: def _vtt_s_to_ts(x: float) -> str: x = max(0.0, x) - h = int(x // 3600); x -= h * 3600 - m = int(x // 60); x -= m * 60 - s = int(x); ms = int(round((x - s) * 1000)) + h = int(x // 3600) + x -= h * 3600 + m = int(x // 60) + x -= m * 60 + s = int(x) + ms = int(round((x - s) * 1000)) if ms >= 1000: - s += 1; ms -= 1000 + s += 1 + ms -= 1000 return f"{h:02d}:{m:02d}:{s:02d}.{ms:03d}" @@ -1834,8 +1830,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), ) diff --git a/backend/app/routes/public.py b/backend/app/routes/public.py index b36f1e9..7c5b9ea 100644 --- a/backend/app/routes/public.py +++ b/backend/app/routes/public.py @@ -5,7 +5,6 @@ credential (a capability URL). They only ever expose one file's stream + minimal the app or any account data. Password-protected links exchange the password once at `/unlock` for a short-lived signed grant that the media URL carries (a `