"""Adds Spotify release and track search support to the autotagger. Also includes Spotify playlist construction. """ from __future__ import annotations import base64 import collections import json import re import threading import time import webbrowser from http import HTTPStatus from typing import TYPE_CHECKING, Any, ClassVar, Literal, TypedDict import confuse import requests from beets import ui from beets.autotag import AlbumInfo, TrackInfo from beets.dbcore import types from beets.exceptions import UserError from beets.library import Library from beets.metadata_plugins import IDResponse, SearchApiMetadataSourcePlugin from beets.util import chunks if TYPE_CHECKING: from collections.abc import Sequence from beets.dbcore.db import Results from beets.library import Item, Library from beets.metadata_plugins import QueryType, SearchParams from beetsplug._typing import JSONDict DEFAULT_WAITING_TIME = 5 class TrackDetails(TypedDict): """Popularity and external IDs returned by the /v1/tracks batch endpoint.""" spotify_track_popularity: int | None isrc: str | None ean: str | None upc: str | None class AudioFeatures(TypedDict, total=False): """Audio feature fields returned by the /v1/audio-features endpoint.""" id: str acousticness: float danceability: float energy: float instrumentalness: float key: int liveness: float loudness: float mode: int speechiness: float tempo: float time_signature: int valence: float class SearchResponseAlbums(IDResponse): """A response returned by the Spotify API. We only use items and disregard the pagination information. i.e. res["albums"]["items"][0]. There are more fields in the response, but we only type the ones we currently use. see https://developer.spotify.com/documentation/web-api/reference/search """ album_type: str available_markets: Sequence[str] name: str class SearchResponseTracks(IDResponse): """A track response returned by the Spotify API.""" album: SearchResponseAlbums available_markets: Sequence[str] popularity: int name: str class APIError(Exception): pass class AudioFeaturesUnavailableError(Exception): """Raised when audio features API returns 403 (deprecated).""" class SpotifyPlugin( SearchApiMetadataSourcePlugin[SearchResponseAlbums | SearchResponseTracks] ): item_types: ClassVar[dict[str, types.Type]] = { "spotify_track_popularity": types.INTEGER, "spotify_acousticness": types.FLOAT, "spotify_danceability": types.FLOAT, "spotify_energy": types.FLOAT, "spotify_instrumentalness": types.FLOAT, "spotify_key": types.FLOAT, "spotify_liveness": types.FLOAT, "spotify_loudness": types.FLOAT, "spotify_mode": types.INTEGER, "spotify_speechiness": types.FLOAT, "spotify_tempo": types.FLOAT, "spotify_time_signature": types.INTEGER, "spotify_valence": types.FLOAT, "spotify_updated": types.DATE, } # Base URLs for the Spotify API # Documentation: https://developer.spotify.com/web-api oauth_token_url = "https://accounts.spotify.com/api/token" open_track_url = "https://open.spotify.com/track/" search_url = "https://api.spotify.com/v1/search" album_url = "https://api.spotify.com/v1/albums/" track_url = "https://api.spotify.com/v1/tracks" audio_features_url = "https://api.spotify.com/v1/audio-features" spotify_audio_features: ClassVar[dict[str, str]] = { "acousticness": "spotify_acousticness", "danceability": "spotify_danceability", "energy": "spotify_energy", "instrumentalness": "spotify_instrumentalness", "key": "spotify_key", "liveness": "spotify_liveness", "loudness": "spotify_loudness", "mode": "spotify_mode", "speechiness": "spotify_speechiness", "tempo": "spotify_tempo", "time_signature": "spotify_time_signature", "valence": "spotify_valence", } def __init__(self): super().__init__() self.config.add( { "mode": "list", "tiebreak": "popularity", "show_failures": False, "region_filter": None, "regex": [], "client_id": "4e414367a1d14c75a5c5129a627fcab8", "client_secret": "4a9b5b7848e54e118a7523b1c7c3e1e5", "tokenfile": "spotify_token.json", } ) self.config["client_id"].redact = True self.config["client_secret"].redact = True self.audio_features_available = ( True # Track if audio features API is available ) self._audio_features_lock = ( threading.Lock() ) # Protects audio_features_available self.setup() def setup(self): """Retrieve previously saved OAuth token or generate a new one.""" try: with open(self._tokenfile()) as f: token_data = json.load(f) except OSError: self._authenticate() else: self.access_token = token_data["access_token"] def _tokenfile(self) -> str: """Get the path to the JSON file for storing the OAuth token.""" return self.config["tokenfile"].get(confuse.Filename(in_app_dir=True)) def _authenticate(self) -> None: """Request an access token via the Client Credentials Flow: https://developer.spotify.com/documentation/general/guides/authorization-guide/#client-credentials-flow""" c_id: str = self.config["client_id"].as_str() c_secret: str = self.config["client_secret"].as_str() headers = { "Authorization": ( "Basic" f" {base64.b64encode(f'{c_id}:{c_secret}'.encode()).decode()}" ) } response = requests.post( self.oauth_token_url, data={"grant_type": "client_credentials"}, headers=headers, timeout=10, ) try: response.raise_for_status() except requests.exceptions.HTTPError as e: raise UserError( f"Spotify authorization failed: {e}\n{response.text}" ) self.access_token = response.json()["access_token"] # Save the token for later use. self._log.debug("{0.data_source} access token: {0.access_token}", self) with open(self._tokenfile(), "w") as f: json.dump({"access_token": self.access_token}, f) def _handle_response( self, method: Literal["get", "post", "put", "delete"], url: str, params: Any = None, retry_count: int = 0, max_retries: int = 3, ) -> JSONDict: """Send a request, reauthenticating if necessary. :param method: HTTP method to use for the request. :param url: URL for the new :class:`Request` object. :param dict params: (optional) list of tuples or bytes to send in the query string for the :class:`Request`. """ if retry_count > max_retries: raise APIError("Maximum retries reached.") try: response = requests.request( method, url, headers={"Authorization": f"Bearer {self.access_token}"}, params=params, timeout=10, ) response.raise_for_status() return response.json() except requests.exceptions.ReadTimeout: self._log.error("ReadTimeout.") raise APIError("Request timed out.") except requests.exceptions.ConnectionError as e: self._log.error("Network error: {}", e) raise APIError("Network error.") except requests.exceptions.RequestException as e: if e.response is None: self._log.error("Request failed: {}", e) raise APIError("Request failed.") if e.response.status_code == 401: self._log.debug( "{.data_source} access token has expired. Reauthenticating.", self, ) self._authenticate() return self._handle_response( method, url, params=params, retry_count=retry_count + 1 ) if e.response.status_code == 404: raise APIError( f"API Error: {e.response.status_code}\n" f"URL: {url}\nparams: {params}" ) if e.response.status_code == 403: # Check if this is the audio features endpoint if url.startswith(self.audio_features_url): raise AudioFeaturesUnavailableError( "Audio features API returned 403 " "(deprecated or unavailable)" ) raise APIError( f"API Error: {e.response.status_code}\n" f"URL: {url}\nparams: {params}" ) if e.response.status_code == 429: seconds = e.response.headers.get( "Retry-After", DEFAULT_WAITING_TIME ) self._log.debug( "Too many API requests. Retrying after {} seconds.", seconds ) time.sleep(int(seconds) + 1) return self._handle_response( method, url, params=params, retry_count=retry_count + 1 ) if e.response.status_code == 503: self._log.error("Service Unavailable.") raise APIError("Service Unavailable.") if e.response.status_code == 502: self._log.error("Bad Gateway.") raise APIError("Bad Gateway.") if e.response is not None: raise APIError( f"{self.data_source} API error:\n" f"{e.response.text}\n" f"URL:\n{url}\nparams:\n{params}" ) self._log.error("Request failed. Error: {}", e) raise APIError("Request failed.") def _multi_artist_credit( self, artists: list[dict[str | int, str]] ) -> tuple[list[str], list[str]]: """Given a list of artist dictionaries, accumulate data into a pair of lists: the first being the artist names, and the second being the artist IDs. """ artist_names = [] artist_ids = [] for artist in artists: artist_names.append(artist["name"]) artist_ids.append(artist["id"]) return artist_names, artist_ids def album_for_id(self, album_id: str) -> AlbumInfo | None: """Fetch an album by its Spotify ID or URL and return an AlbumInfo object or None if the album is not found. :param str album_id: Spotify ID or URL for the album :returns: AlbumInfo object for album :rtype: beets.autotag.hooks.AlbumInfo or None """ if not (spotify_id := self._extract_id(album_id)): return None album_data = self._handle_response( "get", f"{self.album_url}{spotify_id}" ) if album_data["name"] == "": self._log.debug("Album removed from Spotify: {}", album_id) return None artists_names, artists_ids = self._multi_artist_credit( album_data["artists"] ) artist = ", ".join(artists_names) date_parts = [ int(part) for part in album_data["release_date"].split("-") ] release_date_precision = album_data["release_date_precision"] if release_date_precision == "day": year, month, day = date_parts elif release_date_precision == "month": year, month = date_parts day = None elif release_date_precision == "year": year = date_parts[0] month = None day = None else: raise UserError( "Invalid `release_date_precision` returned " f"by {self.data_source} API: '{release_date_precision}'" ) tracks_data = album_data["tracks"] tracks_items = tracks_data["items"] while tracks_data["next"]: tracks_data = self._handle_response("get", tracks_data["next"]) tracks_items.extend(tracks_data["items"]) tracks = [] medium_totals: dict[int | None, int] = collections.defaultdict(int) for i, track_data in enumerate(tracks_items, start=1): track = self._get_track(track_data) track.index = i medium_totals[track.medium] += 1 tracks.append(track) for track in tracks: track.medium_total = medium_totals[track.medium] return AlbumInfo( album=album_data["name"], album_id=spotify_id, spotify_album_id=spotify_id, artist=artist, artist_id=artists_ids[0] if len(artists_ids) > 0 else None, spotify_artist_id=artists_ids[0] if len(artists_ids) > 0 else None, artists=artists_names, artists_ids=artists_ids, tracks=tracks, albumtype=album_data["album_type"], va=len(album_data["artists"]) == 1 and artist.lower() == "various artists", year=year, month=month, day=day, label=album_data.get("label"), mediums=max(filter(None, medium_totals.keys())), data_source=self.data_source, data_url=album_data["external_urls"]["spotify"], ) def _get_track(self, track_data: JSONDict) -> TrackInfo: """Convert a Spotify track object dict to a TrackInfo object. :param track_data: Simplified track object (https://developer.spotify.com/documentation/web-api/reference/object-model/#track-object-simplified) :returns: TrackInfo object for track """ artists_names, artists_ids = self._multi_artist_credit( track_data["artists"] ) artist = ", ".join(artists_names) # Get album information for spotify tracks try: album = track_data["album"]["name"] except (KeyError, TypeError): album = None return TrackInfo( title=track_data["name"], track_id=track_data["id"], spotify_track_id=track_data["id"], artist=artist, album=album, artist_id=artists_ids[0] if len(artists_ids) > 0 else None, spotify_artist_id=artists_ids[0] if len(artists_ids) > 0 else None, artists=artists_names, artists_ids=artists_ids, length=track_data["duration_ms"] / 1000, index=track_data["track_number"], medium=track_data["disc_number"], medium_index=track_data["track_number"], data_source=self.data_source, data_url=track_data["external_urls"]["spotify"], ) def track_for_id(self, track_id: str) -> None | TrackInfo: """Fetch a track by its Spotify ID or URL. Returns a TrackInfo object or None if the track is not found. """ if not (spotify_id := self._extract_id(track_id)): self._log.debug("Invalid Spotify ID: {}", track_id) return None if not ( track_data := self._handle_response( "get", f"{self.track_url}/{spotify_id}" ) ): self._log.debug("Track not found: {}", track_id) return None track = self._get_track(track_data) # Get album's tracks to set `track.index` (position on the entire # release) and `track.medium_total` (total number of tracks on # the track's disc). album_data = self._handle_response( "get", f"{self.album_url}{track_data['album']['id']}" ) medium_total = 0 for i, track_data in enumerate(album_data["tracks"]["items"], start=1): if track_data["disc_number"] == track.medium: medium_total += 1 if track_data["id"] == track.track_id: track.index = i track.medium_total = medium_total return track def get_search_query_with_filters( self, query_type: QueryType, items: Sequence[Item], artist: str, name: str, va_likely: bool, ) -> tuple[str, dict[str, str]]: query = f'album:"{name}"' if query_type == "album" else name if query_type == "track" or not va_likely: query += f' artist:"{artist}"' return query, {} def get_search_response( self, params: SearchParams ) -> Sequence[SearchResponseAlbums | SearchResponseTracks]: """Search Spotify and return raw album or track result items. Unauthorized responses trigger one token refresh attempt before the method gives up and falls back to an empty result set. """ for _ in range(2): response = requests.get( self.search_url, headers={"Authorization": f"Bearer {self.access_token}"}, params={ **params.filters, "q": params.query.replace('"', "'"), "type": params.query_type, "limit": str(params.limit), }, timeout=10, ) try: response.raise_for_status() except requests.exceptions.HTTPError: if response.status_code == HTTPStatus.UNAUTHORIZED: self._authenticate() continue raise return ( response.json() .get(f"{params.query_type}s", {}) .get("items", []) ) return () def commands(self) -> list[ui.Subcommand]: # autotagger import command def queries(lib, opts, args): success = self._parse_opts(opts) if success: results = self._match_library_tracks(lib, args) self._output_match_results(results) spotify_cmd = ui.Subcommand( "spotify", help=f"build a {self.data_source} playlist" ) spotify_cmd.parser.add_option( "-m", "--mode", action="store", help=( f'"open" to open {self.data_source} with playlist, ' '"list" to print (default)' ), ) spotify_cmd.parser.add_option( "-f", "--show-failures", action="store_true", dest="show_failures", help=f"list tracks that did not match a {self.data_source} ID", ) spotify_cmd.func = queries # spotifysync command sync_cmd = ui.Subcommand( "spotifysync", help="fetch track attributes from Spotify" ) sync_cmd.parser.add_option( "-f", "--force", dest="force_refetch", action="store_true", default=False, help="re-download data when already present", ) def func(lib, opts, args): items = lib.items(args) self._fetch_info(lib, items, ui.should_write(), opts.force_refetch) sync_cmd.func = func return [spotify_cmd, sync_cmd] def _parse_opts(self, opts): if opts.mode: self.config["mode"].set(opts.mode) if opts.show_failures: self.config["show_failures"].set(True) if self.config["mode"].get() not in ["list", "open"]: self._log.warning( "{} is not a valid mode", self.config["mode"].get() ) return False self.opts = opts return True def _match_library_tracks(self, library: Library, keywords: str): """Get simplified track object dicts for library tracks. Matches tracks based on the specified ``keywords``. :param library: beets library object to query. :param keywords: Query to match library items against. :returns: List of simplified track object dicts for library items matching the specified query. """ results = [] failures = [] items = library.items(keywords) if not items: self._log.debug( "Your beets query returned no items, skipping {.data_source}.", self, ) return None self._log.info("Processing {} tracks...", len(items)) for item in items: # Apply regex transformations if provided for regex in self.config["regex"].get(): if ( not regex["field"] or not regex["search"] or not regex["replace"] ): continue value = item[regex["field"]] item[regex["field"]] = re.sub( regex["search"], regex["replace"], value ) artist = item["artist"] or item["albumartist"] album = item["album"] query_string = item["title"] # Query the Web API for each track, look for the items' JSON data query = query_string if artist: query += f" artist:'{artist}'" if album: query += f" album:'{album}'" response_data_tracks = self._search_api("track", query, {}) if not response_data_tracks: failures.append(query) continue # Apply market filter if requested region_filter: str = self.config["region_filter"].get() if region_filter: response_data_tracks = [ track_data for track_data in response_data_tracks if region_filter in track_data["available_markets"] ] if ( len(response_data_tracks) == 1 or self.config["tiebreak"].get() == "first" ): self._log.debug( "{.data_source} track(s) found, count: {}", self, len(response_data_tracks), ) chosen_result = response_data_tracks[0] else: # Use the popularity filter self._log.debug( "Most popular track chosen, count: {}", len(response_data_tracks), ) chosen_result = max( response_data_tracks, key=lambda x: x[ # We are sure this is a track response! "popularity" # type: ignore[typeddict-item] ], ) results.append(chosen_result) failure_count = len(failures) if failure_count > 0: if self.config["show_failures"].get(): self._log.info( "{} track(s) did not match a {.data_source} ID:", failure_count, self, ) for track in failures: self._log.info("track: {}", track) self._log.info("") else: self._log.warning( "{} track(s) did not match a {.data_source} ID:\n" "use --show-failures to display", failure_count, self, ) return results def _output_match_results(self, results): """Open a playlist or print Spotify URLs. Uses the provided track object dicts. :param list[dict] results: List of simplified track object dicts (https://developer.spotify.com/documentation/web-api/ reference/object-model/#track-object-simplified) """ if results: spotify_ids = [track_data["id"] for track_data in results] if self.config["mode"].get() == "open": self._log.info( "Attempting to open {.data_source} with playlist", self ) spotify_url = ( f"spotify:trackset:Playlist:{','.join(spotify_ids)}" ) webbrowser.open(spotify_url) else: for spotify_id in spotify_ids: print(f"{self.open_track_url}{spotify_id}") else: self._log.warning( "No {.data_source} tracks found from beets query", self ) def _disable_audio_features(self) -> None: """Disable audio features globally and warn only once.""" should_log = False with self._audio_features_lock: if self.audio_features_available: self.audio_features_available = False should_log = True if should_log: self._log.warning( "Audio features API is unavailable (403 error). " "Skipping audio features for remaining tracks." ) def get_track_details_by_id( self, track_ids: Sequence[str] ) -> dict[str, TrackDetails]: """Fetch popularity and external IDs in batches of 50 tracks.""" if not track_ids: return {} details_by_id: dict[str, TrackDetails] = {} for chunk in chunks(track_ids, 50): track_data = self._handle_response( "get", self.track_url, params={"ids": ",".join(chunk)} ) for idx, track in enumerate(track_data.get("tracks", [])): if track is None: continue external_ids = track.get("external_ids", {}) track_id = track.get("id") or chunk[idx] details_by_id[track_id] = TrackDetails( spotify_track_popularity=track.get("popularity"), isrc=external_ids.get("isrc"), ean=external_ids.get("ean"), upc=external_ids.get("upc"), ) return details_by_id def track_audio_features_batch( self, track_ids: Sequence[str] ) -> dict[str, AudioFeatures]: """Fetch track audio features in batches of 100 tracks.""" if not track_ids: return {} with self._audio_features_lock: if not self.audio_features_available: return {} features_by_id: dict[str, AudioFeatures] = {} for chunk in chunks(track_ids, 100): try: features_data = self._handle_response( "get", self.audio_features_url, params={"ids": ",".join(chunk)}, ) except AudioFeaturesUnavailableError: self._disable_audio_features() break except APIError as e: self._log.debug("Spotify API error: {}", e) continue for idx, feature_data in enumerate( features_data.get("audio_features", []) ): if feature_data: track_id = feature_data.get("id") or chunk[idx] features_by_id[track_id] = feature_data return features_by_id def _fetch_info( self, lib: Library, items: Results[Item] | Sequence[Item], write: bool, force: bool, ) -> None: """Obtain track information from Spotify.""" self._log.debug("Total {} tracks", len(items)) items_to_update: list[tuple[Item, str]] = [] for index, item in enumerate(items, start=1): self._log.info( "Processing {}/{} tracks - {} ", index, len(items), item ) # If we're not forcing re-downloading for all tracks, check # whether the popularity data is already present if not force: if "spotify_track_popularity" in item: self._log.debug("Popularity already present for: {}", item) continue try: spotify_track_id = item.spotify_track_id except AttributeError: self._log.debug("No track_id present for: {}", item) continue items_to_update.append((item, spotify_track_id)) if not items_to_update: return unique_track_ids = list( dict.fromkeys(track_id for _, track_id in items_to_update) ) track_details_by_id = self.get_track_details_by_id(unique_track_ids) audio_features_by_id = self.track_audio_features_batch(unique_track_ids) for item, spotify_track_id in items_to_update: if track_details := track_details_by_id.get(spotify_track_id): item.update(track_details) if self.audio_features_available: audio_features = audio_features_by_id.get(spotify_track_id) if audio_features is None: self._log.info("No audio features found for: {}", item) else: for feature, value in audio_features.items(): if feature in self.spotify_audio_features: item[self.spotify_audio_features[feature]] = value item["spotify_updated"] = time.time() if write: item.try_write() with lib.transaction(): for item, _ in items_to_update: item.store() def track_info(self, track_id: str): """Fetch a track's popularity and external IDs using its Spotify ID.""" track_data = self._handle_response( "get", f"{self.track_url}/{track_id}" ) external_ids = track_data.get("external_ids", {}) popularity = track_data.get("popularity") self._log.debug( "track_popularity: {} and track_isrc: {}", popularity, external_ids.get("isrc"), ) return ( popularity, external_ids.get("isrc"), external_ids.get("ean"), external_ids.get("upc"), ) def track_audio_features(self, track_id: str): """Fetch track audio features by its Spotify ID. Thread-safe: avoids redundant API calls and logs the 403 warning only once. """ # Fast path: if we've already detected unavailability, skip the call. with self._audio_features_lock: if not self.audio_features_available: return None try: return self._handle_response( "get", f"{self.audio_features_url}/{track_id}" ) except AudioFeaturesUnavailableError: self._disable_audio_features() return None except APIError as e: self._log.debug("Spotify API error: {}", e) return None