Skip to content

Spotify

This module contains the configuration object for the spotify package. The attributes of this configuration object correspond with the "spotify" key of config.yaml

SpotifyConfig

Bases: BaseConfigFormatter

Configuration object for the spotify package.

Source code in djtools/spotify/config.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
class SpotifyConfig(BaseConfigFormatter):
    """Configuration object for the spotify package."""

    spotify_playlist_default_limit: NonNegativeInt = 50
    spotify_playlist_default_period: SubredditPeriod = SubredditPeriod.WEEK
    spotify_playlist_default_type: SubredditType = SubredditType.HOT
    spotify_playlist_from_upload: bool = False
    spotify_playlist_fuzz_ratio: NonNegativeInt = 70
    spotify_playlist_post_limit: NonNegativeInt = 100
    spotify_playlist_subreddits: List[SubredditConfig] = []
    spotify_playlists: bool = False
    reddit_client_id: str = ""
    reddit_client_secret: str = ""
    reddit_user_agent: str = ""
    spotify_client_id: str = ""
    spotify_client_secret: str = ""
    spotify_redirect_uri: str = ""
    spotify_username: str = ""

    def __init__(self, *args, **kwargs):
        """Constructor.

        Raises:
            RuntimeError: Spotify API credentials must exit.
            RuntimeError: Spotify API credentials must be valid.
            RuntimeError: Reddit API credentials must exist.
        """
        super().__init__(*args, **kwargs)

        if (
            self.spotify_playlists or self.spotify_playlist_from_upload
        ) and not all(
            [
                self.spotify_client_id,
                self.spotify_client_secret,
                self.spotify_redirect_uri,
                self.spotify_username,
            ]
        ):
            raise RuntimeError(
                "Without all the configuration options spotify_client_id, "
                "spotify_client_secret, spotify_redirect_uri, and "
                "spotify_username set to valid values, you cannot use "
                "spotify_playlists or spotify_playlist_from_upload"
            )
        if self.spotify_playlists or self.spotify_playlist_from_upload:
            # pylint: disable=cyclic-import
            from djtools.spotify.helpers import get_spotify_client

            spotify = get_spotify_client(self)
            try:
                spotify.current_user()
            except Exception as exc:
                raise RuntimeError("Spotify credentials are invalid!") from exc

        if self.spotify_playlists and not all(
            [
                self.reddit_client_id,
                self.reddit_client_secret,
                self.reddit_user_agent,
            ]
        ):
            raise RuntimeError(
                "Without all the configuration options reddit_client_id, "
                "reddit_client_secret, and reddit_user_agent, set to valid "
                "values, you cannot use spotify_playlists"
            )

__init__(*args, **kwargs)

Constructor.

Raises:

Type Description
RuntimeError

Spotify API credentials must exit.
RuntimeError

Spotify API credentials must be valid.
RuntimeError

Reddit API credentials must exist.
Source code in djtools/spotify/config.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def __init__(self, *args, **kwargs):
    """Constructor.

    Raises:
        RuntimeError: Spotify API credentials must exit.
        RuntimeError: Spotify API credentials must be valid.
        RuntimeError: Reddit API credentials must exist.
    """
    super().__init__(*args, **kwargs)

    if (
        self.spotify_playlists or self.spotify_playlist_from_upload
    ) and not all(
        [
            self.spotify_client_id,
            self.spotify_client_secret,
            self.spotify_redirect_uri,
            self.spotify_username,
        ]
    ):
        raise RuntimeError(
            "Without all the configuration options spotify_client_id, "
            "spotify_client_secret, spotify_redirect_uri, and "
            "spotify_username set to valid values, you cannot use "
            "spotify_playlists or spotify_playlist_from_upload"
        )
    if self.spotify_playlists or self.spotify_playlist_from_upload:
        # pylint: disable=cyclic-import
        from djtools.spotify.helpers import get_spotify_client

        spotify = get_spotify_client(self)
        try:
            spotify.current_user()
        except Exception as exc:
            raise RuntimeError("Spotify credentials are invalid!") from exc

    if self.spotify_playlists and not all(
        [
            self.reddit_client_id,
            self.reddit_client_secret,
            self.reddit_user_agent,
        ]
    ):
        raise RuntimeError(
            "Without all the configuration options reddit_client_id, "
            "reddit_client_secret, and reddit_user_agent, set to valid "
            "values, you cannot use spotify_playlists"
        )

