Skip to content

Spotify

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 src/djtools/spotify/config.py 
27
28
29
30
31
32
33
34
35
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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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] = Field(
        default_factory=list
    )
    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 exist and 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:
            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 exist and be valid.
RuntimeError

Reddit API credentials must exist.
Source code in src/djtools/spotify/config.py 
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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def __init__(self, *args, **kwargs):
    """Constructor.

    Raises:
        RuntimeError: Spotify API credentials must exist and 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:
        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 src/djtools/spotify/config.py 
18
19
20
21
22
23
24
class SubredditConfig(BaseModel):
    """Configuration object for spotify_playlists."""

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

This module creates or updates Spotify playlists.

Playlists can be created from: 1. Discord webhook output from djtools.sync.sync_operations.upload_music 2. Subreddit posts (directly linked Spotify tracks or fuzzy-matched titles)

async_spotify_playlists(config) async

Updates Spotify playlists from subreddit posts.

Parameters:

Name Type Description Default
config BaseConfig

Configuration object.

 required
Source code in src/djtools/spotify/playlist_builder.py 
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
76
77
78
79
80
81
82
async def async_spotify_playlists(config: BaseConfig):
    """Updates Spotify playlists from subreddit posts.

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

    # Load praw cache
    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 {}

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

    # Process results as they complete
    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()

    # Save state
    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 from Discord webhook output.

Uses clipboard contents containing upload_music webhook output to create a Spotify playlist of those tracks.

Parameters:

Name Type Description Default
config BaseConfig

Configuration object.

 required

Raises:

Type Description
RuntimeError

If clipboard is empty.
Source code in src/djtools/spotify/playlist_builder.py 
 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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def spotify_playlist_from_upload(config: BaseConfig):
    """Generates a Spotify playlist from Discord webhook output.

    Uses clipboard contents containing upload_music webhook output to
    create a Spotify playlist of those tracks.

    Args:
        config: Configuration object.

    Raises:
        RuntimeError: If clipboard is empty.
    """
    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()

    # Parse (track title, artist name) tuples from upload output
    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) == TRACK_ARTIST_TUPLE_LENGTH, files))

    # Search Spotify for each file
    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}"))

    # Populate playlist
    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)

Entry point for async Spotify playlist updates.

Parameters:

Name Type Description Default
config BaseConfig

Configuration object.

 required
