peter/siftlode
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(youtube): a related page is empty only on YouTube's own word

Browse Source 
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.
This commit is contained in:
peter
2026-10-25 00:32:27 +02:00
parent b04b51c909
commit d53729a63d
4 changed files with 109 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/youtube/innertube.py
+49
View File
@@ -54,6 +54,55 @@ def runs_text(node: dict | None) -> str | None:
return None
def scraped_video(
vid: str, channel_id: object, channel_title: object, title: object, thumbnails: object
) -> dict:
"""One scraped video in the shape `YouTubeClient.search_videos` answers, so a page of either
source is ingested the same way. `published_at` is None (the website shows only a relative
"3 years ago"; the enrich fills the real date) and `live_broadcast` "none" (live/upcoming are
judged after the enrich, by `live_status`). The largest thumbnail is the last."""
thumbs = thumbnails if isinstance(thumbnails, list) else []
last = thumbs[-1] if thumbs else None
return {
"id": vid,
"channel_id": channel_id if isinstance(channel_id, str) and channel_id else None,
"channel_title": channel_title if isinstance(channel_title, str) else None,
"title": title if isinstance(title, str) else None,
"published_at": None,
"thumbnail_url": last.get("url") if isinstance(last, dict) else None,
"live_broadcast": "none",
}
def video_renderer(node: dict) -> dict | None:
"""A `videoRenderer` (search) or `compactVideoRenderer` (the older related column) flattened —
the two draw a video the same way. None without a video id; the channel is None when the byline
names none."""
vid = node.get("videoId")
if not isinstance(vid, str) or not vid:
return None
owner = node.get("ownerText") or node.get("longBylineText") or {}
runs = owner.get("runs") if isinstance(owner, dict) else None
first = runs[0] if isinstance(runs, list) and runs and isinstance(runs[0], dict) else {}
endpoint = (first.get("navigationEndpoint") or {}).get("browseEndpoint") or {}
return scraped_video(
vid,
endpoint.get("browseId"),
first.get("text"),
runs_text(node.get("title")),
(node.get("thumbnail") or {}).get("thumbnails"),
)
def continuation_token(node: object) -> str | None:
"""The "load more" cursor of a `continuationItemRenderer`, or None when `node` is not one."""
if not isinstance(node, dict):
return None
command = (node.get("continuationEndpoint") or {}).get("continuationCommand") or {}
token = command.get("token")
return token if isinstance(token, str) and token else 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.
backend/app/youtube/related_scrape.py
+35 -44
View File
@@ -13,7 +13,13 @@ channel up front, and the stack does not say which one uploaded it.
`hl`/`gl` pick the locale YouTube fills the list in: ``en``/``US`` keeps it on the video's topic,
``hu``/``HU`` mixes in what is trending in Hungary (measured)."""
from app.youtube.innertube import ScrapeError, post, runs_text
from app.youtube.innertube import (
ScrapeError,
continuation_token,
post,
scraped_video,
video_renderer,
)
__all__ = ["ScrapeError", "related"]
@@ -40,21 +46,6 @@ def _at(node: object, *path: str | int) -> object:
return node
def _item(vid: str, channel_id: object, channel_title: object, title: object, sources: object):
if not isinstance(channel_id, str) or not channel_id:
return None
thumbs = sources if isinstance(sources, list) else []
return {
"id": vid,
"channel_id": channel_id,
"channel_title": channel_title if isinstance(channel_title, str) else None,
"title": title if isinstance(title, str) else None,
"published_at": None, # only a relative "1y ago" here; the enrich fills the real date
"thumbnail_url": _at(thumbs, -1, "url"),
"live_broadcast": "none", # live/upcoming are judged after the enrich, by live_status
}
def _lockup(node: dict) -> dict | None:
if node.get("contentType") != "LOCKUP_CONTENT_TYPE_VIDEO" or not node.get("contentId"):
return None
@@ -73,7 +64,7 @@ def _lockup(node: dict) -> dict | None:
channel_title = _at(
meta, "metadata", "contentMetadataViewModel", "metadataRows", 0, "metadataParts", 0
)
return _item(
return scraped_video(
node["contentId"],
channel_id,
_at(channel_title, "text", "content"),
@@ -82,18 +73,17 @@ def _lockup(node: dict) -> dict | None:
)
def _compact(node: dict) -> dict | None:
vid = node.get("videoId")
if not vid:
return None
byline = _at(node, "longBylineText", "runs", 0)
return _item(
vid,
_at(byline, "navigationEndpoint", "browseEndpoint", "browseId"),
_at(byline, "text"),
runs_text(node.get("title")),
_at(node, "thumbnail", "thumbnails"),
)
def _unavailable(data: dict) -> bool:
"""Whether the watch page says the video is not available: its main column draws a
`backgroundPromoRenderer` ("This video isn't available anymore") and there is no related column.
YouTube's own word for "nothing related" — the one empty page that is not a shape we lost."""
sections = _at(data, "contents", "twoColumnWatchNextResults", "results", "results", "contents")
for section in sections if isinstance(sections, list) else []:
inner = _at(section, "itemSectionRenderer", "contents")
for node in inner if isinstance(inner, list) else []:
if isinstance(node, dict) and "backgroundPromoRenderer" in node:
return True
return False
def _items(data: dict) -> list | None:
@@ -101,9 +91,7 @@ def _items(data: dict) -> list | None:
first = _at(data, *_FIRST)
if isinstance(first, list):
return first
# A video YouTube no longer has keeps the watch page's frame but draws an "isn't available"
# promo in it and no related column: nothing related, said by YouTube itself.
if isinstance(_at(data, *_FIRST[:2]), dict):
if _unavailable(data):
return []
for endpoint in data.get("onResponseReceivedEndpoints") or []:
for key in _APPENDS:
@@ -118,8 +106,9 @@ def related(
) -> dict:
"""One page of `video_id`'s related videos (`continuation` = the token a previous page gave).
A video YouTube no longer has is an empty page. An answer without the watch page's frame is a
`ScrapeError` — not an empty page, which would read as "nothing related"."""
A video YouTube says it no longer has is an empty page. An answer of a shape this does not know
— no related column, or one whose entries are all renderers it cannot read — is a `ScrapeError`:
not an empty page, which would read as "nothing related"."""
body = {"continuation": continuation} if continuation else {"videoId": video_id}
data = post("next", body, hl=hl, gl=gl)
items = _items(data)
@@ -128,23 +117,25 @@ def related(
out: list[dict] = []
seen: set[str] = set()
token = None
known = 0 # entries drawn by a renderer read here (a video's, or the cursor)
for entry in items:
if not isinstance(entry, dict):
continue
if "lockupViewModel" in entry:
item = _lockup(entry["lockupViewModel"])
elif "compactVideoRenderer" in entry:
item = _compact(entry["compactVideoRenderer"])
item = video_renderer(entry["compactVideoRenderer"])
else:
token = token or _at(
entry,
"continuationItemRenderer",
"continuationEndpoint",
"continuationCommand",
"token",
)
found = continuation_token(entry.get("continuationItemRenderer"))
known += found is not None
token = token or found
continue
if item and item["id"] not in seen:
known += 1
# A video by several channels draws an avatar stack, not one: no channel, left out.
if item and item["channel_id"] and item["id"] not in seen:
seen.add(item["id"])
out.append(item)
return {"items": out, "next_page_token": token if isinstance(token, str) else None}
# A list with entries but none of them one we read: YouTube draws its items some new way.
if items and not known:
raise ScrapeError("InnerTube's related list holds no renderer this reads")
return {"items": out, "next_page_token": token}
backend/app/youtube/search_scrape.py
+5 -30
View File
@@ -17,7 +17,7 @@ only difference is the cursor is an InnerTube *continuation token*, not a Data A
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
from app.youtube.innertube import ScrapeError, continuation_token, post, video_renderer
__all__ = ["ScrapeError", "search"]
@@ -29,29 +29,6 @@ _TYPE_VIDEO = "EgIQAQ=="
_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
@@ -59,7 +36,7 @@ def _collect_videos(obj, out: list[dict], seen: set[str]) -> None:
if isinstance(obj, dict):
vr = obj.get("videoRenderer")
if isinstance(vr, dict):
item = _video_item(vr)
item = video_renderer(vr)
if item and item["id"] not in seen:
seen.add(item["id"])
out.append(item)
@@ -73,11 +50,9 @@ def _collect_videos(obj, out: list[dict], seen: set[str]) -> None:
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
tok = continuation_token(obj.get("continuationItemRenderer"))
if tok:
return tok
for v in obj.values():
tok = _find_continuation(v)
if tok: