siftlode/backend/app/youtube/related_scrape.py

"""A video's related videos, read from the watch page's InnerTube `next` endpoint — no quota.

The Data API has no related-videos call any more, so this is the only source: the same list the
website shows beside a playing video. A page holds about 20 videos and a continuation token for the
next 20. The returned page has the search scrape's shape (``{"items": [...], "next_page_token"}``)
so the catalog ingests it the same way (`catalog.search.ingest_candidates`).

YouTube draws each item as a `lockupViewModel` now; the older `compactVideoRenderer` is read too, so
a rollback on its side does not empty the list. An item that names no single channel — a video by
several channels draws an avatar stack, not one avatar — is left out: the catalog row needs its
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,
    continuation_token,
    post,
    scraped_video,
    video_renderer,
)

__all__ = ["ScrapeError", "related"]

# Where a first page and a continuation page put their item list.
_FIRST = (
    "contents",
    "twoColumnWatchNextResults",
    "secondaryResults",
    "secondaryResults",
    "results",
)
_APPENDS = ("appendContinuationItemsAction", "reloadContinuationItemsCommand")


def _at(node: object, *path: str | int) -> object:
    """The value under `path` in nested dicts/lists, or None where the path stops."""
    for key in path:
        if isinstance(node, dict) and isinstance(key, str):
            node = node.get(key)
        elif isinstance(node, list) and isinstance(key, int) and -len(node) <= key < len(node):
            node = node[key]
        else:
            return None
    return node


def _lockup(node: dict) -> dict | None:
    if node.get("contentType") != "LOCKUP_CONTENT_TYPE_VIDEO" or not node.get("contentId"):
        return None
    meta = _at(node, "metadata", "lockupMetadataViewModel")
    channel_id = _at(
        meta,
        "image",
        "decoratedAvatarViewModel",
        "rendererContext",
        "commandContext",
        "onTap",
        "innertubeCommand",
        "browseEndpoint",
        "browseId",
    )
    channel_title = _at(
        meta, "metadata", "contentMetadataViewModel", "metadataRows", 0, "metadataParts", 0
    )
    return scraped_video(
        node["contentId"],
        channel_id,
        _at(channel_title, "text", "content"),
        _at(meta, "title", "content"),
        _at(node, "contentImage", "thumbnailViewModel", "image", "sources"),
    )


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:
    """The page's item list (first page or continuation), or None when it has neither frame."""
    first = _at(data, *_FIRST)
    if isinstance(first, list):
        return first
    if _unavailable(data):
        return []
    for endpoint in data.get("onResponseReceivedEndpoints") or []:
        for key in _APPENDS:
            items = _at(endpoint, key, "continuationItems")
            if isinstance(items, list):
                return items
    return None


def related(
    video_id: str, continuation: str | None = None, *, hl: str = "en", gl: str = "US"
) -> dict:
    """One page of `video_id`'s related videos (`continuation` = the token a previous page gave).

    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)
    if items is None:
        raise ScrapeError("InnerTube returned no related list")
    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 = video_renderer(entry["compactVideoRenderer"])
        else:
            found = continuation_token(entry.get("continuationItemRenderer"))
            known += found is not None
            token = token or found
            continue
        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)
    # 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}