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/routes/channels.py b/backend/app/routes/channels.py index c7dcdb7..adfa4b8 100644 --- a/backend/app/routes/channels.py +++ b/backend/app/routes/channels.py @@ -12,6 +12,7 @@ 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) 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/search.py b/backend/app/routes/search.py index c84a509..139b6f7 100644 --- a/backend/app/routes/search.py +++ b/backend/app/routes/search.py @@ -24,6 +24,7 @@ from app import quota, sysconfig from app.auth import require_human from app.db import get_db from app.models import ( + LIVE_OR_UPCOMING, BlockedChannel, Channel, Playlist, @@ -45,8 +46,6 @@ router = APIRouter(prefix="/api/search", tags=["search"]) # search.list costs this many quota units per page. SEARCH_COST = 100 -# live_status values we never surface from a live search (was_live VODs are real content). -_LIVE_HIDDEN = ("live", "upcoming") def _classify_shorts(db: Session, videos: list[Video]) -> None: @@ -165,7 +164,7 @@ def _ingest_candidates( # 4) Confirm/deny Shorts (no quota) so none enter via search. _classify_shorts(db, videos) - keep = {v.id for v in videos if not v.is_short and v.live_status not in _LIVE_HIDDEN} + keep = {v.id for v in videos if not v.is_short and v.live_status not in LIVE_OR_UPCOMING} return [vid for vid in ids if vid in keep] diff --git a/backend/app/sync/videos.py b/backend/app/sync/videos.py index 89d6865..1b3178e 100644 --- a/backend/app/sync/videos.py +++ b/backend/app/sync/videos.py @@ -2,6 +2,7 @@ uploads playlist, and enrichment via videos.list (duration, stats, category, Shorts and livestream classification).""" import re +from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta, timezone @@ -10,7 +11,7 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.orm import Session from app import progress, sysconfig -from app.models import Channel, Video +from app.models import LIVE_OR_UPCOMING, Channel, Video from app.titles import normalize_title from app.youtube.client import YouTubeClient, best_thumbnail from app.youtube.rss import fetch_channel_feed @@ -260,6 +261,37 @@ def apply_video_details(video: Video, item: dict) -> None: # is_short is decided separately by the youtube.com/shorts probe (run_shorts_classification). +def _apply_video_batch( + db: Session, + yt: YouTubeClient, + videos: list[Video], + on_missing: Callable[[Video], None], +) -> int: + """Fetch videos.list details for the given rows, apply them (stamping enriched_at), and + return how many were applied. Rows YouTube omits (deleted/private) are handled by the + caller-supplied `on_missing` so each caller decides how to retire them.""" + if not videos: + return 0 + items = {it["id"]: it for it in yt.get_videos([v.id for v in videos])} + now = _now() + applied = 0 + for video in videos: + item = items.get(video.id) + if item is None: + on_missing(video) + continue + apply_video_details(video, item) + video.enriched_at = now + applied += 1 + db.commit() + return applied + + +def _mark_enriched(video: Video) -> None: + # Unavailable (deleted/private): stamp so we don't keep retrying. + video.enriched_at = _now() + + def enrich_pending(db: Session, yt: YouTubeClient, limit: int | None = None) -> int: limit = limit or sysconfig.get_int(db, "enrich_batch_size") videos = ( @@ -267,22 +299,7 @@ def enrich_pending(db: Session, yt: YouTubeClient, limit: int | None = None) -> .scalars() .all() ) - if not videos: - return 0 - items = {it["id"]: it for it in yt.get_videos([v.id for v in videos])} - now = _now() - enriched = 0 - for video in videos: - item = items.get(video.id) - if item is None: - # Unavailable (deleted/private): stamp so we don't keep retrying. - video.enriched_at = now - continue - apply_video_details(video, item) - video.enriched_at = now - enriched += 1 - db.commit() - return enriched + return _apply_video_batch(db, yt, list(videos), _mark_enriched) def refresh_live(db: Session, yt: YouTubeClient, limit: int | None = None) -> int: @@ -300,7 +317,7 @@ def refresh_live(db: Session, yt: YouTubeClient, limit: int | None = None) -> in select(Video) .where( or_( - Video.live_status.in_(("live", "upcoming")), + Video.live_status.in_(LIVE_OR_UPCOMING), and_( Video.live_status == "was_live", Video.duration_seconds.is_(None), @@ -312,23 +329,13 @@ def refresh_live(db: Session, yt: YouTubeClient, limit: int | None = None) -> in .scalars() .all() ) - if not videos: - return 0 - items = {it["id"]: it for it in yt.get_videos([v.id for v in videos])} - now = _now() - updated = 0 - for video in videos: - item = items.get(video.id) - if item is None: - # The broadcast vanished (deleted/private). Stop treating it as live so it - # doesn't get refreshed forever; it just becomes an ordinary (dead) row. - video.live_status = "none" - continue - apply_video_details(video, item) - video.enriched_at = now - updated += 1 - db.commit() - return updated + + def _retire(video: Video) -> None: + # The broadcast vanished (deleted/private). Stop treating it as live so it + # doesn't get refreshed forever; it just becomes an ordinary (dead) row. + video.live_status = "none" + + return _apply_video_batch(db, yt, list(videos), _retire) def run_shorts_classification(db: Session, limit: int | None = None) -> dict: diff --git a/backend/app/youtube/client.py b/backend/app/youtube/client.py index 2c32405..dc63043 100644 --- a/backend/app/youtube/client.py +++ b/backend/app/youtube/client.py @@ -167,9 +167,9 @@ class YouTubeClient: if not page_token: break - def iter_my_playlist_video_ids(self, playlist_id: str) -> Iterator[str]: - """Yield the video ids of one of the user's own playlists, in playlist order - (OAuth, so private/unlisted playlists work).""" + def _iter_playlist_items(self, playlist_id: str) -> Iterator[dict]: + """Paginate one of the user's playlists' items (playlistItems, contentDetails part; + OAuth so private/unlisted work), yielding each raw item in playlist order.""" page_token = None while True: params = { @@ -180,14 +180,19 @@ class YouTubeClient: if page_token: params["pageToken"] = page_token data = self._get("playlistItems", params, cost=1, allow_key=False) - for item in data.get("items", []): - vid = item.get("contentDetails", {}).get("videoId") - if vid: - yield vid + yield from data.get("items", []) page_token = data.get("nextPageToken") if not page_token: break + def iter_my_playlist_video_ids(self, playlist_id: str) -> Iterator[str]: + """Yield the video ids of one of the user's own playlists, in playlist order + (OAuth, so private/unlisted playlists work).""" + for item in self._iter_playlist_items(playlist_id): + vid = item.get("contentDetails", {}).get("videoId") + if vid: + yield vid + def get_my_channel_id(self) -> str | None: """The authenticated user's own channel id (channels.list?mine=true). OAuth only; 1 quota unit. Returns None if the account has no channel.""" @@ -264,23 +269,10 @@ class YouTubeClient: """Yield {item_id, video_id} for each item of one of the user's playlists, in playlist order. `item_id` is the playlistItem resource id, needed to delete or reposition the item via the write API. OAuth (private playlists work).""" - page_token = None - while True: - params = { - "part": "contentDetails", - "playlistId": playlist_id, - "maxResults": 50, - } - if page_token: - params["pageToken"] = page_token - data = self._get("playlistItems", params, cost=1, allow_key=False) - for item in data.get("items", []): - vid = item.get("contentDetails", {}).get("videoId") - if vid: - yield {"item_id": item.get("id"), "video_id": vid} - page_token = data.get("nextPageToken") - if not page_token: - break + for item in self._iter_playlist_items(playlist_id): + vid = item.get("contentDetails", {}).get("videoId") + if vid: + yield {"item_id": item.get("id"), "video_id": vid} # --- write endpoints (OAuth only, never the API key; each costs 50 quota units) --- def _write(self, method: str, path: str, *, params: dict, json: dict | None = None) -> dict: