EmTpro01 commited on
Commit
86cd934
·
verified ·
1 Parent(s): afa1a06

Update app/main.py

Browse files
Files changed (1) hide show
  1. app/main.py +178 -169
app/main.py CHANGED
@@ -1,170 +1,179 @@
1
- from fastapi import FastAPI, Query
2
- from fastapi.middleware.cors import CORSMiddleware
3
- from fastapi.staticfiles import StaticFiles
4
- from fastapi.responses import FileResponse
5
- from typing import List, Optional
6
- import pandas as pd
7
- import joblib
8
- from scipy.spatial.distance import cdist
9
- from .models.schemas import Song, RecommendationWithPreview
10
- from .api.itunes import search_itunes_tracks
11
-
12
- app = FastAPI(title="Music Recommendation API")
13
-
14
- app.add_middleware(
15
- CORSMiddleware,
16
- allow_origins=["*"],
17
- allow_credentials=True,
18
- allow_methods=["*"],
19
- allow_headers=["*"],
20
- )
21
-
22
- # Mount static files (frontend)
23
- app.mount("/static", StaticFiles(directory="static"), name="static")
24
-
25
- # Load data and model
26
- numeric_features = ['acousticness', 'danceability', 'energy', 'instrumentalness',
27
- 'liveness', 'loudness', 'speechiness', 'tempo', 'valence',
28
- 'popularity', 'year', 'cluster_label']
29
-
30
- model = joblib.load('data/song_cluster_pipeline.joblib')
31
- df = pd.read_csv('data/processed_songs.csv', dtype={col: float for col in numeric_features})
32
- df['artists'] = df['artists'].apply(eval)
33
-
34
- # Serve frontend at root
35
- @app.get("/")
36
- async def read_root():
37
- return FileResponse('static/index.html')
38
-
39
- @app.get("/search/", response_model=List[Song])
40
- async def search_songs(q: str = Query(..., min_length=1), limit: int = 5):
41
- q = q.lower()
42
-
43
- # Perform separate searches
44
- name_matches = df[df['name'].str.lower().str.contains(q, na=False)]
45
- artist_matches = df[df['artists'].apply(lambda x: any(q in artist.lower() for artist in x))]
46
-
47
- # Convert the artists lists to strings for deduplication
48
- name_matches = name_matches.copy()
49
- artist_matches = artist_matches.copy()
50
-
51
- name_matches['artists_str'] = name_matches['artists'].apply(lambda x: ','.join(sorted(x)))
52
- artist_matches['artists_str'] = artist_matches['artists'].apply(lambda x: ','.join(sorted(x)))
53
-
54
- # Concatenate and drop duplicates based on name and artists_str
55
- results = pd.concat([name_matches, artist_matches])
56
- results = results.drop_duplicates(subset=['name', 'artists_str'])
57
-
58
- # Get top matches by popularity
59
- top_matches = results.nlargest(limit, 'popularity')
60
-
61
- return [
62
- Song(
63
- name=row['name'],
64
- artists=row['artists'],
65
- year=int(row['year']),
66
- popularity=int(row['popularity'])
67
- )
68
- for _, row in top_matches.iterrows()
69
- ]
70
-
71
- @app.get("/recommendations/", response_model=List[RecommendationWithPreview])
72
- async def get_recommendations(song_name: str, artist_name: Optional[str] = None, number_songs: int = 6):
73
- try:
74
- if artist_name:
75
- mask = (df['name'].str.lower() == song_name.lower()) & \
76
- (df['artists'].apply(lambda x: artist_name.lower() in str(x).lower()))
77
- song = df[mask].iloc[0]
78
- else:
79
- matches = df[df['name'].str.lower() == song_name.lower()]
80
- if len(matches) > 1:
81
- return {"error": f"Multiple songs found with name '{song_name}'. Please specify an artist."}
82
- song = matches.iloc[0]
83
-
84
- cluster_label = song['cluster_label']
85
- cluster_songs = df[df['cluster_label'] == cluster_label]
86
- cluster_songs = cluster_songs[cluster_songs['name'] != song_name]
87
-
88
- audio_features = ['acousticness', 'danceability', 'energy', 'instrumentalness',
89
- 'liveness', 'loudness', 'speechiness', 'tempo', 'valence']
90
-
91
- song_features = song[audio_features].astype(float).values.reshape(1, -1)
92
- cluster_features = cluster_songs[audio_features].astype(float).values
93
-
94
- distances = cdist(song_features, cluster_features, metric='euclidean')
95
- closest_indices = distances.argsort()[0][:number_songs]
96
-
97
- recommendations = cluster_songs.iloc[closest_indices]
98
-
99
- result = []
100
- for _, row in recommendations.iterrows():
101
- # Create search query for iTunes
102
- search_query = f"{row['name']} {row['artists'][0]}"
103
- preview_info = await search_itunes_tracks(search_query)
104
-
105
- rec = RecommendationWithPreview(
106
- name=row['name'],
107
- artists=row['artists'],
108
- year=int(row['year']),
109
- popularity=int(row['popularity']),
110
- danceability=float(row['danceability']),
111
- energy=float(row['energy']),
112
- valence=float(row['valence']),
113
- preview_info=preview_info
114
- )
115
- result.append(rec)
116
-
117
- return result
118
-
119
- except IndexError:
120
- return {"error": f"Song '{song_name}' {'by ' + artist_name if artist_name else ''} not found."}
121
-
122
- @app.get("/song_details/")
123
- async def get_song_details(song_name: str, artist_name: Optional[str] = None):
124
- """
125
- Get both song data and iTunes preview info for a specific song
126
- """
127
- try:
128
- # Find the song in our dataset
129
- if artist_name:
130
- mask = (df['name'].str.lower() == song_name.lower()) & \
131
- (df['artists'].apply(lambda x: artist_name.lower() in str(x).lower()))
132
- song = df[mask].iloc[0]
133
- else:
134
- matches = df[df['name'].str.lower() == song_name.lower()]
135
- if len(matches) > 1:
136
- return {"error": f"Multiple songs found with name '{song_name}'. Please specify an artist."}
137
- song = matches.iloc[0]
138
-
139
- # Get iTunes preview info
140
- search_query = f"{song_name} {artist_name if artist_name else song['artists'][0]}"
141
- preview_info = await search_itunes_tracks(search_query)
142
-
143
- # Return flattened response
144
- return {
145
- "name": song['name'],
146
- "artists": song['artists'],
147
- "year": int(song['year']),
148
- "popularity": int(song['popularity']),
149
- "danceability": float(song['danceability']),
150
- "energy": float(song['energy']),
151
- "valence": float(song['valence']),
152
- "acousticness": float(song['acousticness']),
153
- "instrumentalness": float(song['instrumentalness']),
154
- "liveness": float(song['liveness']),
155
- "speechiness": float(song['speechiness']),
156
- "tempo": float(song['tempo']),
157
- "preview_info": preview_info
158
- }
159
-
160
- except IndexError:
161
- return {"error": f"Song '{song_name}' {'by ' + artist_name if artist_name else ''} not found."}
162
-
163
- @app.get("/health")
164
- @app.head("/health")
165
- async def health_check():
166
- return {"status": "ok"}
167
-
168
- if __name__ == "__main__":
169
- import uvicorn
 
 
 
 
 
 
 
 
 
