peter
8b37c470ba
GET /api/search/related/{video_id} reads the list YouTube shows beside a
video from the website's InnerTube "next" endpoint (no quota), and brings
it into the catalog the way live-search results come in. It leaves out
what the user watched or hid, here or in their Takeout history (unless
taken back to "new" since), blocked channels and the video itself, and
pages on until 20 are gathered (at most 4 pages). region=topical|hu picks
the locale the list is filled in. Its own per-user daily cap
(related_daily_limit_per_user, one event per opened list); a failed
scrape answers 422 youtube/unavailable.
The InnerTube request (live key, consent cookie, errors) moves into
youtube/innertube.py, shared by the search scrape and the related one.
111 lines
5.0 KiB
Python
111 lines
5.0 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, post, runs_text
|
|
|
|
__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 _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."""
|
|
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)}
|