siftlode/backend/app/routes/search.py

"""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,
    }