Update app.py

app.py

@@ -2,40 +2,24 @@ import gradio as gr
 import requests
 import pandas as pd
 import time
+import shutil
+import numpy as np
 from tempfile import NamedTemporaryFile
+from sklearn.neighbors import NearestNeighbors
 from openpyxl import Workbook
-import shutil
 
 # Lista de credenciales de API de Spotify
-client_ids =[
+client_ids = [
     '9dfc90506fa04938b05d7913f8b13bad',
     '0ef8df0b57864accb36251fb0b741935',
     '066bb9a3e0ac40aba89732f9a97249bf',
-    'cfbe3754c86048d1a82542a5ab432b9a',
-    '37492f29c7fe478391e8fd8fe66f3f1b',
-    'ac3c1dd6252740e68aeb0d7dd4e6e37d',
-    '807008cb2ce041178c1871973fc81716',
-    '107286416613436ab5c63b56d3564f9e',
-    '528d1d6733ba46eb8e9532e56ced61eb',
-    'aed6435c51484dc18b8c18aa74dd51f8',
-    '9df51caba5d247dc921b21de35a47c44',
-    '191227c66e0d4be692bc8ee73ea6eb3d',
-    'e272c0705c7c4fd68937c58adaa446ed'
- ]   
+    'cfbe3754c86048d1a82542a5ab432b9a'
+]
 client_secrets = [
     '62a96c883b564f1e8c9f3af935f9f88e',
     '8a1bb6f0f8f14feb9be2dff4b603bb5f',
     '9cdae5b56ec24aed91bb3958823ff39e',
-    'c55801afd5c24df3b8673cc07468c7b6',
-    'b61fa17afff64c8693ea5147b79562be',
-    'ff49073ebd324e8cac2d997a8e34644c',
-    'b6c3fbe2304145e4b268f05eefd6ab2a',
-    '4de4c14ed109420c9517c083b8018f8c',
-    '3dbdcf9a12634cfca9e6aca8d3093020',
-    '8662b33d594f4d198ea025d4aa9f0b98',
-    '0e39502ec7e74fe99bb74245678d5f0d',
-    '2d2a895d85874c088897dd9894dc64ad',
-    '9fdfa58ea0a94ce7a0cb34fa19fb7d74'
+    'c55801afd5c24df3b8673cc07468c7b6'
 ]
 current_api_index = 0
 
@@ -44,7 +28,6 @@ solicitudes_totales = 0
 
 # Funciones para Spotify
 def obtener_token(client_id, client_secret):
-    print(f"Obteniendo token de Spotify con client_id: {client_id}")
     url = 'https://accounts.spotify.com/api/token'
     headers = {'Content-Type': 'application/x-www-form-urlencoded'}
     payload = {'grant_type': 'client_credentials'}
@@ -52,23 +35,14 @@ def obtener_token(client_id, client_secret):
     global solicitudes_totales
     solicitudes_totales += 1  # Contando solicitud
     if response.status_code == 200:
-        print(f"Token obtenido exitosamente para client_id: {client_id}")
         return response.json().get('access_token')
     else:
-        print(f"Error al obtener token para client_id: {client_id}, status_code: {response.status_code}, response: {response.text}")
         return None
 
-def cambiar_api_key():
-    global current_api_index
-    current_api_index = (current_api_index + 1) % len(client_ids)
-    print(f"Cambiando a la siguiente API Key, índice actual: {current_api_index}")
-    return obtener_token(client_ids[current_api_index], client_secrets[current_api_index])
-
 def manejar_rate_limit(response, intento):
     if response.status_code == 429:
         retry_after = int(response.headers.get('Retry-After', 1))
         wait_time = retry_after + 10 * (2 ** intento)  # Retroceso exponencial
-        print(f"Límite de peticiones alcanzado, Retry-After: {retry_after} segundos. Esperando {wait_time} segundos.")
         time.sleep(wait_time)
         return True
     return False
@@ -83,81 +57,14 @@ def hacer_request_con_reintento(url, headers, params=None, max_retries=5):
         if response.status_code == 200:
             return response
         else:
-            print(f"Error en la solicitud: {response.status_code}, response: {response.text}")
             break
     return None
 
