chore(feed): Phase 2 #2 backend cleanup — dead return, dup helpers, shared const

- feed.py _filtered_query: drop the dead 2nd return element (status_expr was
  used only internally for WHERE filters; all 3 callers discarded it as _status).
  Now returns (query, rank_expr); fixed the stale docstring.
- youtube/client.py: extract _iter_playlist_items() — iter_my_playlist_video_ids
  and iter_playlist_items_with_ids were near-identical playlistItems paging loops
  (jscpd [173-187]≈[267-281]); both now map over the shared generator.
- sync/videos.py: extract _apply_video_batch() with an on_missing callback —
  enrich_pending and refresh_live shared the fetch-map-apply skeleton, differing
  only in the query and how they retire rows YouTube omits.
- models.py: add LIVE_OR_UPCOMING = ("live","upcoming"); replace the 4 duplicated
  copies (feed.HIDDEN_LIVE, search._LIVE_HIDDEN, videos.py + channels.py inline).

Behavior-neutral. ruff clean on touched files, localdev boots, feed/worker healthy.
This commit is contained in:
npeter83 2026-07-11 17:37:52 +02:00
parent 477056bfa8
commit 505f0e5650
6 changed files with 75 additions and 71 deletions

View file

@ -232,6 +232,11 @@ class Subscription(Base, TimestampMixin):
channel: Mapped["Channel"] = relationship(back_populates="subscriptions") 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): class Video(Base, TimestampMixin):
"""A YouTube video. Shared across users; per-user state lives elsewhere.""" """A YouTube video. Shared across users; per-user state lives elsewhere."""

View file

