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.
233 lines
8.3 KiB
Python
233 lines
8.3 KiB
Python
"""youtube/related_scrape — a video's related videos from the InnerTube `next` endpoint.
|
|
|
|
The two pages under `fixtures/` are real answers (2026-10-24), trimmed to three items and the
|
|
continuation: they hold the `lockupViewModel` shape YouTube draws today. Everything here breaks
|
|
silently when YouTube moves — a parser that stops finding videos, a cursor that stops paging — so a
|
|
shape it does not know must say so, never read as "nothing related". The network is an httpx
|
|
MockTransport."""
|
|
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.youtube import innertube, related_scrape
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures"
|
|
FIRST = json.loads((FIXTURES / "innertube_next_first.json").read_text(encoding="utf-8"))
|
|
CONTINUATION = json.loads(
|
|
(FIXTURES / "innertube_next_continuation.json").read_text(encoding="utf-8")
|
|
)
|
|
RESULTS_HTML = (
|
|
'<script>{"INNERTUBE_API_KEY":"live-key","INNERTUBE_CLIENT_VERSION":"2.20261001"}</script>'
|
|
)
|
|
|
|
|
|
def _results(page: dict) -> list:
|
|
return page["contents"]["twoColumnWatchNextResults"]["secondaryResults"]["secondaryResults"][
|
|
"results"
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def net(monkeypatch):
|
|
"""Route every httpx.Client the InnerTube module opens through a handler the test sets."""
|
|
calls: list[httpx.Request] = []
|
|
state: dict = {"handler": None, "calls": calls}
|
|
|
|
def transport(request: httpx.Request) -> httpx.Response:
|
|
calls.append(request)
|
|
if request.url.path == "/results":
|
|
return httpx.Response(200, text=RESULTS_HTML)
|
|
return state["handler"](request)
|
|
|
|
real = httpx.Client
|
|
monkeypatch.setattr(
|
|
innertube.httpx,
|
|
"Client",
|
|
lambda **kw: real(transport=httpx.MockTransport(transport), **kw),
|
|
)
|
|
monkeypatch.setattr(innertube, "_config", None)
|
|
return state
|
|
|
|
|
|
def _answer(net, page, status=200):
|
|
net["handler"] = lambda request: (
|
|
httpx.Response(status, text=page)
|
|
if isinstance(page, str)
|
|
else httpx.Response(status, json=page)
|
|
)
|
|
|
|
|
|
def test_a_first_page_gives_its_videos_with_their_channel_and_the_next_cursor(net):
|
|
_answer(net, FIRST)
|
|
page = related_scrape.related("SdTZAMDKrNY")
|
|
assert [it["id"] for it in page["items"]] == ["to-lHJfK4pw", "R8V8CbuxryI", "HLKo4fJx_7k"]
|
|
first = page["items"][0]
|
|
assert first["channel_id"] == "UCtslD4DGH6PKyG_1gFAX7sg"
|
|
assert first["channel_title"] == "Alexander Amini"
|
|
assert first["title"] == "MIT 6.S191 (2025): Reinforcement Learning"
|
|
assert first["thumbnail_url"].startswith("https://i.ytimg.com/vi/to-lHJfK4pw/")
|
|
assert first["published_at"] is None and first["live_broadcast"] == "none"
|
|
assert page["next_page_token"]
|
|
body = json.loads(net["calls"][-1].content)
|
|
assert body["videoId"] == "SdTZAMDKrNY" and "continuation" not in body
|
|
assert net["calls"][-1].url.path == "/youtubei/v1/next"
|
|
|
|
|
|
def test_the_largest_thumbnail_is_taken(net):
|
|
_answer(net, FIRST)
|
|
item = related_scrape.related("SdTZAMDKrNY")["items"][0]
|
|
sources = _results(FIRST)[0]["lockupViewModel"]["contentImage"]["thumbnailViewModel"]["image"][
|
|
"sources"
|
|
]
|
|
assert item["thumbnail_url"] == sources[-1]["url"]
|
|
|
|
|
|
def test_a_continuation_page_reads_its_videos_and_sends_only_the_cursor(net):
|
|
_answer(net, CONTINUATION)
|
|
page = related_scrape.related("SdTZAMDKrNY", continuation="TOKEN", hl="hu", gl="HU")
|
|
assert len(page["items"]) == 2
|
|
assert all(it["channel_id"].startswith("UC") for it in page["items"])
|
|
assert page["next_page_token"]
|
|
body = json.loads(net["calls"][-1].content)
|
|
assert body["continuation"] == "TOKEN" and "videoId" not in body
|
|
assert body["context"]["client"]["hl"] == "hu" and body["context"]["client"]["gl"] == "HU"
|
|
|
|
|
|
def test_the_locale_defaults_to_the_topical_one(net):
|
|
_answer(net, FIRST)
|
|
related_scrape.related("SdTZAMDKrNY")
|
|
client = json.loads(net["calls"][-1].content)["context"]["client"]
|
|
assert (client["hl"], client["gl"]) == ("en", "US")
|
|
|
|
|
|
def test_a_video_by_several_channels_is_left_out(net):
|
|
page = copy.deepcopy(FIRST)
|
|
meta = _results(page)[1]["lockupViewModel"]["metadata"]["lockupMetadataViewModel"]
|
|
meta["image"] = {"avatarStackViewModel": {"avatars": []}}
|
|
_answer(net, page)
|
|
ids = [it["id"] for it in related_scrape.related("SdTZAMDKrNY")["items"]]
|
|
assert ids == ["to-lHJfK4pw", "HLKo4fJx_7k"]
|
|
|
|
|
|
def test_what_is_not_a_video_is_left_out_and_a_repeat_counts_once(net):
|
|
page = copy.deepcopy(FIRST)
|
|
results = _results(page)
|
|
playlist = copy.deepcopy(results[0])
|
|
playlist["lockupViewModel"]["contentType"] = "LOCKUP_CONTENT_TYPE_PLAYLIST"
|
|
playlist["lockupViewModel"]["contentId"] = "PLxyz"
|
|
results.insert(0, playlist)
|
|
results.insert(2, copy.deepcopy(results[1]))
|
|
results.insert(0, {"adSlotRenderer": {}})
|
|
_answer(net, page)
|
|
ids = [it["id"] for it in related_scrape.related("SdTZAMDKrNY")["items"]]
|
|
assert ids == ["to-lHJfK4pw", "R8V8CbuxryI", "HLKo4fJx_7k"]
|
|
|
|
|
|
def test_the_previous_compact_renderer_is_still_read(net):
|
|
page = copy.deepcopy(FIRST)
|
|
_results(page).insert(
|
|
0,
|
|
{
|
|
"compactVideoRenderer": {
|
|
"videoId": "oldshape001",
|
|
"title": {"simpleText": "Old shape"},
|
|
"longBylineText": {
|
|
"runs": [
|
|
{
|
|
"text": "Old Chan",
|
|
"navigationEndpoint": {"browseEndpoint": {"browseId": "UColdchan"}},
|
|
}
|
|
]
|
|
},
|
|
"thumbnail": {"thumbnails": [{"url": "s.jpg"}, {"url": "b.jpg"}]},
|
|
}
|
|
},
|
|
)
|
|
_answer(net, page)
|
|
first = related_scrape.related("SdTZAMDKrNY")["items"][0]
|
|
assert first == {
|
|
"id": "oldshape001",
|
|
"channel_id": "UColdchan",
|
|
"channel_title": "Old Chan",
|
|
"title": "Old shape",
|
|
"published_at": None,
|
|
"thumbnail_url": "b.jpg",
|
|
"live_broadcast": "none",
|
|
}
|
|
|
|
|
|
def test_the_last_page_has_no_cursor(net):
|
|
page = copy.deepcopy(FIRST)
|
|
_results(page).pop() # the continuationItemRenderer
|
|
_answer(net, page)
|
|
assert related_scrape.related("SdTZAMDKrNY")["next_page_token"] is None
|
|
|
|
|
|
def test_a_video_youtube_no_longer_has_is_an_empty_page(net):
|
|
_answer(
|
|
net,
|
|
{
|
|
"contents": {
|
|
"twoColumnWatchNextResults": {
|
|
"results": {
|
|
"results": {
|
|
"contents": [
|
|
{
|
|
"itemSectionRenderer": {
|
|
"contents": [{"backgroundPromoRenderer": {}}]
|
|
}
|
|
}
|
|
]
|
|
}
|
|
}
|
|
}
|
|
}
|
|
},
|
|
)
|
|
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",
|
|
[
|
|
{"responseContext": {}}, # no watch page frame at all
|
|
{"onResponseReceivedEndpoints": [{"somethingElse": {}}]}, # a continuation we do not know
|
|
"<html>consent</html>", # not JSON
|
|
[1, 2], # JSON, not an object
|
|
],
|
|
)
|
|
def test_an_answer_of_an_unknown_shape_is_an_error_not_an_empty_list(net, answer):
|
|
_answer(net, answer)
|
|
with pytest.raises(related_scrape.ScrapeError):
|
|
related_scrape.related("SdTZAMDKrNY")
|
|
|
|
|
|
def test_a_refused_request_is_an_error_whatever_its_body_says(net):
|
|
_answer(net, FIRST, status=429)
|
|
with pytest.raises(related_scrape.ScrapeError):
|
|
related_scrape.related("SdTZAMDKrNY")
|