SubredditConfig

Bases: BaseModel

Configuration object for spotify_playlists.

Source code in djtools/spotify/config.py 
66
67
68
69
70
71
72
class SubredditConfig(BaseModel):
    """Configuration object for spotify_playlists."""

    name: str
    limit: NonNegativeInt = 50
    period: SubredditPeriod = SubredditPeriod.WEEK
    type: SubredditType = SubredditType.HOT

This module is responsible for creating or updating Spotify playlists. This can be done in a couple of ways.

The first way is by using the posted output from djtools.sync.sync_operations.upload_music. When another user uploads music to the Beatcloud, you may want to generate a Spotify playlist from those tracks so they may be previewed before you decide whether or not to download them.

The second way is by querying subreddit posts. Posts are first checked to see if they are direct links to a Spotify track. If this is not the case, then the post title is parsed in an attempt to interpret it as either 'ARTIST NAME - TRACK TITLE' or 'TRACK TITLE - ARTIST NAME'. These components are then used to search the Spotify API for tracks. The resulting tracks have their title and artist fields compared with the reddit post title and are added to the respective playlist if the Levenshtein similarity passes a threshold.

async_spotify_playlists(config) async

This function updates the contents of one or more Spotify playlists with the posts of one or more subreddits (currently only supports one subreddit per playlist).

Parameters:

Name Type Description Default
config BaseConfig

Configuration object.

 required
Source code in djtools/spotify/playlist_builder.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
async def async_spotify_playlists(config: BaseConfig):
    """This function updates the contents of one or more Spotify playlists with
        the posts of one or more subreddits (currently only supports one
        subreddit per playlist).

    Args:
        config: Configuration object.
    """
    spotify = get_spotify_client(config)
    reddit = get_reddit_client(config)
    playlist_ids = get_playlist_ids()

    praw_cache = {}
    cache_file = Path(__file__).parent / ".praw.cache"
    if cache_file.exists():
        with open(cache_file, mode="r", encoding="utf-8") as _file:
            praw_cache = yaml.load(_file, Loader=yaml.FullLoader) or {}

    tasks = [
        asyncio.create_task(
            get_subreddit_posts(
                spotify,
                reddit,
                subreddit,
                config,
                praw_cache,
            )
        )
        for subreddit in config.spotify.spotify_playlist_subreddits
    ]

    for task in asyncio.as_completed(tasks):
        tracks, subreddit = await task
        playlist_ids = populate_playlist(
            playlist_name=subreddit.name,
            playlist_ids=playlist_ids,
            spotify_username=config.spotify.spotify_username,
            spotify=spotify,
            tracks=tracks,
            playlist_limit=subreddit.limit,
            verbosity=config.verbosity,
        )

    await reddit.close()

    write_playlist_ids(playlist_ids)

    with open(cache_file, mode="w", encoding="utf-8") as _file:
        yaml.dump(praw_cache, _file)

spotify_playlist_from_upload(config)

Generates a Spotify playlist using a Discord webhook output.

If "upload_output", a path to a text file containing the pasted output of the upload_music Discord webhook output, is provided, this is used to generate a Spotify playlist of those uploaded tracks. If this is not provided, then the clipboard contents are used instead.

Parameters:

Name Type Description Default
config BaseConfig

Configuration object.

 required

Raises:

Type Description
RuntimeError

