"""Live YouTube search. Searches YouTube directly 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). The source is admin-selectable (``search_source``): **scrape** (InnerTube, the website's own endpoint, zero API quota — the default) or **api** (search.list, **100 quota units per page**). Either way it's gated behind a per-user daily cap, the API source additionally behind a budget pre-check, and only real users (never the shared demo account) may call it. The result ids are enriched through the cheap shared videos.list path (1 unit / 50) regardless of source. 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.dialects.postgresql import insert as pg_insert 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, SearchFind, 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 import search_scrape 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.") # Source: "scrape" (InnerTube, no API quota) or "api" (search.list, 100 units/page). source = (sysconfig.get_str(db, "search_source") or "scrape").lower() if source != "scrape": source = "api" # Per-user daily cap applies to both sources (anti-abuse); with scrape it's the only guard # since the search itself spends no quota. 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.", ) # The 100-unit budget pre-check only applies to the API source. if source == "api" and 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: if source == "api": with quota.attribute(user.id, quota.QuotaAction.VIDEOS_SEARCH): page = yt.search_videos(term, page_token=cursor) else: page = search_scrape.search(term, continuation=cursor) # Zero-cost, but logged so the per-user daily cap above keeps counting it. quota.log_action(db, user.id, quota.QuotaAction.VIDEOS_SEARCH) except (YouTubeError, search_scrape.ScrapeError) 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"]), "source": source, } # 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] # 6) Record these as THIS user's search finds so they surface in their own Mine feed # (and the Mine Source filter), independent of the global via_search flag. Idempotent. if ordered: db.execute( pg_insert(SearchFind) .values([{"user_id": user.id, "video_id": vid} for vid in ordered]) .on_conflict_do_nothing(index_elements=["user_id", "video_id"]) ) db.commit() return { "items": _serialize_ids(db, user, ordered), "next_cursor": page["next_page_token"], "has_more": bool(page["next_page_token"]), "source": source, }