From 74799f7706fbf0415dac51fc61136c3bfaa2c8de Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Sat, 24 Jan 2026 01:54:59 -0300 Subject: [PATCH 01/36] Fix --- votify/spotify_api.py | 158 ++++++++++++++++++++++++++++++------------ 1 file changed, 113 insertions(+), 45 deletions(-) diff --git a/votify/spotify_api.py b/votify/spotify_api.py index cb637eb..3323e5f 100644 --- a/votify/spotify_api.py +++ b/votify/spotify_api.py @@ -24,10 +24,11 @@ class SpotifyApi: 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.spotify.com/v1/{type}/{item_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/v7/json/sources/{gid}/options/supports_drm" + TRACK_PLAYBACK_API_URL = "https://gue1-spclient.spotify.com/track-playback/v1/media/spotify:{type}:{id}" PLAYPLAY_LICENSE_API_URL = ( "https://gew4-spclient.spotify.com/playplay/v1/key/{file_id}" ) @@ -43,6 +44,7 @@ class SpotifyApi: 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, @@ -108,10 +110,12 @@ class SpotifyApi: check_response(response) return 1e3 * response.json()["serverTime"] - def set_authorization_header(self, token: str) -> None: + 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 } ) @@ -124,18 +128,36 @@ class SpotifyApi: "reason": "init", "productType": "web-player", "totp": totp, + "totpServer": totp, "totpVer": str(self.totp.version), - "ts": str(server_time), }, ) check_response(response) authorization_info = response.json() if not authorization_info.get("accessToken"): raise ValueError("Failed to retrieve access token.") - self.set_authorization_header(authorization_info["accessToken"]) + + 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: @@ -146,20 +168,34 @@ class SpotifyApi: def _setup_authorization_with_device_flow(self) -> None: device_flow = SpotifyDeviceFlow(self.sp_dc) token_data = device_flow.get_token() - self.set_authorization_header(token_data["access_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() * 1000 + timestamp_now = time.time() if timestamp_now < timestamp_session_expire: return self._setup_authorization() def _setup_user_profile(self) -> None: - response = self.session.get(self.METADATA_API_URL.format(type="me", item_id="")) + 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() @@ -177,12 +213,21 @@ class SpotifyApi: media_type: str, ) -> dict: self._refresh_session_auth() - response = self.session.get( - self.GID_METADATA_API_URL.format(gid=gid, media_type=media_type) - ) + 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() + + params = {"manifestFileFormat": ["file_ids_mp4"]} + response = self.session.get(f'https://gue1-spclient.spotify.com/track-playback/v1/media/spotify:track:{media_id}?manifestFileFormat=file_ids_mp4') + 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)) @@ -193,8 +238,22 @@ class SpotifyApi: def get_track(self, track_id: str) -> dict: self._refresh_session_auth() - response = self.session.get( - self.METADATA_API_URL.format(type="tracks", item_id=track_id) + 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() @@ -212,51 +271,60 @@ class SpotifyApi: time.sleep(self.EXTEND_TRACK_COLLECTION_WAIT_TIME) @functools.lru_cache() + def get_album( self, album_id: str, extend: bool = True, ) -> dict: - self._refresh_session_auth() - response = self.session.get( - self.METADATA_API_URL.format(type="albums", item_id=album_id) + 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, ) - check_response(response) - album = response.json() - if extend: - album["tracks"]["items"].extend( - [ - item - for extended_collection in self.extended_media_collection( - album["tracks"]["next"], - ) - for item in extended_collection["items"] - ] - ) - return album + + return response.json()['data']['albumUnion']['tracksV2']['items'] + def get_playlist( self, playlist_id: str, extend: bool = True, ) -> dict: - self._refresh_session_auth() - response = self.session.get( - self.METADATA_API_URL.format(type="playlists", item_id=playlist_id) + 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, ) - check_response(response) - playlist = response.json() - if extend: - playlist["tracks"]["items"].extend( - [ - item - for extended_collection in self.extended_media_collection( - playlist["tracks"]["next"], - ) - for item in extended_collection["items"] - ] - ) - return playlist + return response.json()['data']['playlistV2'] + def get_track_credits(self, track_id: str) -> dict: self._refresh_session_auth() From 5160be1c771e899c824c651ab73307cdfa48daf8 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Sat, 24 Jan 2026 01:55:31 -0300 Subject: [PATCH 02/36] Fix --- votify/downloader_song.py | 262 ++++++++++++++++++++++++++------------ 1 file changed, 179 insertions(+), 83 deletions(-) diff --git a/votify/downloader_song.py b/votify/downloader_song.py index fac7332..d258748 100644 --- a/votify/downloader_song.py +++ b/votify/downloader_song.py @@ -23,86 +23,123 @@ class DownloaderSong(DownloaderAudio): self.no_lrc = no_lrc def get_tags( - self, - track_metadata: dict, - album_metadata: dict, - track_credits: dict, - lyrics_unsynced: str, + self, + track_metadata: dict, + album_metadata: list, + track_credits: dict, + lyrics_unsynced: str, ) -> dict: - external_ids = track_metadata.get("external_ids") - external_urls = (track_metadata.get("linked_from") or track_metadata)[ - "external_urls" - ] - track_id = self.downloader.get_media_id(track_metadata) + track_info = track_metadata.get("data", {}).get("trackUnion", {}) + album_info = track_info.get("albumOfTrack", {}) + + external_url = track_info.get("sharingInfo", {}).get("shareUrl") + + isrc = None + + release_date_iso = album_info.get("date", {}).get("isoString") + precision_raw = album_info.get("date", {}).get("precision") + + precision = precision_raw.lower() if precision_raw else "day" + + release_date_str = release_date_iso.split("T")[0] if release_date_iso else None + release_date_datetime_obj = self.downloader.get_release_date_datetime_obj( - album_metadata["release_date"], - album_metadata["release_date_precision"], + release_date_str, + precision, ) - producers = next( - role - for role in track_credits["roleCredits"] - if role["roleTitle"] == "Producers" - )["artists"] - composers = next( - role - for role in track_credits["roleCredits"] - if role["roleTitle"] == "Writers" - )["artists"] - disc = next( - ( - track["disc_number"] - for track in album_metadata["tracks"]["items"] - if self.downloader.get_media_id(track) == track_id - ), + + track_artists = [] + + raw_artist_items = track_info.get("firstArtist", {}).get("items", []) + + other_artist_items = track_info.get("otherArtists", {}).get("items", []) + + all_artist_items = raw_artist_items + other_artist_items + + for item in all_artist_items: + name = item.get("profile", {}).get("name") + if name: + track_artists.append({"name": name}) + + if not track_artists: + alt_artists = track_info.get("artists", {}).get("items", []) + for item in alt_artists: + name = item.get("profile", {}).get("name") or item.get("name") + if name: + track_artists.append({"name": name}) + + if not track_artists: + track_artists = [{"name": "Unknown Artist"}] + + album_artists = track_artists + + def get_credits_artists(role_title): + try: + role = next( + r for r in track_credits.get("roleCredits", []) + if r.get("roleTitle") == role_title + ) + return role.get("artists", []) + except StopIteration: + return [] + + producers = get_credits_artists("Producers") + composers = get_credits_artists("Writers") + + current_uri = track_info.get("uri") + + current_track_entry = next( + (item.get("track", {}) for item in album_metadata if item.get("track", {}).get("uri") == current_uri), + {} ) + + disc_number = current_track_entry.get("discNumber", 1) + + try: + disc_total = max( + (item.get("track", {}).get("discNumber", 1) for item in album_metadata), + default=1 + ) + except Exception: + disc_total = 1 + + track_total = album_info.get("tracks", {}).get("totalCount") + if not track_total: + track_total = len(album_metadata) + + copyright_text = next( + (c["text"] for c in album_info.get("copyright", {}).get("items", []) if c.get("type") == "P"), + None + ) + label = album_info.get("courtesyLine") or album_info.get("label") + tags = { - "album": album_metadata["name"], - "album_artist": self.downloader.get_artist_string( - album_metadata["artists"] - ), - "artist": self.downloader.get_artist_string(track_metadata["artists"]), - "compilation": ( - True if album_metadata["album_type"] == "compilation" else False - ), + "album": album_info.get("name"), + "album_artist": self.downloader.get_artist_string(album_artists), + "artist": self.downloader.get_artist_string(track_artists), + "compilation": False, "composer": ( self.downloader.get_artist_string(composers) if composers else None ), - "copyright": next( - (i["text"] for i in album_metadata["copyrights"] if i["type"] == "P"), - None, - ), - "disc": int(disc), - "disc_total": int(album_metadata["tracks"]["items"][-1]["disc_number"]), - "isrc": external_ids.get("isrc") if external_ids is not None else None, - "label": album_metadata.get("label"), + "copyright": copyright_text, + "disc": int(disc_number), + "disc_total": int(disc_total), + "isrc": isrc, + "label": label, "lyrics": lyrics_unsynced, "media_type": "Song", "producer": ( self.downloader.get_artist_string(producers) if producers else None ), - "rating": "Explicit" if track_metadata.get("explicit") else "Unknown", + "rating": "Explicit" if track_info.get("contentRating", {}).get("label") == "EXPLICIT" else "Unknown", "release_date": self.downloader.get_release_date_tag( release_date_datetime_obj ), "release_year": str(release_date_datetime_obj.year), - "title": track_metadata["name"], - "track": int( - next( - ( - track["track_number"] - for track in album_metadata["tracks"]["items"] - if self.downloader.get_media_id(track) == track_id - ), - ) - ), - "track_total": int( - max( - track["track_number"] - for track in album_metadata["tracks"]["items"] - if track["disc_number"] == disc - ) - ), - "url": external_urls["spotify"], + "title": track_info.get("name"), + "track": int(track_info.get("trackNumber", 1)), + "track_total": int(track_total), + "url": external_url, } return tags @@ -140,54 +177,107 @@ class DownloaderSong(DownloaderAudio): self.downloader.cleanup_temp_path() def _download( - self, - track_id: str, - track_metadata: dict = None, - album_metadata: dict = None, - gid_metadata: dict = None, - stream_info: StreamInfoAudio = None, - playlist_metadata: dict = None, - playlist_track: int = None, - decryption_key: bytes = None, + self, + track_id: str, + track_metadata: dict = None, + album_metadata: dict = None, + gid_metadata: dict = None, + stream_info: StreamInfoAudio = None, + playlist_metadata: dict = None, + product_name: dict = None, + playlist_track: int = None, + decryption_key: bytes = None, ): - if not track_metadata: + if not album_metadata: logger.debug("Getting track metadata") track_metadata = self.downloader.spotify_api.get_track(track_id) + + try: + current_date = track_metadata['data']['trackUnion']['albumOfTrack'].get('release_date') + + if current_date == '1970-01-01': + fresh_data = self.downloader.spotify_api.get_track(track_id) + + real_date = None + real_precision = 'day' + real_album_name = None + + if 'album' in fresh_data: + real_date = fresh_data['album']['release_date'] + real_precision = fresh_data['album'].get('release_date_precision', 'day') + real_album_name = fresh_data['album']['name'] + + elif 'data' in fresh_data: + src_album = fresh_data['data']['trackUnion']['albumOfTrack'] + if 'date' in src_album and 'isoString' in src_album['date']: + real_date = src_album['date']['isoString'] + real_precision = 'day' + + if real_date and not str(real_date).startswith('1970'): + track_metadata['data']['trackUnion']['albumOfTrack']['release_date'] = real_date + track_metadata['data']['trackUnion']['albumOfTrack']['release_date_precision'] = real_precision + + if not album_metadata: + album_metadata = {} + + album_metadata['release_date'] = real_date + album_metadata['release_date_precision'] = real_precision + if real_album_name: + album_metadata['name'] = real_album_name + + except Exception as e: + pass + if not album_metadata: logger.debug("Getting album metadata") - album_metadata = self.downloader.spotify_api.get_album( - track_metadata["album"]["id"] - ) + try: + alb_node = track_metadata['data']['trackUnion']['albumOfTrack'] + album_id = alb_node['id'] + album_metadata = self.downloader.spotify_api.get_album(album_id) + except (KeyError, TypeError): + pass + if not gid_metadata: logger.debug("Getting GID metadata") - gid_metadata = self.downloader.get_gid_metadata(track_id, "track") + try: + gid_metadata = self.downloader.get_gid_metadata(track_id, "track", product_name) + except TypeError: + gid_metadata = self.downloader.get_gid_metadata(track_id, "track") + if gid_metadata.get("original_video"): logger.warning("Track is a music video, skipping.") return + if not stream_info: logger.debug("Getting stream info") - stream_info = self.get_stream_info(gid_metadata, "track") + stream_info = self.get_stream_info(gid_metadata, product_name, "track") + if not stream_info.file_id: logger.warning( "Track is not available on Spotify's " "servers and no alternative found, skipping" ) return + if stream_info.quality != self.audio_quality: - logger.warning(f"Quality has been changed to {stream_info.quality.value}") - if gid_metadata.get("has_lyrics"): + logger.warning(f"Quality has been changed to {stream_info.quality}") + + if gid_metadata: logger.debug("Getting lyrics") lyrics = self.get_lyrics(track_id) else: lyrics = Lyrics() + logger.debug("Getting track credits") track_credits = self.downloader.spotify_api.get_track_credits(track_id) + tags = self.get_tags( track_metadata, album_metadata, track_credits, lyrics.unsynced, ) + if playlist_metadata: tags = { **tags, @@ -196,6 +286,7 @@ class DownloaderSong(DownloaderAudio): playlist_track, ), } + file_extension = self.get_file_extension() final_path = self.downloader.get_final_path( "track", @@ -205,11 +296,13 @@ class DownloaderSong(DownloaderAudio): lrc_path = self.downloader.get_lrc_path(final_path) cover_path = self.get_cover_path(final_path) cover_url = self.downloader.get_cover_url( - album_metadata, + track_metadata, COVER_SIZE_X_KEY_MAPPING_SONG, ) + decrypted_path = None remuxed_path = None + if self.lrc_only: pass elif final_path.exists() and not self.downloader.overwrite: @@ -244,18 +337,21 @@ class DownloaderSong(DownloaderAudio): decrypted_path, remuxed_path, ) - if self.no_lrc or not lyrics.synced: + + if self.no_lrc: pass elif lrc_path.exists() and not self.downloader.overwrite: logger.debug(f'Synced lyrics already exists at "{lrc_path}", skipping') else: logger.debug(f'Saving synced lyrics to "{lrc_path}"') - self.downloader.save_lrc(lrc_path, lyrics.synced) + lyrics = None + media_temp_path = ( remuxed_path if remuxed_path is not None and remuxed_path.exists() else decrypted_path ) + self.downloader._final_processing( cover_path, cover_url, From 62a9d7dd21df6b80106e2ddc1097cdb4e7f9f2e1 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Sat, 24 Jan 2026 01:56:02 -0300 Subject: [PATCH 03/36] Fix --- votify/downloader_audio.py | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/votify/downloader_audio.py b/votify/downloader_audio.py index 0ec21fa..cb7b51a 100644 --- a/votify/downloader_audio.py +++ b/votify/downloader_audio.py @@ -60,11 +60,12 @@ class DownloaderAudio: def get_stream_info( self, gid_metadata: dict, + product_name: dict, media_type: str, ) -> StreamInfoAudio: stream_info = StreamInfoAudio() if media_type == "track": - audio_files = gid_metadata.get("file") + audio_files = gid_metadata elif media_type == "episode": audio_files = gid_metadata.get("audio") else: @@ -72,20 +73,23 @@ class DownloaderAudio: audio_files = audio_files or gid_metadata.get("alternative") if not audio_files: return stream_info - if audio_files[0].get("gid"): - audio_files = audio_files[0]["file"] - quality, audio_file = self.get_audio_file(audio_files) - if not audio_file: - return stream_info - file_id = audio_file["file_id"] + if audio_files: + audio_files = audio_files + file_id = audio_files stream_url = self.downloader.spotify_api.get_stream_urls(file_id)["cdnurl"][0] stream_info.stream_url = stream_url stream_info.file_id = file_id + product = product_name['data']['me']['account']['product'] + if 'FREE' == product: + quality = 128000 + else: + quality = 256000 stream_info.quality = quality if self.audio_quality in AAC_AUDIO_QUALITIES: seek_table = self.downloader.spotify_api.get_seek_table(file_id) pssh = seek_table["pssh"] stream_info.widevine_pssh = pssh + return stream_info def get_decryption_key(self, stream_info: StreamInfoAudio) -> str: From 43e8f3b811949143090667780b693fda8add0073 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Sat, 24 Jan 2026 01:57:00 -0300 Subject: [PATCH 04/36] Fix --- votify/downloader.py | 303 ++++++++++++++++++++++++++++++++++++------- 1 file changed, 256 insertions(+), 47 deletions(-) diff --git a/votify/downloader.py b/votify/downloader.py index b77827a..22385d0 100644 --- a/votify/downloader.py +++ b/votify/downloader.py @@ -142,31 +142,122 @@ class Downloader: return UrlInfo(type=url_regex_result.group(1), id=url_regex_result.group(2)) def get_download_queue( - self, - media_type: str, - media_id: str, + self, + media_type: str, + media_id: str, ) -> list[DownloadQueueItem]: download_queue = [] if media_type == "album": album = self.spotify_api.get_album(media_id) - for track in album["tracks"]["items"]: - download_queue.append( - DownloadQueueItem( - album_metadata=album, - media_metadata=track, - ) - ) + for item in album: + t = item['track'] + track_id = t['uri'].split(':')[-1] + download_queue.append({ + 'data': { + 'trackUnion': { + 'id': track_id, + 'uri': t['uri'], + 'name': t['name'], + 'trackNumber': t['trackNumber'], + 'duration': { + 'totalMilliseconds': t['duration']['totalMilliseconds'] + }, + 'is_playable': t.get('is_playable', True), + 'artists': { + 'items': [ + { + 'profile': { + 'name': t['artists']['items'][0]['profile']['name'] + } + } + ] + } + } + } + }) elif media_type == "playlist": - playlist = self.spotify_api.get_playlist(media_id) - for track in playlist["tracks"]["items"]: - if track["track"] is None: - continue - download_queue.append( - DownloadQueueItem( - playlist_metadata=playlist, - media_metadata=track["track"], - ) - ) + playlist = self.spotify_api.get_playlist(media_id) + playlist_metadata = playlist + if 'content' in playlist and 'items' in playlist['content']: + items_list = playlist['content']['items'] + else: + items_list = [] + for item in items_list: + if 'itemV2' not in item or 'data' not in item['itemV2']: + continue + + t = item['itemV2']['data'] + + if 'uri' not in t: + continue + + track_id = t['uri'].split(':')[-1] + + duration_ms = 0 + if 'duration' in t and isinstance(t['duration'], dict): + duration_ms = t['duration'].get('totalMilliseconds', 0) + elif 'trackDuration' in t and isinstance(t['trackDuration'], dict): + duration_ms = t['trackDuration'].get('totalMilliseconds', 0) + + album_data = t.get('albumOfTrack', {}) + if 'uri' in album_data and 'id' not in album_data: + album_data['id'] = album_data['uri'].split(':')[-1] + + if 'date' in album_data and isinstance(album_data['date'], dict): + date_obj = album_data['date'] + if 'isoString' in date_obj and date_obj['isoString']: + album_data['release_date'] = date_obj['isoString'] + album_data['release_date_precision'] = 'day' + elif 'year' in date_obj and date_obj['year']: + album_data['release_date'] = str(date_obj['year']) + album_data['release_date_precision'] = 'year' + + if 'release_date' not in album_data: + album_data['release_date'] = '1970-01-01' + album_data['release_date_precision'] = 'day' + + artist_name = "Unknown Artist" + try: + if 'artists' in t and 'items' in t['artists'] and len(t['artists']['items']) > 0: + profile = t['artists']['items'][0].get('profile') + if profile: + artist_name = profile.get('name', "Unknown Artist") + except Exception: + pass + + is_playable = t.get('isPlayable', True) + + track_object = { + 'data': { + 'trackUnion': { + 'id': track_id, + 'uri': t['uri'], + '__typename': 'Track', + 'name': t.get('name', 'Unknown Track'), + 'trackNumber': t.get('trackNumber', 1), + 'duration': { + 'totalMilliseconds': duration_ms + }, + 'isPlayable': is_playable, + 'albumOfTrack': album_data, + 'artists': { + 'items': [ + { + 'profile': { + 'name': artist_name + } + } + ] + } + } + } + } + + download_queue.append({ + 'media_metadata': track_object, + 'playlist_metadata': playlist_metadata + }) + elif media_type == "track": download_queue.append( DownloadQueueItem( @@ -260,9 +351,30 @@ class Downloader: return download_queue def get_cover_url(self, metadata: dict, cover_size_mapping: dict) -> str | None: - if not metadata.get("images"): + if not isinstance(metadata, dict): return None - return self._get_cover_url(metadata["images"], cover_size_mapping) + images = None + try: + images = ( + metadata.get("data", {}) + .get("trackUnion", {}) + .get("albumOfTrack", {}) + .get("coverArt", {}) + .get("sources") + ) + except AttributeError: + pass + + if not images: + images = metadata.get("images") + + if not images: + images = metadata.get("coverArt", {}).get("sources") + + if not images: + return None + + return self._get_cover_url(images, cover_size_mapping) def _get_cover_url(self, images_dict: list[dict], cover_size_mapping: dict) -> str: original_cover_url = images_dict[0]["url"] @@ -273,15 +385,59 @@ class Downloader: return cover_url def get_media_id(self, media_metadata: dict) -> str: - return (media_metadata.get("linked_from") or media_metadata)["id"] + linked_from = media_metadata.get("linked_from") + if linked_from: + media_metadata = linked_from + + if "data" in media_metadata and isinstance(media_metadata["data"], dict): + track_union = media_metadata["data"].get("trackUnion") + if track_union and "id" in track_union: + return track_union["id"] + + if "id" in media_metadata: + return media_metadata["id"] + + if "itemV2" in media_metadata: + return media_metadata["itemV2"]["data"]["id"] + + if "media_metadata" in media_metadata: + nested = media_metadata["media_metadata"] + if "data" in nested and "trackUnion" in nested["data"]: + return nested["data"]["trackUnion"]["id"] + if "id" in nested: + return nested["id"] + + if "uri" in media_metadata: + return media_metadata["uri"].split(":")[-1] + + raise ValueError("Could not find media ID in metadata") def get_playlist_tags(self, playlist_metadata: dict, playlist_track: int) -> dict: + playlist_title = playlist_metadata.get("name", "Unknown Playlist") + playlist_artist = "Unknown Artist" + try: + if "ownerV2" in playlist_metadata: + playlist_artist = playlist_metadata['ownerV2']['data']['name'] + except (KeyError, TypeError): + pass + + if playlist_artist == "Unknown Artist" and "owner" in playlist_metadata: + owner_obj = playlist_metadata["owner"] + if isinstance(owner_obj, dict): + playlist_artist = owner_obj.get("display_name") or owner_obj.get("id", "Unknown Artist") + elif isinstance(owner_obj, str): + playlist_artist = owner_obj + + if playlist_artist == "Unknown Artist": + playlist_artist = playlist_metadata.get("owner_name", "Unknown Artist") + return { - "playlist_artist": playlist_metadata["owner"]["display_name"], - "playlist_title": playlist_metadata["name"], + "playlist_artist": playlist_artist, + "playlist_title": playlist_title, "playlist_track": playlist_track, } + def get_playlist_file_path( self, tags: dict, @@ -309,11 +465,14 @@ class Downloader: def get_final_path(self, media_type: str, tags: dict, file_extension: str) -> Path: if media_type == "track": - template_folder = ( - self.template_folder_compilation.split("/") - if tags.get("compilation") - else self.template_folder_album.split("/") - ) + if tags.get("playlist_title"): + template_folder = self.template_file_playlist.split("/") + else: + template_folder = ( + self.template_folder_compilation.split("/") + if tags.get("compilation") + else self.template_folder_album.split("/") + ) template_file = ( self.template_file_multi_disc.split("/") if tags["disc_total"] > 1 @@ -327,7 +486,9 @@ class Downloader: template_file = self.template_file_music_video.split("/") else: raise RuntimeError() + template_final = template_folder + template_file + return Path( self.output_path, *[ @@ -335,8 +496,8 @@ class Downloader: for i in template_final[0:-1] ], ( - self.get_sanitized_string(template_final[-1].format(**tags), False) - + file_extension + self.get_sanitized_string(template_final[-1].format(**tags), False) + + file_extension ), ) @@ -367,12 +528,38 @@ class Downloader: playlist_file.writelines(playlist_file_lines) def get_gid_metadata( - self, - media_id: str, - media_type: str, - ) -> dict: - gid = self.spotify_api.media_id_to_gid(media_id) - return self.spotify_api.get_gid_metadata(gid, media_type) + self, + media_id: str, + media_type: str, + audio_quality: dict, + ) -> str: + playback_info = self.spotify_api.get_track_playback_info(media_id, media_type) + user_product = audio_quality.get('data', {}).get('me', {}).get('account', {}).get('product', 'free') + + if user_product == 'PREMIUM': + target_bitrate = 256000 + else: + target_bitrate = 128000 + + expected_key = f"spotify:{media_type}:{media_id}" + + if expected_key not in playback_info.get("media", {}): + raise ValueError(f"No avaliable Track") + + track_manifest = playback_info["media"][expected_key] + + try: + audio_file_id = next( + f["file_id"] + for files in track_manifest["item"]["manifest"].values() + if isinstance(files, list) + for f in files + if isinstance(f, dict) and f.get("bitrate") == int(target_bitrate) + ) + except StopIteration: + raise ValueError(f"No audio file found with bitrate {target_bitrate}") + + return audio_file_id def get_playplay_decryption_key(self, file_id: str) -> bytes: raise NotImplementedError() @@ -413,24 +600,46 @@ class Downloader: return dirty_string.strip() def get_release_date_datetime_obj( - self, - release_date: str, - release_date_precision: str, + self, + release_date: str, + release_date_precision: str, ) -> datetime.datetime: - return datetime.datetime.strptime( - release_date, - self.RELEASE_DATE_PRECISION_MAPPING[release_date_precision], - ) + if not release_date: + return datetime.datetime(1970, 1, 1) + + if not release_date_precision: + if len(release_date) == 4: + release_date_precision = 'year' + elif len(release_date) == 7: + release_date_precision = 'month' + else: + release_date_precision = 'day' + + try: + precision_key = str(release_date_precision).lower() + fmt = self.RELEASE_DATE_PRECISION_MAPPING.get(precision_key, "%Y-%m-%d") + return datetime.datetime.strptime(release_date, fmt) + except (ValueError, TypeError, KeyError, AttributeError): + for fmt in ["%Y", "%Y-%m", "%Y-%m-%d"]: + try: + return datetime.datetime.strptime(release_date, fmt) + except ValueError: + continue + return datetime.datetime(1970, 1, 1) def get_release_date_tag(self, datetime_obj: datetime.datetime) -> str: return datetime_obj.strftime(self.date_tag_template) def get_artist_string(self, artist_list: list[dict]) -> str: + if not artist_list: + return "" + if len(artist_list) == 1: return artist_list[0]["name"] + return ( - ", ".join(i["name"] for i in artist_list[:-1]) - + f' & {artist_list[-1]["name"]}' + ", ".join(i["name"] for i in artist_list[:-1]) + + f' & {artist_list[-1]["name"]}' ) def get_file_temp_path( From e4067aa4a07b7ec340fec0c4c2f04f48ee3d2eb1 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Sat, 24 Jan 2026 01:57:38 -0300 Subject: [PATCH 05/36] Fix Refactor media metadata handling and improve error logging. --- votify/cli.py | 149 ++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 119 insertions(+), 30 deletions(-) diff --git a/votify/cli.py b/votify/cli.py index b643b25..c559de5 100644 --- a/votify/cli.py +++ b/votify/cli.py @@ -1,5 +1,7 @@ from __future__ import annotations +import traceback + import inspect import logging import time @@ -489,7 +491,7 @@ def main( ) is_premium = ( - True if force_premium else spotify_api.user_profile["product"] == "premium" + True if force_premium else spotify_api.user_profile['data']['me']['account']['product'] == "premium" ) if not lrc_only: if download_mode == DownloadMode.ARIA2C and not downloader.aria2c_path_full: @@ -566,11 +568,19 @@ def main( urls = _urls error_count = 0 + + global_playlist_metadata = None + for url_index, url in enumerate(urls, start=1): url_progress = color_text(f"URL {url_index}/{len(urls)}", colorama.Style.DIM) logger.info(f'({url_progress}) Checking "{url}"') try: url_info = downloader.get_url_info(url) + + if url_info.type == 'playlist': + if hasattr(url_info, 'name') and url_info.name: + global_playlist_metadata = {'name': url_info.name, 'owner_name': getattr(url_info, 'owner', 'Unknown')} + if url_info.type == "artist": download_queue = downloader.get_download_queue_from_artist(url_info.id) else: @@ -582,76 +592,154 @@ def main( error_count += 1 logger.error( f'({url_progress}) Failed to check "{url}"', - exc_info=no_exceptions, + exc_info=not no_exceptions, ) continue + for index, download_queue_item in enumerate(download_queue, start=1): queue_progress = color_text( f"Track {index}/{len(download_queue)} from URL {url_index}/{len(urls)}", colorama.Style.DIM, ) - media_metadata = download_queue_item.media_metadata + + media_metadata = None + if isinstance(download_queue_item, dict): + media_metadata = download_queue_item.get("media_metadata", download_queue_item) + else: + media_metadata = getattr(download_queue_item, "media_metadata", download_queue_item) + + if not isinstance(media_metadata, dict): + media_metadata = getattr(media_metadata, "__dict__", {}) + + if isinstance(media_metadata, dict) and 'track' in media_metadata and isinstance(media_metadata['track'],dict): + clean_track_metadata = media_metadata['track'] + else: + clean_track_metadata = media_metadata + + track_name = "Unknown Track" try: - logger.info( - f'({queue_progress}) Downloading "{media_metadata["name"]}"' - ) - media_id = downloader.get_media_id(media_metadata) - media_type = media_metadata["type"] - gid_metadata = downloader.get_gid_metadata(media_id, media_type) + track_name = media_metadata["data"]["trackUnion"]["name"] + except (KeyError, TypeError): + track_name = clean_track_metadata.get("name", "Unknown Track") + + is_playable = True + + if isinstance(clean_track_metadata, dict): + + if clean_track_metadata.get('isPlayable') is False: + is_playable = False + elif clean_track_metadata.get('is_playable') is False: + is_playable = False + elif clean_track_metadata.get('playable') is False: + is_playable = False + + elif 'data' in clean_track_metadata: + try: + union = clean_track_metadata['data']['trackUnion'] + if union.get('isPlayable') is False or union.get('playable') is False: + is_playable = False + except: + pass + + if not is_playable: + logger.warning(f'({queue_progress}) Skipping "{track_name}": Track unavailable/unplayable.') + continue + + try: + logger.info(f'({queue_progress}) Downloading "{track_name}"') + + try: + media_id = downloader.get_media_id(clean_track_metadata) + except ValueError: + if isinstance(download_queue_item, dict) and 'id' in download_queue_item: + media_id = download_queue_item['id'] + else: + raise + + media_metadata_for_download = media_metadata + + media_type = "track" + try: + type_from_meta = media_metadata['data']['trackUnion']['__typename'] + media_type = type_from_meta.lower() + except (KeyError, TypeError, AttributeError): + if isinstance(media_metadata, dict) and 'episode' in media_metadata: + media_type = "episode" + media_metadata_for_download = None + else: + media_type = "track" + media_metadata_for_download = None + + gid_metadata = downloader.get_gid_metadata(media_id, media_type, spotify_api.user_profile) + + def get_meta(item, key): + val = getattr(item, key, None) + if val is None and isinstance(item, dict): + val = item.get(key) + return val + + safe_album_metadata = get_meta(download_queue_item, "album_metadata") + safe_playlist_metadata = get_meta(download_queue_item, "playlist_metadata") + safe_show_metadata = get_meta(download_queue_item, "show_metadata") + + if not safe_playlist_metadata and global_playlist_metadata: + safe_playlist_metadata = global_playlist_metadata + if media_type == "track": if disable_wvd: - logger.warning( - "Widevine decryption is disabled, " - "skipping Widevine protected content" - ) + logger.warning("Widevine decryption is disabled, skipping.") continue if audio_quality in VORBIS_AUDIO_QUALITIES: - logger.warning( - "Vorbis audio quality is only supported for podcasts, " - "skipping" - ) + logger.warning("Vorbis audio quality is only supported for podcasts.") continue + if download_music_videos: downloader_music_video.download( music_video_id=media_id, - music_video_metadata=media_metadata, - album_metadata=download_queue_item.album_metadata, + music_video_metadata=media_metadata_for_download, + album_metadata=safe_album_metadata, gid_metadata=gid_metadata, - playlist_metadata=download_queue_item.playlist_metadata, + playlist_metadata=safe_playlist_metadata, playlist_track=index, ) else: downloader_song.download( track_id=media_id, - track_metadata=media_metadata, - album_metadata=download_queue_item.album_metadata, + track_metadata=media_metadata_for_download, + album_metadata=safe_album_metadata, gid_metadata=gid_metadata, - playlist_metadata=download_queue_item.playlist_metadata, + playlist_metadata=safe_playlist_metadata, + product_name=spotify_api.user_profile, playlist_track=index, ) + elif media_type == "episode": if download_podcast_videos: downloader_episode_video.download( episode_id=media_id, - episode_metadata=media_metadata, - show_metadata=download_queue_item.show_metadata, + episode_metadata=media_metadata_for_download, + show_metadata=safe_show_metadata, gid_metadata=gid_metadata, - playlist_metadata=download_queue_item.playlist_metadata, + playlist_metadata=safe_playlist_metadata, playlist_track=index, ) else: downloader_episode.download( episode_id=media_id, - episode_metadata=media_metadata, - show_metadata=download_queue_item.show_metadata, + episode_metadata=media_metadata_for_download, + show_metadata=safe_show_metadata, gid_metadata=gid_metadata, - playlist_metadata=download_queue_item.playlist_metadata, + playlist_metadata=safe_playlist_metadata, playlist_track=index, ) + #except ValueError: + # error_count += 1 + # continue + except Exception as e: error_count += 1 logger.error( - f'({queue_progress}) Failed to download "{media_metadata["name"]}"', + f'({queue_progress}) Failed to download "{track_name}"', exc_info=not no_exceptions, ) finally: @@ -660,4 +748,5 @@ def main( f"Waiting for {wait_interval} second(s) before continuing" ) time.sleep(wait_interval) + logger.info(f"Done ({error_count} error(s))") From 852363da6fd5736590aa0788bb346ab05b114912 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Sat, 24 Jan 2026 01:58:08 -0300 Subject: [PATCH 06/36] Fix --- votify/totp.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/votify/totp.py b/votify/totp.py index 9e6ea32..3b7d15f 100644 --- a/votify/totp.py +++ b/votify/totp.py @@ -5,10 +5,8 @@ import hmac import math import requests - -# thanks to https://github.com/glomatico/votify/pull/42#issuecomment-2727036757 class TOTP: - SPOTIFY_SECRETS_JSON = "https://raw.githubusercontent.com/Thereallo1026/spotify-secrets/refs/heads/main/secrets/secretDict.json" + SPOTIFY_SECRETS_JSON = "https://code.thetadev.de/ThetaDev/spotify-secrets/raw/branch/main/secrets/secretDict.json" PERIOD = 30 DIGITS = 6 From d92244fccf632662b67b36c537c61dacdf38ac07 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Sat, 24 Jan 2026 02:21:02 -0300 Subject: [PATCH 07/36] To install fix Updated installation instructions for votify package. --- README.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index a6c10d4..ace7911 100644 --- a/README.md +++ b/README.md @@ -44,7 +44,9 @@ The following tools are optional but required for specific features. Add them to 1. Install the package `votify` using pip: ```bash - pip install votify + pip uninstall votify votify-fix -y + pip cache purge + pip install git+https://github.com/GladistonXD/votify-fix.git ``` 2. Set up the cookies file. - Move the cookies file to the directory where you'll run Votify and rename it to `cookies.txt`. From cbffc492c32c5ea12d9e19ac5c000cdc445c1c5a Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Sun, 25 Jan 2026 19:22:01 -0300 Subject: [PATCH 08/36] Fix episode / show --- votify/cli.py | 50 ++++++++++++++++++++++++++++++-------------------- 1 file changed, 30 insertions(+), 20 deletions(-) diff --git a/votify/cli.py b/votify/cli.py index c559de5..1db25af 100644 --- a/votify/cli.py +++ b/votify/cli.py @@ -601,7 +601,6 @@ def main( f"Track {index}/{len(download_queue)} from URL {url_index}/{len(urls)}", colorama.Style.DIM, ) - media_metadata = None if isinstance(download_queue_item, dict): media_metadata = download_queue_item.get("media_metadata", download_queue_item) @@ -620,7 +619,10 @@ def main( try: track_name = media_metadata["data"]["trackUnion"]["name"] except (KeyError, TypeError): - track_name = clean_track_metadata.get("name", "Unknown Track") + try: + track_name = media_metadata["data"]["episodeUnionV2"]["name"] + except (KeyError, TypeError): + track_name = clean_track_metadata.get("name", "Unknown Track") is_playable = True @@ -659,18 +661,30 @@ def main( media_metadata_for_download = media_metadata media_type = "track" - try: - type_from_meta = media_metadata['data']['trackUnion']['__typename'] - media_type = type_from_meta.lower() - except (KeyError, TypeError, AttributeError): - if isinstance(media_metadata, dict) and 'episode' in media_metadata: - media_type = "episode" - media_metadata_for_download = None - else: - media_type = "track" - media_metadata_for_download = None - gid_metadata = downloader.get_gid_metadata(media_id, media_type, spotify_api.user_profile) + target_uri = "" + try: + target_uri = media_metadata['data']['trackUnion']['uri'] + except (KeyError, TypeError): + target_uri = clean_track_metadata.get('uri', '') + + if "episode" in target_uri: + media_type = "episode" + elif "track" in target_uri: + media_type = "track" + + else: + try: + type_from_meta = media_metadata['data']['trackUnion']['__typename'] + media_type = type_from_meta.lower() + except (KeyError, TypeError, AttributeError): + if isinstance(media_metadata, dict): + if media_metadata.get('type') == 'episode' or 'episode' in media_metadata: + media_type = "episode" + if 'episode' in media_metadata: + media_metadata_for_download = None + + gid_metadata = downloader.get_gid_metadata(media_id, media_type, spotify_api.user_profile, track_name) def get_meta(item, key): val = getattr(item, key, None) @@ -730,18 +744,14 @@ def main( show_metadata=safe_show_metadata, gid_metadata=gid_metadata, playlist_metadata=safe_playlist_metadata, + product_name=spotify_api.user_profile, playlist_track=index, ) - #except ValueError: - # error_count += 1 - # continue except Exception as e: error_count += 1 - logger.error( - f'({queue_progress}) Failed to download "{track_name}"', - exc_info=not no_exceptions, - ) + logger.error(f'({queue_progress}) Failed to download "{track_name}"', exc_info=False) + #logger.error(f'({queue_progress}) Failed to download "{track_name}"', exc_info=not no_exceptions) finally: if wait_interval > 0 and index != len(download_queue): logger.debug( From 6c4ac9fbbfe8e5bb0638b05c483520e777e0b6e6 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Sun, 25 Jan 2026 19:22:43 -0300 Subject: [PATCH 09/36] Fix episode / show --- votify/downloader_episode.py | 263 +++++++++++++++++++++-------------- 1 file changed, 162 insertions(+), 101 deletions(-) diff --git a/votify/downloader_episode.py b/votify/downloader_episode.py index 0e0e759..eef6c37 100644 --- a/votify/downloader_episode.py +++ b/votify/downloader_episode.py @@ -9,7 +9,6 @@ from .models import StreamInfoAudio logger = logging.getLogger("votify") - class DownloaderEpisode(DownloaderAudio): def __init__( self, @@ -18,33 +17,55 @@ class DownloaderEpisode(DownloaderAudio): self.__dict__.update(downloader_audio.__dict__) def get_tags( - self, - episode_metadata: dict, - show_metadata: dict, + self, + episode_metadata: dict, + show_metadata: dict, ) -> dict: - release_date_datetime_obj = self.downloader.get_release_date_datetime_obj( - episode_metadata["release_date"], - episode_metadata["release_date_precision"], - ) + try: + release_date = episode_metadata.get("release_date") + if release_date and "T" in release_date: + release_date = release_date.split("T")[0] + + release_date_datetime_obj = self.downloader.get_release_date_datetime_obj( + release_date, + episode_metadata.get("release_date_precision", "day"), + ) + tag_release_date = self.downloader.get_release_date_tag(release_date_datetime_obj) + tag_release_year = str(release_date_datetime_obj.year) + except Exception: + tag_release_date = str(episode_metadata.get("release_date")) + tag_release_year = tag_release_date.split("-")[0] if "-" in tag_release_date else "2024" + + track_number = 1 + try: + if 'trackNumber' in episode_metadata and isinstance(episode_metadata['trackNumber'], int): + track_number = episode_metadata['trackNumber'] + + elif show_metadata and "episodes" in show_metadata and "items" in show_metadata["episodes"]: + items = show_metadata["episodes"]["items"] + total_items = len(items) + + for index, item in enumerate(items): + item_id = item.get('id') + if not item_id and 'entity' in item: + item_id = item['entity'].get('data', {}).get('id') + + if item_id == episode_metadata['id']: + track_number = total_items - index + break + except (KeyError, TypeError, IndexError): + track_number = 1 + tags = { - "album": show_metadata["name"], - "description": episode_metadata["description"], + "album": show_metadata.get("name"), + "description": episode_metadata.get("description"), "media_type": "Podcast", "publisher": show_metadata.get("publisher"), "rating": "Explicit" if episode_metadata.get("explicit") else "Unknown", - "release_date": self.downloader.get_release_date_tag( - release_date_datetime_obj - ), - "release_year": str(release_date_datetime_obj.year), - "title": episode_metadata["name"], - "track": next( - index - for index in range(1, len(show_metadata["episodes"]["items"]) + 1) - if show_metadata["episodes"]["items"][ - len(show_metadata["episodes"]["items"]) - index - ]["id"] - == episode_metadata["id"] - ), + "release_date": tag_release_date, + "release_year": tag_release_year, + "title": episode_metadata.get("name"), + "track": track_number, "url": f"https://open.spotify.com/episode/{episode_metadata['id']}", } return tags @@ -63,102 +84,142 @@ class DownloaderEpisode(DownloaderAudio): self.downloader.cleanup_temp_path() def _download( - self, - episode_id: str, - episode_metadata: dict = None, - show_metadata: dict = None, - gid_metadata: dict = None, - stream_info: StreamInfoAudio = None, - playlist_metadata: dict = None, - playlist_track: int = None, - decryption_key: bytes = None, + self, + episode_id: str, + episode_metadata: dict = None, + show_metadata: dict = None, + gid_metadata: dict = None, + stream_info: StreamInfoAudio = None, + playlist_metadata: dict = None, + playlist_track: int = None, + product_name: dict = None, + decryption_key: bytes = None, ): if not episode_metadata: - logger.debug("Getting episode metadata") episode_metadata = self.downloader.spotify_api.get_episode(episode_id) + + if 'data' in episode_metadata and 'trackUnion' in episode_metadata['data']: + episode_metadata = episode_metadata['data']['trackUnion'] + + if 'release_date' in episode_metadata: + + raw_date = str(episode_metadata['release_date']) + if 'T' in raw_date: + episode_metadata['release_date'] = raw_date.split('T')[0] + + precision = str(episode_metadata.get('release_date_precision', '')) + if 'T' in precision or len(precision) > 10: + episode_metadata['release_date_precision'] = 'day' + + else: + + if 'releaseDate' in episode_metadata and isinstance(episode_metadata['releaseDate'], dict): + r_node = episode_metadata['releaseDate'] + episode_metadata['release_date'] = r_node.get('isoString', '1970-01-01').split('T')[0] + episode_metadata['release_date_precision'] = r_node.get('precision', 'day') + else: + episode_metadata['release_date'] = '1970-01-01' + episode_metadata['release_date_precision'] = 'year' + + if not show_metadata or show_metadata.get('name') == 'Unknown Show': + logger.debug("Attempting to extract Show Metadata from alternative fields...") + found_show_id = None + + try: + if 'podcastV2' in episode_metadata: + uri = episode_metadata['podcastV2']['data']['uri'] + found_show_id = uri.split(':')[-1] + except (KeyError, IndexError, AttributeError): + pass + + if not found_show_id and 'show' in episode_metadata: + found_show_id = episode_metadata['show'].get('id') + + if found_show_id: + logger.debug(f"Found Show ID: {found_show_id}, fetching details...") + try: + show_metadata = self.downloader.spotify_api.get_show(found_show_id) + except Exception as e: + logger.warning(f"Failed to fetch show details: {e}") + if not show_metadata: - logger.debug("Getting show metadata") - show_metadata = self.downloader.spotify_api.get_show( - episode_metadata["show"]["id"] - ) + show_name = "Unknown Show" + try: + show_name = episode_metadata['data']['trackUnion']['artists']['items'][0]['profile']['name'] + except: + pass + show_metadata = {'name': show_name, 'publisher': 'Unknown', 'episodes': {'items': []}} + if not gid_metadata: logger.debug("Getting GID metadata") - gid_metadata = self.downloader.get_gid_metadata(episode_id, "episode") + try: + gid_metadata = self.downloader.get_gid_metadata(episode_id, "episode", None) + except TypeError: + gid_metadata = self.downloader.get_gid_metadata(episode_id, "episode") + if not stream_info: logger.debug("Getting stream info") - stream_info = self.get_stream_info(gid_metadata, "episode") + stream_info = self.get_stream_info(gid_metadata, product_name, "episode") + if not stream_info.file_id: - logger.warning( - "Episode is not available on Spotify's " - "servers and no alternative found, skipping" - ) + logger.warning("Episode not available, skipping") return - if stream_info.quality != self.audio_quality: - logger.warning(f"Quality has been changed to {stream_info.quality.value}") - tags = self.get_tags( - episode_metadata, - show_metadata, - ) + + tags = self.get_tags(episode_metadata, show_metadata) + if playlist_metadata: - tags = { - **tags, - **self.downloader.get_playlist_tags( - playlist_metadata, - playlist_track, - ), - } + tags = {**tags, **self.downloader.get_playlist_tags(playlist_metadata, playlist_track)} + file_extension = self.get_file_extension() - final_path = self.downloader.get_final_path( - "episode", - tags, - file_extension, - ) + final_path = self.downloader.get_final_path("episode", tags, file_extension) + + cover_url = None + + if 'cover_url' in episode_metadata: + cover_url = episode_metadata['cover_url'] + + if not cover_url and 'coverArt' in episode_metadata and 'sources' in episode_metadata['coverArt']: + try: + cover_url = episode_metadata['coverArt']['sources'][0]['url'] + except (IndexError, KeyError, TypeError): + pass + + if not cover_url and 'albumOfTrack' in episode_metadata: + alb = episode_metadata['albumOfTrack'] + if 'coverArt' in alb and 'sources' in alb['coverArt']: + try: + cover_url = alb['coverArt']['sources'][0]['url'] + except (IndexError, KeyError, TypeError): + pass + + elif 'images' in alb and isinstance(alb['images'], list) and len(alb['images']) > 0: + cover_url = alb['images'][0]['url'] + + + if not cover_url: + key_mapping = COVER_SIZE_X_KEY_MAPPING_EPISODE if 'COVER_SIZE_X_KEY_MAPPING_EPISODE' in globals() else COVER_SIZE_X_KEY_MAPPING_SONG + cover_url = self.downloader.get_cover_url(episode_metadata, key_mapping) + cover_path = self.get_cover_path(final_path) - cover_url = self.downloader.get_cover_url( - episode_metadata, - COVER_SIZE_X_KEY_MAPPING_EPISODE, - ) + decrypted_path = None remuxed_path = None + if final_path.exists() and not self.downloader.overwrite: - logger.warning(f'Track already exists at "{final_path}", skipping') + logger.warning(f'Episode already exists at "{final_path}", skipping') else: - decryption_key = ( - self.DEFAULT_EPISODE_DECRYPTION_KEY.hex() - if self.audio_quality in AAC_AUDIO_QUALITIES - else self.DEFAULT_EPISODE_DECRYPTION_KEY - ) - encrypted_path = self.downloader.get_file_temp_path( - episode_id, - "_encrypted", - file_extension, - ) - decrypted_path = self.downloader.get_file_temp_path( - episode_id, - "_decrypted", - file_extension, - ) - remuxed_path = self.downloader.get_file_temp_path( - episode_id, - "_remuxed", - file_extension, - ) - logger.debug(f'Downloading to "{encrypted_path}"') + if not decryption_key: + decryption_key = self.get_decryption_key(stream_info) + + encrypted_path = self.downloader.get_file_temp_path(episode_id, "_encrypted", file_extension) + decrypted_path = self.downloader.get_file_temp_path(episode_id, "_decrypted", file_extension) + remuxed_path = self.downloader.get_file_temp_path(episode_id, "_remuxed", file_extension) + self.download_stream_url(encrypted_path, stream_info.stream_url) - logger.debug( - f'Decrypting to "{decrypted_path}" and remuxing to "{remuxed_path}"' - ) - self.decrypt( - decryption_key, - encrypted_path, - decrypted_path, - remuxed_path, - ) - media_temp_path = ( - remuxed_path - if remuxed_path is not None and remuxed_path.exists() - else decrypted_path - ) + self.decrypt(decryption_key, encrypted_path, decrypted_path, remuxed_path) + + media_temp_path = remuxed_path if remuxed_path and remuxed_path.exists() else decrypted_path + self.downloader._final_processing( cover_path, cover_url, From a5ac9236d445f61308a52a226a7f2c2bbe7e85a7 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Sun, 25 Jan 2026 19:23:58 -0300 Subject: [PATCH 10/36] Fix episode / show --- votify/downloader_audio.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/votify/downloader_audio.py b/votify/downloader_audio.py index cb7b51a..efce78f 100644 --- a/votify/downloader_audio.py +++ b/votify/downloader_audio.py @@ -67,7 +67,7 @@ class DownloaderAudio: if media_type == "track": audio_files = gid_metadata elif media_type == "episode": - audio_files = gid_metadata.get("audio") + audio_files = gid_metadata else: raise RuntimeError() audio_files = audio_files or gid_metadata.get("alternative") From 45ce4b5d9b9f105875df55040e4d5aa95a30f2e2 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Sun, 25 Jan 2026 19:25:25 -0300 Subject: [PATCH 11/36] Fix episode / show --- votify/downloader.py | 187 +++++++++++++++++++++++++++++++++++++++---- 1 file changed, 172 insertions(+), 15 deletions(-) diff --git a/votify/downloader.py b/votify/downloader.py index 22385d0..dd5aca6 100644 --- a/votify/downloader.py +++ b/votify/downloader.py @@ -265,18 +265,159 @@ class Downloader: ) ) elif media_type == "episode": - download_queue.append( - DownloadQueueItem( - media_metadata=self.spotify_api.get_episode(media_id), - ) - ) + episode = self.spotify_api.get_episode(media_id) + + podcast_name = "Unknown Podcast" + podcast_uri = "" + podcast_cover = "" + + if 'podcastV2' in episode and 'data' in episode['podcastV2']: + p_data = episode['podcastV2']['data'] + podcast_name = p_data.get('name', 'Unknown Podcast') + podcast_uri = p_data.get('uri', '') + if 'coverArt' in p_data and 'sources' in p_data['coverArt']: + podcast_cover = p_data['coverArt']['sources'][0]['url'] + + release_date = "" + precision = "" + if 'releaseDate' in episode and episode['releaseDate']: + release_date = episode['releaseDate'].get('isoString') + precision = episode['releaseDate'].get('isoString') + + cover_url = podcast_cover + try: + if 'coverArt' in episode and 'sources' in episode['coverArt']: + sources = episode['coverArt']['sources'] + cover_url = max(sources, key=lambda x: x.get('width', 0))['url'] + except Exception: + pass + + download_queue.append({ + 'data': { + 'trackUnion': { + 'id': episode['id'], + 'uri': episode['uri'], + 'name': episode['name'], + 'description': episode.get('description', ''), + 'trackNumber': 1, + 'duration': { + 'totalMilliseconds': episode['duration']['totalMilliseconds'] + }, + 'is_playable': episode.get('playability', {}).get('playable', True), + + 'release_date': release_date, + 'release_date_precision': precision, + + 'cover_url': cover_url, + + 'artists': { + 'items': [ + { + 'profile': { + 'name': podcast_name + } + } + ] + }, + + 'albumOfTrack': { + 'uri': podcast_uri, + 'name': podcast_name, + 'coverArt': { + 'sources': [{'url': cover_url}] + }, + 'id': podcast_uri.split(':')[-1] if ':' in podcast_uri else None, + 'release_date': release_date + } + } + }, + 'show_metadata': { + 'name': podcast_name, + 'publisher': 'Unknown', + 'uri': podcast_uri + } + }) + + elif media_type == "show": show = self.spotify_api.get_show(media_id) - for episode in show["episodes"]["items"]: + items_list = [] + if 'episodesV2' in show and 'items' in show['episodesV2']: + items_list = show['episodesV2']['items'] + elif 'episodes' in show and 'items' in show['episodes']: + items_list = show['episodes']['items'] + total_items = len(items_list) + for index, item in enumerate(items_list): + episode_raw = None + if 'entity' in item and 'data' in item['entity']: + episode_raw = item['entity']['data'] + elif 'data' in item: + episode_raw = item['data'] + else: + episode_raw = item + if not episode_raw or not isinstance(episode_raw, dict): + continue + podcast_name = "Unknown Podcast" + podcast_uri = "" + if 'podcastV2' in episode_raw and 'data' in episode_raw['podcastV2']: + p_data = episode_raw['podcastV2']['data'] + podcast_name = p_data.get('name', 'Unknown Podcast') + podcast_uri = p_data.get('uri', '') + elif show.get('name'): + podcast_name = show.get('name') + podcast_uri = show.get('uri') + cover_url = None + if 'coverArt' in episode_raw and episode_raw['coverArt'].get('sources'): + try: + sources = episode_raw['coverArt']['sources'] + cover_url = max(sources, key=lambda x: x.get('width', 0))['url'] + except Exception: + pass + release_date = "" + precision = "day" + if 'releaseDate' in episode_raw and isinstance(episode_raw['releaseDate'], dict): + release_date = episode_raw['releaseDate'].get('isoString') + if 'T' in release_date: + release_date = release_date.split('T')[0] + precision = episode_raw['releaseDate'].get('precision', 'day') + elif 'release_date' in episode_raw: + release_date = episode_raw['release_date'] + current_track_number = total_items - index + formatted_episode = { + 'data': { + 'trackUnion': { + 'id': episode_raw.get('id'), + 'uri': episode_raw.get('uri'), + 'name': episode_raw.get('name'), + 'description': episode_raw.get('description', ''), + 'trackNumber': current_track_number, + 'duration': { + 'totalMilliseconds': episode_raw.get('duration', {}).get('totalMilliseconds', 0) + }, + 'is_playable': episode_raw.get('playability', {}).get('playable', True), + 'release_date': release_date, + 'release_date_precision': precision, + 'cover_url': cover_url, + 'artists': {'items': [{'profile': {'name': podcast_name}}]}, + 'albumOfTrack': { + 'uri': podcast_uri, + 'name': podcast_name, + 'coverArt': {'sources': [{'url': cover_url}]}, + 'id': podcast_uri.split(':')[-1] if ':' in podcast_uri else None, + 'release_date': release_date + } + } + } + } + current_show_metadata = { + 'name': podcast_name, + 'publisher': 'Unknown', + 'uri': podcast_uri + } download_queue.append( DownloadQueueItem( - show_metadata=show, - media_metadata=episode, + show_metadata=current_show_metadata, + media_metadata=formatted_episode, ) ) return download_queue @@ -532,21 +673,37 @@ class Downloader: media_id: str, media_type: str, audio_quality: dict, + track_name: str, ) -> str: playback_info = self.spotify_api.get_track_playback_info(media_id, media_type) user_product = audio_quality.get('data', {}).get('me', {}).get('account', {}).get('product', 'free') - if user_product == 'PREMIUM': - target_bitrate = 256000 - else: - target_bitrate = 128000 + target_bitrate = 256000 if user_product == 'PREMIUM' else 128000 + + media_items = playback_info.get("media", {}) + if not media_items: + raise ValueError("No available Track (Empty response)") expected_key = f"spotify:{media_type}:{media_id}" - if expected_key not in playback_info.get("media", {}): - raise ValueError(f"No avaliable Track") + track_manifest = media_items.get(expected_key) - track_manifest = playback_info["media"][expected_key] + if not track_manifest: + track_manifest = list(media_items.values())[0] + + metadata = track_manifest.get("item", {}).get("metadata", {}) + server_name = metadata.get("name") + + if not server_name: + raise ValueError(f"Track Unavailable: No name returned") + + req_clean = track_name.lower().strip() + srv_clean = server_name.lower().strip() + + if (req_clean in srv_clean) or (srv_clean in req_clean): + track_manifest = track_manifest + else: + raise ValueError(f"Track Unavailable: No name returned") try: audio_file_id = next( From ed09c433dafbf8b7080c69b6cc39e05c006d8e75 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Sun, 25 Jan 2026 19:26:01 -0300 Subject: [PATCH 12/36] Fix episode / show --- votify/spotify_api.py | 49 ++++++++++++++++++++++++++++++++++--------- 1 file changed, 39 insertions(+), 10 deletions(-) diff --git a/votify/spotify_api.py b/votify/spotify_api.py index 3323e5f..a0976c2 100644 --- a/votify/spotify_api.py +++ b/votify/spotify_api.py @@ -223,9 +223,8 @@ class SpotifyApi: media_type: str ) -> dict | None: self._refresh_session_auth() - params = {"manifestFileFormat": ["file_ids_mp4"]} - response = self.session.get(f'https://gue1-spclient.spotify.com/track-playback/v1/media/spotify:track:{media_id}?manifestFileFormat=file_ids_mp4') + response = self.session.get(f'https://gue1-spclient.spotify.com/track-playback/v1/media/spotify:{media_type}:{media_id}?manifestFileFormat=file_ids_mp4') return response.json() def get_lyrics(self, track_id: str) -> dict | None: @@ -303,6 +302,7 @@ class SpotifyApi: playlist_id: str, extend: bool = True, ) -> dict: + self._refresh_session_auth() payload = { "variables": { "uri": f"spotify:playlist:{playlist_id}", @@ -336,20 +336,49 @@ class SpotifyApi: def get_episode(self, episode_id: str) -> dict: self._refresh_session_auth() - response = self.session.get( - self.METADATA_API_URL.format(type="episodes", item_id=episode_id) + 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() + return response.json()['data']['episodeUnionV2'] def get_show(self, show_id: str, extend: bool = True) -> dict: self._refresh_session_auth() - response = self.session.get( - self.METADATA_API_URL.format(type="shows", item_id=show_id) + 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: + '''if extend: show["episodes"]["items"].extend( [ item @@ -358,8 +387,8 @@ class SpotifyApi: ) for item in extended_collection["items"] ] - ) - return show + )''' + return show["data"]['podcastUnionV2'] def get_artist_albums( self, From fbe044473883d2e1b2010cad5afce9d43c5b7711 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Wed, 28 Jan 2026 10:08:38 -0300 Subject: [PATCH 13/36] Implement wvd_cdrm method for DRM decryption Add wvd_cdrm method to handle DRM decryption requests. --- votify/spotify_api.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/votify/spotify_api.py b/votify/spotify_api.py index a0976c2..dd812e4 100644 --- a/votify/spotify_api.py +++ b/votify/spotify_api.py @@ -235,6 +235,16 @@ class SpotifyApi: check_response(response) return response.json() + def wvd_cdrm(self, pssh): + cmd = self.session.post('https://cdrm-project.com/api/decrypt', + headers={'Content-Type': 'application/json'}, + json={ + 'pssh': pssh, + 'licurl': 'https://gue1-spclient.spotify.com/widevine-license/v1/audio/license', + 'headers': str(self.session.headers) + }).json() + return cmd['message'] + def get_track(self, track_id: str) -> dict: self._refresh_session_auth() payload = { From 57aa8d20b6371719724d1e7cfe4a9e3b66ca88e4 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Wed, 28 Jan 2026 10:09:30 -0300 Subject: [PATCH 14/36] Refactor Widevine path handling --- votify/downloader.py | 44 ++++++++++++++++++++++++++++++-------------- 1 file changed, 30 insertions(+), 14 deletions(-) diff --git a/votify/downloader.py b/votify/downloader.py index dd5aca6..4a0ed10 100644 --- a/votify/downloader.py +++ b/votify/downloader.py @@ -9,7 +9,7 @@ import shutil import subprocess from io import BytesIO from pathlib import Path - +from .utils import prompt_path import requests from InquirerPy import inquirer from InquirerPy.base.control import Choice @@ -75,7 +75,12 @@ class Downloader: self.spotify_api = spotify_api self.output_path = output_path self.temp_path = temp_path - self.wvd_path = wvd_path + self.wvd_path = prompt_path( + is_file=True, + initial_path=wvd_path, + description="device.wvd", + optional=True + ) self.aria2c_path = aria2c_path self.ffmpeg_path = ffmpeg_path self.mp4box_path = mp4box_path @@ -133,7 +138,10 @@ class Downloader: self.subprocess_additional_args = {} def set_cdm(self) -> None: - self.cdm = Cdm.from_device(Device.load(self.wvd_path)) + if self.wvd_path.exists(): + self.cdm = Cdm.from_device(Device.load(self.wvd_path)) + else: + self.cmd = None def get_url_info(self, url: str) -> UrlInfo: url_regex_result = re.search(self.URL_RE, url) @@ -727,18 +735,26 @@ class Downloader: media_type: str, ) -> tuple[str, str]: try: - pssh = PSSH(pssh) - cdm_session = self.cdm.open() - challenge = self.cdm.get_license_challenge(cdm_session, pssh) - license = self.spotify_api.get_widevine_license(challenge, media_type) - self.cdm.parse_license(cdm_session, license) - keys = next( - i for i in self.cdm.get_keys(cdm_session) if i.type == "CONTENT" - ) - decryption_key = keys.key.hex() - key_id = keys.kid.hex + if self.cdm == None: + cmd = self.spotify_api.wvd_cdrm(pssh) + key_id, decryption_key = cmd.split(':') + else: + pssh = PSSH(pssh) + cdm_session = self.cdm.open() + challenge = self.cdm.get_license_challenge(cdm_session, pssh) + license = self.spotify_api.get_widevine_license(challenge, media_type) + self.cdm.parse_license(cdm_session, license) + keys = next( + i for i in self.cdm.get_keys(cdm_session) if i.type == "CONTENT" + ) + decryption_key = keys.key.hex() + key_id = keys.kid.hex + finally: - self.cdm.close(cdm_session) + try: + self.cdm.close(cdm_session) + except: + pass return key_id, decryption_key def get_sanitized_string(self, dirty_string: str, is_folder: bool) -> str: From 62293b15c28c65da02db027013c0eb73cf49fb34 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Wed, 28 Jan 2026 10:10:24 -0300 Subject: [PATCH 15/36] Add optional parameter to prompt_path function Modify prompt_path function to add optional parameter for path existence check. --- votify/utils.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/votify/utils.py b/votify/utils.py index 20474e8..81b306e 100644 --- a/votify/utils.py +++ b/votify/utils.py @@ -22,9 +22,9 @@ def _raise_response_exception(response: requests.Response): ) -def prompt_path(is_file: bool, initial_path: Path, description: str) -> Path: +def prompt_path(is_file: bool, initial_path: Path, description: str, optional: bool = False) -> Path: path_validator = click.Path( - exists=True, + exists=not optional, file_okay=is_file, dir_okay=not is_file, path_type=Path, From 6ac93d39aaa92affe6d11a791294a7c5b79d5735 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Wed, 28 Jan 2026 10:11:02 -0300 Subject: [PATCH 16/36] Update wvd_path prompt to be optional Make the .wvd file prompt optional. --- votify/cli.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/votify/cli.py b/votify/cli.py index 1db25af..848592e 100644 --- a/votify/cli.py +++ b/votify/cli.py @@ -557,7 +557,7 @@ def main( ) if not disable_wvd: - wvd_path = prompt_path(True, wvd_path, ".wvd file") + wvd_path = prompt_path(True, wvd_path, ".wvd file", optional=True) downloader.set_cdm() if read_urls_as_txt: From c0939e3063a165484ab21440abf4034cd397e1c0 Mon Sep 17 00:00:00 2001 From: Cynthia Date: Thu, 29 Jan 2026 17:21:13 +0530 Subject: [PATCH 17/36] Make Spotify secrets URL modifiable --- votify/cli.py | 9 ++++++++- votify/spotify_api.py | 16 +++++++++++----- votify/totp.py | 37 ++++++++++++------------------------- 3 files changed, 31 insertions(+), 31 deletions(-) diff --git a/votify/cli.py b/votify/cli.py index 848592e..c42489a 100644 --- a/votify/cli.py +++ b/votify/cli.py @@ -227,6 +227,12 @@ def load_config_file( default=downloader_sig.parameters["packager_path"].default, help="Path to Shaka Packager binary.", ) +@click.option( + "--spotify-secrets-url", + type=str, + default="https://code.thetadev.de/ThetaDev/spotify-secrets/raw/branch/main/secrets/secretDict.json", + help="Spotify secrets for TOTP generation" +) @click.option( "--template-folder-album", type=str, @@ -395,6 +401,7 @@ def main( mp4box_path: str, mp4decrypt_path: str, packager_path: str, + spotify_secrets_url: str, template_folder_album: str, template_folder_compilation: str, template_file_single_disc: str, @@ -428,7 +435,7 @@ def main( cookies_path = prompt_path(True, cookies_path, "Cookies file") logger.info("Starting Votify") - spotify_api = SpotifyApi.from_cookies_file(cookies_path) + spotify_api = SpotifyApi.from_cookies_file(cookies_path, secrets_url=spotify_secrets_url) downloader = Downloader( spotify_api, diff --git a/votify/spotify_api.py b/votify/spotify_api.py index dd812e4..3d92433 100644 --- a/votify/spotify_api.py +++ b/votify/spotify_api.py @@ -47,16 +47,24 @@ class SpotifyApi: CLIENT_TOKEN_URL = "https://clienttoken.spotify.com/v1/clienttoken" def __init__( - self, + 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) -> SpotifyApi: + 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( @@ -73,11 +81,9 @@ class SpotifyApi: '"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) + return cls(sp_dc=sp_dc, **kwargs) def _set_session(self) -> None: - self.totp = TOTP() - self.session = requests.Session() self._setup_session_headers() self._setup_authorization() self._setup_user_profile() diff --git a/votify/totp.py b/votify/totp.py index 3b7d15f..5c070b8 100644 --- a/votify/totp.py +++ b/votify/totp.py @@ -1,39 +1,26 @@ -from __future__ import annotations - import hashlib import hmac -import math -import requests + +from typing import Collection + class TOTP: - SPOTIFY_SECRETS_JSON = "https://code.thetadev.de/ThetaDev/spotify-secrets/raw/branch/main/secrets/secretDict.json" PERIOD = 30 DIGITS = 6 - def __init__(self) -> None: - self._setup() - - def _setup(self) -> None: - version, secret_cipher_bytes = self.get_latest_secret() + def __init__(self, *, version: int, ciphertext: Collection[int]) -> None: self.version = version - self.secret = self.derive_secret_number(secret_cipher_bytes).encode() + self.secret = self.derive(ciphertext) - def derive_secret_number(self, secret_cipher_bytes: list[int]) -> str: - transformed = [ - byte ^ ((i % 33) + 9) for i, byte in enumerate(secret_cipher_bytes) - ] - return "".join(str(n) for n in transformed) - - def get_latest_secret(self) -> tuple[int, list[int]]: - response = requests.get(self.SPOTIFY_SECRETS_JSON) - response.raise_for_status() - secrets = response.json() - latest_version = max(int(v) for v in secrets.keys()) - return latest_version, secrets[str(latest_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 = math.floor(timestamp / 1000 / self.PERIOD) - counter_bytes = counter.to_bytes(8, byteorder="big") + counter = int(timestamp) // 1000 // self.PERIOD + counter_bytes = counter.to_bytes(8) h = hmac.new(self.secret, counter_bytes, hashlib.sha1) hmac_result = h.digest() From 75c26acd74c76d1a0d59a549ccb9b32f45773211 Mon Sep 17 00:00:00 2001 From: Cynthia Date: Thu, 29 Jan 2026 17:32:33 +0530 Subject: [PATCH 18/36] Fix CDRM integration to with other mediatypes --- votify/downloader.py | 6 ++---- votify/spotify_api.py | 15 ++++++++------- 2 files changed, 10 insertions(+), 11 deletions(-) diff --git a/votify/downloader.py b/votify/downloader.py index 4a0ed10..63574dd 100644 --- a/votify/downloader.py +++ b/votify/downloader.py @@ -140,8 +140,6 @@ class Downloader: def set_cdm(self) -> None: if self.wvd_path.exists(): self.cdm = Cdm.from_device(Device.load(self.wvd_path)) - else: - self.cmd = None def get_url_info(self, url: str) -> UrlInfo: url_regex_result = re.search(self.URL_RE, url) @@ -735,8 +733,8 @@ class Downloader: media_type: str, ) -> tuple[str, str]: try: - if self.cdm == None: - cmd = self.spotify_api.wvd_cdrm(pssh) + if self.cdm is None: + cmd = self.spotify_api.extract_keys_with_cdrm(pssh, media_type) key_id, decryption_key = cmd.split(':') else: pssh = PSSH(pssh) diff --git a/votify/spotify_api.py b/votify/spotify_api.py index 3d92433..4142b52 100644 --- a/votify/spotify_api.py +++ b/votify/spotify_api.py @@ -241,14 +241,15 @@ class SpotifyApi: check_response(response) return response.json() - def wvd_cdrm(self, pssh): + def extract_keys_with_cdrm(self, pssh, media_type): cmd = self.session.post('https://cdrm-project.com/api/decrypt', - headers={'Content-Type': 'application/json'}, - json={ - 'pssh': pssh, - 'licurl': 'https://gue1-spclient.spotify.com/widevine-license/v1/audio/license', - 'headers': str(self.session.headers) - }).json() + headers={'Accept': 'application/json', + 'Content-Type': 'application/json'}, + json={ + 'pssh': pssh, + 'licurl': self.WIDEVINE_LICENSE_API_URL.format(type=media_type), + 'headers': json.dumps(self.session.headers) + }).json() return cmd['message'] def get_track(self, track_id: str) -> dict: From 96e40b9ddf050ab88aea8bd8c8d41fa319e4a724 Mon Sep 17 00:00:00 2001 From: Cynthia Date: Thu, 29 Jan 2026 17:42:10 +0530 Subject: [PATCH 19/36] Mention that WVD files are optional --- README.md | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index ace7911..0d70c55 100644 --- a/README.md +++ b/README.md @@ -22,13 +22,13 @@ A command-line app for downloading songs, podcasts and videos from Spotify. - **FFmpeg** on your system PATH. Use one of the recommended builds: - **Windows**: [AnimMouse's FFmpeg Builds](https://github.com/AnimMouse/ffmpeg-stable-autobuild/releases). - **Linux**: [John Van Sickle's FFmpeg Builds](https://johnvansickle.com/ffmpeg/). -- A **.wvd file**. - - A .wvd file contains the Widevine keys from a device and is required to decrypt music videos and songs in AAC. The easiest method of obtaining one is using KeyDive, which extracts it from an Android device. Detailed instructions can be found here: https://github.com/hyugogirubato/KeyDive. **.wvd files extracted from emulated devices may not work**. +- **(Optional)** A **.wvd file**. + - A `.wvd` file contains the Widevine keys from a device and is required to decrypt music videos and songs in AAC. The easiest method of obtaining one is using KeyDive, which extracts it from an Android device. Detailed instructions can be found here: https://github.com/hyugogirubato/KeyDive. **.wvd files extracted from emulated devices may not work**. #### Notes - **Some users have reported that Spotify suspended their accounts after using Votify**. Use it at your own risk. -- The .wvd file is not required if you plan on only downloading podcasts and can be skipped by enabling the `disable_wvd` option. +- The .wvd file is not required, but expect latency otherwise. - FFmpeg is not required if you plan on only downloading podcasts in Vorbis, but it's needed for downloading podcasts in AAC. ### Optional dependencies @@ -51,7 +51,8 @@ The following tools are optional but required for specific features. Add them to 2. Set up the cookies file. - Move the cookies file to the directory where you'll run Votify and rename it to `cookies.txt`. - Alternatively, specify the path to the cookies file using command-line arguments or the config file. -3. Set up the .wvd file. +3. **(Optional**) Set up the .wvd file. + - Move the .wvd file file to the directory where you'll run Votify and rename it to `device.wvd`. - Alternatively, specify the path to the .wvd file using command-line arguments or the config file. From 9c7a76a4779a9654639c5f500654cc4725c1142c Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Thu, 29 Jan 2026 09:14:56 -0300 Subject: [PATCH 20/36] Fix assignment of cdm to None in set_cdm method --- votify/downloader.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/votify/downloader.py b/votify/downloader.py index 4a0ed10..e39f115 100644 --- a/votify/downloader.py +++ b/votify/downloader.py @@ -141,7 +141,7 @@ class Downloader: if self.wvd_path.exists(): self.cdm = Cdm.from_device(Device.load(self.wvd_path)) else: - self.cmd = None + self.cdm = None def get_url_info(self, url: str) -> UrlInfo: url_regex_result = re.search(self.URL_RE, url) From e903e5e8f322928e697d3818b14490f859e8c04b Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Thu, 29 Jan 2026 09:16:43 -0300 Subject: [PATCH 21/36] Log error and warning for invalid device.wvd file --- votify/spotify_api.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/votify/spotify_api.py b/votify/spotify_api.py index dd812e4..db60a6f 100644 --- a/votify/spotify_api.py +++ b/votify/spotify_api.py @@ -470,6 +470,9 @@ class SpotifyApi: 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 From 349d69b63801af7f402bd307c2b0faaf2a246650 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Thu, 29 Jan 2026 09:17:17 -0300 Subject: [PATCH 22/36] Log audio quality change with bit rate info Log the current bit rate when audio quality changes. --- votify/downloader_song.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/votify/downloader_song.py b/votify/downloader_song.py index d258748..918a6c8 100644 --- a/votify/downloader_song.py +++ b/votify/downloader_song.py @@ -260,7 +260,11 @@ class DownloaderSong(DownloaderAudio): return if stream_info.quality != self.audio_quality: - logger.warning(f"Quality has been changed to {stream_info.quality}") + if stream_info.quality == 128000: + info = '128 kb/s' + else: + info = '256 kb/s' + logger.info(f"The current bit rate is {info}") if gid_metadata: logger.debug("Getting lyrics") From af9f8d6c3910a42d605c71e9a83c832bba572531 Mon Sep 17 00:00:00 2001 From: Cynthia Date: Thu, 29 Jan 2026 18:16:38 +0530 Subject: [PATCH 23/36] Fix a minor mistake in CDRM integration --- votify/spotify_api.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/votify/spotify_api.py b/votify/spotify_api.py index 4142b52..d7f8c48 100644 --- a/votify/spotify_api.py +++ b/votify/spotify_api.py @@ -248,7 +248,7 @@ class SpotifyApi: json={ 'pssh': pssh, 'licurl': self.WIDEVINE_LICENSE_API_URL.format(type=media_type), - 'headers': json.dumps(self.session.headers) + 'headers': str(self.session.headers) }).json() return cmd['message'] From b3dd8160d348391594ca9b922dd77d9d6bc1054d Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Fri, 30 Jan 2026 14:55:07 -0300 Subject: [PATCH 24/36] Update API URLs and parameters in spotify_api.py --- votify/spotify_api.py | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/votify/spotify_api.py b/votify/spotify_api.py index b982856..6590121 100644 --- a/votify/spotify_api.py +++ b/votify/spotify_api.py @@ -27,8 +27,8 @@ class SpotifyApi: 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/v7/json/sources/{gid}/options/supports_drm" - TRACK_PLAYBACK_API_URL = "https://gue1-spclient.spotify.com/track-playback/v1/media/spotify:{type}:{id}" + 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}" ) @@ -229,8 +229,19 @@ class SpotifyApi: media_type: str ) -> dict | None: self._refresh_session_auth() - params = {"manifestFileFormat": ["file_ids_mp4"]} - response = self.session.get(f'https://gue1-spclient.spotify.com/track-playback/v1/media/spotify:{media_type}:{media_id}?manifestFileFormat=file_ids_mp4') + 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: From d38a28b0e9169327843166dd57f4e656e0027998 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Fri, 30 Jan 2026 14:55:44 -0300 Subject: [PATCH 25/36] Implement config file check and fix functionality Add check_and_fix_config function to update config file format. --- votify/cli.py | 34 ++++++++++++++++++++++++++++++++-- 1 file changed, 32 insertions(+), 2 deletions(-) diff --git a/votify/cli.py b/votify/cli.py index c42489a..0d4fd50 100644 --- a/votify/cli.py +++ b/votify/cli.py @@ -79,6 +79,9 @@ class Csv(click.ParamType): return result + + + def load_config_file( ctx: click.Context, param: click.Parameter, @@ -110,6 +113,32 @@ def load_config_file( return ctx +def check_and_fix_config(): + config_path = Path.home() / ".votify" / "config.ini" + if config_path.exists(): + try: + content = config_path.read_text(encoding='utf-8') + if "template_folder_music_video" in content and "Unknown Album" in content: + lines = content.splitlines() + new_lines = [] + updated = False + + for line in lines: + if "template_folder_music_video" in line and "Unknown Album" in line: + new_lines.append("template_folder_music_video = {artist}/{album}") + updated = True + else: + new_lines.append(line) + if updated: + new_content = "\n".join(new_lines) + config_path.write_text(new_content, encoding='utf-8') + except Exception as e: + pass +try: + check_and_fix_config() +except: + pass + @click.command() @click.help_option("-h", "--help") @click.version_option(__version__, "-v", "--version") @@ -380,6 +409,8 @@ def load_config_file( callback=load_config_file, help="Do not use a config file.", ) + + def main( urls: list[str], wait_interval: float, @@ -691,7 +722,7 @@ def main( if 'episode' in media_metadata: media_metadata_for_download = None - gid_metadata = downloader.get_gid_metadata(media_id, media_type, spotify_api.user_profile, track_name) + gid_metadata = downloader.get_gid_metadata(media_id, media_type, spotify_api.user_profile, track_name, download_music_videos, download_podcast_videos) def get_meta(item, key): val = getattr(item, key, None) @@ -713,7 +744,6 @@ def main( if audio_quality in VORBIS_AUDIO_QUALITIES: logger.warning("Vorbis audio quality is only supported for podcasts.") continue - if download_music_videos: downloader_music_video.download( music_video_id=media_id, From d4082edf7d4879890e135e4e300285a26027db1b Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Fri, 30 Jan 2026 14:56:31 -0300 Subject: [PATCH 26/36] Update music video and podcast video download logic --- votify/downloader.py | 87 ++++++++++++++++++++++++++------------------ 1 file changed, 52 insertions(+), 35 deletions(-) diff --git a/votify/downloader.py b/votify/downloader.py index 63574dd..f66d300 100644 --- a/votify/downloader.py +++ b/votify/downloader.py @@ -59,7 +59,7 @@ class Downloader: template_file_multi_disc: str = "{disc}-{track:02d} {title}", template_folder_episode: str = "Podcasts/{album}", template_file_episode: str = "{track:02d} {title}", - template_folder_music_video: str = "{artist}/Unknown Album", + template_folder_music_video: str = "{artist}/{album}", template_file_music_video: str = "{title}", template_file_playlist: str = "Playlists/{playlist_artist}/{playlist_title}", date_tag_template: str = "%Y-%m-%dT%H:%M:%SZ", @@ -680,49 +680,66 @@ class Downloader: media_type: str, audio_quality: dict, track_name: str, + download_music_video: str, + download_podcast_videos: str, ) -> str: playback_info = self.spotify_api.get_track_playback_info(media_id, media_type) user_product = audio_quality.get('data', {}).get('me', {}).get('account', {}).get('product', 'free') - target_bitrate = 256000 if user_product == 'PREMIUM' else 128000 + if media_type == 'episode': + target_bitrate = 128000 + else: + target_bitrate = 256000 if user_product == 'PREMIUM' else 128000 media_items = playback_info.get("media", {}) - if not media_items: - raise ValueError("No available Track (Empty response)") - - expected_key = f"spotify:{media_type}:{media_id}" - - track_manifest = media_items.get(expected_key) - - if not track_manifest: - track_manifest = list(media_items.values())[0] - - metadata = track_manifest.get("item", {}).get("metadata", {}) - server_name = metadata.get("name") - - if not server_name: - raise ValueError(f"Track Unavailable: No name returned") - - req_clean = track_name.lower().strip() - srv_clean = server_name.lower().strip() - - if (req_clean in srv_clean) or (srv_clean in req_clean): - track_manifest = track_manifest + if download_music_video == True or download_podcast_videos == True: + for track_key, track_data in media_items.items(): + manifest = track_data['item']['manifest'] + if 'manifest_ids_video' in manifest: + video_file_id = manifest['manifest_ids_video'][0]['file_id'] + else: + logger.error("There is no video for this media.") + return + return video_file_id else: - raise ValueError(f"Track Unavailable: No name returned") + if not media_items: + raise ValueError("No available Track (Empty response)") + + expected_key = f"spotify:{media_type}:{media_id}" + + track_manifest = media_items.get(expected_key) + + if not track_manifest: + track_manifest = list(media_items.values())[0] + + metadata = track_manifest.get("item", {}).get("metadata", {}) + server_name = metadata.get("name") + + if not server_name: + raise ValueError(f"Track Unavailable: No name returned") + + req_clean = track_name.lower().strip() + srv_clean = server_name.lower().strip() + + if (req_clean in srv_clean) or (srv_clean in req_clean): + track_manifest = track_manifest + else: + raise ValueError(f"Track Unavailable: No name returned") + + try: + audio_file_id = next( + f["file_id"] + for files in track_manifest["item"]["manifest"].values() + if isinstance(files, list) + for f in files + if isinstance(f, dict) and f.get("bitrate") == int(target_bitrate) + ) + except StopIteration: + raise ValueError(f"No audio file found with bitrate {target_bitrate}") + + return audio_file_id - try: - audio_file_id = next( - f["file_id"] - for files in track_manifest["item"]["manifest"].values() - if isinstance(files, list) - for f in files - if isinstance(f, dict) and f.get("bitrate") == int(target_bitrate) - ) - except StopIteration: - raise ValueError(f"No audio file found with bitrate {target_bitrate}") - return audio_file_id def get_playplay_decryption_key(self, file_id: str) -> bytes: raise NotImplementedError() From fa3e5a0e3cc88427548c9ff08931eca0f797aba5 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Fri, 30 Jan 2026 14:57:09 -0300 Subject: [PATCH 27/36] Fix download music video --- votify/downloader_music_video.py | 103 +++++++++++++++++++++---------- 1 file changed, 69 insertions(+), 34 deletions(-) diff --git a/votify/downloader_music_video.py b/votify/downloader_music_video.py index bceb2aa..cebadbb 100644 --- a/votify/downloader_music_video.py +++ b/votify/downloader_music_video.py @@ -19,58 +19,92 @@ class DownloaderMusicVideo(DownloaderVideo): self.__dict__.update(downloader_video.__dict__) def get_video_gid(self, gid_metadata: dict) -> str | None: - if not gid_metadata.get("original_video"): + if not gid_metadata: return None - return gid_metadata["original_video"][0]["gid"] + return gid_metadata def get_tags( - self, - track_metadata: dict, - album_metadata: dict, - track_credits: dict, + self, + track_metadata: dict, + album_metadata: dict, + track_credits: dict, ) -> dict: - external_ids = track_metadata.get("external_ids") - external_urls = (track_metadata.get("linked_from") or track_metadata)[ - "external_urls" - ] + track_data = track_metadata.get('data', {}).get('trackUnion', {}) + if not track_data: + track_data = track_metadata + + album_info = track_data.get('albumOfTrack', {}) + + spotify_url = track_data.get('sharingInfo', {}).get('shareUrl') + if not spotify_url and 'id' in track_data: + spotify_url = f"https://open.spotify.com/track/{track_data['id']}" + + release_date_iso = album_info.get("date", {}).get("isoString") + precision_raw = album_info.get("date", {}).get("precision") + + precision = precision_raw.lower() if precision_raw else "day" + + release_date_str = release_date_iso.split("T")[0] if release_date_iso else None + release_date_datetime_obj = self.downloader.get_release_date_datetime_obj( - album_metadata["release_date"], - album_metadata["release_date_precision"], + release_date_str, + precision, ) - producers = next( - role - for role in track_credits["roleCredits"] - if role["roleTitle"] == "Producers" - )["artists"] - composers = next( - role - for role in track_credits["roleCredits"] - if role["roleTitle"] == "Writers" - )["artists"] + + try: + main_artists = next( + role + for role in track_credits.get("roleCredits", []) + if role["roleTitle"] == "Performers" + )["artists"] + artist_string = self.downloader.get_artist_string(main_artists) + except (StopIteration, KeyError): + artist_string = "Unknown Artist" + + def get_credits_by_role(role_name): + try: + return next( + role + for role in track_credits.get("roleCredits", []) + if role["roleTitle"] == role_name + )["artists"] + except (StopIteration, KeyError): + return None + + producers = get_credits_by_role("Producers") + composers = get_credits_by_role("Writers") + + copyright_text = None + copyrights_list = album_info.get('copyright', {}).get('items', []) + if copyrights_list: + copyright_obj = next((i for i in copyrights_list if i.get("type") == "P"), copyrights_list[0]) + copyright_text = copyright_obj.get('text') + + album_name = album_info.get("name") + if not album_name and isinstance(album_metadata, dict): + album_name = album_metadata.get("name") tags = { - "artist": self.downloader.get_artist_string(track_metadata["artists"]), + "album": album_name or "Unknown Album", + "artist": artist_string, "composer": ( self.downloader.get_artist_string(composers) if composers else None ), - "copyright": next( - (i["text"] for i in album_metadata["copyrights"] if i["type"] == "P"), - None, - ), - "isrc": external_ids.get("isrc") if external_ids is not None else None, - "label": album_metadata.get("label"), + "copyright": copyright_text, + "isrc": None, + "label": None, "media_type": "Music video", "producer": ( self.downloader.get_artist_string(producers) if producers else None ), - "rating": "Explicit" if track_metadata.get("explicit") else "Unknown", - "title": track_metadata["name"], + "rating": "Explicit" if track_data.get("contentRating", {}).get("label") == "EXPLICIT" else "Clean", + "title": track_data.get("name"), "release_date": self.downloader.get_release_date_tag( release_date_datetime_obj ), "release_year": str(release_date_datetime_obj.year), - "url": external_urls["spotify"], + "url": spotify_url, } - tags["release_year"] = str(release_date_datetime_obj.year) + return tags def get_music_video_id_from_song_id( @@ -146,12 +180,13 @@ class DownloaderMusicVideo(DownloaderVideo): if not album_metadata: logger.debug("Getting album metadata") album_metadata = self.downloader.spotify_api.get_album( - music_video_metadata["album"]["id"] + music_video_metadata['data']['trackUnion']['albumOfTrack']["id"] ) if not gid_metadata: logger.debug("Getting GID metadata") gid_metadata = self.downloader.get_gid_metadata(music_video_id, "track") video_gid = self.get_video_gid(gid_metadata) + if not video_gid: logger.debug("Getting equivalent music video ID from song ID") music_video_id = self.get_music_video_id_from_song_id( From 76d001ff184bdd390e120a8d461bd7353f5ee542 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Fri, 30 Jan 2026 14:57:42 -0300 Subject: [PATCH 28/36] Fix episode video Update get_video_gid method to handle empty gid_metadata and adjust download method to use trackUnion. --- votify/downloader_episode_video.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/votify/downloader_episode_video.py b/votify/downloader_episode_video.py index 89f2947..1708abb 100644 --- a/votify/downloader_episode_video.py +++ b/votify/downloader_episode_video.py @@ -19,9 +19,9 @@ class DownloaderEpisodeVideo(DownloaderVideo): self.downloader_episode = downloader_episode def get_video_gid(self, gid_metadata: dict) -> str | None: - if not gid_metadata.get("video"): + if not gid_metadata: return None - return gid_metadata["video"][0]["file_id"] + return gid_metadata def download( self, @@ -65,7 +65,7 @@ class DownloaderEpisodeVideo(DownloaderVideo): ) return tags = self.downloader_episode.get_tags( - episode_metadata, + episode_metadata['data']['trackUnion'], show_metadata, ) file_extension = self.get_file_extension( From 5e473c5e8393f84ecbfbc506fbb9a46460bbd3ae Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Fri, 30 Jan 2026 23:13:50 -0300 Subject: [PATCH 29/36] Refactor video file ID extraction logic --- votify/downloader.py | 22 ++++++++++++++-------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/votify/downloader.py b/votify/downloader.py index f66d300..cb6705c 100644 --- a/votify/downloader.py +++ b/votify/downloader.py @@ -692,14 +692,20 @@ class Downloader: target_bitrate = 256000 if user_product == 'PREMIUM' else 128000 media_items = playback_info.get("media", {}) - if download_music_video == True or download_podcast_videos == True: - for track_key, track_data in media_items.items(): - manifest = track_data['item']['manifest'] - if 'manifest_ids_video' in manifest: - video_file_id = manifest['manifest_ids_video'][0]['file_id'] - else: - logger.error("There is no video for this media.") - return + if download_music_video or download_podcast_videos: + video_file_id = next( + ( + data['item']['manifest']['manifest_ids_video'][0]['file_id'] + for data in playback_info['media'].values() + if 'item' in data + and 'manifest' in data['item'] + and 'manifest_ids_video' in data['item']['manifest'] + ), + None + ) + if video_file_id is None: + logger.error("There is no video for this media.") + return return video_file_id else: if not media_items: From ff0d759f4b2bc5152d3f4bff0277a2da57932cc3 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Mon, 2 Feb 2026 11:41:54 -0300 Subject: [PATCH 30/36] Refactor artist album retrieval methods --- votify/spotify_api.py | 174 ++++++++++++++++++++++++++++++++++++++---- 1 file changed, 158 insertions(+), 16 deletions(-) diff --git a/votify/spotify_api.py b/votify/spotify_api.py index 6590121..dc20a35 100644 --- a/votify/spotify_api.py +++ b/votify/spotify_api.py @@ -418,28 +418,170 @@ class SpotifyApi: )''' return show["data"]['podcastUnionV2'] - def get_artist_albums( + def get_artist_albums_selection( self, artist_id: str, extend: bool = True, ) -> dict: self._refresh_session_auth() - response = self.session.get( - self.METADATA_API_URL.format(type="artists", item_id=artist_id) + "/albums" - ) + 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() - if extend: - artist_albums["items"].extend( - [ - item - for extended_collection in self.extended_media_collection( - artist_albums["next"], - ) - for item in extended_collection["items"] - ] - ) - return artist_albums + 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, From d5a7cd3708512ec5a1e66ecead63a7ee152a2a97 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Mon, 2 Feb 2026 11:42:41 -0300 Subject: [PATCH 31/36] Fix album type filtering and update selection logic Refactor album filtering and selection process in downloader.py. --- votify/downloader.py | 84 +++++++++++++++----------------------------- 1 file changed, 28 insertions(+), 56 deletions(-) diff --git a/votify/downloader.py b/votify/downloader.py index cb6705c..7349977 100644 --- a/votify/downloader.py +++ b/votify/downloader.py @@ -433,70 +433,42 @@ class Downloader: artist_albums: list[dict], album_type: str, ) -> list[dict]: - return [album for album in artist_albums if album["album_type"] == album_type] + return [album for album in artist_albums if album == album_type] - def get_download_queue_from_artist( - self, - artist_id: str, - ) -> list[DownloadQueueItem]: + + def get_download_queue_from_artist(self, artist_id: str) -> list[DownloadQueueItem]: download_queue = [] - artist_albums = self.spotify_api.get_artist_albums(artist_id) - artist_albums_all = {} - for album_type in ["album", "single", "compilation", "appears_on"]: - artist_albums_all[album_type] = self._filter_artist_albums( - artist_albums["items"], - album_type, - ) - selected_album_type = inquirer.select( - message=f"Select which album type to download for artist:", + + selected_option = inquirer.select( + message="Select which album type to download for artist:", choices=[ - Choice( - name="Albums", - value="album", - ), - Choice( - name="Singles", - value="single", - ), - Choice( - name="Compilations", - value="compilation", - ), - Choice( - name="Collaborations", - value="appears_on", - ), + Choice(name="Albuns", value="album"), + Choice(name="Singles", value="single"), + Choice(name="Compilations", value="compilation"), + Choice(name="Collaborations", value="appears_on"), ], - validate=lambda result: artist_albums_all.get(result), - invalid_message="The artist doesn't have any albums of this type", ).execute() - choices = [ - Choice( - name=" | ".join( - [ - f"{album['total_tracks']:03d}", - album["release_date"][:4], - album["name"], - ], - ), - value=album["id"], - ) - for album in artist_albums_all[selected_album_type] - ] - selected = inquirer.select( - message="Select which albums to download: (Total tracks | Release year | Album name)", - choices=choices, - multiselect=True, - ).execute() - for album_id in selected: - download_queue.extend( - self.get_download_queue( - "album", - album_id, + + selected_ids = [] + if selected_option == "album": + selected_ids = self.spotify_api.get_artist_albums_selection(artist_id) + elif selected_option == "single": + selected_ids = self.spotify_api.get_artist_singles_selection(artist_id) + elif selected_option == "compilation": + selected_ids = self.spotify_api.get_artist_compilations_selection(artist_id) + elif selected_option == "appears_on": + selected_ids = self.spotify_api.get_artist_collaborations_selection(artist_id) + + if selected_ids: + print(f"Processing {len(selected_ids)} items...") + for album_id in selected_ids: + download_queue.extend( + self.get_download_queue("album", album_id) ) - ) + return download_queue + def get_cover_url(self, metadata: dict, cover_size_mapping: dict) -> str | None: if not isinstance(metadata, dict): return None From 4dff2049fed30ae841e005d76d182ee45f87bcfe Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Wed, 4 Feb 2026 12:50:18 -0300 Subject: [PATCH 32/36] Add delay for download queue processing --- votify/cli.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/votify/cli.py b/votify/cli.py index 0d4fd50..7ffbcf7 100644 --- a/votify/cli.py +++ b/votify/cli.py @@ -795,5 +795,7 @@ def main( f"Waiting for {wait_interval} second(s) before continuing" ) time.sleep(wait_interval) + if len(download_queue) > 1 and index != len(download_queue): + time.sleep(15) logger.info(f"Done ({error_count} error(s))") From f13fb97f53325b6e652c124a2189b4fe69c597d6 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Thu, 5 Feb 2026 20:25:40 -0300 Subject: [PATCH 33/36] Add WebSocket support for Spotify playback control --- votify/spotify_api.py | 169 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 169 insertions(+) diff --git a/votify/spotify_api.py b/votify/spotify_api.py index dc20a35..8d1fa72 100644 --- a/votify/spotify_api.py +++ b/votify/spotify_api.py @@ -12,6 +12,10 @@ 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 @@ -229,6 +233,171 @@ class SpotifyApi: 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", From 0ebcf15f9112232a6a62d30fe409c28e2c9fc450 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Thu, 5 Feb 2026 20:26:05 -0300 Subject: [PATCH 34/36] Update downloader_audio.py --- votify/downloader_audio.py | 28 +++++++++++++++++++--------- 1 file changed, 19 insertions(+), 9 deletions(-) diff --git a/votify/downloader_audio.py b/votify/downloader_audio.py index efce78f..420c517 100644 --- a/votify/downloader_audio.py +++ b/votify/downloader_audio.py @@ -1,6 +1,7 @@ from __future__ import annotations import subprocess +import time from pathlib import Path from Crypto.Cipher import AES @@ -92,16 +93,25 @@ class DownloaderAudio: return stream_info + def get_decryption_key(self, stream_info: StreamInfoAudio) -> str: - if self.audio_quality in AAC_AUDIO_QUALITIES: - _, decryption_key = self.downloader.get_widevine_decryption_key( - stream_info.widevine_pssh, - "audio", - ) - else: - decryption_key = self.downloader.get_playplay_decryption_key( - stream_info.file_id - ) + decryption_key = None + for attempt in range(10): + try: + if self.audio_quality in AAC_AUDIO_QUALITIES: + _, decryption_key = self.downloader.get_widevine_decryption_key( + stream_info.widevine_pssh, + "audio", + ) + else: + decryption_key = self.downloader.get_playplay_decryption_key( + stream_info.file_id + ) + if decryption_key: + return decryption_key + + except Exception as e: + time.sleep(attempt + 10) return decryption_key def download_stream_url(self, input_path: Path, stream_url: str): From 02f6f78ec88f1c2d5ef6f41700401db2db038560 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Thu, 5 Feb 2026 20:26:30 -0300 Subject: [PATCH 35/36] Remove unnecessary sleep in download processing --- votify/cli.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/votify/cli.py b/votify/cli.py index 7ffbcf7..0d4fd50 100644 --- a/votify/cli.py +++ b/votify/cli.py @@ -795,7 +795,5 @@ def main( f"Waiting for {wait_interval} second(s) before continuing" ) time.sleep(wait_interval) - if len(download_queue) > 1 and index != len(download_queue): - time.sleep(15) logger.info(f"Done ({error_count} error(s))") From d32e26a932e557e952bb7a845ae3f4e3c851dcc6 Mon Sep 17 00:00:00 2001 From: GladistonXD <50533550+GladistonXD@users.noreply.github.com> Date: Thu, 5 Feb 2026 20:29:44 -0300 Subject: [PATCH 36/36] Update requirements.txt --- requirements.txt | 1 + 1 file changed, 1 insertion(+) diff --git a/requirements.txt b/requirements.txt index 9794f97..b1ae04f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,3 +8,4 @@ pybase62 pycryptodome pywidevine yt-dlp +websocket-client