Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _get_xsrf_token(self):
"""
Get the xsrf_token used as the session token.
video_id must be initialized.
Sets the session token(xsrf_token).
"""
if not self.video_id:
raise ValueError("Can't start a session without the video_id.")
response = requests.get(self.video_url)
response.raise_for_status()
token = re.search(YOUTUBE_SESSION_TOKEN_REGEX, str(response.content))
if not token:
raise YoutubeSessionError("Could not fetch the xsrf token")
self._xsrf_token = token.group(1)
def add_to_queue(self, youtube_id):
"""
Adds a video to the queue video will play after the currently playing video ends.
If video is buffering it will not be added!
:param youtube_id: The video id to add to the queue
"""
if not self.in_session:
raise YoutubeSessionError("Session must be initialized to add to queue")
if not self.playlist:
self.playlist = [self.video_id]
elif youtube_id in self.playlist:
raise YoutubeControllerError("Video already in queue")
self.update_screen_id()
# if self.status.player_is_idle:
# raise YoutubeControllerError("Can't add to queue while video is idle")
if self.status.player_state == "BUFFERING":
raise YoutubeControllerError("Can't add to queue while video is buffering")
self._ofs += 1
self._manage_playlist(data=REQUEST_DATA_ADD_TO_PLAYLIST % (self._ofs, youtube_id))
self.playlist.append(youtube_id)
"""
Sends a POST to start the session.
Uses loung_token and video id as parameters.
Sets session SID and gsessionid on success.
"""
if not self.video_id:
raise ValueError("Can't start a session without the video_id.")
if not self._lounge_token:
raise ValueError("lounge token is None. _get_lounge_token must be called")
url_params = REQUEST_PARAMS_SET_PLAYLIST.copy()
url_params["loungeIdToken"] = self._lounge_token
url_params["params"] = VIDEO_ID_PARAM.format(video_id=self.video_id)
response = self._do_post(REQUEST_URL_SET_PLAYLIST, data=REQUEST_DATA_SET_PLAYLIST, params=url_params)
content = str(response.content)
if response.status_code == 401 and content.find(EXPIRED_LOUNGE_ID_RESPONSE_CONTENT) != -1:
raise YoutubeSessionError("The lounge token expired.")
response.raise_for_status()
if not self.in_session:
self._extract_session_parameters(content)
def _get_lounge_id(self):
"""
Gets the lounge_token.
session_token(xsrf_token) and screenId must be initialized.
Sets the lounge token.
"""
if not self.screen_id:
raise ValueError("Screen id is None. update_screen_id must be called.")
if not self._xsrf_token:
raise ValueError("xsrf token is None. Get xsrf token must be called.")
data = REQUEST_DATA_LOUNGE_TOKEN.format(screenId=self.screen_id, XSRFToken=self._xsrf_token)
response = self._do_post(REQUEST_URL_LOUNGE_TOKEN, data=data)
if response.status_code == 401:
# Screen id is not None and it is updated with a message from the Chromecast.
# It is very unlikely that screen_id caused the problem.
raise YoutubeSessionError("Could not get lounge id. XSRF token has expired or is not valid.")
response.raise_for_status()
try:
lounge_token = response.json()["screens"][0]["loungeToken"]
except JSONDecodeError:
raise YoutubeSessionError("Could not get lounge id. XSRF token has expired or not valid.")
self._lounge_token = lounge_token
"""
if not self.screen_id:
raise ValueError("Screen id is None. update_screen_id must be called.")
if not self._xsrf_token:
raise ValueError("xsrf token is None. Get xsrf token must be called.")
data = REQUEST_DATA_LOUNGE_TOKEN.format(screenId=self.screen_id, XSRFToken=self._xsrf_token)
response = self._do_post(REQUEST_URL_LOUNGE_TOKEN, data=data)
if response.status_code == 401:
# Screen id is not None and it is updated with a message from the Chromecast.
# It is very unlikely that screen_id caused the problem.
raise YoutubeSessionError("Could not get lounge id. XSRF token has expired or is not valid.")
response.raise_for_status()
try:
lounge_token = response.json()["screens"][0]["loungeToken"]
except JSONDecodeError:
raise YoutubeSessionError("Could not get lounge id. XSRF token has expired or not valid.")
self._lounge_token = lounge_token
def _extract_session_parameters(self, response_packet_content):
"""
Extracts the playlist id, SID, gsession id, first video(the playlist base video)
and now playing from a session response.
:param response_packet_content: (str) the response packet content
"""
content = response_packet_content
playlist_id = re.search(PLAYLIST_ID_REGEX, content)
sid = re.search(SID_REGEX, content)
gsession = re.search(GSESSION_ID_REGEX, content)
first_video = re.search(FIRST_VIDEO_ID_REGEX, content)
now_playing = re.search(NOW_PLAYING_REGEX, content)
if not (sid and gsession and playlist_id):
raise YoutubeSessionError("Could not parse session parameters.")
self._sid = sid.group(1)
self._gsession_id = gsession.group(1)
self._playlist_id = playlist_id.group(1)
if first_video:
self._first_video = first_video.group(1)
else:
self._first_video = None
if now_playing:
self._now_playing = now_playing.group(1)
else:
self._now_playing = None