1from settings import SPOTIFY_PROVIDER, SPOTIFY_CLIENT_ID, SPOTIFY_CLIENT_SECRET, SPOTIFY_REDIRECT_URI, SPOTIFY_URL_TOKEN, SPOTIFY_SCOPE, SPOTIFY_USER_URL 

2from urllib.parse import urlencode 

3import base64 

4import requests 

5import dateutil.parser 

6from flask_jwt_extended import create_access_token 

7 

8 

9class Spotify: 

10 @staticmethod 

11 def oauth_url(user): 

12 access_token = create_access_token(identity=user) 

13 params = urlencode({ 

14 'client_id': SPOTIFY_CLIENT_ID, 

15 'scope': SPOTIFY_SCOPE, 

16 'redirect_uri': SPOTIFY_REDIRECT_URI, 

17 'response_type': 'code', 

18 'state': access_token 

19 }) 

20 

21 url = SPOTIFY_PROVIDER + '?' + params 

22 return url 

23 

24 @staticmethod 

25 def get_tokens(code): 

26 payload = { 

27 'grant_type': 'authorization_code', 

28 'code': code, 

29 'redirect_uri': SPOTIFY_REDIRECT_URI 

30 } 

31 auth_header = base64.b64encode( 

32 (SPOTIFY_CLIENT_ID + ':' + SPOTIFY_CLIENT_SECRET).encode('ascii')) 

33 headers = {'Authorization': 'Basic %s' % auth_header.decode('ascii')} 

34 

35 response = requests.post( 

36 SPOTIFY_URL_TOKEN, data=payload, headers=headers) 

37 

38 return response.json() 

39 

40 @staticmethod 

41 def refresh_token(refresh_token): 

42 

43 payload = { 

44 'refresh_token': refresh_token, 

45 'grant_type': 'refresh_token' 

46 } 

47 

48 auth_header = base64.b64encode( 

49 (SPOTIFY_CLIENT_ID + ':' + SPOTIFY_CLIENT_SECRET).encode('ascii')) 

50 headers = {'Authorization': 'Basic %s' % auth_header.decode('ascii')} 

51 

52 response = requests.post( 

53 SPOTIFY_URL_TOKEN, data=payload, headers=headers) 

54 

55 return response.json() 

56 

57 """ @staticmethod 

58 def get_user_top_artist(access_token): 

59 limit = 20 

60 access_token = access_token # your function to pull the token 

61 # from the place where you saved it 

62 items_list = [] 

63 url = SPOTIFY_USER_URL + f"top/artists?limit={limit}" 

64 headers = {'Authorization': "Bearer "+access_token} 

65 all_loaded = False 

66 while not all_loaded: 

67 response = requests.get(url, headers=headers).json() 

68 items = response['items'] 

69 for item in items: 

70 elements = {} 

71 elements['genres'] = item['genres'] 

72 elements['artist_name'] = item['name'] 

73 items_list.append(elements) 

74 if response['next']: 

75 url = response['next'] # + f"&access_token={access_token}" 

76 else: 

77 all_loaded = True 

78 return items_list 

79 """ 

80 @staticmethod 

81 def get_user_top_track(access_token): 

82 limit = 20 

83 

84 items_list = [] 

85 url = SPOTIFY_USER_URL + f"top/tracks?limit={limit}" 

86 headers = {'Authorization': "Bearer "+access_token} 

87 all_loaded = False 

88 while not all_loaded: 

89 response = requests.get(url, headers=headers).json() 

90 items = response['items'] 

91 for item in items: 

92 elements = {} 

93 elements['artist_name'] = [] 

94 elements['release'] = item['album']['name'] 

95 if isinstance(item['album']['images'], list) and len(item['album']['images']) > 0: 

96 elements['cover_art_url'] = item['album']['images'][0]["url"] if (isinstance( 

97 item['album']['images'], list) and len(item['album']['images']) > 0) else None 

98 else: 

99 elements['cover_art_url'] = None 

