1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
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,
)
)
|
