From 9c768e6bf47a5eb6f179815de4783f513af211ea Mon Sep 17 00:00:00 2001 From: Rafael Moraes <50295204+glomatico@users.noreply.github.com> Date: Tue, 10 Feb 2026 21:45:43 -0300 Subject: [PATCH] Refactor Spotify API into votify.api package --- votify/api/__init__.py | 3 + votify/api/api.py | 628 +++++++++++++++++++++++++++++ votify/api/constants.py | 30 ++ votify/api/exceptions.py | 19 + votify/api/totp.py | 69 ++++ votify/spotify_api.py | 839 --------------------------------------- votify/totp.py | 36 -- 7 files changed, 749 insertions(+), 875 deletions(-) create mode 100644 votify/api/__init__.py create mode 100644 votify/api/api.py create mode 100644 votify/api/constants.py create mode 100644 votify/api/exceptions.py create mode 100644 votify/api/totp.py delete mode 100644 votify/spotify_api.py delete mode 100644 votify/totp.py diff --git a/votify/api/__init__.py b/votify/api/__init__.py new file mode 100644 index 0000000..119fdc6 --- /dev/null +++ b/votify/api/__init__.py @@ -0,0 +1,3 @@ +from .api import * +from .exceptions import * +from .totp import * diff --git a/votify/api/api.py b/votify/api/api.py new file mode 100644 index 0000000..9fe22ba --- /dev/null +++ b/votify/api/api.py @@ -0,0 +1,628 @@ +import base64 +import logging +import time +from http.cookiejar import MozillaCookieJar + +import base62 +import httpx + +from ..utils import safe_json +from .constants import ( + CLIENT_TOKEN_URL, + CLIENT_VERSION, + COOKIE_DOMAIN, + GID_METADATA_URL, + HOME_PAGE_URL, + LYRICS_API_URL, + PATHFINDER_API_URL, + SEEK_TABLE_API_URL, + SERVER_TIME_URL, + SESSION_TOKEN_URL, + AUDIO_STREAM_URLS_API_URL, + TRACK_CREDITS_API_URL, + TRACK_PLAYBACK_API_URL, + VIDEO_MANIFEST_API_URL, + WIDEVINE_LICENSE_API_URL, +) +from .exceptions import VotifyRequestException +from .totp import Totp + +logger = logging.getLogger(__name__) + + +class SpotifyApi: + def __init__( + self, + sp_dc: str | None = None, + ) -> None: + self.sp_dc = sp_dc + + @property + def premium_session(self) -> bool: + return ( + getattr(self, "user_profile", {}) + .get("data", {}) + .get("account", {}) + .get("product") + == "PREMIUM" + ) + + @property + def anonymous_session(self) -> bool: + return self.user_profile is None + + @staticmethod + def _parse_cookies(cookies_path: str) -> dict[str, str]: + cookies = MozillaCookieJar(cookies_path) + cookies.load(ignore_discard=True, ignore_expires=True) + + cookie_dict = { + cookie.name: cookie.value + for cookie in cookies + if cookie.domain == COOKIE_DOMAIN + } + + logger.debug(f"Parsed cookies: {cookie_dict}") + + return cookie_dict + + @classmethod + async def create_from_netscape_cookies(cls, cookies_path: str) -> "SpotifyApi": + cookies = cls._parse_cookies(cookies_path) + sp_dc = cookies.get("sp_dc") + if sp_dc is None: + raise ValueError( + "'sp_dc' cookie not found in cookies. " + "Make sure you have exported the cookies " + "from the Spotify homepage and are logged in." + ) + + return await cls.create(sp_dc=sp_dc) + + @classmethod + async def create( + cls, + *args, + **kwargs, + ) -> "SpotifyApi": + api = cls(*args, **kwargs) + + await api._initialize() + + return api + + async def _initialize(self) -> None: + self._initialize_client() + await self._initialize_totp() + await self._initialize_authorization() + await self._initialize_user_profile() + + def _initialize_client(self) -> None: + self.client = httpx.AsyncClient() + + self.client.headers.update( + { + "accept": "application/json", + "accept-language": "en-US", + "content-type": "application/json", + "origin": HOME_PAGE_URL, + "priority": "u=1, i", + "referer": HOME_PAGE_URL, + "sec-ch-ua": '"Not)A;Brand";v="99", "Google Chrome";v="127", "Chromium";v="127"', + "sec-ch-ua-mobile": "?0", + "sec-ch-ua-platform": '"Windows"', + "sec-fetch-dest": "empty", + "sec-fetch-mode": "cors", + "sec-fetch-site": "same-site", + "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36", + "spotify-app-version": CLIENT_VERSION, + "app-platform": "WebPlayer", + } + ) + + if self.sp_dc: + self.client.cookies.update({"sp_dc": self.sp_dc}) + + async def _initialize_totp(self) -> Totp: + self.totp = await Totp.initialize() + + async def _initialize_authorization(self) -> None: + session_info = await self._get_session_token() + client_token = await self._get_client_token(session_info["clientId"]) + + access_token = session_info["accessToken"] + granted_token = client_token["granted_token"]["token"] + + self._authorization_expire_time = ( + session_info["accessTokenExpirationTimestampMs"] / 1000 + ) + + self.client.headers.update( + { + "authorization": f"Bearer {access_token}", + "client-token": granted_token, + } + ) + + async def _initialize_user_profile(self) -> None: + self.user_profile = await self._get_user_profile() if self.sp_dc else None + + async def _get_server_time(self) -> int: + response = await self.client.get(SERVER_TIME_URL) + server_time = safe_json(response) + if response.status_code != 200 or not server_time: + raise VotifyRequestException( + name="Server time", + response_status_code=response.status_code, + response_text=response.text, + ) + + logger.debug(f"Received server time: {server_time}") + + return 1e3 * server_time["serverTime"] + + async def _get_session_token(self) -> None: + server_time = await self._get_server_time() + + generated_totp = self.totp.generate(timestamp=server_time) + + response = await self.client.get( + SESSION_TOKEN_URL, + params={ + "reason": "init", + "productType": "web-player", + "totp": generated_totp, + "totpServer": generated_totp, + "totpVer": self.totp.version, + }, + ) + session_info = safe_json(response) + if response.status_code != 200 or not session_info: + raise VotifyRequestException( + name="Session info", + response_status_code=response.status_code, + response_text=response.text, + ) + + logger.debug(f"Received session info: {session_info}") + + return session_info + + async def _get_client_token(self, client_id: str) -> None: + response = await self.client.post( + CLIENT_TOKEN_URL, + json={ + "client_data": { + "client_version": CLIENT_VERSION, + "client_id": client_id, + "js_sdk_data": {}, + } + }, + headers={ + "Accept": "application/json", + }, + ) + client_token = safe_json(response) + if response.status_code != 200 or not client_token: + raise VotifyRequestException( + name="Client token", + response_status_code=response.status_code, + response_text=response.text, + ) + + logger.debug(f"Received client token: {client_token}") + + return client_token + + async def _refresh_authorization_if_needed(self) -> None: + timestamp_session_expire = int(self._authorization_expire_time) + timestamp_now = time.time() + if timestamp_now < timestamp_session_expire: + return + await self._initialize_authorization() + + @staticmethod + def media_id_to_gid(media_id: str) -> str: + return hex(base62.decode(media_id, base62.CHARSET_INVERTED))[2:].zfill(32) + + @staticmethod + def gid_to_media_id(gid: str) -> str: + return base62.encode(int(gid, 16), charset=base62.CHARSET_INVERTED).zfill(22) + + async def _pathfinder_request( + self, + operation_name: str, + persisted_query_hash: str, + variables: dict = {}, + ) -> dict: + await self._refresh_authorization_if_needed() + + response = await self.client.post( + PATHFINDER_API_URL, + json={ + "variables": variables, + "operationName": operation_name, + "extensions": { + "persistedQuery": { + "version": 1, + "sha256Hash": persisted_query_hash, + } + }, + }, + ) + response_json = safe_json(response) + + if ( + response.status_code != 200 + or not response_json + or "errors" in response_json + ): + raise VotifyRequestException( + name="Pathfinder", + response_status_code=response.status_code, + response_text=response.text, + ) + + return response_json + + async def _get_user_profile(self) -> dict: + user_profile = await self._pathfinder_request( + operation_name="accountAttributes", + persisted_query_hash="24aaa3057b69fa91492de26841ad199bd0b330ca95817b7a4d6715150de01827", + ) + + logger.debug(f"Received user profile: {user_profile}") + + return user_profile + + async def get_track(self, track_id: str) -> dict: + result = await self._pathfinder_request( + operation_name="getTrack", + persisted_query_hash="612585ae06ba435ad26369870deaae23b5c8800a256cd8a57e08eddc25a37294", + variables={"uri": f"spotify:track:{track_id}"}, + ) + + logger.debug(f"Received track: {result}") + + return result + + async def get_album( + self, + album_id: str, + offset: int = 0, + limit: int = 300, + ) -> dict: + album = await self._pathfinder_request( + operation_name="getAlbum", + persisted_query_hash="b9bfabef66ed756e5e13f68a942deb60bd4125ec1f1be8cc42769dc0259b4b10", + variables={ + "uri": f"spotify:album:{album_id}", + "offset": offset, + "limit": limit, + }, + ) + + logger.debug(f"Received album: {album}") + + return album + + async def get_playlist( + self, + playlist_id: str, + offset: int = 0, + limit: int = 300, + ) -> dict: + playlist = await self._pathfinder_request( + operation_name="fetchPlaylist", + persisted_query_hash="bb67e0af06e8d6f52b531f97468ee4acd44cd0f82b988e15c2ea47b1148efc77", + variables={ + "uri": f"spotify:playlist:{playlist_id}", + "offset": offset, + "limit": limit, + "enableWatchFeedEntrypoint": True, + }, + ) + + logger.debug(f"Received playlist: {playlist}") + + return playlist + + async def get_episode(self, episode_id: str) -> dict: + episode = await self._pathfinder_request( + operation_name="getEpisodeOrChapter", + persisted_query_hash="8a62dbdeb7bd79605d7d68b01bcdf83f08bc6c6287ee1665ba012c748a4cf1f3", + variables={"uri": f"spotify:episode:{episode_id}"}, + ) + + logger.debug(f"Received episode: {episode}") + + return episode + + async def get_show( + self, + show_id: str, + offset: int = 0, + limit: int = 300, + ) -> dict: + show = await self._pathfinder_request( + operation_name="queryPodcastEpisodes", + persisted_query_hash="8e2826c5993383566cc08bf9f5d3301b69513c3f6acb8d706286855e57bf44b2", + variables={ + "uri": f"spotify:show:{show_id}", + "offset": offset, + "limit": limit, + }, + ) + + logger.debug(f"Received show: {show}") + + return show + + async def _get_artist_discography( + self, + artist_id: str, + type: str, + offeset: int, + limit: int, + ) -> dict: + result = await self._pathfinder_request( + operation_name=f"queryArtistDiscography{type.capitalize()}s", + persisted_query_hash="5e07d323febb57b4a56a42abbf781490e58764aa45feb6e3dc0591564fc56599", + variables={ + "uri": f"spotify:artist:{artist_id}", + "offset": offeset, + "limit": limit, + "order": "DATE_DESC", + }, + ) + + logger.debug(f"Received artist {type}s: {result}") + + return result + + async def get_artist_albums( + self, + artist_id: str, + offset: int = 0, + limit: int = 300, + ) -> dict: + return await self._get_artist_discography( + artist_id=artist_id, + type="album", + offeset=offset, + limit=limit, + ) + + async def get_artist_singles( + self, + artist_id: str, + offset: int = 0, + limit: int = 300, + ) -> dict: + return await self._get_artist_discography( + artist_id=artist_id, + type="single", + offeset=offset, + limit=limit, + ) + + async def get_artist_compilations( + self, + artist_id: str, + offset: int = 0, + limit: int = 300, + ) -> dict: + return await self._get_artist_discography( + artist_id=artist_id, + type="compilation", + offeset=offset, + limit=limit, + ) + + async def get_artist_videos( + self, + artist_id: str, + offset: int = 0, + limit: int = 300, + ) -> dict: + artist_videos = await self._pathfinder_request( + operation_name="queryArtistRelatedVideos", + persisted_query_hash="8958042d3dd127ec7882a7117fafa4df21af27ff1560af51e55061e8451de67b", + variables={ + "uri": f"spotify:artist:{artist_id}", + "showMapped": True, + "showUnmapped": True, + "offset": offset, + "limit": limit, + }, + ) + + logger.debug(f"Received artist videos: {artist_videos}") + + return artist_videos + + async def get_video_manifest( + self, + file_id: str, + ) -> dict: + await self._refresh_authorization_if_needed() + + response = await self.client.get(VIDEO_MANIFEST_API_URL.format(file_id=file_id)) + video_manifest = safe_json(response) + + if response.status_code != 200 or not video_manifest: + raise VotifyRequestException( + name="Video manifest", + response_status_code=response.status_code, + response_text=response.text, + ) + + logger.debug(f"Received video manifest: {video_manifest}") + + return video_manifest + + async def get_seek_table(self, file_id: str) -> dict: + async with httpx.AsyncClient() as client: + response = await client.get( + SEEK_TABLE_API_URL.format(file_id=file_id), + headers={ + "Accept": "*/*", + "Accept-Encoding": "gzip, deflate, br, zstd", + "Accept-Language": "en-US", + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "Origin": HOME_PAGE_URL, + "Pragma": "no-cache", + "Priority": "u=4", + "Referer": HOME_PAGE_URL, + "Sec-Fetch-Dest": "empty", + "Sec-Fetch-Mode": "cors", + "Sec-Fetch-Site": "cross-site", + "User-Agent": self.client.headers["user-agent"], + }, + ) + seek_table = safe_json(response) + + if response.status_code != 200 or not seek_table: + raise VotifyRequestException( + name="Seek table", + response_status_code=response.status_code, + response_text=response.text, + ) + + logger.debug(f"Received seek table: {seek_table}") + + return seek_table + + async def get_track_playback_info( + self, + media_id: str, + media_type: str, + file_formats: list[str] = ["file_ids_mp4", "manifest_ids_video"], + ) -> dict: + await self._refresh_authorization_if_needed() + + response = await self.client.get( + TRACK_PLAYBACK_API_URL.format( + media_id=media_id, + media_type=media_type, + ), + params={ + "manifestFileFormat": file_formats, + }, + ) + track_playback_info = safe_json(response) + + if response.status_code != 200 or not track_playback_info: + raise VotifyRequestException( + name="Track playback info", + response_status_code=response.status_code, + response_text=response.text, + ) + + logger.debug(f"Received track playback info: {track_playback_info}") + + return track_playback_info + + async def get_gid_metadata( + self, + gid: str, + media_type: str, + ) -> dict: + await self._refresh_authorization_if_needed() + + response = await self.client.get( + GID_METADATA_URL.format( + gid=gid, + media_type=media_type, + ) + ) + gid_metadata = safe_json(response) + + if response.status_code != 200 or not gid_metadata: + raise VotifyRequestException( + name="GID metadata", + response_status_code=response.status_code, + response_text=response.text, + ) + + logger.debug(f"Received GID metadata: {gid_metadata}") + + return gid_metadata + + async def get_lyrics(self, track_id: str) -> dict: + await self._refresh_authorization_if_needed() + + response = await self.client.get(LYRICS_API_URL.format(track_id=track_id)) + lyrics = safe_json(response) + + if response.status_code != 200 or not lyrics: + raise VotifyRequestException( + name="Lyrics", + response_status_code=response.status_code, + response_text=response.text, + ) + + logger.debug(f"Received lyrics: {lyrics}") + + return lyrics + + async def get_track_credits(self, track_id: str) -> dict: + await self._refresh_authorization_if_needed() + + response = await self.client.get( + TRACK_CREDITS_API_URL.format(track_id=track_id) + ) + track_credits = safe_json(response) + + if response.status_code != 200 or not track_credits: + raise VotifyRequestException( + name="Track credits", + response_status_code=response.status_code, + response_text=response.text, + ) + + logger.debug(f"Received track credits: {track_credits}") + + return track_credits + + async def get_widevine_license(self, challenge: bytes, media_type: str) -> bytes: + await self._refresh_authorization_if_needed() + + response = await self.client.post( + WIDEVINE_LICENSE_API_URL.format(type=media_type), + data=challenge, + ) + widevine_license = response.content + + if response.status_code != 200 or not widevine_license: + raise VotifyRequestException( + name="Widevine license", + response_status_code=response.status_code, + response_text=response.text, + ) + + logger.debug( + f"Received Widevine license: {base64.b64encode(widevine_license).decode()}" + ) + + return widevine_license + + async def get_audio_stream_urls(self, file_id: str) -> dict: + await self._refresh_authorization_if_needed() + + response = await self.client.get( + AUDIO_STREAM_URLS_API_URL.format(file_id=file_id), + ) + audio_stream_urls = safe_json(response) + + if response.status_code != 200 or not audio_stream_urls: + raise VotifyRequestException( + name="Audio stream URLs", + response_status_code=response.status_code, + response_text=response.text, + ) + + logger.debug(f"Received audio stream URLs: {audio_stream_urls}") + + return audio_stream_urls diff --git a/votify/api/constants.py b/votify/api/constants.py new file mode 100644 index 0000000..dbb78fe --- /dev/null +++ b/votify/api/constants.py @@ -0,0 +1,30 @@ +HOME_PAGE_URL = "https://open.spotify.com/" +COOKIE_DOMAIN = ".spotify.com" +CLIENT_VERSION = "1.2.70.61.g856ccd63" +LYRICS_API_URL = "https://spclient.wg.spotify.com/color-lyrics/v2/track/{track_id}" +GID_METADATA_URL = ( + "https://spclient.wg.spotify.com/metadata/4/{media_type}/{gid}?market=from_token" +) +PATHFINDER_API_URL = "https://api-partner.spotify.com/pathfinder/v2/query" +VIDEO_MANIFEST_API_URL = "https://gue1-spclient.spotify.com/manifests/v9/json/sources/{file_id}/options/supports_drm" +TRACK_PLAYBACK_API_URL = "https://gue1-spclient.spotify.com/track-playback/v1/media/spotify:{media_type}:{media_id}" +PLAYPLAY_LICENSE_API_URL = "https://gew4-spclient.spotify.com/playplay/v1/key/{file_id}" +WIDEVINE_LICENSE_API_URL = ( + "https://gue1-spclient.spotify.com/widevine-license/v1/{type}/license" +) +SEEK_TABLE_API_URL = "https://seektables.scdn.co/seektable/{file_id}.json" +TRACK_CREDITS_API_URL = "https://spclient.wg.spotify.com/track-credits-view/v0/experimental/{track_id}/credits" +AUDIO_STREAM_URLS_API_URL = ( + "https://gue1-spclient.spotify.com/storage-resolve/v2/files/audio/interactive/11/" + "{file_id}?version=10000000&product=9&platform=39&alt=json" +) +EXTEND_TRACK_COLLECTION_WAIT_TIME = 0.5 +SERVER_TIME_URL = "https://open.spotify.com/api/server-time" +SESSION_TOKEN_URL = "https://open.spotify.com/api/token" +CLIENT_TOKEN_URL = "https://clienttoken.spotify.com/v1/clienttoken" + +TOTP_PERIOD = 30 +TOTP_DIGITS = 6 +TOTP_SECRETS_URL = ( + "https://git.gay/thereallo/totp-secrets/raw/branch/main/secrets/secretDict.json" +) diff --git a/votify/api/exceptions.py b/votify/api/exceptions.py new file mode 100644 index 0000000..0cf39e1 --- /dev/null +++ b/votify/api/exceptions.py @@ -0,0 +1,19 @@ +from ..utils import VotiyException + + +class VotifyApiException(VotiyException): + pass + + +class VotifyRequestException(VotifyApiException): + def __init__( + self, + name: str, + response_status_code: int, + response_text: str, + ): + super().__init__( + f"{name} request failed with status code {response_status_code}: {response_text}" + ) + self.response_status_code = response_status_code + self.response_text = response_text diff --git a/votify/api/totp.py b/votify/api/totp.py new file mode 100644 index 0000000..1689dd2 --- /dev/null +++ b/votify/api/totp.py @@ -0,0 +1,69 @@ +import hashlib +import hmac +import logging +from typing import Collection + +import httpx + +from ..utils import safe_json +from .constants import TOTP_DIGITS, TOTP_PERIOD, TOTP_SECRETS_URL +from .exceptions import VotifyRequestException + +logger = logging.getLogger(__name__) + + +class Totp: + def __init__( + self, + version: str, + secret: bytes, + ) -> None: + self.version = version + self.secret = secret + + @classmethod + async def initialize(cls) -> "Totp": + async with httpx.AsyncClient() as client: + response = await client.get(TOTP_SECRETS_URL) + secrets = safe_json(response) + if response.status_code != 200 or not secrets: + raise VotifyRequestException( + name="TOTP secrets", + response_status_code=response.status_code, + response_text=response.text, + ) + + logger.debug(f"Received TOTP secrets: {secrets}") + + version = max(secrets.keys(), key=int) + + return cls( + version=version, + secret=cls.derive(secrets[version]), + ) + + @staticmethod + def derive(ciphertext: Collection[int]) -> bytes: + return "".join( + str(byte ^ ((i % 33) + 9)) for i, byte in enumerate(ciphertext) + ).encode("ascii") + + def generate(self, timestamp: int) -> str: + counter = int(timestamp) // 1000 // TOTP_PERIOD + counter_bytes = counter.to_bytes(8, "big") + + h = hmac.new(self.secret, counter_bytes, hashlib.sha1) + hmac_result = h.digest() + + offset = hmac_result[-1] & 0x0F + binary = ( + (hmac_result[offset] & 0x7F) << 24 + | (hmac_result[offset + 1] & 0xFF) << 16 + | (hmac_result[offset + 2] & 0xFF) << 8 + | (hmac_result[offset + 3] & 0xFF) + ) + result = str(binary % (10**TOTP_DIGITS)).zfill(TOTP_DIGITS) + + logger.debug(f"Generated TOTP code: {result}") + + return result diff --git a/votify/spotify_api.py b/votify/spotify_api.py deleted file mode 100644 index 8d1fa72..0000000 --- a/votify/spotify_api.py +++ /dev/null @@ -1,839 +0,0 @@ -from __future__ import annotations - -import functools -import json -import logging -import re -import time -import typing -from http.cookiejar import MozillaCookieJar -from pathlib import Path - -import base62 -import requests - -import secrets -import websocket -import threading - -from .device_flow import SpotifyDeviceFlow -from .totp import TOTP -from .utils import check_response - -logger = logging.getLogger("votify") - - -class SpotifyApi: - SPOTIFY_HOME_PAGE_URL = "https://open.spotify.com/" - SPOTIFY_COOKIE_DOMAIN = ".spotify.com" - CLIENT_VERSION = "1.2.70.61.g856ccd63" - LYRICS_API_URL = "https://spclient.wg.spotify.com/color-lyrics/v2/track/{track_id}" - METADATA_API_URL = "https://api-partner.spotify.com/pathfinder/v2/query" - GID_METADATA_API_URL = "https://spclient.wg.spotify.com/metadata/4/{media_type}/{gid}?market=from_token" - PATHFINDER_API_URL = "https://api-partner.spotify.com/pathfinder/v1/query" - VIDEO_MANIFEST_API_URL = "https://gue1-spclient.spotify.com/manifests/v9/json/sources/{gid}/options/supports_drm" - TRACK_PLAYBACK_API_URL = "https://gue1-spclient.spotify.com/track-playback/v1/media/spotify:{media_type}:{media_id}" - PLAYPLAY_LICENSE_API_URL = ( - "https://gew4-spclient.spotify.com/playplay/v1/key/{file_id}" - ) - WIDEVINE_LICENSE_API_URL = ( - "https://gue1-spclient.spotify.com/widevine-license/v1/{type}/license" - ) - SEEK_TABLE_API_URL = "https://seektables.scdn.co/seektable/{file_id}.json" - TRACK_CREDITS_API_URL = "https://spclient.wg.spotify.com/track-credits-view/v0/experimental/{track_id}/credits" - STREAM_URLS_API_URL = ( - "https://gue1-spclient.spotify.com/storage-resolve/v2/files/audio/interactive/11/" - "{file_id}?version=10000000&product=9&platform=39&alt=json" - ) - EXTEND_TRACK_COLLECTION_WAIT_TIME = 0.5 - SERVER_TIME_URL = "https://open.spotify.com/api/server-time" - SESSION_TOKEN_URL = "https://open.spotify.com/api/token" - CLIENT_TOKEN_URL = "https://clienttoken.spotify.com/v1/clienttoken" - - def __init__( - self, *, - secrets_url: str, - sp_dc: str | None = None, - use_device_flow: bool = False, - ) -> None: - self.session = requests.Session() - - secrets = self.session.get(secrets_url) - check_response(secrets) - totp_version, secrets_ciphertext = max(secrets.json().items(), key=lambda item: int(item[0])) - - self.totp = TOTP(version=totp_version, ciphertext=secrets_ciphertext) - self.sp_dc = sp_dc - self.use_device_flow = use_device_flow - self._set_session() - - @classmethod - def from_cookies_file(cls, cookies_path: Path, **kwargs) -> SpotifyApi: - cookies = MozillaCookieJar(cookies_path) - cookies.load(ignore_discard=True, ignore_expires=True) - parse_cookie = lambda name: next( - ( - cookie.value - for cookie in cookies - if cookie.name == name and cookie.domain == cls.SPOTIFY_COOKIE_DOMAIN - ), - None, - ) - sp_dc = parse_cookie("sp_dc") - if sp_dc is None: - raise ValueError( - '"sp_dc" cookie not found in cookies. ' - "Make sure you have exported the cookies from the Spotify homepage and are logged in." - ) - return cls(sp_dc=sp_dc, **kwargs) - - def _set_session(self) -> None: - self._setup_session_headers() - self._setup_authorization() - self._setup_user_profile() - - def _setup_session_headers(self) -> None: - headers = { - "accept": "application/json", - "accept-language": "en-US", - "content-type": "application/json", - "origin": self.SPOTIFY_HOME_PAGE_URL, - "priority": "u=1, i", - "referer": self.SPOTIFY_HOME_PAGE_URL, - "sec-ch-ua": '"Not)A;Brand";v="99", "Google Chrome";v="127", "Chromium";v="127"', - "sec-ch-ua-mobile": "?0", - "sec-ch-ua-platform": '"Windows"', - "sec-fetch-dest": "empty", - "sec-fetch-mode": "cors", - "sec-fetch-site": "same-site", - "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36", - "spotify-app-version": self.CLIENT_VERSION, - "app-platform": "WebPlayer", - } - self.session.headers.update(headers) - - if self.sp_dc: - self.session.cookies.update({"sp_dc": self.sp_dc}) - - def _get_server_time(self) -> int: - response = self.session.get(self.SERVER_TIME_URL) - check_response(response) - return 1e3 * response.json()["serverTime"] - - def set_token_headers(self, token: str, client_token: str | None = None) -> None: - - self.session.headers.update( - { - "authorization": f"Bearer {token}", - "client-token": client_token - } - ) - - def _setup_authorization_with_totp(self) -> None: - server_time = self._get_server_time() - totp = self.totp.generate(timestamp=server_time) - response = self.session.get( - self.SESSION_TOKEN_URL, - params={ - "reason": "init", - "productType": "web-player", - "totp": totp, - "totpServer": totp, - "totpVer": str(self.totp.version), - }, - ) - check_response(response) - authorization_info = response.json() - if not authorization_info.get("accessToken"): - raise ValueError("Failed to retrieve access token.") - - response = self.session.post(self.CLIENT_TOKEN_URL, - json = { - 'client_data': { - 'client_version': self.CLIENT_VERSION, - 'client_id': authorization_info['clientId'], - 'js_sdk_data': {} - } - }, - headers = { - 'Accept': 'application/json', - } - ) - check_response(response) - client_token = response.json() - if not client_token.get("granted_token"): - raise ValueError("Failed to retrieve granted token.") - - self.set_token_headers(authorization_info["accessToken"], client_token["granted_token"]["token"]) - self.session_auth_expire_time = ( - authorization_info["accessTokenExpirationTimestampMs"] / 1000 - ) - - def _setup_authorization(self) -> None: - if self.use_device_flow: - self._setup_authorization_with_device_flow() - else: - self._setup_authorization_with_totp() - - def _setup_authorization_with_device_flow(self) -> None: - device_flow = SpotifyDeviceFlow(self.sp_dc) - token_data = device_flow.get_token() - self.set_token_headers(token_data["access_token"]) - self.session_auth_expire_time = ( - int(time.time()) + token_data["expires_in"] - ) * 1000 - - def _refresh_session_auth(self) -> None: - timestamp_session_expire = int(self.session_auth_expire_time) - timestamp_now = time.time() - if timestamp_now < timestamp_session_expire: - return - self._setup_authorization() - - def _setup_user_profile(self) -> None: - payload = { - "variables": {}, - "operationName": "accountAttributes", - "extensions": { - "persistedQuery": { - "version": 1, - "sha256Hash": "24aaa3057b69fa91492de26841ad199bd0b330ca95817b7a4d6715150de01827" - } - } - } - - response = self.session.post( - self.METADATA_API_URL, - json=payload, - ) - check_response(response) - self.user_profile = response.json() - - @staticmethod - def media_id_to_gid(media_id: str) -> str: - return hex(base62.decode(media_id, base62.CHARSET_INVERTED))[2:].zfill(32) - - @staticmethod - def gid_to_media_id(gid: str) -> str: - return base62.encode(int(gid, 16), charset=base62.CHARSET_INVERTED).zfill(22) - - def get_gid_metadata( - self, - gid: str, - media_type: str, - ) -> dict: - self._refresh_session_auth() - response = self.session.get(self.GID_METADATA_API_URL.format(gid=gid, media_type=media_type)) - check_response(response) - return response.json() - - def get_track_playback_info( - self, - media_id: str, - media_type: str - ) -> dict | None: - self._refresh_session_auth() - - TRACK_ID = media_id - if not hasattr(self, 'cached_device_id'): - self.cached_device_id = secrets.token_hex(20) - - DEVICE_ID = self.cached_device_id - - ACCESS_TOKEN = self.session.headers["Authorization"].replace("Bearer ", "") - WS_URL = f"wss://dealer.spotify.com/?access_token={ACCESS_TOKEN}" - URL_REGISTRO = "https://spclient.wg.spotify.com/track-playback/v1/devices" - URL_COMMAND = f"https://spclient.wg.spotify.com/connect-state/v1/player/command/from/{DEVICE_ID}/to/{DEVICE_ID}" - URL = f"https://gue1-spclient.spotify.com/track-playback/v1/devices/{DEVICE_ID}/state" - - class Context: - conn_id = None - machine_id = None - state_id = None - stop_event = threading.Event() - - ctx = Context() - - def find_key(data, target): - if isinstance(data, dict): - for k, v in data.items(): - if k == target: return v - res = find_key(v, target) - if res: return res - elif isinstance(data, list): - for item in data: - res = find_key(item, target) - if res: return res - return None - - def on_message(ws, message): - try: - data = json.loads(message) - if "headers" in data and "Spotify-Connection-Id" in data["headers"]: - ctx.conn_id = data["headers"]["Spotify-Connection-Id"] - - new_mid = find_key(data, "state_machine_id") - if new_mid: - if new_mid != ctx.machine_id: - ctx.machine_id = new_mid - - new_sid = find_key(data, "state_id") - if new_sid and new_sid != ctx.state_id: - ctx.state_id = new_sid - except: - pass - - def start_socket(): - ws = websocket.WebSocketApp(WS_URL, on_message=on_message) - ws.run_forever() - - def keep_alive_loop(): - seq_num = 1 - last_mid = None - attempt_count = 0 - wait_ids_count = 0 - - while not ctx.stop_event.is_set(): - if ctx.machine_id and ctx.state_id and ctx.conn_id: - wait_ids_count = 0 - - if ctx.machine_id != last_mid: - seq_num = 1 - last_mid = ctx.machine_id - - payload = { - "seq_num": 0, - "state_ref": { - "state_machine_id": ctx.machine_id, - "state_id": ctx.state_id, - "paused": False, - }, - "sub_state": { - "playback_speed": 1, - "position": 1258, - "duration": 199954, - "media_type": "AUDIO", - "bitrate": 128000, - "audio_quality": "HIGH", - "format": 10, - "is_video_on": False, - }, - "previous_position": 1258, - "debug_source": "started_playing", - } - - try: - r = self.session.put( - URL, - data=json.dumps(payload, separators=(",", ":")) - ) - - if r.status_code == 200: - ctx.stop_event.set() - break - else: - attempt_count += 1 - - except Exception as e: - attempt_count += 1 - - if attempt_count >= 3: - ctx.stop_event.set() - break - - seq_num += 1 - time.sleep(3) - - else: - wait_ids_count += 1 - time.sleep(0.5) - - if wait_ids_count > 20: - ctx.stop_event.set() - break - - t_ws = threading.Thread(target=start_socket, daemon=True) - t_ws.start() - - while not ctx.conn_id: - time.sleep(0.1) - - self.session.headers.update({ - "x-spotify-connection-id": ctx.conn_id - }) - - reg_payload = { - "device": { - "brand": "spotify", - "capabilities": {"change_volume": True, "enable_play_token": True, "supports_file_media_type": True, - "manifest_formats": ["file_ids_mp4", "file_ids_mp4_dual"], "audio_podcasts": True, - "video_playback": True}, - "device_id": DEVICE_ID, "device_type": "computer", "metadata": {}, "model": "web_player", - "name": "Python Final", "platform_identifier": "web_player windows 10;chrome 124.0.0.0;desktop", - "is_group": False - }, - "connection_id": ctx.conn_id, "client_version": "harmony:4.62.1-5dc29b8a7", "volume": 65535 - } - self.session.post(URL_REGISTRO, json=reg_payload) - - t_put = threading.Thread(target=keep_alive_loop, daemon=True) - t_put.start() - - track_uri = f"spotify:track:{TRACK_ID}" - cmd_payload = { - "command": { - "context": {"uri": track_uri, "url": f"context://{track_uri}", "metadata": {}}, - "play_origin": {"feature_identifier": "harmony", "feature_version": "4.26.0"}, - "options": {"license": "premium", "skip_to": {"track_uri": track_uri}, "player_options_override": {}}, - "endpoint": "play" - } - } - self.session.post(URL_COMMAND, json=cmd_payload) - - try: - while not ctx.stop_event.is_set(): - time.sleep(1) - except KeyboardInterrupt: - ctx.stop_event.set() - - ################################################## - - params = { - "manifestFileFormat": [ - "file_ids_mp4", - "manifest_ids_video" - ] - } - response = self.session.get( - self.TRACK_PLAYBACK_API_URL.format( - media_type=media_type, - media_id=media_id - ), - params=params - ) - return response.json() - - def get_lyrics(self, track_id: str) -> dict | None: - self._refresh_session_auth() - response = self.session.get(self.LYRICS_API_URL.format(track_id=track_id)) - if response.status_code == 404: - return None - check_response(response) - return response.json() - - def extract_keys_with_cdrm(self, pssh, media_type): - cmd = self.session.post('https://cdrm-project.com/api/decrypt', - headers={'Accept': 'application/json', - 'Content-Type': 'application/json'}, - json={ - 'pssh': pssh, - 'licurl': self.WIDEVINE_LICENSE_API_URL.format(type=media_type), - 'headers': str(self.session.headers) - }).json() - return cmd['message'] - - def get_track(self, track_id: str) -> dict: - self._refresh_session_auth() - payload = { - "variables": { - "uri": f"spotify:track:{track_id}" - }, - "operationName": "getTrack", - "extensions": { - "persistedQuery": { - "version": 1, - "sha256Hash": "612585ae06ba435ad26369870deaae23b5c8800a256cd8a57e08eddc25a37294" - } - } - } - - response = self.session.post( - self.METADATA_API_URL, - json=payload, - ) - check_response(response) - return response.json() - - def extended_media_collection( - self, - next_url: str, - ) -> typing.Generator[dict, None, None]: - while next_url is not None: - response = self.session.get(next_url) - check_response(response) - extended_collection = response.json() - yield extended_collection - next_url = extended_collection["next"] - time.sleep(self.EXTEND_TRACK_COLLECTION_WAIT_TIME) - - @functools.lru_cache() - - def get_album( - self, - album_id: str, - extend: bool = True, - ) -> dict: - payload = { - "variables": { - "uri": f"spotify:album:{album_id}", "locale": "intl-pt", "offset": 0, "limit": 5000 - }, - "operationName": "getAlbum", - "extensions": { - "persistedQuery": { - "version": 1, - "sha256Hash": "b9bfabef66ed756e5e13f68a942deb60bd4125ec1f1be8cc42769dc0259b4b10" - } - } - } - - response = self.session.post( - self.METADATA_API_URL, - json=payload, - ) - - return response.json()['data']['albumUnion']['tracksV2']['items'] - - - def get_playlist( - self, - playlist_id: str, - extend: bool = True, - ) -> dict: - self._refresh_session_auth() - payload = { - "variables": { - "uri": f"spotify:playlist:{playlist_id}", - "offset": 0, - "limit": 5000, - "enableWatchFeedEntrypoint": True - }, - "operationName": "fetchPlaylist", - "extensions": { - "persistedQuery": { - "version": 1, - "sha256Hash": "bb67e0af06e8d6f52b531f97468ee4acd44cd0f82b988e15c2ea47b1148efc77" - } - } - } - - response = self.session.post( - self.METADATA_API_URL, - json=payload, - ) - return response.json()['data']['playlistV2'] - - - def get_track_credits(self, track_id: str) -> dict: - self._refresh_session_auth() - response = self.session.get( - self.TRACK_CREDITS_API_URL.format(track_id=track_id) - ) - check_response(response) - return response.json() - - def get_episode(self, episode_id: str) -> dict: - self._refresh_session_auth() - payload = { - "variables": { - "uri": f"spotify:episode:{episode_id}" - }, - "operationName": "getEpisodeOrChapter", - "extensions": { - "persistedQuery": { - "version": 1, - "sha256Hash": "8a62dbdeb7bd79605d7d68b01bcdf83f08bc6c6287ee1665ba012c748a4cf1f3" - } - } - } - - response = self.session.post( - self.METADATA_API_URL, - json=payload, - ) - check_response(response) - return response.json()['data']['episodeUnionV2'] - - def get_show(self, show_id: str, extend: bool = True) -> dict: - self._refresh_session_auth() - payload = { - "variables": { - "uri": f"spotify:show:{show_id}", - "offset": 0, - "limit": 5000 - }, - "operationName": "queryPodcastEpisodes", - "extensions": { - "persistedQuery": { - "version": 1, - "sha256Hash": "8e2826c5993383566cc08bf9f5d3301b69513c3f6acb8d706286855e57bf44b2" - } - } - } - response = self.session.post( - self.METADATA_API_URL, - json=payload, - ) - check_response(response) - show = response.json() - '''if extend: - show["episodes"]["items"].extend( - [ - item - for extended_collection in self.extended_media_collection( - show["episodes"]["next"], - ) - for item in extended_collection["items"] - ] - )''' - return show["data"]['podcastUnionV2'] - - def get_artist_albums_selection( - self, - artist_id: str, - extend: bool = True, - ) -> dict: - self._refresh_session_auth() - payload = { - "variables": { - "uri": f"spotify:artist:{artist_id}", - "offset": 0, - "limit": 5000, - "order": "DATE_DESC" - }, - "operationName": "queryArtistDiscographyAlbums", - "extensions": { - "persistedQuery": { - "version": 1, - "sha256Hash": "5e07d323febb57b4a56a42abbf781490e58764aa45feb6e3dc0591564fc56599" - } - } - } - response = self.session.post(self.METADATA_API_URL, json=payload) - check_response(response) - artist_albums = response.json()['data']['artistUnion']['discography']['albums'] - return self._select_and_queue(artist_albums, "Albuns") - - def get_artist_singles_selection( - self, - artist_id: str, - extend: bool = True, - ) -> dict: - self._refresh_session_auth() - payload = { - "variables": { - "uri": f"spotify:artist:{artist_id}", - "offset": 0, - "limit": 5000, - "order": "DATE_DESC" - }, - "operationName": "queryArtistDiscographySingles", - "extensions": { - "persistedQuery": { - "version": 1, - "sha256Hash": "5e07d323febb57b4a56a42abbf781490e58764aa45feb6e3dc0591564fc56599" - } - } - } - response = self.session.post(self.METADATA_API_URL, json=payload) - check_response(response) - artist_albums = response.json()['data']['artistUnion']['discography']['singles'] - return self._select_and_queue(artist_albums, "Singles") - - def get_artist_compilations_selection( - self, - artist_id: str, - extend: bool = True, - ) -> dict: - self._refresh_session_auth() - payload = { - "variables": { - "uri": f"spotify:artist:{artist_id}", - "offset": 0, - "limit": 5000, - "order": "DATE_DESC" - }, - "operationName": "queryArtistDiscographyCompilations", - "extensions": { - "persistedQuery": { - "version": 1, - "sha256Hash": "5e07d323febb57b4a56a42abbf781490e58764aa45feb6e3dc0591564fc56599" - } - } - } - response = self.session.post(self.METADATA_API_URL, json=payload) - check_response(response) - artist_albums = response.json()['data']['artistUnion']['discography']['compilations'] - return self._select_and_queue(artist_albums, "Compilations") - - def get_artist_collaborations_selection( - self, - artist_id: str, - extend: bool = True, - ) -> dict: - self._refresh_session_auth() - payload = { - "variables": { - "uri": f"spotify:artist:{artist_id}", - "offset": 0, - "limit": 5000, - "order": "DATE_DESC" - }, - "operationName": "queryArtistAppearsOn", - "extensions": { - "persistedQuery": { - "version": 1, - "sha256Hash": "9a4bb7a20d6720fe52d7b47bc001cfa91940ddf5e7113761460b4a288d18a4c1" - } - } - } - response = self.session.post(self.METADATA_API_URL, json=payload) - check_response(response) - artist_albums = response.json()['data']['artistUnion']['relatedContent']['appearsOn'] - return self._select_and_queue(artist_albums, "Collaborations") - - def _select_and_queue(self, data_input: any, category_name: str) -> list[str]: - raw_items = [] - if isinstance(data_input, dict) and 'items' in data_input: - raw_items = data_input['items'] - elif isinstance(data_input, list): - raw_items = data_input - - if not raw_items: - print(f"No {category_name} found for this artist.") - return [] - - clean_list = [] - for entry in raw_items: - try: - if 'releases' in entry and 'items' in entry['releases']: - album_data = entry['releases']['items'][0] - clean_obj = { - 'name': album_data.get('name', 'Unknown'), - 'id': album_data.get('id'), - 'year': str(album_data.get('date', {}).get('year', '????')), - 'total_tracks': album_data.get('tracks', {}).get('totalCount', 0) - } - clean_list.append(clean_obj) - elif 'name' in entry: - clean_list.append({ - 'name': entry.get('name'), - 'id': entry.get('id'), - 'year': entry.get('release_date', '????')[:4], - 'total_tracks': entry.get('total_tracks', 0) - }) - except Exception: - continue - - if not clean_list: - print(f"Could not parse items for {category_name}.") - return [] - - print(f"\n--- Select {category_name} ---") - print("Index | Tracks | Year | Name") - for index, album in enumerate(clean_list): - display_str = f"{album['total_tracks']:03d} | {album['year']} | {album['name']}" - print(f"[{index + 1}] {display_str}") - - print("\nType numbers separated by space (e.g. '1 3') or 'all'.") - user_input = input("Selection: ").strip().lower() - - selected_ids = [] - - if user_input == 'all': - selected_ids = [album['id'] for album in clean_list] - else: - parts = user_input.replace(',', ' ').split() - for part in parts: - if part.isdigit(): - idx = int(part) - 1 - if 0 <= idx < len(clean_list): - selected_ids.append(clean_list[idx]['id']) - - return selected_ids - - - def get_video_manifest( - self, - gid: str, - ) -> dict: - self._refresh_session_auth() - response = self.session.get(self.VIDEO_MANIFEST_API_URL.format(gid=gid)) - check_response(response) - return response.json() - - def get_seek_table(self, file_id: str) -> dict: - headers = { - "Accept": "*/*", - "Accept-Encoding": "gzip, deflate, br, zstd", - "Accept-Language": "en-US", - "Cache-Control": "no-cache", - "Connection": "keep-alive", - "Origin": self.SPOTIFY_HOME_PAGE_URL, - "Pragma": "no-cache", - "Priority": "u=4", - "Referer": self.SPOTIFY_HOME_PAGE_URL, - "Sec-Fetch-Dest": "empty", - "Sec-Fetch-Mode": "cors", - "Sec-Fetch-Site": "cross-site", - "User-Agent": self.session.headers["user-agent"], - } - response = requests.get( - self.SEEK_TABLE_API_URL.format(file_id=file_id), - headers=headers, - ) - check_response(response) - return response.json() - - def get_playplay_license(self, file_id: str, challenge: bytes) -> bytes: - self._refresh_session_auth() - response = self.session.post( - self.PLAYPLAY_LICENSE_API_URL.format(file_id=file_id), - challenge, - ) - check_response(response) - return response.content - - def get_widevine_license(self, challenge: bytes, media_type: str) -> bytes: - self._refresh_session_auth() - response = self.session.post( - self.WIDEVINE_LICENSE_API_URL.format(type=media_type), - challenge, - ) - if response.status_code == 403: - logger.error(f"The device.wvd file is invalid or banned.") - logger.warning(f'Delete the device.wvd file to use the alternate key.') - check_response(response) - return response.content - - def get_stream_urls(self, file_id: str) -> str: - self._refresh_session_auth() - response = self.session.get(self.STREAM_URLS_API_URL.format(file_id=file_id)) - check_response(response) - return response.json() - - def get_now_playing_view(self, track_id: str, artist_id: str) -> dict: - self._refresh_session_auth() - response = self.session.get( - self.PATHFINDER_API_URL, - params={ - "operationName": "queryNpvArtist", - "variables": json.dumps( - { - "artistUri": f"spotify:artist:{artist_id}", - "trackUri": f"spotify:track:{track_id}", - "enableCredits": True, - "enableRelatedVideos": True, - } - ), - "extensions": json.dumps( - { - "persistedQuery": { - "version": 1, - "sha256Hash": "4ec4ae302c609a517cab6b8868f601cd3457c751c570ab12e988723cc036284f", - } - } - ), - }, - ) - check_response(response) - return response.json() diff --git a/votify/totp.py b/votify/totp.py deleted file mode 100644 index be071d7..0000000 --- a/votify/totp.py +++ /dev/null @@ -1,36 +0,0 @@ -import hashlib -import hmac - -from typing import Collection - - -class TOTP: - PERIOD = 30 - DIGITS = 6 - - def __init__(self, *, version: int, ciphertext: Collection[int]) -> None: - self.version = version - self.secret = self.derive(ciphertext) - - @staticmethod - def derive(ciphertext: Collection[int]) -> bytes: - return "".join( - str(byte ^ ((i % 33) + 9)) for i, byte in enumerate(ciphertext) - ).encode("ascii") - - def generate(self, timestamp: int) -> str: - counter = int(timestamp) // 1000 // self.PERIOD - counter_bytes = counter.to_bytes(8, "big") - - h = hmac.new(self.secret, counter_bytes, hashlib.sha1) - hmac_result = h.digest() - - offset = hmac_result[-1] & 0x0F - binary = ( - (hmac_result[offset] & 0x7F) << 24 - | (hmac_result[offset + 1] & 0xFF) << 16 - | (hmac_result[offset + 2] & 0xFF) << 8 - | (hmac_result[offset + 3] & 0xFF) - ) - - return str(binary % (10**self.DIGITS)).zfill(self.DIGITS)