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_, 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.enriched_at is None:
            # Enrichment failed (proxy blip) or YouTube omitted the row — duration is unknown,
            # so DON'T finalise shorts_probed off a null duration; leave it for the scheduler
            # (which only classifies enriched rows) to retry. Otherwise a real Short would leak.
            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 v.enriched_at is not None
        and 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}