trjhtr

The result is just a list of youtube links.

I'm requesting an access token every time and throwing it away, so I'm thinking of a way to save it for future use. Maybe I'll use a tempfile, but even then I'll have hard time making it compatible across different OSes.

Also, I don't have much experience with concurrent programming, so I hope I'm using multiprocessing properly.

#!/usr/bin/env python3

# Usage Instructions (You need a media player able to play youtube urls to use this script):

# $ export SIMLR_CLIENT_ID=[your client id]
# $ export SIMLR_CLIENT_SECRET=[your client secret]
# $ mpv --no-video --msg-level=cplayer=no --term-playing-msg='Playing: $media-title' --playlist=<(./simlr.py ARTIST)

import os
import sys
import random
import argparse
import multiprocessing
import functools
from itertools import chain

import requests
from bs4 import BeautifulSoup


def init_argparser():
 help_text = """
 Generates a playlist using Spotify based on ARTIST.

 You need to hava a Spotify account to use this script.
 Register a new app with spotify to obtain the client id and secret.
 Either export your client id and client secret as environment variables:

 export SIMLR_CLIENT_ID=<your client id>
 export SIMLR_CLIENT_SECRET=<your client secret>

 or pass them as commandline arguments:
 ... --id=<your client id> --secret=<your client secret>
 """

 parser = argparse.ArgumentParser(formatter_class=argparse.RawDescriptionHelpFormatter, description=help_text)
 parser.add_argument('artist', metavar='ARTIST', type=str, help="artist seed to generate playlist")
 parser.add_argument('-r', '--related', action='store_true', help="generate playlist from tracks by related artists; playlist will not have tracks by ARTIST")
 parser.add_argument('-c', '--country', dest='country', default='US', help="ISO 3166-1 alpha-2 country code, default is 'US'")
 parser.add_argument('-l', '--limit', dest='limit', type=int, default=20, help="limit the number of tracks, valid values are numbers from 1 to 100, default is 20")
 parser.add_argument('--id', dest='client_id', default=os.environ.get('SIMLR_CLIENT_ID', None), help="Client id of your spotify app")
 parser.add_argument('--secret', dest='client_secret', default=os.environ.get('SIMLR_CLIENT_SECRET', None), help="Client secret of your spotify app")
 return parser

def _make_http_request(url, method="get", headers=None, params=None, data=None, auth=None):
 """
 Make an HTTP request and return a json object of the response.
 """
 try:
 request_method = requests.post if method == "post" else requests.get
 res = request_method(url, headers=headers, params=params, data=data, auth=auth)
 responsejson = res.json()

 if res.status_code != 200:
 raise Exception(res.text)
 except ValueError:
 # if the response isn't JSON, .json() method will raise JSONDecodeError,
 # which is a subclass of ValueError
 return res.text
 except Exception as err:
 sys.exit("Error during HTTP request to : ".format(url, err))
 return responsejson

def request_access_token(client_id, client_secret):
 """
 Request a new access token from Spotify.
 Returns the new access token on success, exits on failure.
 """

 headers = 'Accept': 'application/json'
 data = [('grant_type', 'client_credentials')]
 res = _make_http_request('https://accounts.spotify.com/api/token', method="post", headers=headers, data=data, auth=(client_id, client_secret))
 try:
 return res['access_token']

 except (KeyError, TypeError):
 sys.exit("Error while requesting an access token")

def get_spotify_artist(name, access_token):
 """
 Get the spotify id for the artist.
 Returns spotify artist id when successful, exits on failure.
 """

 headers = 'Authorization': 'Bearer ' + access_token,
 params = (('q', name), ('type', 'artist'),)
 res = _make_http_request('https://api.spotify.com/v1/search', headers=headers, params=params)
 items = res.get('artists', ).get('items', None)
 try:
 return items[0]['id']
 except (IndexError, KeyError):
 sys.exit("Error while retriving the Spotify artist id. Maybe the artist isn't on spotify")

