"""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_, delete, func, or_, select, update 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 ( LIVE_OR_UPCOMING, BlockedChannel, Channel, Playlist, PlaylistItem, SearchFind, Subscription, 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 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] def _ingest_candidates( db: Session, yt: YouTubeClient, user: User, candidates: list[dict] ) -> list[str]: """Materialise one page of search candidates into the catalog (channel stubs + enrich, video stubs flagged via_search, enrich, Shorts classification) and return the KEEPABLE video ids — non-Short, non-live — in the input relevance order.""" ids = [it["id"] for it in candidates] if not ids: return [] # 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) 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] def _new_first(db: Session, user: User, ids: list[str]) -> list[str]: """Reorder results so genuinely-new discoveries come first and things the user already has (a subscribed channel, or a video they've watched / in-progress / saved / playlisted) sink to the back — keeping the relevance order within each group.""" if not ids: return ids sub_ch = select(Subscription.channel_id).where(Subscription.user_id == user.id) has_state = ( select(VideoState.id) .where(VideoState.video_id == Video.id, VideoState.user_id == user.id) .exists() ) in_playlist = ( select(PlaylistItem.id) .join(Playlist, Playlist.id == PlaylistItem.playlist_id) .where(PlaylistItem.video_id == Video.id, Playlist.user_id == user.id) .exists() ) have = set( db.scalars( select(Video.id).where( Video.id.in_(ids), or_(Video.channel_id.in_(sub_ch), has_state, in_playlist), ) ).all() ) return [i for i in ids if i not in have] + [i for i in ids if i in have] @router.get("/youtube") def search_youtube( q: str | None = None, cursor: str | None = None, limit: int = 20, user: User = Depends(require_human), db: Session = Depends(get_db), ) -> dict: """Live YouTube search results materialised into the catalog. `limit` is how many results to gather (the count selector); with the free scrape source we page through continuations until that many are collected, so there's no manual "load more". The API source costs 100 units per page, so it returns a single page (≤50) regardless. `cursor` lets the caller fetch further.""" term = (q or "").strip() if not term: raise HTTPException(status_code=400, detail="A search term is required.") limit = max(1, min(limit, 100)) # 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.", ) # Channels this user has blocked are dropped before ingestion — never shown, never stored. blocked = set( db.scalars( select(BlockedChannel.channel_id).where(BlockedChannel.user_id == user.id) ).all() ) collected: list[str] = [] next_cursor = cursor MAX_PAGES = 12 # scrape safety bound with YouTubeClient(db, user) as yt: for _ in range(MAX_PAGES): try: if source == "api": with quota.attribute(user.id, quota.QuotaAction.VIDEOS_SEARCH): page = yt.search_videos(term, page_token=next_cursor) else: page = search_scrape.search(term, continuation=next_cursor) except (YouTubeError, search_scrape.ScrapeError) as exc: if collected: break # keep what we already gathered raise HTTPException(status_code=502, detail=f"YouTube search failed: {exc}") # Drop currently-live/upcoming, channel-less, and blocked-channel results. candidates = [ it for it in page["items"] if it["live_broadcast"] == "none" and it["channel_id"] and it["channel_id"] not in blocked ] collected.extend(_ingest_candidates(db, yt, user, candidates)) next_cursor = page["next_page_token"] # The API source pages cost 100 units each — never auto-page; one page per request. if source == "api" or len(collected) >= limit or not next_cursor: break # Count this as ONE per-user search against the daily cap. The scrape source spends no # quota, so it never logs an event on its own; the API source already logged exactly one # (via record_usage — it never auto-pages). Logging once here, after a search that produced # at least one page (a total failure raises 502 above and never reaches this), keeps the cap # counting user searches instead of internal continuation pages. if source != "api": quota.log_action(db, user.id, quota.QuotaAction.VIDEOS_SEARCH) ordered = collected[:limit] if ordered: # Record these as THIS user's search finds (Mine feed / Source filter), and fold the # query into each result's search_terms so local full-text search inherits YouTube's # relevance. Idempotent; the generated search_vector updates automatically. 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.execute( update(Video) .where(Video.id.in_(ordered)) .where(func.coalesce(Video.search_terms, "").notilike(f"%{term}%")) .values( search_terms=func.trim( func.coalesce(Video.search_terms, "").op("||")(" ").op("||")(term) ) ) ) db.commit() ordered = _new_first(db, user, ordered) return { "items": _serialize_ids(db, user, ordered), "next_cursor": next_cursor, "has_more": bool(next_cursor), "source": source, } @router.post("/clear") def clear_search( payload: dict, user: User = Depends(require_human), db: Session = Depends(get_db), ) -> dict: """Discard the given live-search results "as if never added": drop this user's search-finds for them, then hard-delete the now-orphaned ones — a via_search video nobody KEPT (no watch state, not playlisted, channel not subscribed) and that no other user still has as a search find. Kept videos and organically-owned ones are left untouched.""" ids = [str(v) for v in (payload.get("video_ids") or [])] if not ids: return {"removed": 0} db.execute( delete(SearchFind).where( SearchFind.user_id == user.id, SearchFind.video_id.in_(ids) ) ) db.commit() has_state = select(VideoState.id).where(VideoState.video_id == Video.id).exists() in_playlist = select(PlaylistItem.id).where(PlaylistItem.video_id == Video.id).exists() subscribed = ( select(Subscription.id).where(Subscription.channel_id == Video.channel_id).exists() ) other_find = select(SearchFind.id).where(SearchFind.video_id == Video.id).exists() removed = db.execute( delete(Video).where( Video.id.in_(ids), Video.via_search.is_(True), ~has_state, ~in_playlist, ~subscribed, ~other_find, ) ).rowcount db.commit() return {"removed": removed}