From 5f09259bbfcc98cc4e19750dc468423c0e317cee Mon Sep 17 00:00:00 2001 From: npeter83 Date: Mon, 17 Aug 2026 17:42:01 +0200 Subject: [PATCH] refactor(feed): extract read-model to app.feedquery + admin_user sweep MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit B-core-3 (2026-08-17 review): decouple two things from router modules. 1. Feed read-model → new app/feedquery.py: the pure query-building and row-serialization layer (feed_columns, serialize, _filtered_query, to_tsquery_str, and the keyset-pagination helpers) moves out of routes/feed.py, which keeps only the route handlers, _feed_params, and the request models. The three external consumers (playlists, search, plex/browse) now import from app.feedquery instead of reaching into the feed router. 2. admin_user sweep: 7 modules imported admin_user via app.routes.admin (an incidental re-export of app.auth.admin_user). Repoint them all to the definition in app.auth (audit, channels, config, quota, scheduler, plex/admin, plex/browse); routes.admin keeps its own import for its use. Pure code motion — no behaviour change. Full DB-lane suite 287 passed / 2 skipped; ruff clean. --- backend/app/feedquery.py | 437 ++++++++++++++++++++++++++++++ backend/app/routes/audit.py | 2 +- backend/app/routes/channels.py | 3 +- backend/app/routes/config.py | 2 +- backend/app/routes/feed.py | 430 +---------------------------- backend/app/routes/playlists.py | 2 +- backend/app/routes/plex/admin.py | 3 +- backend/app/routes/plex/browse.py | 5 +- backend/app/routes/quota.py | 3 +- backend/app/routes/scheduler.py | 2 +- backend/app/routes/search.py | 2 +- 11 files changed, 459 insertions(+), 432 deletions(-) create mode 100644 backend/app/feedquery.py diff --git a/backend/app/feedquery.py b/backend/app/feedquery.py new file mode 100644 index 00000000..8698b6ef --- /dev/null +++ b/backend/app/feedquery.py @@ -0,0 +1,437 @@ +"""Feed read-model: the pure query-building and row-serialization layer behind the feed. + +Extracted from `routes/feed.py` so the read-model has one home instead of living inside the +router module — the feed route handlers, the playlist-detail query, and the Plex search box all +share these helpers (the projection, the filter builder, `to_tsquery_str`, and keyset paging). +Nothing here defines a route or reaches for request state; it's all `(db, user, params) -> query`. +""" + +import base64 +import binascii +import json +import re +from datetime import date, datetime, timedelta, timezone + +from fastapi import HTTPException +from sqlalchemy import BigInteger, Select, and_, cast, false, func, or_, select +from sqlalchemy.orm import Session, aliased + +from app.models import ( + LIVE_OR_UPCOMING, + BlockedChannel, + Channel, + ChannelTag, + ExploredChannel, + Playlist, + PlaylistItem, + SearchFind, + Subscription, + Tag, + User, + Video, + VideoState, +) +from app.utils import iso + + +def _channel_url(channel_id: str, handle: str | None) -> str: + if handle and handle.startswith("@"): + return f"https://www.youtube.com/{handle}" + return f"https://www.youtube.com/channel/{channel_id}" + + +def _saved_expr(user: User): + """A correlated EXISTS labelled `saved`: is this video in the user's Watch later list? + (Watch later is the built-in playlist that replaced the old per-video 'saved' status.)""" + return ( + select(1) + .select_from(PlaylistItem) + .join(Playlist, Playlist.id == PlaylistItem.playlist_id) + .where( + Playlist.user_id == user.id, + Playlist.kind == "watch_later", + PlaylistItem.video_id == Video.id, + ) + .exists() + .label("saved") + ) + + +def feed_columns(user: User, state) -> list: + """The shared feed-row projection (joined Channel fields + per-user watch state + + `saved`), consumed by both the feed query and the playlist-detail query so a new column + is added in exactly one place. `state` is the caller's aliased VideoState (outer-joined).""" + return [ + Video.id, + Video.title, + Video.channel_id, + Channel.title.label("channel_title"), + Channel.thumbnail_url.label("channel_thumbnail"), + Channel.handle.label("channel_handle"), + Video.published_at, + Video.thumbnail_url, + Video.duration_seconds, + Video.view_count, + Video.is_short, + Video.live_status, + func.coalesce(state.status, "new").label("status"), + func.coalesce(state.position_seconds, 0).label("position_seconds"), + _saved_expr(user), + ] + + +def serialize(row) -> dict: + return { + "id": row.id, + "title": row.title, + "channel_id": row.channel_id, + "channel_title": row.channel_title, + "channel_thumbnail": row.channel_thumbnail, + "channel_url": _channel_url(row.channel_id, row.channel_handle), + "published_at": iso(row.published_at), + "thumbnail_url": row.thumbnail_url, + "duration_seconds": row.duration_seconds, + "view_count": row.view_count, + "is_short": row.is_short, + "live_status": row.live_status, + "status": row.status or "new", + "position_seconds": row.position_seconds or 0, + "saved": bool(getattr(row, "saved", False)), + "watch_url": f"https://www.youtube.com/watch?v={row.id}", + } + + +def _filtered_query( + db: Session, + user: User, + *, + tags: list[int], + tag_mode: str, + channel_ids: list[str], + q: str | None, + min_duration: int | None, + max_duration: int | None, + max_age_days: int | None, + published_after: date | None, + published_before: date | None, + show_normal: bool, + include_shorts: bool, + include_live: bool, + show: str, + scope: str = "my", + library_source: str = "organic", + 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 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); + the subscription is then only LEFT-joined so per-channel priority still resolves for + channels the user happens to be subscribed to. Per-user watch state stays private in + either mode via the VideoState outer join.""" + state = aliased(VideoState) + status_expr = func.coalesce(state.status, "new") + position_expr = func.coalesce(state.position_seconds, 0) + + query = select(*feed_columns(user, state)).join( + Channel, Channel.id == Video.channel_id + ) + + if scope == "all": + # Whole shared catalog; subscription is optional (only for priority sort). + query = query.outerjoin( + Subscription, + and_( + Subscription.channel_id == Video.channel_id, + Subscription.user_id == user.id, + ), + ) + else: + # "Mine": the user's own videos = their non-hidden subscriptions OR videos they + # surfaced via their own live YouTube search. Both are LEFT-joined so the Source + # filter below can include either or both, and the subscription row stays available + # for the priority sort. + query = query.outerjoin( + Subscription, + and_( + Subscription.channel_id == Video.channel_id, + Subscription.user_id == user.id, + Subscription.hidden.is_(False), + ), + ).outerjoin( + SearchFind, + and_(SearchFind.video_id == Video.id, SearchFind.user_id == user.id), + ) + + query = query.outerjoin( + state, and_(state.video_id == Video.id, state.user_id == user.id) + ) + + # Hide videos the maintenance job has flagged as currently unplayable (deleted/private/ + # abandoned). They're either deleted after a grace period or unhidden if they recover; + # either way they shouldn't show in any feed view meanwhile. + query = query.where(Video.unavailable_since.is_(None)) + + # Explored videos are PER-EXPLORER private: a via_explore video (ingested when someone + # opened an un-subscribed channel's page) is visible only to users who explored or + # subscribe to its channel, so one user's curiosity never leaks into everyone else's + # catalog. Applied in BOTH scopes; the channel page reaches its ephemeral videos by + # asking for scope=all + channel_ids (the channel is in this accessible set). + accessible_explore = ( + select(Subscription.channel_id).where(Subscription.user_id == user.id) + ).union( + select(ExploredChannel.channel_id).where(ExploredChannel.user_id == user.id) + ) + query = query.where( + or_( + Video.via_explore.is_(False), + Video.channel_id.in_(accessible_explore), + ) + ) + + # Per-user channel blocklist: never show this user videos from a channel they blocked + # (applies to feed, Library, explore, and the channel page alike). + query = query.where( + Video.channel_id.notin_( + select(BlockedChannel.channel_id).where(BlockedChannel.user_id == user.id) + ) + ) + + # Source filter. In the shared Library ("all" scope) it uses the GLOBAL via_search flag + # (search-discovered by anyone). In "my" scope it's the user's OWN provenance: + # "organic" = your subscriptions, "search" = videos you found via your own search, + # "all" = both. Default "organic" keeps the Mine feed = subscriptions only. + if scope == "all": + if library_source == "organic": + query = query.where(Video.via_search.is_(False)) + elif library_source == "search": + query = query.where(Video.via_search.is_(True)) + # "all" → both organic and search-discovered videos + else: + subscribed = Subscription.channel_id.isnot(None) + searched = SearchFind.video_id.isnot(None) + if library_source == "search": + query = query.where(searched) + elif library_source == "all": + query = query.where(or_(subscribed, searched)) + else: # "organic" (default): subscriptions only + query = query.where(subscribed) + + if channel_ids: + # OR across the picked channels; every other filter still applies on top. + query = query.where(Video.channel_id.in_(channel_ids)) + if min_duration is not None: + query = query.where(Video.duration_seconds >= min_duration) + if max_duration is not None: + query = query.where(Video.duration_seconds <= max_duration) + if max_age_days is not None: + cutoff = datetime.now(timezone.utc).timestamp() - max_age_days * 86400 + query = query.where( + Video.published_at >= datetime.fromtimestamp(cutoff, tz=timezone.utc) + ) + if published_after is not None: + start = datetime.combine( + published_after, datetime.min.time(), tzinfo=timezone.utc + ) + query = query.where(Video.published_at >= start) + if published_before is not None: + end = datetime.combine( + published_before, datetime.min.time(), tzinfo=timezone.utc + ) + timedelta(days=1) + query = query.where(Video.published_at < end) + # Full-text relevance search over the weighted search document (title > creator keywords + + # the queries that surfaced the video > description; see Video.search_vector). YouTube-like + # "fuzzy" matching: word-order-independent, multi-word AND, prefix on the word being typed, + # accent-insensitive (unaccent_simple). The channel name still matches as a substring so + # channel searches work. `rank_expr` (weight-aware ts_rank) drives the "relevance" sort. + rank_expr = None + if q: + ts_str = to_tsquery_str(q) + if ts_str: + tsq = func.to_tsquery("public.unaccent_simple", ts_str) + query = query.where( + or_( + Video.search_vector.op("@@")(tsq), + func.unaccent(Channel.title).ilike(func.unaccent(f"%{q}%")), + ) + ) + # Scale ts_rank (a float4) to an integer score so the keyset cursor compares it + # EXACTLY — a raw float4 vs the float8 cursor value mismatches on round-trip and + # breaks paging (the same page repeats). 1e6 granularity is ample; ties break on + # published_at then id. + rank_expr = cast(func.ts_rank(Video.search_vector, tsq) * 1000000, BigInteger) + else: + # No usable tokens (e.g. only punctuation): fall back to a plain substring match. + like = func.unaccent(f"%{q}%") + query = query.where( + or_( + func.unaccent(Video.title).ilike(like), + func.unaccent(Channel.title).ilike(like), + ) + ) + + if tags: + # AND across tag categories (e.g. language AND topic narrows), OR within a + # category; the any/all toggle controls multiple topic tags. + cat_rows = db.execute(select(Tag.id, Tag.category).where(Tag.id.in_(tags))).all() + by_category: dict[str, list[int]] = {} + for tag_id, category in cat_rows: + by_category.setdefault(category, []).append(tag_id) + # Facet counting drops the category being counted so its own chips don't zero each + # other out (standard drill-down faceting: a category's count ignores its own + # selections but still honours the other categories' filters). + if exclude_tag_category: + by_category.pop(exclude_tag_category, None) + visible = or_(ChannelTag.user_id.is_(None), ChannelTag.user_id == user.id) + for category, ids in by_category.items(): + if category == "topic" and tag_mode == "and" and len(ids) > 1: + sub = ( + select(ChannelTag.channel_id) + .where(ChannelTag.tag_id.in_(ids), visible) + .group_by(ChannelTag.channel_id) + .having(func.count(func.distinct(ChannelTag.tag_id)) == len(set(ids))) + ) + else: + sub = select(ChannelTag.channel_id).where( + ChannelTag.tag_id.in_(ids), visible + ) + query = query.where(Video.channel_id.in_(sub)) + + # Content type: Normal / Shorts / Live·Upcoming as a union of enabled types. + # Explicit Watched/Saved/Hidden views show every type so nothing goes missing. + explicit_view = show in ("watched", "hidden") + if not explicit_view: + type_clauses = [] + if show_normal: + type_clauses.append( + 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_(LIVE_OR_UPCOMING)) + query = query.where(or_(*type_clauses) if type_clauses else false()) + + if show == "unwatched": + query = query.where(status_expr.notin_(("watched", "hidden"))) + elif show == "in_progress": + # Started but not finished: a stored resume position, not yet watched/hidden. + query = query.where( + and_(position_expr > 0, status_expr.notin_(("watched", "hidden"))) + ) + elif show == "watched": + query = query.where(status_expr == "watched") + elif show == "hidden": + query = query.where(status_expr == "hidden") + else: # all + query = query.where(status_expr != "hidden") + + return query, rank_expr + + +def to_tsquery_str(q: str) -> str | None: + """Turn free-text into a to_tsquery() string: word runs ANDed together, the LAST word + prefix-matched (`:*`) for as-you-type search — e.g. "ti amo magyar" → "ti & amo & magyar:*". + Only `\\w` runs containing an alphanumeric are kept, so no tsquery operators can leak in + (the string is built from sanitised tokens, never the raw query).""" + tokens = [t for t in re.findall(r"\w+", q, flags=re.UNICODE) if any(c.isalnum() for c in t)] + if not tokens: + return None + tokens[-1] = tokens[-1] + ":*" + return " & ".join(tokens) + + +# --- Keyset (cursor) pagination --------------------------------------------- +# +# Each sort is described as an ordered list of key columns (expression + direction +# + whether it can be NULL + a JSON round-trip kind), always with Video.id as the +# final, unique tiebreaker. This drives BOTH the ORDER BY and the keyset WHERE so +# they can never drift apart — the prerequisite for correct cursor paging. Keyset +# replaces OFFSET/LIMIT: deep scrolling no longer scans-and-discards skipped rows, +# and the page boundary is stable even as the scheduler ingests new videos mid-scroll +# (OFFSET would shift and duplicate/skip rows). +def _sort_keys(sort: str, seed: int, rank_expr=None) -> list[dict]: + pub = {"expr": Video.published_at, "asc": False, "nullable": True, "kind": "dt"} + table: dict[str, list[dict]] = { + "newest": [pub], + "oldest": [{**pub, "asc": True}], + "views": [{"expr": Video.view_count, "asc": False, "nullable": True, "kind": "num"}], + "views_asc": [{"expr": Video.view_count, "asc": True, "nullable": True, "kind": "num"}], + "duration_desc": [{"expr": Video.duration_seconds, "asc": False, "nullable": True, "kind": "num"}], + "duration_asc": [{"expr": Video.duration_seconds, "asc": True, "nullable": True, "kind": "num"}], + "title": [{"expr": func.lower(Video.title), "asc": True, "nullable": True, "kind": "str"}], + "title_desc": [{"expr": func.lower(Video.title), "asc": False, "nullable": True, "kind": "str"}], + "subscribers": [{"expr": Channel.subscriber_count, "asc": False, "nullable": True, "kind": "num"}], + "subscribers_asc": [{"expr": Channel.subscriber_count, "asc": True, "nullable": True, "kind": "num"}], + # Your per-channel priority (set in the channel manager), newest first within a tier. + # coalesce keeps it null-safe in "all" scope where unsubscribed channels have no row. + "priority": [{"expr": func.coalesce(Subscription.priority, 0), "asc": False, "nullable": False, "kind": "num"}, pub], + "priority_asc": [{"expr": func.coalesce(Subscription.priority, 0), "asc": True, "nullable": False, "kind": "num"}, pub], + # Deterministic per-seed shuffle: the same seed yields the same total order, so + # the cursor stays valid across pages of one shuffled run. + "shuffle": [{"expr": func.md5(func.concat(Video.id, str(seed))), "asc": True, "nullable": False, "kind": "str"}], + } + # Relevance (search-result ranking) only exists when there's a query to rank against; + # without `rank_expr` it falls through to newest. Ties break by newest, then id. + if rank_expr is not None: + table["relevance"] = [ + {"expr": rank_expr, "asc": False, "nullable": False, "kind": "num"}, + pub, + ] + return table.get(sort) or table["newest"] + + +def _order_by(keys: list[dict]): + cols = [] + for k in keys: + c = k["expr"].asc() if k["asc"] else k["expr"].desc() + cols.append(c.nulls_last() if k["nullable"] else c) + cols.append(Video.id.asc()) # unique final tiebreaker + return cols + + +def _keyset_predicate(keys: list[dict], cursor_id: str): + """WHERE clause selecting rows strictly AFTER the cursor in the sort order. + + Lexicographic over the key columns then id, NULL-aware for `nulls_last`: a NULL + sorts after every non-NULL value, so a cursor sitting on a non-NULL value must + also pull in the NULL group, and equality treats NULL == NULL.""" + def eq(k): + return k["expr"].is_(None) if k["value"] is None else k["expr"] == k["value"] + + def after(k): + if k["value"] is None: + return false() # nulls are last; nothing sorts strictly after them + cmp = k["expr"] > k["value"] if k["asc"] else k["expr"] < k["value"] + return or_(cmp, k["expr"].is_(None)) if k["nullable"] else cmp + + branches = [] + for j in range(len(keys)): + branches.append(and_(*[eq(keys[i]) for i in range(j)], after(keys[j]))) + branches.append(and_(*[eq(k) for k in keys], Video.id > cursor_id)) + return or_(*branches) + + +def _encode_cursor(keys: list[dict], row) -> str: + vals = [] + for i, k in enumerate(keys): + v = getattr(row, f"_k{i}") + vals.append(v.isoformat() if (v is not None and k["kind"] == "dt") else v) + payload = json.dumps({"k": vals, "id": row.id}) + return base64.urlsafe_b64encode(payload.encode()).decode() + + +def _decode_cursor(keys: list[dict], cursor: str) -> tuple[list, str]: + try: + payload = json.loads(base64.urlsafe_b64decode(cursor.encode())) + raw = payload["k"] + if len(raw) != len(keys): + raise ValueError("key arity mismatch") + vals = [ + datetime.fromisoformat(v) if (v is not None and k["kind"] == "dt") else v + for k, v in zip(keys, raw) + ] + return vals, payload["id"] + except (binascii.Error, json.JSONDecodeError, KeyError, ValueError, TypeError): + raise HTTPException(status_code=400, detail="Invalid feed cursor") diff --git a/backend/app/routes/audit.py b/backend/app/routes/audit.py index 9e7a4ef5..493f68db 100644 --- a/backend/app/routes/audit.py +++ b/backend/app/routes/audit.py @@ -9,9 +9,9 @@ from sqlalchemy.orm import Session from app import audit from app.audit import AuditAction +from app.auth import admin_user from app.db import get_db from app.models import AuditLog, User -from app.routes.admin import admin_user from app.utils import iso router = APIRouter(prefix="/api/admin/audit", tags=["admin"]) diff --git a/backend/app/routes/channels.py b/backend/app/routes/channels.py index 7b81fb49..fa4148e5 100644 --- a/backend/app/routes/channels.py +++ b/backend/app/routes/channels.py @@ -10,7 +10,7 @@ from sqlalchemy import func, select, update from sqlalchemy.orm import Session from app import quota -from app.auth import current_user, require_human, require_write_scope +from app.auth import admin_user, current_user, require_human, require_write_scope from app.db import get_db from app.models import ( LIVE_OR_UPCOMING, @@ -26,7 +26,6 @@ from app.models import ( Video, ) from app.routes._common import owned_or_404 -from app.routes.admin import admin_user from app.sync.explore import explore_ingest_page from app.sync.runner import backfill_channel_now, run_recent_backfill from app.sync.subscriptions import apply_channel_details diff --git a/backend/app/routes/config.py b/backend/app/routes/config.py index 95131624..e2763337 100644 --- a/backend/app/routes/config.py +++ b/backend/app/routes/config.py @@ -11,9 +11,9 @@ from app import audit from app import email as email_mod from app import sysconfig from app.audit import AuditAction +from app.auth import admin_user from app.db import get_db from app.models import User -from app.routes.admin import admin_user router = APIRouter(prefix="/api/admin/config", tags=["admin"]) diff --git a/backend/app/routes/feed.py b/backend/app/routes/feed.py index 7e555f11..c76790f2 100644 --- a/backend/app/routes/feed.py +++ b/backend/app/routes/feed.py @@ -1,29 +1,26 @@ -import base64 -import binascii -import json -import re -from datetime import date, datetime, timedelta, timezone +from datetime import date, datetime, timezone from fastapi import APIRouter, Depends, HTTPException, Query from pydantic import BaseModel -from sqlalchemy import BigInteger, Select, and_, cast, false, func, or_, select +from sqlalchemy import and_, func, or_, select from sqlalchemy.dialects.postgresql import insert as pg_insert -from sqlalchemy.orm import Session, aliased +from sqlalchemy.orm import Session from sqlalchemy.orm.exc import StaleDataError from app import quota from app.auth import current_user from app.db import get_db +from app.feedquery import ( + _decode_cursor, + _encode_cursor, + _filtered_query, + _keyset_predicate, + _order_by, + _sort_keys, + serialize, +) from app.models import ( - LIVE_OR_UPCOMING, - BlockedChannel, - Channel, ChannelTag, - ExploredChannel, - Playlist, - PlaylistItem, - SearchFind, - Subscription, Tag, User, Video, @@ -47,314 +44,6 @@ PROGRESS_MIN_SECONDS = 5 FINISH_MARGIN_SECONDS = 10 -def _channel_url(channel_id: str, handle: str | None) -> str: - if handle and handle.startswith("@"): - return f"https://www.youtube.com/{handle}" - return f"https://www.youtube.com/channel/{channel_id}" - - -def _saved_expr(user: User): - """A correlated EXISTS labelled `saved`: is this video in the user's Watch later list? - (Watch later is the built-in playlist that replaced the old per-video 'saved' status.)""" - return ( - select(1) - .select_from(PlaylistItem) - .join(Playlist, Playlist.id == PlaylistItem.playlist_id) - .where( - Playlist.user_id == user.id, - Playlist.kind == "watch_later", - PlaylistItem.video_id == Video.id, - ) - .exists() - .label("saved") - ) - - -def feed_columns(user: User, state) -> list: - """The shared feed-row projection (joined Channel fields + per-user watch state + - `saved`), consumed by both the feed query and the playlist-detail query so a new column - is added in exactly one place. `state` is the caller's aliased VideoState (outer-joined).""" - return [ - Video.id, - Video.title, - Video.channel_id, - Channel.title.label("channel_title"), - Channel.thumbnail_url.label("channel_thumbnail"), - Channel.handle.label("channel_handle"), - Video.published_at, - Video.thumbnail_url, - Video.duration_seconds, - Video.view_count, - Video.is_short, - Video.live_status, - func.coalesce(state.status, "new").label("status"), - func.coalesce(state.position_seconds, 0).label("position_seconds"), - _saved_expr(user), - ] - - -def serialize(row) -> dict: - return { - "id": row.id, - "title": row.title, - "channel_id": row.channel_id, - "channel_title": row.channel_title, - "channel_thumbnail": row.channel_thumbnail, - "channel_url": _channel_url(row.channel_id, row.channel_handle), - "published_at": iso(row.published_at), - "thumbnail_url": row.thumbnail_url, - "duration_seconds": row.duration_seconds, - "view_count": row.view_count, - "is_short": row.is_short, - "live_status": row.live_status, - "status": row.status or "new", - "position_seconds": row.position_seconds or 0, - "saved": bool(getattr(row, "saved", False)), - "watch_url": f"https://www.youtube.com/watch?v={row.id}", - } - - -def _filtered_query( - db: Session, - user: User, - *, - tags: list[int], - tag_mode: str, - channel_ids: list[str], - q: str | None, - min_duration: int | None, - max_duration: int | None, - max_age_days: int | None, - published_after: date | None, - published_before: date | None, - show_normal: bool, - include_shorts: bool, - include_live: bool, - show: str, - scope: str = "my", - library_source: str = "organic", - 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 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); - the subscription is then only LEFT-joined so per-channel priority still resolves for - channels the user happens to be subscribed to. Per-user watch state stays private in - either mode via the VideoState outer join.""" - state = aliased(VideoState) - status_expr = func.coalesce(state.status, "new") - position_expr = func.coalesce(state.position_seconds, 0) - - query = select(*feed_columns(user, state)).join( - Channel, Channel.id == Video.channel_id - ) - - if scope == "all": - # Whole shared catalog; subscription is optional (only for priority sort). - query = query.outerjoin( - Subscription, - and_( - Subscription.channel_id == Video.channel_id, - Subscription.user_id == user.id, - ), - ) - else: - # "Mine": the user's own videos = their non-hidden subscriptions OR videos they - # surfaced via their own live YouTube search. Both are LEFT-joined so the Source - # filter below can include either or both, and the subscription row stays available - # for the priority sort. - query = query.outerjoin( - Subscription, - and_( - Subscription.channel_id == Video.channel_id, - Subscription.user_id == user.id, - Subscription.hidden.is_(False), - ), - ).outerjoin( - SearchFind, - and_(SearchFind.video_id == Video.id, SearchFind.user_id == user.id), - ) - - query = query.outerjoin( - state, and_(state.video_id == Video.id, state.user_id == user.id) - ) - - # Hide videos the maintenance job has flagged as currently unplayable (deleted/private/ - # abandoned). They're either deleted after a grace period or unhidden if they recover; - # either way they shouldn't show in any feed view meanwhile. - query = query.where(Video.unavailable_since.is_(None)) - - # Explored videos are PER-EXPLORER private: a via_explore video (ingested when someone - # opened an un-subscribed channel's page) is visible only to users who explored or - # subscribe to its channel, so one user's curiosity never leaks into everyone else's - # catalog. Applied in BOTH scopes; the channel page reaches its ephemeral videos by - # asking for scope=all + channel_ids (the channel is in this accessible set). - accessible_explore = ( - select(Subscription.channel_id).where(Subscription.user_id == user.id) - ).union( - select(ExploredChannel.channel_id).where(ExploredChannel.user_id == user.id) - ) - query = query.where( - or_( - Video.via_explore.is_(False), - Video.channel_id.in_(accessible_explore), - ) - ) - - # Per-user channel blocklist: never show this user videos from a channel they blocked - # (applies to feed, Library, explore, and the channel page alike). - query = query.where( - Video.channel_id.notin_( - select(BlockedChannel.channel_id).where(BlockedChannel.user_id == user.id) - ) - ) - - # Source filter. In the shared Library ("all" scope) it uses the GLOBAL via_search flag - # (search-discovered by anyone). In "my" scope it's the user's OWN provenance: - # "organic" = your subscriptions, "search" = videos you found via your own search, - # "all" = both. Default "organic" keeps the Mine feed = subscriptions only. - if scope == "all": - if library_source == "organic": - query = query.where(Video.via_search.is_(False)) - elif library_source == "search": - query = query.where(Video.via_search.is_(True)) - # "all" → both organic and search-discovered videos - else: - subscribed = Subscription.channel_id.isnot(None) - searched = SearchFind.video_id.isnot(None) - if library_source == "search": - query = query.where(searched) - elif library_source == "all": - query = query.where(or_(subscribed, searched)) - else: # "organic" (default): subscriptions only - query = query.where(subscribed) - - if channel_ids: - # OR across the picked channels; every other filter still applies on top. - query = query.where(Video.channel_id.in_(channel_ids)) - if min_duration is not None: - query = query.where(Video.duration_seconds >= min_duration) - if max_duration is not None: - query = query.where(Video.duration_seconds <= max_duration) - if max_age_days is not None: - cutoff = datetime.now(timezone.utc).timestamp() - max_age_days * 86400 - query = query.where( - Video.published_at >= datetime.fromtimestamp(cutoff, tz=timezone.utc) - ) - if published_after is not None: - start = datetime.combine( - published_after, datetime.min.time(), tzinfo=timezone.utc - ) - query = query.where(Video.published_at >= start) - if published_before is not None: - end = datetime.combine( - published_before, datetime.min.time(), tzinfo=timezone.utc - ) + timedelta(days=1) - query = query.where(Video.published_at < end) - # Full-text relevance search over the weighted search document (title > creator keywords + - # the queries that surfaced the video > description; see Video.search_vector). YouTube-like - # "fuzzy" matching: word-order-independent, multi-word AND, prefix on the word being typed, - # accent-insensitive (unaccent_simple). The channel name still matches as a substring so - # channel searches work. `rank_expr` (weight-aware ts_rank) drives the "relevance" sort. - rank_expr = None - if q: - ts_str = to_tsquery_str(q) - if ts_str: - tsq = func.to_tsquery("public.unaccent_simple", ts_str) - query = query.where( - or_( - Video.search_vector.op("@@")(tsq), - func.unaccent(Channel.title).ilike(func.unaccent(f"%{q}%")), - ) - ) - # Scale ts_rank (a float4) to an integer score so the keyset cursor compares it - # EXACTLY — a raw float4 vs the float8 cursor value mismatches on round-trip and - # breaks paging (the same page repeats). 1e6 granularity is ample; ties break on - # published_at then id. - rank_expr = cast(func.ts_rank(Video.search_vector, tsq) * 1000000, BigInteger) - else: - # No usable tokens (e.g. only punctuation): fall back to a plain substring match. - like = func.unaccent(f"%{q}%") - query = query.where( - or_( - func.unaccent(Video.title).ilike(like), - func.unaccent(Channel.title).ilike(like), - ) - ) - - if tags: - # AND across tag categories (e.g. language AND topic narrows), OR within a - # category; the any/all toggle controls multiple topic tags. - cat_rows = db.execute(select(Tag.id, Tag.category).where(Tag.id.in_(tags))).all() - by_category: dict[str, list[int]] = {} - for tag_id, category in cat_rows: - by_category.setdefault(category, []).append(tag_id) - # Facet counting drops the category being counted so its own chips don't zero each - # other out (standard drill-down faceting: a category's count ignores its own - # selections but still honours the other categories' filters). - if exclude_tag_category: - by_category.pop(exclude_tag_category, None) - visible = or_(ChannelTag.user_id.is_(None), ChannelTag.user_id == user.id) - for category, ids in by_category.items(): - if category == "topic" and tag_mode == "and" and len(ids) > 1: - sub = ( - select(ChannelTag.channel_id) - .where(ChannelTag.tag_id.in_(ids), visible) - .group_by(ChannelTag.channel_id) - .having(func.count(func.distinct(ChannelTag.tag_id)) == len(set(ids))) - ) - else: - sub = select(ChannelTag.channel_id).where( - ChannelTag.tag_id.in_(ids), visible - ) - query = query.where(Video.channel_id.in_(sub)) - - # Content type: Normal / Shorts / Live·Upcoming as a union of enabled types. - # Explicit Watched/Saved/Hidden views show every type so nothing goes missing. - explicit_view = show in ("watched", "hidden") - if not explicit_view: - type_clauses = [] - if show_normal: - type_clauses.append( - 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_(LIVE_OR_UPCOMING)) - query = query.where(or_(*type_clauses) if type_clauses else false()) - - if show == "unwatched": - query = query.where(status_expr.notin_(("watched", "hidden"))) - elif show == "in_progress": - # Started but not finished: a stored resume position, not yet watched/hidden. - query = query.where( - and_(position_expr > 0, status_expr.notin_(("watched", "hidden"))) - ) - elif show == "watched": - query = query.where(status_expr == "watched") - elif show == "hidden": - query = query.where(status_expr == "hidden") - else: # all - query = query.where(status_expr != "hidden") - - return query, rank_expr - - -def to_tsquery_str(q: str) -> str | None: - """Turn free-text into a to_tsquery() string: word runs ANDed together, the LAST word - prefix-matched (`:*`) for as-you-type search — e.g. "ti amo magyar" → "ti & amo & magyar:*". - Only `\\w` runs containing an alphanumeric are kept, so no tsquery operators can leak in - (the string is built from sanitised tokens, never the raw query).""" - tokens = [t for t in re.findall(r"\w+", q, flags=re.UNICODE) if any(c.isalnum() for c in t)] - if not tokens: - return None - tokens[-1] = tokens[-1] + ":*" - return " & ".join(tokens) - - # Shared query parameters for /feed and /feed/count. def _feed_params( tags: list[int] = Query(default=[]), @@ -392,101 +81,6 @@ def _feed_params( } -# --- Keyset (cursor) pagination --------------------------------------------- -# -# Each sort is described as an ordered list of key columns (expression + direction -# + whether it can be NULL + a JSON round-trip kind), always with Video.id as the -# final, unique tiebreaker. This drives BOTH the ORDER BY and the keyset WHERE so -# they can never drift apart — the prerequisite for correct cursor paging. Keyset -# replaces OFFSET/LIMIT: deep scrolling no longer scans-and-discards skipped rows, -# and the page boundary is stable even as the scheduler ingests new videos mid-scroll -# (OFFSET would shift and duplicate/skip rows). -def _sort_keys(sort: str, seed: int, rank_expr=None) -> list[dict]: - pub = {"expr": Video.published_at, "asc": False, "nullable": True, "kind": "dt"} - table: dict[str, list[dict]] = { - "newest": [pub], - "oldest": [{**pub, "asc": True}], - "views": [{"expr": Video.view_count, "asc": False, "nullable": True, "kind": "num"}], - "views_asc": [{"expr": Video.view_count, "asc": True, "nullable": True, "kind": "num"}], - "duration_desc": [{"expr": Video.duration_seconds, "asc": False, "nullable": True, "kind": "num"}], - "duration_asc": [{"expr": Video.duration_seconds, "asc": True, "nullable": True, "kind": "num"}], - "title": [{"expr": func.lower(Video.title), "asc": True, "nullable": True, "kind": "str"}], - "title_desc": [{"expr": func.lower(Video.title), "asc": False, "nullable": True, "kind": "str"}], - "subscribers": [{"expr": Channel.subscriber_count, "asc": False, "nullable": True, "kind": "num"}], - "subscribers_asc": [{"expr": Channel.subscriber_count, "asc": True, "nullable": True, "kind": "num"}], - # Your per-channel priority (set in the channel manager), newest first within a tier. - # coalesce keeps it null-safe in "all" scope where unsubscribed channels have no row. - "priority": [{"expr": func.coalesce(Subscription.priority, 0), "asc": False, "nullable": False, "kind": "num"}, pub], - "priority_asc": [{"expr": func.coalesce(Subscription.priority, 0), "asc": True, "nullable": False, "kind": "num"}, pub], - # Deterministic per-seed shuffle: the same seed yields the same total order, so - # the cursor stays valid across pages of one shuffled run. - "shuffle": [{"expr": func.md5(func.concat(Video.id, str(seed))), "asc": True, "nullable": False, "kind": "str"}], - } - # Relevance (search-result ranking) only exists when there's a query to rank against; - # without `rank_expr` it falls through to newest. Ties break by newest, then id. - if rank_expr is not None: - table["relevance"] = [ - {"expr": rank_expr, "asc": False, "nullable": False, "kind": "num"}, - pub, - ] - return table.get(sort) or table["newest"] - - -def _order_by(keys: list[dict]): - cols = [] - for k in keys: - c = k["expr"].asc() if k["asc"] else k["expr"].desc() - cols.append(c.nulls_last() if k["nullable"] else c) - cols.append(Video.id.asc()) # unique final tiebreaker - return cols - - -def _keyset_predicate(keys: list[dict], cursor_id: str): - """WHERE clause selecting rows strictly AFTER the cursor in the sort order. - - Lexicographic over the key columns then id, NULL-aware for `nulls_last`: a NULL - sorts after every non-NULL value, so a cursor sitting on a non-NULL value must - also pull in the NULL group, and equality treats NULL == NULL.""" - def eq(k): - return k["expr"].is_(None) if k["value"] is None else k["expr"] == k["value"] - - def after(k): - if k["value"] is None: - return false() # nulls are last; nothing sorts strictly after them - cmp = k["expr"] > k["value"] if k["asc"] else k["expr"] < k["value"] - return or_(cmp, k["expr"].is_(None)) if k["nullable"] else cmp - - branches = [] - for j in range(len(keys)): - branches.append(and_(*[eq(keys[i]) for i in range(j)], after(keys[j]))) - branches.append(and_(*[eq(k) for k in keys], Video.id > cursor_id)) - return or_(*branches) - - -def _encode_cursor(keys: list[dict], row) -> str: - vals = [] - for i, k in enumerate(keys): - v = getattr(row, f"_k{i}") - vals.append(v.isoformat() if (v is not None and k["kind"] == "dt") else v) - payload = json.dumps({"k": vals, "id": row.id}) - return base64.urlsafe_b64encode(payload.encode()).decode() - - -def _decode_cursor(keys: list[dict], cursor: str) -> tuple[list, str]: - try: - payload = json.loads(base64.urlsafe_b64decode(cursor.encode())) - raw = payload["k"] - if len(raw) != len(keys): - raise ValueError("key arity mismatch") - vals = [ - datetime.fromisoformat(v) if (v is not None and k["kind"] == "dt") else v - for k, v in zip(keys, raw) - ] - return vals, payload["id"] - except (binascii.Error, json.JSONDecodeError, KeyError, ValueError, TypeError): - raise HTTPException(status_code=400, detail="Invalid feed cursor") - - @router.get("/feed") def get_feed( params: dict = Depends(_feed_params), diff --git a/backend/app/routes/playlists.py b/backend/app/routes/playlists.py index 08621ea8..43f0aa1f 100644 --- a/backend/app/routes/playlists.py +++ b/backend/app/routes/playlists.py @@ -9,8 +9,8 @@ from sqlalchemy.orm import Session, aliased from app import quota from app.auth import current_user, has_read_scope, has_write_scope, require_write_scope from app.db import get_db +from app.feedquery import feed_columns, serialize from app.models import Channel, Playlist, PlaylistItem, User, Video, VideoState -from app.routes.feed import serialize, feed_columns from app.sync.playlists import ( plan_push, push_playlist, diff --git a/backend/app/routes/plex/admin.py b/backend/app/routes/plex/admin.py index 4681542b..abc1b595 100644 --- a/backend/app/routes/plex/admin.py +++ b/backend/app/routes/plex/admin.py @@ -8,7 +8,7 @@ from sqlalchemy import select from sqlalchemy.orm import Session from app import sysconfig -from app.auth import current_user +from app.auth import admin_user, current_user from app.db import get_db from app.models import ( PlexItem, @@ -19,7 +19,6 @@ from app.plex import paths as plex_paths from app.plex import sync as plex_sync from app.plex import watch_sync as plex_watch from app.plex.client import PlexClient, PlexError, PlexNotConfigured -from app.routes.admin import admin_user from app.utils import iso router = APIRouter(prefix="/api/plex", tags=["plex"]) diff --git a/backend/app/routes/plex/browse.py b/backend/app/routes/plex/browse.py index 2bf98d3f..f6d71218 100644 --- a/backend/app/routes/plex/browse.py +++ b/backend/app/routes/plex/browse.py @@ -25,7 +25,7 @@ from sqlalchemy import ( from sqlalchemy.orm import Session, aliased from app import sysconfig -from app.auth import current_user +from app.auth import admin_user, current_user from app.db import get_db from app.models import ( PlexCollection, @@ -45,9 +45,8 @@ from app.plex import rollup, subtitles from app.plex import sync as plex_sync from app.plex.client import PlexClient, PlexError, PlexNotConfigured from app.plex.lookups import by_rating_key +from app.feedquery import to_tsquery_str from app.routes._common import owned_or_404 -from app.routes.admin import admin_user -from app.routes.feed import to_tsquery_str from app.routes.plex._shared import item_or_404 router = APIRouter(prefix="/api/plex", tags=["plex"]) diff --git a/backend/app/routes/quota.py b/backend/app/routes/quota.py index 6dd6929e..cc390e23 100644 --- a/backend/app/routes/quota.py +++ b/backend/app/routes/quota.py @@ -10,10 +10,9 @@ from sqlalchemy import func, select from sqlalchemy.orm import Session from app import quota -from app.auth import current_user +from app.auth import admin_user, current_user from app.db import get_db from app.models import QuotaEvent, User -from app.routes.admin import admin_user router = APIRouter(prefix="/api/quota", tags=["quota"]) diff --git a/backend/app/routes/scheduler.py b/backend/app/routes/scheduler.py index df20de62..a52ee7cb 100644 --- a/backend/app/routes/scheduler.py +++ b/backend/app/routes/scheduler.py @@ -8,10 +8,10 @@ from sqlalchemy.orm import Session from app import audit, quota, state, sysconfig from app.audit import AuditAction +from app.auth import admin_user from app.config import settings from app.db import get_db from app.models import Channel, Subscription, User, Video -from app.routes.admin import admin_user from app.scheduler import ( JOB_INTERVALS, MAX_INTERVAL, diff --git a/backend/app/routes/search.py b/backend/app/routes/search.py index 8c860ac3..c0e3d012 100644 --- a/backend/app/routes/search.py +++ b/backend/app/routes/search.py @@ -36,7 +36,7 @@ from app.models import ( Video, VideoState, ) -from app.routes.feed import serialize, feed_columns +from app.feedquery import feed_columns, serialize from app.sync.subscriptions import apply_channel_details from app.sync.videos import insert_stubs, apply_video_details, parse_dt from app.youtube import search_scrape