def get_recommended_tracks(artistid, access_token, limit=20, market='US'):
 """
 Get recommended tracks for the artist from Spotify.
 Returns a list of tracks when successful, exits on failure.
 """

 headers = 
 'Accept': 'application/json',
 'Authorization': 'Bearer ' + access_token,
 

 params = (
 ('market', market),
 ('seed_artists', artistid),
 ('limit', limit),
 )

 res = _make_http_request('https://api.spotify.com/v1/recommendations', method="get", headers=headers, params=params)
 items = res.get('tracks', None)
 tracks = 
 try:
 for item in items:
 name = item['name']
 artist = item['artists'][0]['name']
 tracks.append("artist - name".format(name=name, artist=artist))
 except (TypeError, IndexError, KeyError):
 sys.exit("Failed to retrieve Spotify's recommendations.")
 return tracks

def get_related_artists(artistid, access_token):
 """
 Get related artists (according to spotify) to the given artist
 Returns a list of spotify artist ids when successful, exits on failure.
 """

 headers = 'Authorization': 'Bearer ' + access_token,
 res = _make_http_request('https://api.spotify.com/v1/artists/id/related-artists'.format(id=artistid), 
 method="get",
 headers=headers)
 similar_artists = 
 try:
 artists = res['artists']
 for artist in artists:
 aid = artist['id']
 similar_artists.append(aid)
 except (KeyError, TypeError):
 sys.exit("Error while retrieving similar artists")
 return similar_artists

def get_top_tracks(artistid, access_token, limit=1, market='US'):
 """
 Get top tracks of a artist
 Returns a list of spotify track ids when successful, exits on failure.
 """
 headers = 
 'Accept': 'application/json',
 'Authorization': 'Bearer ' + access_token,
 
 params = (
 ('country', market),
 ('limit', limit),) 

 res = _make_http_request('https://api.spotify.com/v1/artists/artistid/top-tracks'.format(artistid=artistid), method="get", headers=headers, params=params)
 try:
 # retrieve at most k random tracks of this artist
 items = random.sample(res['tracks'], k=limit)
 tracks = 
 for item in items:
 name = item['name']
 artist = item['artists'][0]['name']
 tracks.append("artist - name".format(name=name, artist=artist))
 except (ValueError, KeyError) as err:
 sys.exit("Error while retrieving top tracks: ".format(err))
 return tracks

def get_related_tracks(artistid, access_token, limit=20, market='US'):
 """
 Returns a list of tracks by related artists on success, exits on failure.
 """
 related_artists = get_related_artists(artistid, access_token)

 # This is the slowest bit in the code because of multiple http requests
 pool = multiprocessing.Pool(len(related_artists))

 # we'll have 20 artists to fetch the track from, so total tracks fetched will be a multiple of 20
 tracklist = pool.imap_unordered(functools.partial(get_top_tracks, access_token=access_token, limit=limit//20+1, market=market), related_artists)

 # tracklist is a nested list of tracks, so we have to flatten it to get a list of tracks
 return random.sample(list(chain.from_iterable(tracklist)), k=limit)

def get_youtube_url(query):
 """
 Print the most relevant youtube link for the query.
 """
 query = query + ", video" # hint for youtube to only list videos
 res = _make_http_request("https://www.youtube.com/results", "get", params=(('search_query', query),))
 soup = BeautifulSoup(res, "html.parser")
 # we print the first result
 print("https://youtube.com" + soup.find_all(attrs='class':'yt-uix-tile-link')[0]['href'])


def main():
 parser = init_argparser()
 args = parser.parse_args()


 if not args.client_id or not args.client_secret:
 sys.exit("Client Id and/or Client Secret not provided")
 elif args.limit > 100 or args.limit < 1:
 args.limit = parser.get_default('limit') 


 access_token = request_access_token(args.client_id, args.client_secret)

 artist = get_spotify_artist(args.artist, access_token)

 if args.related:
 playlist = get_related_tracks(artist, access_token, limit=args.limit, market=args.country)
 else:
 playlist = get_recommended_tracks(artist, access_token, limit=args.limit, market=args.country)

 for track in playlist:
 multiprocessing.Process(target=get_youtube_url, args=(track,)).start()

 sys.exit()

if __name__ == "__main__":
 main()