Output from an upload_music Discord webhook must be copied to the system's clipboard
Source code in djtools/spotify/playlist_builder.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def spotify_playlist_from_upload(config: BaseConfig):
    """Generates a Spotify playlist using a Discord webhook output.

    If "upload_output", a path to a text file containing the pasted output of
    the upload_music Discord webhook output, is provided, this is used to
    generate a Spotify playlist of those uploaded tracks. If this is not
    provided, then the clipboard contents are used instead.

    Args:
        config: Configuration object.

    Raises:
        RuntimeError: Output from an upload_music Discord webhook must be
            copied to the system's clipboard
    """
    # Load upload output from the system's clipboard.
    data = pyperclip.paste()
    if not data:
        raise RuntimeError(
            "Generating a Spotify playlist from an upload requires output "
            "from an upload_music Discord webhook to be copied to the "
            "system's clipboard"
        )

    spotify = get_spotify_client(config)
    playlist_ids = get_playlist_ids()

    # Get (track title, artist name) tuples from file uploads.
    user = ""
    files = []
    for line in data.split("\n"):
        if not line.startswith(" "):
            if not user:
                user = line.split("/")[0]
            continue
        file_ = Path(line).stem
        try:
            track, artist = file_.strip().split(" - ")
        except ValueError:
            logger.warning(f"{line.strip()} is not a valid file")
            continue
        if config.sync.artist_first:
            track, artist = artist, track
        files.append((track, artist))
    files = list(filter(lambda x: len(x) == 2, files))

    # Query Spotify for files in upload output.
    threshold = config.spotify.spotify_playlist_fuzz_ratio
    tracks = []
    for title, artist in files:
        query = f"track:{title} artist:{artist}"
        try:
            results = spotify.search(q=query, type="track", limit=50)
        except Exception as exc:
            logger.error(f'Error searching for "{title} - {artist}": {exc}')
            continue

        match, _ = filter_results(spotify, results, threshold, title, artist)
        if match:
            artists = ", ".join([y["name"] for y in match["artists"]])
            logger.info(
                f"Matched {match['name']} - {artists} to {title} - {artist}"
            )
        else:
            logger.warning(f"Could not find a match for {title} - {artist}")
            continue
        tracks.append((match["id"], f'{match["name"]} - {artists}'))

    playlist_ids = populate_playlist(
        playlist_name=f"{user} Uploads",
        playlist_ids=playlist_ids,
        spotify_username=config.spotify.spotify_username,
        spotify=spotify,
        tracks=tracks,
        verbosity=config.verbosity,
    )

    write_playlist_ids(playlist_ids)

spotify_playlists(config)

This function asynchronously updates Spotify playlists.

Parameters:

Name Type Description Default
config BaseConfig

Configuration object.

 required
Source code in djtools/spotify/playlist_builder.py 
180
181
182
183
184
185
186
def spotify_playlists(config: BaseConfig):
    """This function asynchronously updates Spotify playlists.

    Args:
        config: Configuration object.
    """
    asyncio.run(async_spotify_playlists(config))

This module contains helper functions used by the "spotify" module.

filter_results(spotify, results, threshold, title, artist)

Helper function for applying filtering logic to find tracks that match the submission title closely enough.

Parameters:

Name Type Description Default
spotify Spotify

Spotify client.

 required
results List[Dict]

Spotify search results.

 required
threshold float

Minimum Levenshtein distance.

 required
title str

Potential title of a track.

 required
artist str

Potential artist of a track.

 required

Returns:

Type Description
Tuple[Dict[str, Any], float]

Tuple of track object and Levenshtein distance.
Source code in djtools/spotify/helpers.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def filter_results(
    spotify: spotipy.Spotify,
    results: List[Dict],
    threshold: float,
    title: str,
    artist: str,
) -> Tuple[Dict[str, Any], float]:
    """Helper function for applying filtering logic to find tracks that
        match the submission title closely enough.

    Args:
        spotify: Spotify client.
        results: Spotify search results.
        threshold: Minimum Levenshtein distance.
        title: Potential title of a track.
        artist: Potential artist of a track.

    Returns:
        Tuple of track object and Levenshtein distance.
    """
    track, dist = {}, 0.0
    tracks = _filter_tracks(
        results["tracks"]["items"], threshold, title, artist
    )
    while results["tracks"]["next"]:
        try:
            results = spotify.next(results["tracks"])
        except Exception:
            logger.warning(f"Failed to get next tracks for {title, artist}")
            break
        tracks.extend(
            _filter_tracks(
                results["tracks"]["items"], threshold, title, artist
            )
        )

    if tracks:
        track, dist = max(tracks, key=itemgetter(1))

    return track, dist

get_playlist_ids()

Load Spotify playlist names -> IDs lookup.

Returns:

Type Description
Dict[str, str]