Source code in src/djtools/spotify/playlist_builder.py 
163
164
165
166
167
168
169
def spotify_playlists(config: BaseConfig):
    """Entry point for async Spotify playlist updates.

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

This module contains helper functions used by the spotify module.

All Spotify API interactions are delegated to the spotify-tools library. This module provides DJ-Tools specific wrappers and configuration handling.

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

Filter Spotify search results to find best matching track.

This function uses spotify-tools' search_track_fuzzy for matching.

Parameters:

Name Type Description Default
spotify Client

Spotify client.

 required
results Dict

Spotify search results (unused, kept for API compatibility).

 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 (as dict) and similarity score.
Source code in src/djtools/spotify/helpers.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
def filter_results(
    spotify: Client,
    results: Dict,
    threshold: float,
    title: str,
    artist: str,
) -> Tuple[Dict[str, Any], float]:
    """Filter Spotify search results to find best matching track.

    This function uses spotify-tools' search_track_fuzzy for matching.

    Args:
        spotify: Spotify client.
        results: Spotify search results (unused, kept for API compatibility).
        threshold: Minimum Levenshtein distance.
        title: Potential title of a track.
        artist: Potential artist of a track.

    Returns:
        Tuple of track object (as dict) and similarity score.
    """
    # Use spotify-tools search with pagination for thorough matching
    result = search_track_fuzzy(
        spotify, title, artist, threshold=threshold, limit=50
    )

    if result and result.track:
        return result.track.model_dump(), result.score

    return {}, 0.0

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 src/djtools/spotify/helpers.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
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 src/djtools/spotify/helpers.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
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 using spotify-tools.

Parameters:

Name Type Description Default
config Union[BaseConfig, DJToolsSpotifyConfig]

Configuration object.

 required

Returns:

Type Description
Client

Spotify API client.
Source code in src/djtools/spotify/helpers.py 
 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
def get_spotify_client(
    config: Union[BaseConfig, DJToolsSpotifyConfig],
) -> Client:
    """Instantiate a Spotify API client using spotify-tools.

    Args:
        config: Configuration object.

    Returns:
        Spotify API client.
    """
    try:
        spotify_config = config.spotify
    except AttributeError:
        spotify_config = config

    st_config = SpotifyConfig(
        CLIENT_ID=spotify_config.spotify_client_id,
        CLIENT_SECRET=spotify_config.spotify_client_secret,
        REDIRECT_URI=spotify_config.spotify_redirect_uri,
    )

    return Client(
        config=st_config,
        scopes=["playlist-modify-public"],
        cache_path=Path(__file__).parent / ".spotify.cache",
    )

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

Filters subreddit submissions and resolves them to Spotify tracks.

Parameters:

Name Type Description Default
spotify Client

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, str]], Dict[str, Union[str, int]]]

List of Spotify track ("id", "name") tuples and SubredditConfig.
Source code in src/djtools/spotify/helpers.py 
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
178
179
180
181
182
183
184
185
186
187
async def get_subreddit_posts(
    spotify: Client,
    reddit: praw.Reddit,
    subreddit: SubredditConfig,
    config: BaseConfig,
    praw_cache: Dict[str, bool],
) -> Tuple[List[Tuple[str, str]], Dict[str, Union[str, int]]]:
    """Filters subreddit submissions and resolves them to Spotify tracks.

    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.
    """
    from concurrent.futures import ThreadPoolExecutor, as_completed

    from tqdm import tqdm

    sub = await reddit.subreddit(subreddit.name)
    func = getattr(sub, subreddit.type.value)
    kwargs = {"limit": config.spotify.spotify_playlist_post_limit}
    from djtools.spotify.enums import SubredditType

    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: List[Tuple[str, str]] = []
    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),
            strict=True,
        )

        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):
                    result = future.result()
                    if result:
                        new_tracks.append(result)
                    pbar.update(1)

        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=None, 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 Client

Spotify client.

 required
tracks List[Tuple[str, str]]

List of (track_id, track_name) tuples.

 required
playlist_limit Optional[int]

Maximum number of tracks allowed in a playlist.

 None
verbosity int

Logging verbosity level.

 0

Returns:

Type Description
Dict[str, str]

Updated playlist IDs.
Source code in src/djtools/spotify/helpers.py 
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
220
221
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
def populate_playlist(
    playlist_name: str,
    playlist_ids: Dict[str, str],
    spotify_username: str,
    spotify: Client,
    tracks: List[Tuple[str, str]],
    playlist_limit: Optional[int] = None,
    verbosity: 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 (track_id, track_name) tuples.
        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)

    # Convert track tuples to PlaylistTrack objects, resolving URLs if needed
    playlist_tracks = _resolve_tracks(spotify, tracks)

    if playlist_id and playlist_tracks:
        # Update existing playlist using spotify-tools
        result = update_playlist(
            spotify,
            playlist_id,
            playlist_tracks,
            max_size=playlist_limit,
            check_duplicates=True,
            duplicate_threshold=90.0,
            verbosity=verbosity,
        )
        _log_update_result(result, verbosity)
        playlist = get_playlist(spotify, playlist_id)
    elif playlist_tracks:
        # Create new playlist using spotify-tools
        logger.warning(
            f"Unable to get ID for {playlist_name}...creating a new playlist"
        )
        playlist = create_playlist(
            spotify,
            name=playlist_name.title(),
            tracks=playlist_tracks,
            public=True,
            user_id=spotify_username,
        )
        if playlist and playlist.id:
            playlist_ids[playlist_name] = playlist.id
    elif playlist_id:
        playlist = get_playlist(spotify, playlist_id)
    else:
        playlist = None

    if playlist:
        url = ""
        if playlist.external_urls:
            url = playlist.external_urls.spotify or ""
        logger.info(f'"{playlist.name}": {url}')

    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 src/djtools/spotify/helpers.py 
259
260
261
262
263
264
265
266
267
268
269
270
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)