peter
d53729a63d
The watch frame alone read as "nothing related" — so a related column YouTube moved, or a list drawn by renderers this does not read, answered an empty list instead of an error. Now only the "isn't available" promo YouTube draws for a video it no longer has is an empty page; anything else it cannot read is a ScrapeError. The flattening of a video renderer and the read of the continuation token are one each, in innertube, for the search and the related scrape.
86 lines
3.9 KiB
Python
86 lines
3.9 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 request itself (the live key, the consent cookie) is `innertube`'s. 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)."""
|
|
|
|
from app.youtube.innertube import ScrapeError, continuation_token, post, video_renderer
|
|
|
|
__all__ = ["ScrapeError", "search"]
|
|
|
|
# sp / params value selecting the "type: video" search filter (base64 protobuf EgIQAQ==).
|
|
_TYPE_VIDEO = "EgIQAQ=="
|
|
|
|
# The top-level keys a first page (contents) and a continuation page (onResponseReceivedCommands)
|
|
# put their result list under.
|
|
_RESULT_FRAMES = {"contents", "onResponseReceivedCommands"}
|
|
|
|
|
|
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_renderer(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):
|
|
tok = continuation_token(obj.get("continuationItemRenderer"))
|
|
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."""
|
|
body: dict = (
|
|
{"continuation": continuation} if continuation else {"query": q, "params": _TYPE_VIDEO}
|
|
)
|
|
data = post("search", body)
|
|
|
|
# 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 _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)}
|