Dictionary of Spotify playlist names mapped to playlist IDs.
Source code in djtools/spotify/helpers.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def get_playlist_ids() -> Dict[str, str]:
    """Load Spotify playlist names -> IDs lookup.

    Returns:
        Dictionary of Spotify playlist names mapped to playlist IDs.
    """
    playlist_ids = {}
    ids_path = (
        Path(__file__).parent.parent / "configs" / "spotify_playlists.yaml"
    )
    if ids_path.exists():
        with open(ids_path, mode="r", encoding="utf-8") as _file:
            playlist_ids = (
                yaml.load(_file, Loader=yaml.FullLoader) or playlist_ids
            )

    return playlist_ids

get_reddit_client(config)

Instantiate a Reddit API client.

Parameters:

Name Type Description Default
config BaseConfig

Configuration object.

 required

Returns:

Type Description
Reddit

Reddit API client.
Source code in djtools/spotify/helpers.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def get_reddit_client(config: BaseConfig) -> praw.Reddit:
    """Instantiate a Reddit API client.

    Args:
        config: Configuration object.

    Returns:
        Reddit API client.
    """
    reddit = praw.Reddit(
        client_id=config.spotify.reddit_client_id,
        client_secret=config.spotify.reddit_client_secret,
        user_agent=config.spotify.reddit_user_agent,
        timeout=30,
    )

    return reddit

get_spotify_client(config)

Instantiate a Spotify API client.

Parameters:

Name Type Description Default
config Union[BaseConfig, SpotifyConfig]

Configuration object.

 required

Returns:

Type Description
Spotify

Spotify API client.
Source code in djtools/spotify/helpers.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def get_spotify_client(
    config: Union[BaseConfig, SpotifyConfig],
) -> spotipy.Spotify:
    """Instantiate a Spotify API client.

    Args:
        config: Configuration object.

    Returns:
        Spotify API client.
    """
    try:
        spotify_config = getattr(config, "spotify")
    except AttributeError:
        spotify_config = config

    spotify = spotipy.Spotify(
        auth_manager=SpotifyOAuth(
            client_id=spotify_config.spotify_client_id,
            client_secret=spotify_config.spotify_client_secret,
            redirect_uri=spotify_config.spotify_redirect_uri,
            scope="playlist-modify-public",
            requests_timeout=30,
            cache_handler=spotipy.CacheFileHandler(
                cache_path=Path(__file__).parent / ".spotify.cache"
            ),
        )
    )

    return spotify

get_subreddit_posts(spotify, reddit, subreddit, config, praw_cache) async

Filters the submissions for the provided subreddit and tries to resolve each into a Spotify track until all the submissions are parsed or the track limit has been met.

Parameters:

Name Type Description Default
spotify Spotify

Spotify client.

 required
reddit Reddit

Reddit client.

 required
subreddit SubredditConfig

SubredditConfig object.

 required
config BaseConfig

Configuration object.

 required
praw_cache Dict[str, bool]

Cached praw submissions.

 required

Returns:

Type Description
Tuple[List[Tuple[str]], Dict[str, Union[str, int]]]

List of Spotify track ("id", "name") tuples and SubredditConfig as a dictionary.
Source code in djtools/spotify/helpers.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
async def get_subreddit_posts(
    spotify: spotipy.Spotify,
    reddit: praw.Reddit,
    subreddit: SubredditConfig,
    config: BaseConfig,
    praw_cache: Dict[str, bool],
) -> Tuple[List[Tuple[str]], Dict[str, Union[str, int]]]:
    """Filters the submissions for the provided subreddit and tries to resolve
        each into a Spotify track until all the submissions are parsed or the
        track limit has been met.

    Args:
        spotify: Spotify client.
        reddit: Reddit client.
        subreddit: SubredditConfig object.
        config: Configuration object.
        praw_cache: Cached praw submissions.

    Returns:
        List of Spotify track ("id", "name") tuples and SubredditConfig as a
            dictionary.
    """
    sub = await reddit.subreddit(subreddit.name)
    func = getattr(sub, subreddit.type.value)
    kwargs = {"limit": config.spotify.spotify_playlist_post_limit}
    if subreddit.type == SubredditType.TOP:
        kwargs["time_filter"] = subreddit.period
    subs = [
        x
        async for x in _catch(
            func(**kwargs), message="Failed to retrieve Reddit submission"
        )
    ]
    msg = f'Filtering {len(subs)} "r/{subreddit.name}" {subreddit.type.value} posts'
    logger.info(msg)
    submissions = []
    for submission in tqdm(subs, desc=msg):
        if submission.id in praw_cache:
            continue
        submissions.append(submission)
        praw_cache[submission.id] = True
    new_tracks = []
    if submissions:
        msg = (
            f"Searching Spotify for {len(submissions)} new submission(s) from "
            f'"r/{subreddit.name}"'
        )
        logger.info(msg)
        payload = zip(
            submissions,
            [spotify] * len(submissions),
            [config.spotify.spotify_playlist_fuzz_ratio] * len(submissions),
        )

        with ThreadPoolExecutor(max_workers=8) as executor:
            futures = [executor.submit(_process, *args) for args in payload]

            with tqdm(total=len(futures), desc=msg) as pbar:
                new_tracks = []
                for future in as_completed(futures):
                    new_tracks.append(future.result())
                    pbar.update(1)

        new_tracks = [track for track in new_tracks if track]
        logger.info(
            f"Got {len(new_tracks)} Spotify track(s) from new "
            f'"r/{subreddit.name}" posts'
        )
    else:
        logger.info(f'No new submissions from "r/{subreddit.name}"')

    return new_tracks, subreddit

