diff --git a/backend/alembic/versions/0028_live_search_provenance.py b/backend/alembic/versions/0028_live_search_provenance.py new file mode 100644 index 0000000..3147193 --- /dev/null +++ b/backend/alembic/versions/0028_live_search_provenance.py @@ -0,0 +1,46 @@ +"""live YouTube search: provenance flags + +Revision ID: 0028_live_search_provenance +Revises: 0027_message_e2ee +Create Date: 2026-06-29 + +Adds provenance flags so videos (and channels) that entered the catalog only because a live +YouTube search surfaced them can be told apart from organically-synced content and hidden from +the Library by default: + - videos.via_search — this video first arrived via search + - channels.from_search — this channel exists only because search surfaced one of its videos +Both default false (every existing row is organic), indexed for the feed's exclude filter. +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0028_live_search_provenance" +down_revision: Union[str, None] = "0027_message_e2ee" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "videos", + sa.Column( + "via_search", sa.Boolean(), nullable=False, server_default="false" + ), + ) + op.create_index("ix_videos_via_search", "videos", ["via_search"]) + op.add_column( + "channels", + sa.Column( + "from_search", sa.Boolean(), nullable=False, server_default="false" + ), + ) + op.create_index("ix_channels_from_search", "channels", ["from_search"]) + + +def downgrade() -> None: + op.drop_index("ix_channels_from_search", table_name="channels") + op.drop_column("channels", "from_search") + op.drop_index("ix_videos_via_search", table_name="videos") + op.drop_column("videos", "via_search") diff --git a/backend/app/config.py b/backend/app/config.py index 5564ee0..933d036 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -71,6 +71,9 @@ class Settings(BaseSettings): youtube_api_proxy: str = "" # Daily quota budget (units). Default leaves headroom under the 10,000/day free limit. quota_daily_budget: int = 9000 + # Per-user cap on live YouTube searches per day. search.list costs 100 units each, so the + # shared budget only affords ~80-90/day total — this stops one user draining it. Admin-tunable. + search_daily_limit_per_user: int = 70 # Recent-first backfill: how far back to fetch on the first pass per channel. backfill_recent_max_videos: int = 100 backfill_recent_max_days: int = 365 diff --git a/backend/app/main.py b/backend/app/main.py index f530847..a404305 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -39,6 +39,7 @@ from app.routes import ( playlists, quota, scheduler as scheduler_routes, + search as search_routes, setup as setup_routes, sync, tags, @@ -113,6 +114,7 @@ app.include_router(auth.router) app.include_router(sync.router) app.include_router(tags.router) app.include_router(feed.router) +app.include_router(search_routes.router) app.include_router(me.router) app.include_router(notifications.router) app.include_router(messages.router) diff --git a/backend/app/models.py b/backend/app/models.py index 4ebf14d..f40653d 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -175,6 +175,12 @@ class Channel(Base, TimestampMixin): backfill_done: Mapped[bool] = mapped_column( Boolean, default=False, server_default="false" ) + # True when this channel only entered the catalog because a live YouTube search + # surfaced one of its videos (nobody subscribes to it / it has no organic upload sync). + # Lets the Library hide search-only "noise" by default. + from_search: Mapped[bool] = mapped_column( + Boolean, default=False, server_default="false", index=True + ) videos: Mapped[list["Video"]] = relationship(back_populates="channel") subscriptions: Mapped[list["Subscription"]] = relationship(back_populates="channel") @@ -238,6 +244,12 @@ class Video(Base, TimestampMixin): shorts_probed: Mapped[bool] = mapped_column( Boolean, default=False, server_default="false", index=True ) + # True when this video first entered the catalog via a live YouTube search (not an + # organic channel/playlist sync). Used to hide search-discovered "noise" from the + # Library by default; left False by the organic ingest paths. + via_search: Mapped[bool] = mapped_column( + Boolean, default=False, server_default="false", index=True + ) # none | live | upcoming | premiere | was_live live_status: Mapped[str] = mapped_column( String(16), default="none", server_default="none", index=True diff --git a/backend/app/quota.py b/backend/app/quota.py index 1a45011..0737675 100644 --- a/backend/app/quota.py +++ b/backend/app/quota.py @@ -36,6 +36,7 @@ class QuotaAction: VIDEOS_BACKFILL_FULL = "videos_backfill_full" VIDEOS_ENRICH = "videos_enrich" VIDEOS_LOOKUP = "videos_lookup" + VIDEOS_SEARCH = "videos_search" CHANNELS_DISCOVER = "channels_discover" CHANNELS_SUBSCRIBE = "channels_subscribe" CHANNELS_UNSUBSCRIBE = "channels_unsubscribe" @@ -109,6 +110,26 @@ def can_spend(db: Session, units: int) -> bool: return remaining_today(db) >= units +def actions_today(db: Session, user_id: int, action: str) -> int: + """How many quota events of `action` this user has logged so far in the current Pacific + day — for per-user, per-action daily caps (e.g. the live-search limit). Counts events, + not units, so it only works for actions charged exactly once per user action.""" + from sqlalchemy import func, select # local import: keep the module's import head lean + + return ( + db.scalar( + select(func.count()) + .select_from(QuotaEvent) + .where( + QuotaEvent.user_id == user_id, + QuotaEvent.action == action, + QuotaEvent.created_at >= pacific_day_start_utc(), + ) + ) + or 0 + ) + + def record_usage(db: Session, units: int) -> None: """Atomically add `units` to today's counter (upsert) and log an attribution event.""" if units <= 0: diff --git a/backend/app/routes/feed.py b/backend/app/routes/feed.py index 7e43f08..7151960 100644 --- a/backend/app/routes/feed.py +++ b/backend/app/routes/feed.py @@ -124,6 +124,7 @@ def _filtered_query( include_live: bool, show: str, scope: str = "my", + exclude_search_discovered: bool = True, exclude_tag_category: str | None = None, ) -> tuple[Select, object]: """Build the feed query (joins + all WHERE filters), shared by /feed and /feed/count. @@ -171,6 +172,12 @@ def _filtered_query( # either way they shouldn't show in any feed view meanwhile. query = query.where(Video.unavailable_since.is_(None)) + # Hide videos that only entered the catalog via a live YouTube search (search-discovered + # "noise") from the Library by default. Only meaningful in "all" scope: "my" already + # restricts to subscribed channels, where a once-searched video is legitimately the user's. + if scope == "all" and exclude_search_discovered: + query = query.where(Video.via_search.is_(False)) + if channel_id: query = query.where(Video.channel_id == channel_id) if min_duration is not None: @@ -271,6 +278,7 @@ def _feed_params( include_live: bool = False, show: str = "unwatched", scope: str = "my", + exclude_search_discovered: bool = True, ) -> dict: return { "tags": tags, @@ -287,6 +295,7 @@ def _feed_params( "include_live": include_live, "show": show, "scope": scope, + "exclude_search_discovered": exclude_search_discovered, } diff --git a/backend/app/routes/search.py b/backend/app/routes/search.py new file mode 100644 index 0000000..bb29119 --- /dev/null +++ b/backend/app/routes/search.py @@ -0,0 +1,204 @@ +"""Live YouTube search. + +Searches YouTube directly (search.list) and materialises the results into the shared catalog +so they render with the normal feed cards + in-app player and gain per-user state (watch / +save / resume). search.list costs **100 quota units per call**, so it's the most expensive +read by far: it's gated behind a per-user daily cap + a budget pre-check, and only real users +(never the shared demo account) may call it. + +Ingested results are flagged (videos.via_search / channels.from_search) so the Library can hide +this search-discovered "noise" by default. Shorts and currently-live/upcoming videos are never +ingested this way (Shorts are confirmed by the youtube.com/shorts probe; live/upcoming are +dropped from the search response up front).""" +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timezone + +from fastapi import APIRouter, Depends, HTTPException +from sqlalchemy import and_, select +from sqlalchemy.orm import Session, aliased + +from app import quota, sysconfig +from app.auth import require_human +from app.db import get_db +from app.models import Channel, User, Video, VideoState +from app.routes.feed import _serialize, feed_columns +from app.sync.subscriptions import apply_channel_details +from app.sync.videos import _insert_stubs, apply_video_details, parse_dt +from app.youtube.client import YouTubeClient, YouTubeError +from app.youtube.shorts import make_client, probe_is_short + +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: + """Confirm/deny Shorts for the freshly-ingested videos so none slip into the feed. + + Mirrors the scheduler's run_shorts_classification, but scoped to just these rows and run + synchronously (the search is an explicit, awaited action). Non-candidates (too long, no + duration, or live) are finalised without a probe; the rest hit youtube.com/shorts (no + quota). Undetermined probes are left for the scheduler to retry.""" + probe_max = sysconfig.get_int(db, "shorts_probe_max_seconds") + to_probe: list[Video] = [] + for v in videos: + if v.shorts_probed: + continue + if ( + v.live_status != "none" + or v.duration_seconds is None + or v.duration_seconds > probe_max + ): + v.is_short = False + v.shorts_probed = True + else: + to_probe.append(v) + if to_probe: + with make_client() as client: + + def work(v: Video): + return v, probe_is_short(client, v.id) + + with ThreadPoolExecutor(max_workers=16) as pool: + for v, result in pool.map(work, to_probe): + if result is None: + continue + v.is_short = result + v.shorts_probed = True + db.commit() + + +def _serialize_ids(db: Session, user: User, ids: list[str]) -> list[dict]: + """Serialize the given video ids with the shared feed projection (joined channel + this + user's watch state + saved), preserving the input (search relevance) order.""" + if not ids: + return [] + state = aliased(VideoState) + rows = db.execute( + select(*feed_columns(user, state)) + .join(Channel, Channel.id == Video.channel_id) + .outerjoin(state, and_(state.video_id == Video.id, state.user_id == user.id)) + .where(Video.id.in_(ids)) + ).all() + by_id = {r.id: _serialize(r) for r in rows} + return [by_id[i] for i in ids if i in by_id] + + +@router.get("/youtube") +def search_youtube( + q: str | None = None, + cursor: str | None = None, + user: User = Depends(require_human), + db: Session = Depends(get_db), +) -> dict: + """One page of live YouTube search results, materialised into the catalog. `cursor` is the + YouTube nextPageToken (each page is another 100-unit search).""" + term = (q or "").strip() + if not term: + raise HTTPException(status_code=400, detail="A search term is required.") + + # Guard the shared quota BEFORE spending: per-user daily cap, then today's budget. + daily_limit = sysconfig.get_int(db, "search_daily_limit_per_user") + used = quota.actions_today(db, user.id, quota.QuotaAction.VIDEOS_SEARCH) + if used >= daily_limit: + raise HTTPException( + status_code=429, + detail=f"You've reached today's YouTube search limit ({daily_limit}). Try again tomorrow.", + ) + if not quota.can_spend(db, SEARCH_COST): + raise HTTPException( + status_code=429, + detail="The shared YouTube quota for today is used up. Try again tomorrow.", + ) + + with YouTubeClient(db, user) as yt: + try: + with quota.attribute(user.id, quota.QuotaAction.VIDEOS_SEARCH): + page = yt.search_videos(term, page_token=cursor) + except YouTubeError as exc: + raise HTTPException(status_code=502, detail=f"YouTube search failed: {exc}") + + # Drop currently-live/upcoming results (search can't filter them and we never ingest + # them this way) and anything missing a channel id. + candidates = [ + it + for it in page["items"] + if it["live_broadcast"] == "none" and it["channel_id"] + ] + ids = [it["id"] for it in candidates] + if not ids: + return { + "items": [], + "next_cursor": page["next_page_token"], + "has_more": bool(page["next_page_token"]), + } + + # 1) Channels are the FK target — create stubs for unknown ones (flagged from_search), + # then enrich the new ones (channels.list, 1 unit). Stubs survive an enrich failure. + channel_ids = list({it["channel_id"] for it in candidates}) + existing = set( + db.scalars(select(Channel.id).where(Channel.id.in_(channel_ids))).all() + ) + new_channel_ids = [c for c in channel_ids if c not in existing] + if new_channel_ids: + titles = {it["channel_id"]: it["channel_title"] for it in candidates} + for cid in new_channel_ids: + db.add(Channel(id=cid, title=titles.get(cid), from_search=True)) + db.commit() + try: + with quota.attribute(user.id, quota.QuotaAction.CHANNELS_DISCOVER): + details = yt.get_channels(new_channel_ids) + apply_channel_details(db, details) + db.commit() + except YouTubeError: + db.rollback() # details can be backfilled later; the stubs remain + + # 2) Video stubs, flagged via_search. Existing rows keep their flags (ON CONFLICT DO + # NOTHING), so an organically-synced video stays organic. + _insert_stubs( + db, + [ + { + "id": it["id"], + "channel_id": it["channel_id"], + "title": it["title"], + "published_at": parse_dt(it["published_at"]), + "thumbnail_url": it["thumbnail_url"], + "via_search": True, + } + for it in candidates + ], + ) + + # 3) Enrich (videos.list, 1 unit/50) so cards have duration/stats and we can judge live. + videos = db.scalars(select(Video).where(Video.id.in_(ids))).all() + try: + with quota.attribute(user.id, quota.QuotaAction.VIDEOS_ENRICH): + detail_by_id = {it["id"]: it for it in yt.get_videos(ids)} + except YouTubeError: + detail_by_id = {} + now = datetime.now(timezone.utc) + for v in videos: + item = detail_by_id.get(v.id) + if item is None: + continue + apply_video_details(v, item) + v.enriched_at = now + db.commit() + + # 4) Confirm/deny Shorts (no quota) so none enter via search. + _classify_shorts(db, videos) + + # 5) Exclude Shorts and still-live results; keep the search's relevance order. + keep = { + v.id for v in videos if not v.is_short and v.live_status not in _LIVE_HIDDEN + } + ordered = [vid for vid in ids if vid in keep] + return { + "items": _serialize_ids(db, user, ordered), + "next_cursor": page["next_page_token"], + "has_more": bool(page["next_page_token"]), + } diff --git a/backend/app/sysconfig.py b/backend/app/sysconfig.py index 8ad8c53..1b45155 100644 --- a/backend/app/sysconfig.py +++ b/backend/app/sysconfig.py @@ -39,6 +39,7 @@ SPECS: tuple[ConfigSpec, ...] = ( # --- Quota (shared YouTube Data API daily budget) --- ConfigSpec("quota_daily_budget", "int", "quota", "quota_daily_budget", min=1, max=1_000_000), ConfigSpec("backfill_quota_reserve", "int", "quota", "backfill_quota_reserve", min=0, max=1_000_000), + ConfigSpec("search_daily_limit_per_user", "int", "quota", "search_daily_limit_per_user", min=0, max=100_000), # --- Backfill (recent-first first pass per channel) --- ConfigSpec("backfill_recent_max_videos", "int", "backfill", "backfill_recent_max_videos", min=1, max=100_000), ConfigSpec("backfill_recent_max_days", "int", "backfill", "backfill_recent_max_days", min=1, max=36_500), diff --git a/backend/app/youtube/client.py b/backend/app/youtube/client.py index aa1490a..b4e5ef6 100644 --- a/backend/app/youtube/client.py +++ b/backend/app/youtube/client.py @@ -351,6 +351,49 @@ class YouTubeClient: """Delete a whole playlist on YouTube. 50 units.""" self._write("DELETE", "playlists", params={"id": playlist_id}) + def search_videos( + self, q: str, page_token: str | None = None, order: str = "relevance" + ) -> dict: + """Live YouTube search (search.list) for embeddable videos. **100 quota units per + call** — by far the most expensive read, so callers must gate it (per-user cap + + budget pre-check). Public read: prefers the API key over OAuth. + + Returns one page (up to 50) as ``{"items": [...], "next_page_token": str|None}`` + where each item is a flat dict ready for stub insertion. `liveBroadcastContent` is + surfaced so the caller can drop currently-live/upcoming results (search can't filter + those out for us). Snippet carries no duration/stats — the caller enriches via + videos.list afterwards.""" + params = { + "part": "snippet", + "q": q, + "type": "video", + "maxResults": 50, + "order": order, + "videoEmbeddable": "true", + "safeSearch": "none", + } + if page_token: + params["pageToken"] = page_token + data = self._get("search", params, cost=100) + items: list[dict] = [] + for it in data.get("items", []): + vid = it.get("id", {}).get("videoId") + if not vid: + continue + sn = it.get("snippet", {}) + items.append( + { + "id": vid, + "channel_id": sn.get("channelId"), + "channel_title": sn.get("channelTitle"), + "title": sn.get("title"), + "published_at": sn.get("publishedAt"), + "thumbnail_url": best_thumbnail(sn.get("thumbnails")), + "live_broadcast": sn.get("liveBroadcastContent") or "none", + } + ) + return {"items": items, "next_page_token": data.get("nextPageToken")} + def get_videos(self, video_ids: list[str]) -> list[dict]: items: list[dict] = [] for batch in _chunks(video_ids, 50):