5
\$\begingroup\$

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()
\$\endgroup\$
  • 1
    \$\begingroup\$ Thank you for including your code, and not asking how to change your code. Your question shouldn't be off-topic now. :) You could improve your question by describing what you code does, amongst other things. This however isn't a requirement. \$\endgroup\$ – Peilonrayz Jan 26 '18 at 10:26

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.