forked from EthanRosenthal/discovered-weekly
-
Notifications
You must be signed in to change notification settings - Fork 0
/
discovered_weekly.py
160 lines (130 loc) · 5.38 KB
/
discovered_weekly.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
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
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
import datetime as dt
import logging
import os
import sys
from typing import List
from dotenv import dotenv_values
import spotipy
from spotipy.oauth2 import SpotifyOAuth
logger = logging.getLogger("discovered-weekly")
handler = logging.StreamHandler(stream=sys.stdout)
handler.setFormatter(logging.Formatter(fmt="%(asctime)s : %(levelname)s : %(message)s"))
logger.addHandler(handler)
logger.setLevel(logging.INFO)
def main():
# Override sample with non-sample file-based env variables,
# and override both with actual env variables
logger.info(os.environ)
config = {**dotenv_values("sample.env"), **dotenv_values(".env"), **os.environ}
logger.info("Start discover weekly archiving")
client = load_client(
config["CLIENT_ID"],
config["CLIENT_SECRET"],
config["REDIRECT_URI"],
config["USERNAME"],
config["REFRESH_TOKEN"],
)
playlist_date, dw_uris = parse_this_week(
client, config["DISCOVER_WEEKLY_PLAYLIST_ID"]
)
logger.info(f"Found this week's playlist for {playlist_date}")
logger.info("Adding to all time playlist")
add_to_all_time_playlist(client, dw_uris, config["ALL_DISCOVERED_PLAYLIST_ID"])
playlist_date, dw_uris = parse_this_week(
client, config["RELEASE_RADAR_PLAYLIST_ID"]
)
logger.info(f"Found this week's release radar playlist for {playlist_date}")
logger.info("Adding to all time playlist")
add_to_all_time_playlist(client, dw_uris, config["ALL_DISCOVERED_PLAYLIST_ID"])
#logger.info("Adding to the weekly archive")
#add_to_weekly_archive(client, config["USERNAME"], playlist_date, dw_uris)
logger.info("Done discover weekly archiving")
def load_client(client_id: str, client_secret: str, redirect_uri: str, username: str, refresh_token: str) -> spotipy.Spotify:
scopes = ["playlist-read-private", "playlist-modify-private"]
# Authenticate
auth_manager = SpotifyOAuth(
scope=scopes,
client_id=client_id,
client_secret=client_secret,
redirect_uri=redirect_uri,
username=username,
)
auth_manager.refresh_access_token(refresh_token)
client = spotipy.Spotify(auth_manager=auth_manager)
return client
def parse_this_week(client: spotipy.Spotify, discover_weekly_playlist_id: str) -> (dt.date, List[str]):
# Grab this week's Discover Weekly (DW) and parse for some info
dw_items = client.playlist_items(discover_weekly_playlist_id)
playlist_created = dt.datetime.strptime(
dw_items["items"][0]["added_at"], "%Y-%m-%dT%H:%M:%S%z"
)
playlist_date = playlist_created.strftime("%Y-%m-%d")
dw_uris = [item["track"]["uri"] for item in dw_items["items"]]
return playlist_date, dw_uris
def add_to_all_time_playlist(client: spotipy.Spotify, dw_uris: List[str], all_discovered_playlist_id: str):
# First, add to the all time DW
# Determine total number of tracks
total = client.playlist(all_discovered_playlist_id)["tracks"]["total"]
# Now, query for the last 5 tracks
offset = max(0, total - 5)
last_five = client.playlist_items(all_discovered_playlist_id, offset=offset, limit=5)
# If the last 5 tracks match the last 5 from the current week, then we've already added
# this week's playlist.
match = len(last_five["items"]) >= 5 and all(
[
dw_uri == item["track"]["uri"]
for dw_uri, item in zip(dw_uris[-5:], last_five["items"])
]
)
if match:
logger.info(
"This script has already been run for this week."
" Skipping add to all time playlist."
)
return
offset = max(0, total - 30 - 5)
last_five = client.playlist_items(all_discovered_playlist_id, offset=offset, limit=5)
# If the last 5 tracks match the last 5 from the current week, then we've already added
# this week's playlist.
match = len(last_five["items"]) >= 5 and all(
[
dw_uri == item["track"]["uri"]
for dw_uri, item in zip(dw_uris[-5:], last_five["items"])
]
)
if match:
logger.info(
"This script has already been run for this week."
" Skipping add to all time playlist."
)
return
client.playlist_add_items(all_discovered_playlist_id, dw_uris)
def add_to_weekly_archive(client: spotipy.Spotify, username: str, playlist_date, dw_uris):
# Second, create the weekly archive playlist
this_weeks_playlist = f"Discovered Week {playlist_date}"
# Need to search all user's playlists to see if this one already exists...
limit = 50
offset = 0
total = 1e9
found = False
while offset + limit < total and not found:
playlists = client.user_playlists(username, limit=limit, offset=offset)
total = playlists["total"]
found = any(
[item["name"] == this_weeks_playlist for item in playlists["items"]]
)
offset += limit
if found:
logger.info(
"This script has already been run for this week."
" Skipping creation of weekly archive playlist."
)
return
logger.info(f"Creating this week's archive playlist: {this_weeks_playlist}")
saved_playlist = client.user_playlist_create(
username, this_weeks_playlist, public=False
)
client.playlist_add_items(saved_playlist["id"], dw_uris)
logger.info("Done creating this week's archive playlist.")
if __name__ == "__main__":
main()