100 elements['year'] = item['album']['release_date'].split( 

101 "-")[0] if item['album']['release_date'] is not None else None 

102 for artist in item['artists']: 

103 elements['artist_name'].append(artist['name']) 

104 elements['artist_name'] = " & ".join(elements['artist_name']) 

105 elements['title'] = item['name'] 

106 elements['spotify_id'] = item['id'] 

107 elements['played_at'] = None 

108 items_list.append(elements) 

109 if response['next']: 

110 url = response['next'] 

111 else: 

112 all_loaded = True 

113 return items_list 

114 

115 @staticmethod 

116 def get_user_recently_played(access_token): 

117 limit = 20 

118 items_list = [] 

119 url = SPOTIFY_USER_URL + f"player/recently-played?limit={limit}" 

120 headers = {'Authorization': "Bearer "+access_token} 

121 all_loaded = False 

122 while not all_loaded: 

123 response = requests.get(url, headers=headers).json() 

124 items = response['items'] 

125 for item in items: 

126 elements = {} 

127 elements['artist_name'] = [] 

128 elements['title'] = item['track']['name'] 

129 elements['spotify_id'] = item['track']['id'] 

130 elements['release'] = item['track']['album']['name'] 

131 elements['year'] = item['track']['album']['release_date'].split( 

132 "-")[0] if item['track']['album']['release_date'] is not None else None 

133 elements['played_at'] = dateutil.parser.isoparse( 

134 item['played_at'][:-1]) if "played_at" in item.keys() else None 

135 for artist in item['track']['artists']: 

136 elements['artist_name'].append(artist['name']) 

137 elements['artist_name'] = " & ".join(elements['artist_name']) 

138 url_cover = requests.get( 

139 item['track']['href'], headers=headers).json() 

140 elements['cover_art_url'] = url_cover['album']['images'][0]['url'] if (isinstance( 

141 url_cover['album']['images'], list) and len(url_cover['album']['images']) > 0) else None 

142 items_list.append(elements) 

143 if response['next']: 

144 url = response['next'] 

145 else: 

146 all_loaded = True 

147 return items_list 

148 

149 @staticmethod 

150 def get_user_playlist(access_token): 

151 limit = 20 

152 items_list = [] 

153 url = SPOTIFY_USER_URL + f"playlists?limit={limit}" 

154 headers = {'Authorization': "Bearer "+access_token} 

155 all_loaded = False 

156 while not all_loaded: 

157 response = requests.get(url, headers=headers).json() 

158 items = response['items'] 

159 for item in items: 

160 elements = {} 

161 elements['tracks'] = [] 

162 url_playlist = requests.get( 

163 item['href'], headers=headers).json() 

164 elements['cover_art_url'] = url_playlist['images'][0]['url'] if (isinstance( 

165 url_playlist['images'], list) and len(url_playlist['images']) > 0) else None 

166 elements['playlist_name'] = url_playlist['name'] 

167 for track in url_playlist["tracks"]["items"]: 

168 tk = {} 

169 tk['artist_name'] = [] 

170 tk['release'] = track['track']['album']['name'] 

171 tk['cover_art_url'] = track['track']['album']['images'][0]["url"] if (isinstance( 

172 track['track']['album']['images'], list) and len(track['track']['album']['images']) > 0) else None 

173 tk['year'] = track['track']['album']['release_date'].split( 

174 "-")[0] if track['track']['album']['release_date'] is not None else None 

175 for artist in track['track']['artists']: 

176 tk['artist_name'].append(artist['name']) 

177 tk['artist_name'] = " & ".join(tk['artist_name']) 

178 tk['title'] = track['track']['name'] 

179 tk['spotify_id'] = track['track']['id'] 

180 tk['played_at'] = None 

181 elements['tracks'].append(tk) 

182 items_list.append(elements) 

183 if response['next']: 

184 url = response['next'] 

185 else: 

186 all_loaded = True 

187 return items_list