populate_playlist(playlist_name, playlist_ids, spotify_username, spotify, tracks, playlist_limit=sys.maxsize, verbosity=0)

Inserts tracks into either a new playlist or an existing one.

Parameters:

Name Type Description Default
playlist_name str

Name of the playlist.

 required
playlist_ids Dict[str, str]

Lookup of playlist IDs.

 required
spotify_username str

Spotify user's username.

 required
spotify Spotify

Spotify client.

 required
tracks List[Tuple[str]]

List of tracks.

 required
playlist_limit Optional[int]

Maximum number of tracks allowed in a playlist.

 maxsize
verbosity Optional[int]

Logging verbosity level.

 0

Returns:

Type Description
Dict[str, str]

Updated playlist IDs.
Source code in djtools/spotify/helpers.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
def populate_playlist(
    playlist_name: str,
    playlist_ids: Dict[str, str],
    spotify_username: str,
    spotify: spotipy.Spotify,
    tracks: List[Tuple[str]],
    playlist_limit: Optional[int] = sys.maxsize,
    verbosity: Optional[int] = 0,
) -> Dict[str, str]:
    """Inserts tracks into either a new playlist or an existing one.

    Args:
        playlist_name: Name of the playlist.
        playlist_ids: Lookup of playlist IDs.
        spotify_username: Spotify user's username.
        spotify: Spotify client.
        tracks: List of tracks.
        playlist_limit: Maximum number of tracks allowed in a playlist.
        verbosity: Logging verbosity level.

    Returns:
        Updated playlist IDs.
    """
    playlist_id = playlist_ids.get(playlist_name)
    playlist = None
    if playlist_id and tracks:
        playlist = _update_existing_playlist(
            spotify,
            playlist_id,
            tracks,
            playlist_limit,
            verbosity,
        )
    elif tracks:
        logger.warning(
            f"Unable to get ID for {playlist_name}...creating a new "
            "playlist"
        )
        playlist = _build_new_playlist(
            spotify, spotify_username, playlist_name, tracks
        )
        playlist_ids[playlist_name] = playlist["id"]
    elif playlist_id:
        playlist = spotify.playlist(playlist_id)
    if playlist:
        logger.info(
            f'"{playlist["name"]}": '
            f'{playlist["external_urls"].get("spotify")}'
        )

    return playlist_ids

write_playlist_ids(playlist_ids)

Write playlist IDs to file.

Parameters:

Name Type Description Default
playlist_ids Dict[str, str]

Dictionary of Spotify playlist names mapped to playlist IDs.

 required
Source code in djtools/spotify/helpers.py 
275
276
277
278
279
280
281
282
283
284
285
286
def write_playlist_ids(playlist_ids: Dict[str, str]):
    """Write playlist IDs to file.

    Args:
        playlist_ids: Dictionary of Spotify playlist names mapped to playlist
            IDs.
    """
    ids_path = (
        Path(__file__).parent.parent / "configs" / "spotify_playlists.yaml"
    )
    with open(ids_path, mode="w", encoding="utf-8") as _file:
        yaml.dump(playlist_ids, _file)