-def buscar_playlists_spotify(token, query, limit=50):
-    print("Buscando playlists en Spotify...")
-    url = 'https://api.spotify.com/v1/search'
-    headers = {'Authorization': f'Bearer {token}'}
-    playlists = []
-
-    if limit <= 50:
-        params = {'q': query, 'type': 'playlist', 'limit': limit}
-        response = hacer_request_con_reintento(url, headers, params)
-        if response:
-            playlists.extend(response.json().get('playlists', {}).get('items', []))
-    else:
-        offset = 0
-        while limit > 0:
-            params = {'q': query, 'type': 'playlist', 'limit': min(50, limit), 'offset': offset}
-            response = hacer_request_con_reintento(url, headers, params)
-            if response:
-                playlists.extend(response.json().get('playlists', {}).get('items', []))
-                limit -= min(50, limit)
-                offset += 50
-                time.sleep(1)  # Pausa de 1 segundo entre las solicitudes para evitar el límite de tasa
-            else:
-                break
-    
-    return [{'playlist_id': playlist['id'], 'playlist_name': playlist['name']} for playlist in playlists]
-
-def obtener_canciones_playlist_spotify(token, playlist_id, playlist_name):
-    print(f"Obteniendo canciones de la playlist {playlist_id} ({playlist_name}) de Spotify...")
-    url = f'https://api.spotify.com/v1/playlists/{playlist_id}/tracks'
-    headers = {'Authorization': f'Bearer {token}'}
-    canciones = []
-
-    response = hacer_request_con_reintento(url, headers)
-    if response:
-        tracks = response.json().get('items')
-        track_ids = [item['track']['id'] for item in tracks if item['track'] and item['track']['id']]
-        album_ids = [item['track']['album']['id'] for item in tracks if item['track'] and item['track']['album']['id']]
-
-        # Obtener características de audio en batch
-        audio_features = obtener_caracteristicas_audio_batch(token, track_ids)
-        # Obtener información de álbum en batch
-        album_info = obtener_record_label_spotify_batch(token, album_ids)
-
-        for item in tracks:
-            track = item['track']
-            if track:
-                audio = audio_features.get(track['id'], {})
-                album = album_info.get(track['album']['id'], {})
-                canciones.append({
-                    'playlist_name': playlist_name,
-                    'artista': track['artists'][0]['name'] if track['artists'] else 'Desconocido',
-                    'titulo': track['name'],
-                    'isrc': track['external_ids'].get('isrc', 'No disponible'),
-                    'popularity': track.get('popularity', 'No disponible'),
-                    'valence': audio.get('valence', 'No disponible'),
-                    'danceability': audio.get('danceability', 'No disponible'),
-                    'energy': audio.get('energy', 'No disponible'),
-                    'tempo': audio.get('tempo', 'No disponible'),
-                    'speechiness': audio.get('speechiness', 'No disponible'),
-                    'instrumentalness': audio.get('instrumentalness', 'No disponible'),
-                    'duration': track.get('duration_ms', 'No disponible'),
-                    'release_year': track.get('album', {}).get('release_date', 'No disponible').split('-')[0] if track.get('album', {}).get('release_date') else 'No disponible',
-                    'record_label': album.get('label', 'No disponible')
-                })
-    return canciones
-
-def obtener_caracteristicas_audio_batch(token, track_ids):
+def obtener_caracteristicas_audio(token, track_ids):
     audio_features = {}
     url = 'https://api.spotify.com/v1/audio-features'
     headers = {'Authorization': f'Bearer {token}'}
-
+    
     for i in range(0, len(track_ids), 100):
         batch_ids = track_ids[i:i+100]
         params = {'ids': ','.join(batch_ids)}
@@ -168,44 +75,77 @@ def obtener_caracteristicas_audio_batch(token, track_ids):
                     audio_features[feature['id']] = feature
     return audio_features
 
-def obtener_record_label_spotify_batch(token, album_ids):
-    album_info = {}
-    url = 'https://api.spotify.com/v1/albums'
+def obtener_tracks_y_caracteristicas(token, url):
+    headers = {'Authorization': f'Bearer {token}'}
+    if "track" in url:
+        track_id = url.split("/")[-1].split("?")[0]
+        track_ids = [track_id]
+    elif "playlist" in url:
+        playlist_id = url.split("/")[-1].split("?")[0]
+        tracks_url = f'https://api.spotify.com/v1/playlists/{playlist_id}/tracks'
+        response = hacer_request_con_reintento(tracks_url, headers)
+        if response:
+            track_ids = [item['track']['id'] for item in response.json().get('items', []) if item['track']]
+    else:
+        return None, None
+
+    audio_features = obtener_caracteristicas_audio(token, track_ids)
+    return track_ids, audio_features
+
+def buscar_canciones_similares(token, audio_features, n_songs=10):
+    features_matrix = np.array([
+        [v['danceability'], v['energy'], v['tempo'], v['valence']] for v in audio_features.values()
+    ])
+    neighbors = NearestNeighbors(n_neighbors=n_songs, algorithm='ball_tree').fit(features_matrix)
+    distances, indices = neighbors.kneighbors(features_matrix)
+
+    similar_tracks = []
+    for idx_list in indices:
+        for idx in idx_list:
+            similar_tracks.append(list(audio_features.keys())[idx])
+    
+    # Eliminar duplicados
+    similar_tracks = list(set(similar_tracks))
+    
+    # Limitar a las n_songs canciones más similares
+    similar_tracks = similar_tracks[:n_songs]
+
+    return obtener_informacion_canciones(token, similar_tracks)
+
+def obtener_informacion_canciones(token, track_ids):
+    tracks_info = []
+    url = 'https://api.spotify.com/v1/tracks'
     headers = {'Authorization': f'Bearer {token}'}
 