@ -12,6 +12,7 @@ from app import quota, sysconfig
from app.auth import current_user, has_write_scope, require_human from app.auth import current_user, has_write_scope, require_human
from app.db import get_db from app.db import get_db
from app.models import ( from app.models import (
LIVE_OR_UPCOMING,
BlockedChannel, BlockedChannel,
Channel, Channel,
ChannelTag, ChannelTag,
@ -59,7 +60,7 @@ def list_channels(
func.max(Video.published_at), func.max(Video.published_at),
func.coalesce(func.sum(Video.duration_seconds), 0), func.coalesce(func.sum(Video.duration_seconds), 0),
func.count(Video.id).filter(Video.is_short.is_(True)), 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)) .where(Video.channel_id.in_(channel_ids))
.group_by(Video.channel_id) .group_by(Video.channel_id)

View file

@ -14,6 +14,7 @@ from app import quota
from app.auth import current_user from app.auth import current_user
from app.db import get_db from app.db import get_db
from app.models import ( from app.models import (
LIVE_OR_UPCOMING,
BlockedChannel, BlockedChannel,
Channel, Channel,
ChannelTag, 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. # "saved" used to be a status; it's now membership in the built-in Watch later playlist.
VALID_STATES = {"new", "watched", "hidden"} VALID_STATES = {"new", "watched", "hidden"}
HIDDEN_LIVE = ("live", "upcoming")
# Resume-position thresholds (mirror the client): positions below this are "didn't # 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 # 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, exclude_tag_category: str | None = None,
) -> tuple[Select, object]: ) -> tuple[Select, object]:
"""Build the feed query (joins + all WHERE filters), shared by /feed and /feed/count. """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="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); `scope="all"` shows every video in the shared catalog (any user's ingested channels);
@ -313,12 +313,12 @@ def _filtered_query(
type_clauses = [] type_clauses = []
if show_normal: if show_normal:
type_clauses.append( 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: if include_shorts:
type_clauses.append(Video.is_short.is_(True)) type_clauses.append(Video.is_short.is_(True))
if include_live: 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()) query = query.where(or_(*type_clauses) if type_clauses else false())
if show == "unwatched": if show == "unwatched":
@ -335,7 +335,7 @@ def _filtered_query(
else: # all else: # all
query = query.where(status_expr != "hidden") query = query.where(status_expr != "hidden")
return query, status_expr, rank_expr return query, rank_expr
def _to_tsquery_str(q: str) -> str | None: def _to_tsquery_str(q: str) -> str | None:
@ -492,7 +492,7 @@ def get_feed(
user: User = Depends(current_user), user: User = Depends(current_user),
db: Session = Depends(get_db), db: Session = Depends(get_db),
) -> dict: ) -> 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) 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. # 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), user: User = Depends(current_user),
db: Session = Depends(get_db), db: Session = Depends(get_db),
) -> dict: ) -> 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())) total = db.scalar(select(func.count()).select_from(query.subquery()))
return {"count": total or 0} 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 # 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). # already-selected topics — and tags that can't co-occur drop to zero (hidden).
conjunctive = category == "topic" and params.get("tag_mode") == "and" conjunctive = category == "topic" and params.get("tag_mode") == "and"
base, _status, _rank = _filtered_query( base, _rank = _filtered_query(
db, db,
user, user,
**{**params, "exclude_tag_category": None if conjunctive else category}, **{**params, "exclude_tag_category": None if conjunctive else category},

View file

@ -24,6 +24,7 @@ from app import quota, sysconfig
from app.auth import require_human from app.auth import require_human
from app.db import get_db from app.db import get_db
from app.models import ( from app.models import (
LIVE_OR_UPCOMING,
BlockedChannel, BlockedChannel,
Channel, Channel,
Playlist, Playlist,
@ -45,8 +46,6 @@ router = APIRouter(prefix="/api/search", tags=["search"])
# search.list costs this many quota units per page. # search.list costs this many quota units per page.
SEARCH_COST = 100 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: 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. # 4) Confirm/deny Shorts (no quota) so none enter via search.
_classify_shorts(db, videos) _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] return [vid for vid in ids if vid in keep]

View file

@ -2,6 +2,7 @@
uploads playlist, and enrichment via videos.list (duration, stats, category, Shorts uploads playlist, and enrichment via videos.list (duration, stats, category, Shorts
and livestream classification).""" and livestream classification)."""
import re import re
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta, timezone 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 sqlalchemy.orm import Session
from app import progress, sysconfig 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.titles import normalize_title
from app.youtube.client import YouTubeClient, best_thumbnail from app.youtube.client import YouTubeClient, best_thumbnail
from app.youtube.rss import fetch_channel_feed 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). # 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: def enrich_pending(db: Session, yt: YouTubeClient, limit: int | None = None) -> int:
limit = limit or sysconfig.get_int(db, "enrich_batch_size") limit = limit or sysconfig.get_int(db, "enrich_batch_size")
videos = ( videos = (
@ -267,22 +299,7 @@ def enrich_pending(db: Session, yt: YouTubeClient, limit: int | None = None) ->
.scalars() .scalars()
.all() .all()
) )
if not videos: return _apply_video_batch(db, yt, list(videos), _mark_enriched)
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
def refresh_live(db: Session, yt: YouTubeClient, limit: int | None = None) -> int: 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) select(Video)
.where( .where(
or_( or_(
Video.live_status.in_(("live", "upcoming")), Video.live_status.in_(LIVE_OR_UPCOMING),
and_( and_(
Video.live_status == "was_live", Video.live_status == "was_live",
Video.duration_seconds.is_(None), Video.duration_seconds.is_(None),
@ -312,23 +329,13 @@ def refresh_live(db: Session, yt: YouTubeClient, limit: int | None = None) -> in
.scalars() .scalars()
.all() .all()
) )
if not videos:
return 0 def _retire(video: Video) -> None:
items = {it["id"]: it for it in yt.get_videos([v.id for v in videos])} # The broadcast vanished (deleted/private). Stop treating it as live so it
now = _now() # doesn't get refreshed forever; it just becomes an ordinary (dead) row.
updated = 0 video.live_status = "none"
for video in videos:
item = items.get(video.id) return _apply_video_batch(db, yt, list(videos), _retire)
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 run_shorts_classification(db: Session, limit: int | None = None) -> dict: def run_shorts_classification(db: Session, limit: int | None = None) -> dict:

View file

@ -167,9 +167,9 @@ class YouTubeClient:
if not page_token: if not page_token:
break break
def iter_my_playlist_video_ids(self, playlist_id: str) -> Iterator[str]: def _iter_playlist_items(self, playlist_id: str) -> Iterator[dict]:
"""Yield the video ids of one of the user's own playlists, in playlist order """Paginate one of the user's playlists' items (playlistItems, contentDetails part;
(OAuth, so private/unlisted playlists work).""" OAuth so private/unlisted work), yielding each raw item in playlist order."""
page_token = None page_token = None
while True: while True:
params = { params = {
@ -180,14 +180,19 @@ class YouTubeClient:
if page_token: if page_token:
params["pageToken"] = page_token params["pageToken"] = page_token
data = self._get("playlistItems", params, cost=1, allow_key=False) data = self._get("playlistItems", params, cost=1, allow_key=False)
for item in data.get("items", []): yield from data.get("items", [])
vid = item.get("contentDetails", {}).get("videoId")
if vid:
yield vid
page_token = data.get("nextPageToken") page_token = data.get("nextPageToken")
if not page_token: if not page_token:
break 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: def get_my_channel_id(self) -> str | None:
"""The authenticated user's own channel id (channels.list?mine=true). OAuth only; """The authenticated user's own channel id (channels.list?mine=true). OAuth only;
1 quota unit. Returns None if the account has no channel.""" 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 """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 playlist order. `item_id` is the playlistItem resource id, needed to delete or
reposition the item via the write API. OAuth (private playlists work).""" reposition the item via the write API. OAuth (private playlists work)."""
page_token = None for item in self._iter_playlist_items(playlist_id):
while True: vid = item.get("contentDetails", {}).get("videoId")
params = { if vid:
"part": "contentDetails", yield {"item_id": item.get("id"), "video_id": vid}
"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
# --- write endpoints (OAuth only, never the API key; each costs 50 quota units) --- # --- 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: def _write(self, method: str, path: str, *, params: dict, json: dict | None = None) -> dict: