Endless radio?

[orrnithogalum]

Hey everyone,

I'm trying to replicate the YTMusic autoplay feature. Right now I can get the radio playlist for a song using get_watch_playlist(...) but that returns a capped amount of songs.

What I've looked into

I noticed by reading the docs that ytmusicapi has a continuations module with methods like:

• get_continuation_params
• get_continuations
• get_continuations_2025

And I've heard about continuation tokens etc. My assumption was that maybe get_watch_playlist(..., radio=True) internally returns continuation data that can be reused? idk.

Would fetching get_watch_playlist on the last song returned by the previous get_watch_playlist call even match autoplay behavior?

[orrnithogalum]

Turns out this can be done pretty easily by looking at the api code. Basically we fetch the 1rst radio page from ytmusic, using the get_watch_playlist code. What we do differently though, is that we also recuperate a continuation token.

Then we reuse that token and that original request body to recuperate the next page in that radio, in a separate method. This allows us to infinitely fetch a single next page at a time instead of fetching 50 pages (which takes a long time) in one go. This does simulate youtube music's radio better imo.

from ytmusicapi.exceptions import YTMusicServerError
from ytmusicapi.parsers.playlists import validate_playlist_id
from ytmusicapi.parsers.watch import (
    NAVIGATION_PLAYLIST_ID,
    TAB_CONTENT,
    nav,
    parse_watch_playlist,
)


async def radio(
    server: YTMusic, videoId: str | None, playlistId: str | None, limit: int = 25
):
    try:
        body = {
            "enablePersistentPlaylistPanel": True,
            "isAudioOnly": True,
            "tunerSettingValue": "AUTOMIX_SETTING_NORMAL",
        }

        if videoId:
            body["videoId"] = videoId
            if not playlistId:
                playlistId = "RDAMVM" + videoId

            body["watchEndpointMusicSupportedConfigs"] = {
                "watchEndpointMusicConfig": {
                    "hasPersistentPlaylistPanel": True,
                    "musicVideoType": "MUSIC_VIDEO_TYPE_ATV",
                }
            }
            body["params"] = "wAEB"

        if playlistId:
            playlist_id = validate_playlist_id(playlistId)
            body["playlistId"] = playlist_id

        endpoint = "next"
        response = server._send_request(endpoint, body)

        watchNextRenderer = nav(
            response,
            [
                "contents",
                "singleColumnMusicWatchNextResultsRenderer",
                "tabbedRenderer",
                "watchNextTabbedResultsRenderer",
            ],
        )

        results = nav(
            watchNextRenderer,
            [*TAB_CONTENT, "musicQueueRenderer", "content", "playlistPanelRenderer"],
            True,
        )

        if not results:
            msg = "No content returned by the server."
            if playlistId:
                msg += f"\nEnsure you have access to {playlistId} - a private playlist may cause this."
            raise YTMusicServerError(msg)

        playlist = next(
            filter(
                bool,
                map(
                    lambda x: nav(
                        x, ["playlistPanelVideoRenderer", *NAVIGATION_PLAYLIST_ID], True
                    ),
                    results["contents"],
                ),
            ),
            None,
        )
        tracks = parse_watch_playlist(results["contents"])

        if videoId:
            tracks = tracks[1:]

        # Extract continuation token
        ctoken = None
        if "continuations" in results:
            cont_key = (
                "nextRadioContinuationData"
                if not body.get("playlistId", "").startswith(("PL", "OLA"))
                else "nextContinuationData"
            )
            ctoken = results["continuations"][0].get(cont_key, {}).get("continuation")

        # Stream tracks
        for track in tracks:
            yield {"type": "radio-track", "status": "ok", "data": track}

        # Final message with continuation
        yield {
            "type": "radio-done",
            "status": "ok",
            "continuation": ctoken,
            "id": videoId if videoId else playlist,
        }

    except Exception as e:
        yield {"type": "error", "message": str(e)}


async def radio_next(
    server: YTMusic,
    video_id: str | None,
    playlist_id: str | None,
    ctoken: str,
    limit: int = 25,
):
    try:
        body = # get original radio request body through some container obj.

        additional_params = f"&ctoken={ctoken}&continuation={ctoken}"
        response = server.ytm._send_request("next", body, additional_params)

        if "continuationContents" not in response:
            yield {
                "type": "radio-done",
                "status": "ok",
                "continuation": None,
                "id": video_id,
            }
            return

        results = response["continuationContents"]["playlistPanelContinuation"]

        tracks = (
            parse_watch_playlist(results.get("contents", results.get("items", [])))
            or []
        )

        # Extract next continuation token
        next_ctoken = None
        if "continuations" in results:
            for key in ["nextRadioContinuationData", "nextContinuationData"]:
                if key in results["continuations"][0]:
                    next_ctoken = results["continuations"][0][key]["continuation"]
                    break

        # Stream tracks
        for track in tracks[:limit]:
            yield {"type": "radio-track", "status": "ok", "data": track}

        # Final message with next continuation
        yield {
            "type": "radio-done",
            "status": "ok",
            "continuation": next_ctoken,
            "id": video_id,
        }

    except Exception as e:
        yield {"type": "error", "message": str(e)}