peter
378c4daf0f
A failed config fetch or a keyless consent page cached the fallback InnerTube key for the life of the process; once it went stale, search returned nothing with no error. The fallback is now used for that call only. A 200 JSON without a result frame returned an empty page that read as no results; it is a ScrapeError now. Parser, cursor and Shorts-probe tests added.
202 lines
8.5 KiB
Python
202 lines
8.5 KiB
Python
"""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
|
|
|
|
from app import logs
|
|
|
|
log = logs.get_logger(__name__)
|
|
|
|
_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",
|
|
}
|
|
|
|
# The top-level keys a first page (contents) and a continuation page (onResponseReceivedCommands)
|
|
# put their result list under.
|
|
_RESULT_FRAMES = {"contents", "onResponseReceivedCommands"}
|
|
_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 that FOUND them.
|
|
|
|
The fallback pair is never cached: it used to be, so one failed fetch (or a consent page with no
|
|
keys on it) at the wrong moment pinned every later search to a hard-coded key and version for
|
|
the life of the process — and when those went stale, search returned nothing, with no error
|
|
anywhere (E32 S2). Now the next search simply tries the live pair again."""
|
|
global _config
|
|
if _config is not None:
|
|
return _config
|
|
try:
|
|
html = client.get(_RESULTS_URL, params={"search_query": "music"}).text
|
|
except httpx.HTTPError as exc:
|
|
log.warning(
|
|
"the InnerTube config could not be fetched; falling back to the pinned key",
|
|
extra=logs.event("youtube.scrape.config_fallback", reason=type(exc).__name__),
|
|
)
|
|
return (_FALLBACK_KEY, _FALLBACK_CLIENT_VERSION)
|
|
key = _KEY_RE.search(html)
|
|
ver = _VER_RE.search(html)
|
|
if key and ver:
|
|
_config = (key.group(1), ver.group(1))
|
|
return _config
|
|
log.warning(
|
|
"the results page carried no InnerTube config; falling back to the pinned key",
|
|
extra=logs.event("youtube.scrape.config_fallback", reason="not_on_page"),
|
|
)
|
|
return (
|
|
key.group(1) if key else _FALLBACK_KEY,
|
|
ver.group(1) if ver else _FALLBACK_CLIENT_VERSION,
|
|
)
|
|
|
|
|
|
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
|
|
|
|
# Even a search with no results carries its result list's frame. Without one this is not "no
|
|
# results" but a shape we do not know (YouTube moved, or a block page in JSON) — say so.
|
|
if not isinstance(data, dict) or not _RESULT_FRAMES & data.keys():
|
|
raise ScrapeError("InnerTube returned an unexpected shape")
|
|
items: list[dict] = []
|
|
_collect_videos(data, items, set())
|
|
return {"items": items, "next_page_token": _find_continuation(data)}
|