From d53729a63dd9857220201c7d8bdb6dcb4f08256c Mon Sep 17 00:00:00 2001 From: npeter83 Date: Sun, 25 Oct 2026 00:32:27 +0200 Subject: [PATCH] fix(youtube): a related page is empty only on YouTube's own word MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- backend/app/youtube/innertube.py | 49 +++++++++++++++++ backend/app/youtube/related_scrape.py | 79 ++++++++++++--------------- backend/app/youtube/search_scrape.py | 35 ++---------- backend/tests/test_related_scrape.py | 20 +++++++ 4 files changed, 109 insertions(+), 74 deletions(-) diff --git a/backend/app/youtube/innertube.py b/backend/app/youtube/innertube.py index 54a8083e..ec6316d8 100644 --- a/backend/app/youtube/innertube.py +++ b/backend/app/youtube/innertube.py @@ -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. diff --git a/backend/app/youtube/related_scrape.py b/backend/app/youtube/related_scrape.py index e66a725c..33f4124e 100644 --- a/backend/app/youtube/related_scrape.py +++ b/backend/app/youtube/related_scrape.py @@ -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} diff --git a/backend/app/youtube/search_scrape.py b/backend/app/youtube/search_scrape.py index e32583e2..22adeac0 100644 --- a/backend/app/youtube/search_scrape.py +++ b/backend/app/youtube/search_scrape.py @@ -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: diff --git a/backend/tests/test_related_scrape.py b/backend/tests/test_related_scrape.py index 2897ba6d..b0efb93c 100644 --- a/backend/tests/test_related_scrape.py +++ b/backend/tests/test_related_scrape.py @@ -191,6 +191,26 @@ def test_a_video_youtube_no_longer_has_is_an_empty_page(net): assert related_scrape.related("aaaaaaaaaaa") == {"items": [], "next_page_token": None} +def test_a_watch_frame_with_no_related_column_and_no_word_why_is_an_error(net): + # The frame alone, its related column moved somewhere this does not look: a lost shape, not + # "nothing related". + page = copy.deepcopy(FIRST) + watch = page["contents"]["twoColumnWatchNextResults"] + watch["sideColumn"] = watch.pop("secondaryResults") + _answer(net, page) + with pytest.raises(related_scrape.ScrapeError): + related_scrape.related("SdTZAMDKrNY") + + +def test_a_related_list_drawn_by_renderers_this_does_not_read_is_an_error(net): + page = copy.deepcopy(FIRST) + results = _results(page) + results[:] = [{"videoTileViewModel": entry} for entry in results] + _answer(net, page) + with pytest.raises(related_scrape.ScrapeError): + related_scrape.related("SdTZAMDKrNY") + + @pytest.mark.parametrize( "answer", [