mirror of
https://github.com/aljazceru/chatgpt-telegram-bot.git
synced 2025-12-19 21:55:06 +01:00
improvements and added initial spotify plugin
This commit is contained in:
ned
302
bot/plugins/spotify.py
Normal file
302
bot/plugins/spotify.py
Normal file
@@ -0,0 +1,302 @@
|
||||
import os
|
||||
from typing import Dict
|
||||
|
||||
import spotipy
|
||||
from spotipy import SpotifyOAuth
|
||||
|
||||
from .plugin import Plugin
|
||||
|
||||
|
||||
class SpotifyPlugin(Plugin):
|
||||
"""
|
||||
A plugin to fetch information from Spotify
|
||||
"""
|
||||
def __init__(self):
|
||||
spotify_client_id = os.getenv('SPOTIFY_CLIENT_ID')
|
||||
spotify_client_secret = os.getenv('SPOTIFY_CLIENT_SECRET')
|
||||
spotify_redirect_uri = os.getenv('SPOTIFY_REDIRECT_URI')
|
||||
if not spotify_client_id or not spotify_client_secret or not spotify_redirect_uri:
|
||||
raise ValueError('SPOTIFY_CLIENT_ID, SPOTIFY_CLIENT_SECRET and SPOTIFY_REDIRECT_URI environment variables'
|
||||
' are required to use SpotifyPlugin')
|
||||
self.spotify = spotipy.Spotify(
|
||||
auth_manager=SpotifyOAuth(
|
||||
client_id=spotify_client_id,
|
||||
client_secret=spotify_client_secret,
|
||||
redirect_uri=spotify_redirect_uri,
|
||||
scope="user-top-read,user-read-currently-playing",
|
||||
open_browser=False
|
||||
)
|
||||
)
|
||||
|
||||
def get_source_name(self) -> str:
|
||||
return "Spotify"
|
||||
|
||||
def get_spec(self) -> [Dict]:
|
||||
time_range_param = {
|
||||
"type": "string",
|
||||
"enum": ["short_term", "medium_term", "long_term"],
|
||||
"description": "The time range of the data to be returned. Short term is the last 4 weeks, "
|
||||
"medium term is last 6 months, long term is last several years. "
|
||||
"Ignore if action is currently_playing",
|
||||
}
|
||||
limit_param = {
|
||||
"type": "integer",
|
||||
"description": "The number of results to return. Max is 50. Default to 10 if not specified."
|
||||
"Ignore if action is currently_playing",
|
||||
}
|
||||
type_param = {
|
||||
"type": "string",
|
||||
"enum": ["album", "artist", "track"],
|
||||
"description": "Type of content to search",
|
||||
}
|
||||
return [
|
||||
{
|
||||
"name": "spotify_get_currently_playing_song",
|
||||
"description": "Get the user's currently playing song",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {}
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "spotify_get_users_top_artists",
|
||||
"description": "Get the user's top artists",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"time_range": time_range_param,
|
||||
"limit": limit_param
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "spotify_get_users_top_tracks",
|
||||
"description": "Get the user's top tracks",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"time_range": time_range_param,
|
||||
"limit": limit_param
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "spotify_search_by_query",
|
||||
"description": "Search spotify content by query",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"query": {
|
||||
"type": "string",
|
||||
"description": "The search query",
|
||||
},
|
||||
"type": type_param
|
||||
},
|
||||
"required": ["query", "type"]
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "spotify_lookup_by_id",
|
||||
"description": "Lookup spotify content by id",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"id": {
|
||||
"type": "string",
|
||||
"description": "The exact id to lookup. Can be a track id, an artist id or an album id",
|
||||
},
|
||||
"type": type_param
|
||||
},
|
||||
"required": ["id", "type"]
|
||||
}
|
||||
}
|
||||
]
|
||||
|
||||
async def execute(self, function_name, **kwargs) -> Dict:
|
||||
time_range = kwargs.get('time_range', 'short_term')
|
||||
limit = kwargs.get('limit', 10)
|
||||
|
||||
if function_name == 'spotify_get_currently_playing_song':
|
||||
return self.fetch_currently_playing()
|
||||
elif function_name == 'spotify_get_users_top_artists':
|
||||
return self.fetch_top_artists(time_range, limit)
|
||||
elif function_name == 'spotify_get_users_top_tracks':
|
||||
return self.fetch_top_tracks(time_range, limit)
|
||||
elif function_name == 'spotify_search_by_query':
|
||||
query = kwargs.get('query', '')
|
||||
search_type = kwargs.get('type', 'track')
|
||||
return self.search_by_query(query, search_type, limit)
|
||||
elif function_name == 'spotify_lookup_by_id':
|
||||
content_id = kwargs.get('id')
|
||||
search_type = kwargs.get('type', 'track')
|
||||
return self.search_by_id(content_id, search_type)
|
||||
|
||||
def fetch_currently_playing(self) -> Dict:
|
||||
"""
|
||||
Fetch user's currently playing song from Spotify
|
||||
"""
|
||||
currently_playing = self.spotify.current_user_playing_track()
|
||||
result = {
|
||||
'name': currently_playing['item']['name'],
|
||||
'artist': currently_playing['item']['artists'][0]['name'],
|
||||
'album': currently_playing['item']['album']['name'],
|
||||
'url': currently_playing['item']['external_urls']['spotify'],
|
||||
'__album_id': currently_playing['item']['album']['id'],
|
||||
'__artist_id': currently_playing['item']['artists'][0]['id'],
|
||||
'__track_id': currently_playing['item']['id'],
|
||||
}
|
||||
return {"result": result}
|
||||
|
||||
def fetch_top_tracks(self, time_range='short_term', limit=5) -> Dict:
|
||||
"""
|
||||
Fetch user's top tracks from Spotify
|
||||
"""
|
||||
results = []
|
||||
top_tracks = self.spotify.current_user_top_tracks(limit=limit, time_range=time_range)
|
||||
for item in top_tracks['items']:
|
||||
results.append({
|
||||
'name': item['name'],
|
||||
'artist': item['artists'][0]['name'],
|
||||
'album': item['album']['name'],
|
||||
'album_release_date': item['album']['release_date'],
|
||||
'url': item['external_urls']['spotify'],
|
||||
'album_url': item['album']['external_urls']['spotify'],
|
||||
'artist_url': item['artists'][0]['external_urls']['spotify'],
|
||||
'__track_id': item['id'],
|
||||
'__album_id': item['album']['id'],
|
||||
'__artist_id': item['artists'][0]['id'],
|
||||
})
|
||||
return {'results': results}
|
||||
|
||||
def fetch_top_artists(self, time_range='short_term', limit=5) -> Dict:
|
||||
"""
|
||||
Fetch user's top artists from Spotify
|
||||
"""
|
||||
results = []
|
||||
top_artists = self.spotify.current_user_top_artists(limit=limit, time_range=time_range)
|
||||
for item in top_artists['items']:
|
||||
results.append({
|
||||
'name': item['name'],
|
||||
'url': item['external_urls']['spotify'],
|
||||
'__artist_id': item['id']
|
||||
})
|
||||
return {'results': results}
|
||||
|
||||
def search_by_query(self, query, search_type, limit=5) -> Dict:
|
||||
"""
|
||||
Search content by query on Spotify
|
||||
"""
|
||||
results = {}
|
||||
search_response = self.spotify.search(q=query, limit=limit, type=search_type)
|
||||
if 'tracks' in search_response:
|
||||
results['tracks'] = []
|
||||
for item in search_response['tracks']['items']:
|
||||
results['tracks'].append({
|
||||
'name': item['name'],
|
||||
'artist': item['artists'][0]['name'],
|
||||
'album': item['album']['name'],
|
||||
'album_release_date': item['album']['release_date'],
|
||||
'url': item['external_urls']['spotify'],
|
||||
'album_url': item['album']['external_urls']['spotify'],
|
||||
'artist_url': item['artists'][0]['external_urls']['spotify'],
|
||||
'__artist_id': item['artists'][0]['id'],
|
||||
'__album_id': item['album']['id'],
|
||||
'__track_id': item['id'],
|
||||
})
|
||||
if 'artists' in search_response:
|
||||
results['artists'] = []
|
||||
for item in search_response['artists']['items']:
|
||||
results['artists'].append({
|
||||
'name': item['name'],
|
||||
'url': item['external_urls']['spotify'],
|
||||
'__artist_id': item['id'],
|
||||
})
|
||||
if 'albums' in search_response:
|
||||
results['albums'] = []
|
||||
for item in search_response['albums']['items']:
|
||||
results['albums'].append({
|
||||
'name': item['name'],
|
||||
'artist': item['artists'][0]['name'],
|
||||
'url': item['external_urls']['spotify'],
|
||||
'artist_url': item['artists'][0]['external_urls']['spotify'],
|
||||
'release_date': item['release_date'],
|
||||
'__artist_id': item['artists'][0]['id'],
|
||||
'__album_id': item['id'],
|
||||
})
|
||||
return {'results': results}
|
||||
|
||||
def search_by_id(self, content_id, search_type) -> Dict:
|
||||
"""
|
||||
Search content by exact id on Spotify
|
||||
"""
|
||||
if search_type == 'track':
|
||||
search_response = self.spotify.track(content_id)
|
||||
return {'result': self._get_track(search_response)}
|
||||
|
||||
elif search_type == 'artist':
|
||||
search_response = self.spotify.artist(content_id)
|
||||
albums_response = self.spotify.artist_albums(artist_id=content_id, limit=3)
|
||||
return {'result': self._get_artist(search_response, albums_response)}
|
||||
|
||||
elif search_type == 'album':
|
||||
search_response = self.spotify.album(content_id)
|
||||
return {'result': self._get_album(search_response)}
|
||||
|
||||
else:
|
||||
return {'error': 'Invalid search type. Must be track, artist or album'}
|
||||
|
||||
def _get_artist(self, response, albums):
|
||||
return {
|
||||
'name': response['name'],
|
||||
'url': response['external_urls']['spotify'],
|
||||
'__artist_id': response['id'],
|
||||
'followers': response['followers']['total'],
|
||||
'genres': response['genres'],
|
||||
'albums': [
|
||||
{
|
||||
'name': album['name'],
|
||||
'__album_id': album['id'],
|
||||
'url': album['external_urls']['spotify'],
|
||||
'release_date': album['release_date'],
|
||||
'total_tracks': album['total_tracks'],
|
||||
}
|
||||
for album in albums['items']
|
||||
],
|
||||
}
|
||||
|
||||
def _get_track(self, response):
|
||||
return {
|
||||
'name': response['name'],
|
||||
'artist': response['artists'][0]['name'],
|
||||
'__artist_id': response['artists'][0]['id'],
|
||||
'album': response['album']['name'],
|
||||
'__album_id': response['album']['id'],
|
||||
'url': response['external_urls']['spotify'],
|
||||
'__track_id': response['id'],
|
||||
'duration_ms': response['duration_ms'],
|
||||
'track_number': response['track_number'],
|
||||
'explicit': response['explicit'],
|
||||
}
|
||||
|
||||
def _get_album(self, response):
|
||||
return {
|
||||
'name': response['name'],
|
||||
'artist': response['artists'][0]['name'],
|
||||
'__artist_id': response['artists'][0]['id'],
|
||||
'url': response['external_urls']['spotify'],
|
||||
'release_date': response['release_date'],
|
||||
'total_tracks': response['total_tracks'],
|
||||
'__album_id': response['id'],
|
||||
'label': response['label'],
|
||||
'tracks': [
|
||||
{
|
||||
'name': track['name'],
|
||||
'url': track['external_urls']['spotify'],
|
||||
'__track_id': track['id'],
|
||||
'duration_ms': track['duration_ms'],
|
||||
'track_number': track['track_number'],
|
||||
'explicit': track['explicit'],
|
||||
}
|
||||
for track in response['tracks']['items']
|
||||
]
|
||||
}
|
||||
Reference in New Issue
Block a user