170
  uvicorn.run(app, host="0.0.0.0", port=7860)
 
1
+ from fastapi import FastAPI, Query
2
+ from fastapi.middleware.cors import CORSMiddleware
3
+ from fastapi.staticfiles import StaticFiles
4
+ from fastapi.responses import FileResponse
5
+ from typing import List, Optional
6
+ import pandas as pd
7
+ import joblib
8
+ from scipy.spatial.distance import cdist
9
+ from .models.schemas import Song, RecommendationWithPreview
10
+ from .api.itunes import search_itunes_tracks
11
+
12
+ app = FastAPI(title="Music Recommendation API")
13
+
14
+ app.add_middleware(
15
+ CORSMiddleware,
16
+ allow_origins=["*"],
17
+ allow_credentials=True,
18
+ allow_methods=["*"],
19
+ allow_headers=["*"],
20
+ )
21
+
22
+ # Mount static files BEFORE other routes
23
+ app.mount("/static", StaticFiles(directory="static"), name="static")
24
+
25
+ # Load data and model
26
+ numeric_features = ['acousticness', 'danceability', 'energy', 'instrumentalness',
27
+ 'liveness', 'loudness', 'speechiness', 'tempo', 'valence',
28
+ 'popularity', 'year', 'cluster_label']
29
+
30
+ model = joblib.load('data/song_cluster_pipeline.joblib')
31
+ df = pd.read_csv('data/processed_songs.csv', dtype={col: float for col in numeric_features})
32
+ df['artists'] = df['artists'].apply(eval)
33
+
34
+ # Serve individual static files at root level for compatibility
35
+ @app.get("/script.js")
36
+ async def get_script():
37
+ return FileResponse('static/script.js', media_type='application/javascript')
38
+
39
+ @app.get("/styles.css")
40
+ async def get_styles():
41
+ return FileResponse('static/styles.css', media_type='text/css')
42
+
43
+ # Serve frontend at root
44
+ @app.get("/")
45
+ async def read_root():
46
+ return FileResponse('static/index.html')
47
+
48
+ @app.get("/search/", response_model=List[Song])
49
+ async def search_songs(q: str = Query(..., min_length=1), limit: int = 5):
50
+ q = q.lower()
51
+
52
+ # Perform separate searches
53
+ name_matches = df[df['name'].str.lower().str.contains(q, na=False)]
54
+ artist_matches = df[df['artists'].apply(lambda x: any(q in artist.lower() for artist in x))]
55
+
56
+ # Convert the artists lists to strings for deduplication
57
+ name_matches = name_matches.copy()
58
+ artist_matches = artist_matches.copy()
59
+
60
+ name_matches['artists_str'] = name_matches['artists'].apply(lambda x: ','.join(sorted(x)))
61
+ artist_matches['artists_str'] = artist_matches['artists'].apply(lambda x: ','.join(sorted(x)))
62
+
63
+ # Concatenate and drop duplicates based on name and artists_str
64
+ results = pd.concat([name_matches, artist_matches])
65
+ results = results.drop_duplicates(subset=['name', 'artists_str'])
66
+
67
+ # Get top matches by popularity
68
+ top_matches = results.nlargest(limit, 'popularity')
69
+
70
+ return [
71
+ Song(
72
+ name=row['name'],
73
+ artists=row['artists'],
74
+ year=int(row['year']),
75
+ popularity=int(row['popularity'])
76
+ )
77
+ for _, row in top_matches.iterrows()
78
+ ]
79
+
80
+ @app.get("/recommendations/", response_model=List[RecommendationWithPreview])
81
+ async def get_recommendations(song_name: str, artist_name: Optional[str] = None, number_songs: int = 6):
82
+ try:
83
+ if artist_name:
84
+ mask = (df['name'].str.lower() == song_name.lower()) & \
85
+ (df['artists'].apply(lambda x: artist_name.lower() in str(x).lower()))
86
+ song = df[mask].iloc[0]
87
+ else:
88
+ matches = df[df['name'].str.lower() == song_name.lower()]
89
+ if len(matches) > 1:
90
+ return {"error": f"Multiple songs found with name '{song_name}'. Please specify an artist."}
91
+ song = matches.iloc[0]
92
+
93
+ cluster_label = song['cluster_label']
94
+ cluster_songs = df[df['cluster_label'] == cluster_label]
95
+ cluster_songs = cluster_songs[cluster_songs['name'] != song_name]
96
+
97
+ audio_features = ['acousticness', 'danceability', 'energy', 'instrumentalness',
98
+ 'liveness', 'loudness', 'speechiness', 'tempo', 'valence']
99
+
100
+ song_features = song[audio_features].astype(float).values.reshape(1, -1)
101
+ cluster_features = cluster_songs[audio_features].astype(float).values
102
+
103
+ distances = cdist(song_features, cluster_features, metric='euclidean')
104
+ closest_indices = distances.argsort()[0][:number_songs]
105
+
106
+ recommendations = cluster_songs.iloc[closest_indices]
107
+
108
+ result = []
109
+ for _, row in recommendations.iterrows():
110
+ # Create search query for iTunes
111
+ search_query = f"{row['name']} {row['artists'][0]}"
112
+ preview_info = await search_itunes_tracks(search_query)
113
+
114
+ rec = RecommendationWithPreview(
115
+ name=row['name'],
116
+ artists=row['artists'],
117
+ year=int(row['year']),
118
+ popularity=int(row['popularity']),
119
+ danceability=float(row['danceability']),
120
+ energy=float(row['energy']),
121
+ valence=float(row['valence']),
122
+ preview_info=preview_info
123
+ )
124
+ result.append(rec)
125
+
126
+ return result
127
+
128
+ except IndexError:
129
+ return {"error": f"Song '{song_name}' {'by ' + artist_name if artist_name else ''} not found."}
130
+
131
+ @app.get("/song_details/")
132
+ async def get_song_details(song_name: str, artist_name: Optional[str] = None):
133
+ """
134
+ Get both song data and iTunes preview info for a specific song
135
+ """
136
+ try:
137
+ # Find the song in our dataset
138
+ if artist_name:
139
+ mask = (df['name'].str.lower() == song_name.lower()) & \
140
+ (df['artists'].apply(lambda x: artist_name.lower() in str(x).lower()))
141
+ song = df[mask].iloc[0]
142
+ else:
143
+ matches = df[df['name'].str.lower() == song_name.lower()]
144
+ if len(matches) > 1:
145
+ return {"error": f"Multiple songs found with name '{song_name}'. Please specify an artist."}
146
+ song = matches.iloc[0]
147
+
148
+ # Get iTunes preview info
149
+ search_query = f"{song_name} {artist_name if artist_name else song['artists'][0]}"
150
+ preview_info = await search_itunes_tracks(search_query)
151
+
152
+ # Return flattened response
153
+ return {
154
+ "name": song['name'],
155
+ "artists": song['artists'],
156
+ "year": int(song['year']),
157
+ "popularity": int(song['popularity']),
158
+ "danceability": float(song['danceability']),
159
+ "energy": float(song['energy']),
160
+ "valence": float(song['valence']),
161
+ "acousticness": float(song['acousticness']),
162
+ "instrumentalness": float(song['instrumentalness']),
163
+ "liveness": float(song['liveness']),
164
+ "speechiness": float(song['speechiness']),
165
+ "tempo": float(song['tempo']),
166
+ "preview_info": preview_info
167
+ }
168
+
169
+ except IndexError:
170
+ return {"error": f"Song '{song_name}' {'by ' + artist_name if artist_name else ''} not found."}
171
+
172
+ @app.get("/health")
173
+ @app.head("/health")
174
+ async def health_check():
175
+ return {"status": "ok"}
176
+
177
+ if __name__ == "__main__":
178
+ import uvicorn
179
  uvicorn.run(app, host="0.0.0.0", port=7860)