Add podcast protobufs + functions to use them

1 files changed, 152 insertions, 0 deletions

diff --git a/yt_dlp_plugins/extractor/radiko_protobufs.py b/yt_dlp_plugins/extractor/radiko_protobufs.py
new file mode 100755
index 0000000..2336f10
--- /dev/null
+++ b/yt_dlp_plugins/extractor/radiko_protobufs.py
@@ -0,0 +1,152 @@
+#!/usr/bin/env python3
+try:
+	import protobug
+except ImportError:
+	protobug = None
+
+import base64
+import struct
+
+import random
+import requests
+
+if protobug:  # i suppose it works lmao
+
+
+	def add_grpc_header(protobuf_data):
+		compression_flag = 0
+		message_length = len(protobuf_data)
+		header = struct.pack('>BI', compression_flag, message_length)
+		return header + protobuf_data
+
+	def strip_grpc_response(response):
+		return response[5:].rpartition(b"grpc-status:")[0]
+
+	def _download_grpc(self, url_or_request, video_id, response_message, note="Downloading GRPC information", *args, **kwargs):
+		urlh = self._request_webpage(url_or_request, video_id,
+			headers={
+				'Content-Type': 'application/grpc-web+proto',
+				'X-User-Agent': 'grpc-web-javascript/0.1',
+				'X-Grpc-Web': '1',
+				**kwargs.pop('headers')
+			},
+			data=add_grpc_header(protobug.dumps(kwargs.pop('data'))), note=note,
+			*args, **kwargs,
+		)
+		response = urlh.read()
+
+		protobuf = strip_grpc_response(response)
+		if len(protobuf) > 0:
+			return protobug.loads(protobuf, response_message)
+
+
+	@protobug.message
+	class SignUpRequest:
+		lsid: protobug.String = protobug.field(1)
+
+	def sign_up(self):
+		lsid = ''.join(random.choices('0123456789abcdef', k=32))
+
+		signup = _download_grpc(self, "https://api.annex.radiko.jp/radiko.UserService/SignUp",
+			"UserService", None, note="Registering ID", headers={'Origin': 'https://radiko.jp'},
+			data=SignUpRequest(lsid=lsid),
+		)
+		# youre meant to only do the sign up ^ once and then keep your lsid for later
+		# so that you can sign in and get the token for the API to work
+		return lsid
+
+
+	@protobug.message
+	class SignInRequest:
+		lsid: protobug.String = protobug.field(2)
+		area: protobug.String = protobug.field(3)
+
+	@protobug.message
+	class SignInResponse:
+		jwt: protobug.String = protobug.field(1)
+
+
+	def sign_in(self, lsid):
+		sign_in = _download_grpc(self, "https://api.annex.radiko.jp/radiko.UserService/SignIn",
+			"UserService", SignInResponse, note="Getting auth token", headers={'Origin': 'https://radiko.jp'},
+			data=SignInRequest(lsid=lsid, area="JP13"),
+		)
+		return sign_in.jwt
+
+
+	def auth_userservice(self):
+		cachedata = self.cache.load("rajiko", "UserService")
+		if cachedata is not None:
+			lsid = cachedata.get("lsid")
+		else:
+			lsid = sign_up(self)
+			self.cache.store("rajiko", "UserService", {"lsid": lsid})
+		jwt = sign_in(self, lsid)
+		return jwt
+
+
+	@protobug.message
+	class ListPodcastEpisodesRequest:
+		channel_id: protobug.String = protobug.field(1)
+		dontknow: protobug.Int32 = protobug.field(2)
+		page_length: protobug.Int32 = protobug.field(4)
+		cursor: protobug.String = protobug.field(5, default=None)
+
+
+	@protobug.message
+	class Audio:
+		revision: protobug.Int32 = protobug.field(1)
+		url: protobug.String = protobug.field(2)
+		fileSize: protobug.Int64 = protobug.field(3)
+		durationSec: protobug.Int64 = protobug.field(4)
+		transcoded: protobug.Bool = protobug.field(5)
+
+	@protobug.message
+	class EpisodeStartAt:
+		seconds: protobug.UInt64 = protobug.field(1)
+		nanos: protobug.UInt64 = protobug.field(2, default=0)
+
+
+	@protobug.message
+	class PodcastEpisode:
+		id: protobug.String = protobug.field(1)
+		workspaceId: protobug.String = protobug.field(2)
+		channelId: protobug.String = protobug.field(3)
+		title: protobug.String = protobug.field(4)
+		description: protobug.String = protobug.field(5)
+
+		audio: Audio = protobug.field(8)
+		channelImageUrl: protobug.String = protobug.field(16)
+		channelTitle: protobug.String = protobug.field(17)
+		channelStationName: protobug.String = protobug.field(18)
+		channelAuthor: protobug.String = protobug.field(19)
+
+		channelThumbnailImageUrl: protobug.String = protobug.field(21)
+		channelStationType: protobug.UInt32 = protobug.field(22)
+		startAt: EpisodeStartAt = protobug.field(27)
+		isEnabled: protobug.Bool = protobug.field(29)
+		hasTranscription: protobug.Bool = protobug.field(32)
+
+		imageUrl: protobug.String = protobug.field(7, default=None)
+		thumbnailImageUrl: protobug.String = protobug.field(20, default=None)
+
+	@protobug.message
+	class ListPodcastEpisodesResponse:
+		episodes: list[PodcastEpisode] = protobug.field(1)
+		hasNextPage: protobug.Bool = protobug.field(2, default=False)
+
+
+	def get_podcast_episodes(self, channel_id, jwt, cursor, page_length=20):
+		# site uses 20 items
+		# cursor is the id of the last episode you've seen in the list
+
+		return _download_grpc(self, 'https://api.annex.radiko.jp/radiko.PodcastService/ListPodcastEpisodes',
+			channel_id, ListPodcastEpisodesResponse, note="Downloading episode listings",
+			headers={'Authorization': f'Bearer {jwt}'},
+			data=ListPodcastEpisodesRequest(
+				channel_id=channel_id,
+				dontknow=1,
+				page_length=page_length,
+				cursor=cursor,
+			)
+		)