-    for i in range(0, len(album_ids), 20):
-        batch_ids = album_ids[i:i+20]
+    for i in range(0, len(track_ids), 50):
+        batch_ids = track_ids[i:i+50]
         params = {'ids': ','.join(batch_ids)}
         response = hacer_request_con_reintento(url, headers, params)
         if response:
-            for album in response.json().get('albums', []):
-                if album:
-                    album_info[album['id']] = {
-                        'label': album.get('label', 'No disponible'),
-                        'release_date': album.get('release_date', 'No disponible'),
-                        'total_tracks': album.get('total_tracks', 'No disponible'),
-                        'name': album.get('name', 'No disponible'),
-                        'artists': [artist['name'] for artist in album.get('artists', [])]
-                    }
-    return album_info
+            tracks = response.json().get('tracks', [])
+            for track in tracks:
+                tracks_info.append({
+                    'artista': track['artists'][0]['name'] if track['artists'] else 'Desconocido',
+                    'titulo': track['name'],
+                    'isrc': track['external_ids'].get('isrc', 'No disponible'),
+                    'popularity': track.get('popularity', 'No disponible'),
+                    'release_year': track.get('album', {}).get('release_date', 'No disponible').split('-')[0] if track.get('album', {}).get('release_date') else 'No disponible',
+                    'duration': track.get('duration_ms', 'No disponible'),
+                    'url': track['external_urls']['spotify']
+                })
+    return tracks_info
 
 # Función principal de la interfaz
-def interface(project_name, query, num_spotify_playlists=50):
-    # Obtener tokens y claves
+def interface(project_name, spotify_url, num_similar_songs=10):
     token_spotify = obtener_token(client_ids[current_api_index], client_secrets[current_api_index])
-    playlists_spotify = buscar_playlists_spotify(token_spotify, query, num_spotify_playlists)
-    canciones_spotify = []
-    for playlist in playlists_spotify:
-        songs = obtener_canciones_playlist_spotify(token_spotify, playlist['playlist_id'], playlist['playlist_name'])
-        canciones_spotify.extend(songs)
-        time.sleep(1)  # Pausa de 1 segundo entre la obtención de canciones para evitar el límite de tasa
+    track_ids, audio_features = obtener_tracks_y_caracteristicas(token_spotify, spotify_url)
+    if not track_ids or not audio_features:
+        return "URL no válida o no se encontraron canciones.", None
 
+    similar_tracks_info = buscar_canciones_similares(token_spotify, audio_features, num_similar_songs)
+    
     # Crear DataFrame
-    df = pd.DataFrame(canciones_spotify)
-    df.rename(columns={'isrc': 'ISRCs'}, inplace=True)
-
-    # Ordenar por popularidad
-    df.sort_values(by=['popularity'], ascending=False, inplace=True)
+    df = pd.DataFrame(similar_tracks_info)
 
     # Guardar DataFrame en un archivo Excel
     tmpfile = NamedTemporaryFile(delete=False, suffix='.xlsx')
@@ -222,11 +162,11 @@ iface = gr.Interface(
     fn=interface,
     inputs=[
         gr.Textbox(label="Nombre del Proyecto"),
-        gr.Textbox(label="Keywords - Palabras Clave para tu búsqueda"),
-        gr.Number(label="Numero de Playlists que vamos a buscar con estas Keywords", value=50, minimum=1, maximum=1000)
+        gr.Textbox(label="Spotify URL (Track o Playlist)"),
+        gr.Number(label="Número de Canciones Similares", value=10, minimum=1, maximum=100)
     ],
     outputs=[gr.Dataframe(), gr.File(label="Download Excel")],
-    title="Spotify Playlist Fetcher",
-    description="Enter a search query to fetch playlists and their songs from Spotify. Client credentials are pre-configured."
+    title="Spotify Similar Track Finder",
+    description="Ingresa una URL de Spotify para encontrar canciones similares basadas en sus características."
 )
 iface.launch()