-
Notifications
You must be signed in to change notification settings - Fork 1
/
query_functions.py
152 lines (123 loc) · 4.39 KB
/
query_functions.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
import spotipy
from spotipy import SpotifyClientCredentials
from spotipy.exceptions import SpotifyException
from sqlalchemy.exc import IntegrityError
from models import storage
from models.artist import Artist
from models.track import Track
auth_manager = SpotifyClientCredentials()
sp = spotipy.Spotify(auth_manager=auth_manager)
def save_json_to_database(obj):
'''
data in obj are being saved to data base from obj
'''
track = {}
artist = {}
if 'artist_id' in obj:
artist['id'] = obj.get('artist_id')
artist['name'] = obj.get('artist_name')
new_artist = Artist(**artist)
try:
storage.new(new_artist)
storage.save()
except IntegrityError:
storage.rollback()
if 'id' in obj:
track['id'] = obj.get('id')
track['title'] = obj.get('title')
track['preview_url'] = obj.get('preview_url')
track['image_url'] = obj.get('image_url')
track['artist_id'] = obj.get('artist_id')
new_track = Track(**track)
try:
storage.new(new_track)
storage.save()
except IntegrityError:
storage.rollback()
def search(query, type="track", limit=10):
'''
search function uses user search query to find songs using spotify search api
the result is a list of all track details
'''
result = sp.search(query, type=type, limit=limit)
if result and type == "track":
tracks = get_details_from_json(result)
for _ in tracks:
save_json_to_database(_)
return tracks
elif type == "artist":
result = sp.search(query, type=type, limit=limit)
result = result.get('artists').get('items')[0]
artist = {}
artist['artist_name'] = result.get('name')
artist['artist_id'] = result.get('id')
save_json_to_database(artist)
return artist
def convert_name_to_id(query, type="track"):
'''
converts a list of names into their ids
'''
result = []
if not isinstance(query, list):
s_result = sp.search(query, type=type, limit=1)
type_ = type + 's'
s_result = s_result.get(type_).get('items')[0]
s_result = s_result.get('id')
result.append(s_result)
else:
type_ = type + 's'
for _ in query:
s_result = sp.search(_, type=type, limit=1)
s_result = s_result.get(type_).get('items')[0]
s_result = s_result.get('id')
result.append(s_result)
return result
def get_details_from_json(dct):
'''
gather track details from api results
dct should be a dictionaries response from the api
'''
details =[]
tracks = dct.get('tracks')
if type(tracks) == dict:
tracks = tracks.get('items', dct.get('tracks'))
for track in tracks:
track_details = {}
track_details['id'] = track.get('id')
track_details['title'] = track.get('name')
track_details['artist_id'] = track.get('artists')[0].get('id')
track_details['artist_name'] = track.get('artists')[0].get('name')
track_details['preview_url'] = track.get('preview_url')
track_details['image_url'] = track.get('album').get(
'images')[-1].get('url')
details.append(track_details)
return details
def recommendation(**kwargs):
'''
recommendation takes user input and gives back music recommendation
the argument is a key value argument containing 'artists', 'tracks',
'genres' and 'limit'
'''
tracks = kwargs.get('tracks')
artists = kwargs.get('artists')
genres = kwargs.get('genres')
limit = kwargs.get('limit', 10)
convert = kwargs.get('convert')
if convert == 'true':
if tracks:
tracks = convert_name_to_id(tracks, 'track')
if artists:
artists = convert_name_to_id(artists, 'artist')
try:
recommendations = sp.recommendations(
seed_artists=artists, seed_tracks=tracks,
seed_genres=genres, limit=limit, type='track'
)
except SpotifyException as e:
return [{"error": "calling recommendation without artist, genre or track"}, 400]
recommendations = get_details_from_json(recommendations)
for _ in recommendations:
save_json_to_database(_)
# append the response to be used by the make_response function
recommendations.append(200)
return recommendations