feat(search): zero-quota InnerTube scrape source + search_source toggle
Live YouTube search can now use YouTube's internal InnerTube endpoint instead of search.list, materialising results through the same enrich/provenance path at zero API quota (search.list costs 100 units/page; scrape costs nothing, only the cheap shared videos.list enrich is charged). - youtube/search_scrape.py: InnerTube search returning the same page shape as YouTubeClient.search_videos (items + continuation cursor); SOCS consent cookie, videoRenderer walk (channels/playlists/Shorts shelves naturally excluded), type:video filter, cached InnerTube key/version with constant fallbacks. - routes/search.py: admin-selectable source (search_source, default scrape). Scrape path skips the 100-unit budget pre-check and logs a zero-cost search event so the per-user daily cap still counts it; api path unchanged. Response carries the active source. - sysconfig/config: new search_source key (scrape|api). - quota.log_action: record a zero-cost action event for per-user rate limits.
This commit is contained in:
npeter83
parent
17c17c1d3c
commit
396e09189b
5 changed files with 228 additions and 12 deletions
184
backend/app/youtube/search_scrape.py
Normal file
184
backend/app/youtube/search_scrape.py
Normal file
|
|
@ -0,0 +1,184 @@
|
|||
"""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)}
|
||||
Loading…
Add table
Add a link
Reference in a new issue