siftlode/backend/app/youtube/search_scrape.py

"""Zero-quota YouTube search via InnerTube (the website's own youtubei/v1/search endpoint).

Live YouTube search **without spending any YouTube Data API quota**: instead of search.list
(100 units per page) we POST to YouTube's internal InnerTube endpoint — the same one the website
calls — and read the result list straight out of the JSON. The returned video ids then go
through the normal enrich path (videos.list, 1 unit / 50), exactly as the official search does,
so the search itself is effectively free and only the (cheap, shared) enrichment is charged.

The SOCS=CAI consent-skip cookie is used like the Shorts probe so a server-side request isn't
bounced to consent.youtube.com. This deliberately scrapes an undocumented endpoint: if YouTube
changes the JSON shape or blocks the host, an admin can flip ``search_source`` back to ``api``.

The returned page mirrors :meth:`YouTubeClient.search_videos` exactly
(``{"items": [...], "next_page_token": str | None}``) so the route is source-agnostic — the
only difference is the cursor is an InnerTube *continuation token*, not a Data API pageToken.
``published_at`` is left ``None`` (the scrape only exposes a relative time like "3 years ago");
the real date is filled by the videos.list enrich. ``live_broadcast`` is reported as ``"none"``
and live/upcoming filtering happens post-enrich via ``live_status`` (same as the API path)."""
import re

import httpx

_SEARCH_URL = "https://www.youtube.com/youtubei/v1/search"
_RESULTS_URL = "https://www.youtube.com/results"
# Public WEB InnerTube key + a recent client version. These are also embedded in every results
# page; we fetch the live pair once and cache it, falling back to these constants.
_FALLBACK_KEY = "AIzaSyAO_FJ2SlqU8Q4STEHLGCilw_Y9_11qcW8"
_FALLBACK_CLIENT_VERSION = "2.20240101.00.00"
# sp / params value selecting the "type: video" search filter (base64 protobuf EgIQAQ==).
_TYPE_VIDEO = "EgIQAQ=="

_HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
    # SOCS skips YouTube's cookie-consent interstitial (otherwise every server-side request is
    # redirected to consent.youtube.com).
    "Cookie": "SOCS=CAI",
    "Accept-Language": "en-US,en;q=0.9",
}

_KEY_RE = re.compile(r'"INNERTUBE_API_KEY":"([^"]+)"')
_VER_RE = re.compile(r'"INNERTUBE_CLIENT_VERSION":"([^"]+)"')

# Cached (api_key, client_version) extracted from a results page; refreshed on demand.
_config: tuple[str, str] | None = None


class ScrapeError(Exception):
    """A scrape-based search could not be completed (network, blocked, or unexpected shape)."""


def _runs_text(node: dict | None) -> str | None:
    if not node:
        return None
    if node.get("simpleText"):
        return node["simpleText"]
    runs = node.get("runs") or []
    if runs:
        return "".join(r.get("text", "") for r in runs)
    return None


def _get_config(client: httpx.Client) -> tuple[str, str]:
    """The current InnerTube (api_key, client_version), cached after the first fetch."""
    global _config
    if _config is not None:
        return _config
    try:
        resp = client.get(_RESULTS_URL, params={"search_query": "music"})
        html = resp.text
        key = _KEY_RE.search(html)
        ver = _VER_RE.search(html)
        _config = (
            key.group(1) if key else _FALLBACK_KEY,
            ver.group(1) if ver else _FALLBACK_CLIENT_VERSION,
        )
    except httpx.HTTPError:
        _config = (_FALLBACK_KEY, _FALLBACK_CLIENT_VERSION)
    return _config


def _video_item(vr: dict) -> dict | None:
    """Flatten a YouTube `videoRenderer` into the same shape as YouTubeClient.search_videos."""
    vid = vr.get("videoId")
    if not vid:
        return None
    owner = vr.get("ownerText") or vr.get("longBylineText") or {}
    runs = owner.get("runs") or []
    channel_id = channel_title = None
    if runs:
        channel_title = runs[0].get("text")
        channel_id = (
            runs[0]
            .get("navigationEndpoint", {})
            .get("browseEndpoint", {})
            .get("browseId")
        )
    thumbs = (vr.get("thumbnail") or {}).get("thumbnails") or []
    return {
        "id": vid,
        "channel_id": channel_id,
        "channel_title": channel_title,
        "title": _runs_text(vr.get("title")),
        "published_at": None,  # scrape exposes only a relative time; enrich fills the real date
        "thumbnail_url": thumbs[-1]["url"] if thumbs else None,
        "live_broadcast": "none",  # live/upcoming filtered post-enrich via live_status
    }


def _collect_videos(obj, out: list[dict], seen: set[str]) -> None:
    """Walk the response collecting videoRenderer items in document (relevance) order. Only
    `videoRenderer` is taken, so channels/playlists/Shorts shelves (reelItemRenderer) are
    naturally excluded."""
    if isinstance(obj, dict):
        vr = obj.get("videoRenderer")
        if isinstance(vr, dict):
            item = _video_item(vr)
            if item and item["id"] not in seen:
                seen.add(item["id"])
                out.append(item)
        for v in obj.values():
            _collect_videos(v, out, seen)
    elif isinstance(obj, list):
        for v in obj:
            _collect_videos(v, out, seen)


def _find_continuation(obj) -> str | None:
    """First continuationItemRenderer token (the 'load more' cursor), or None at the end."""
    if isinstance(obj, dict):
        cir = obj.get("continuationItemRenderer")
        if isinstance(cir, dict):
            tok = (
                cir.get("continuationEndpoint", {})
                .get("continuationCommand", {})
                .get("token")
            )
            if tok:
                return tok
        for v in obj.values():
            tok = _find_continuation(v)
            if tok:
                return tok
    elif isinstance(obj, list):
        for v in obj:
            tok = _find_continuation(v)
            if tok:
                return tok
    return None


def search(q: str, continuation: str | None = None) -> dict:
    """One page of zero-quota YouTube search results.

    `continuation` is the InnerTube continuation token from a previous page (None for the first
    page). Returns ``{"items": [...], "next_page_token": str | None}`` — the same shape as
    YouTubeClient.search_videos so the route handles both sources identically."""
    with httpx.Client(timeout=20.0, headers=_HEADERS) as client:
        api_key, client_version = _get_config(client)
        context = {
            "client": {
                "clientName": "WEB",
                "clientVersion": client_version,
                "hl": "en",
                "gl": "US",
            }
        }
        body: dict = {"context": context}
        if continuation:
            body["continuation"] = continuation
        else:
            body["query"] = q
            body["params"] = _TYPE_VIDEO
        try:
            resp = client.post(_SEARCH_URL, params={"key": api_key}, json=body)
            resp.raise_for_status()
            data = resp.json()
        except httpx.HTTPError as exc:
            raise ScrapeError(f"InnerTube request failed: {exc}") from exc
        except ValueError as exc:  # non-JSON body (blocked / consent / captcha)
            raise ScrapeError("InnerTube returned a non-JSON response") from exc

    items: list[dict] = []
    _collect_videos(data, items, set())
    return {"items": items, "next_page_token": _find_continuation(data)}