<reponame>rootshellz/Beatporter<filename>spotify.py
import sys
import json
import socket
import spotipy
import asyncio
import webbrowser
from time import time
from spotipy import oauth2

from config import *


def listen_for_callback_code():
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind(('localhost', int(redirect_uri.split(":")[-1])))
    s.listen(1)
    while True:
        connection, address = s.accept()
        buf = str(connection.recv(1024))
        if len(buf) > 0:
            break
    start_code = buf.find("?code=") + 6
    end_code = buf.find(" ", start_code)
    if "&" in buf[start_code:end_code]:
        end_code = buf.find("&")
    return buf[start_code:end_code]


async def get_spotify_auth_code():
    auth_url = sp_oauth.get_authorize_url()
    webbrowser.open(auth_url)


async def async_get_auth_code():
    task = asyncio.create_task(get_spotify_auth_code())
    await task
    return listen_for_callback_code()


def do_spotify_oauth():
    try:
        with open("token.json", "r") as fh:
            token = fh.read()
        token = json.loads(token)
    except:
        token = None
    if token:
        if int(time()) > token["expires_at"]:
            token = sp_oauth.refresh_access_token(token["refresh_token"])
    else:
        authorization_code = asyncio.run(async_get_auth_code())
        print(authorization_code)
        if not authorization_code:
            print("\n[!] Unable to authenticate to Spotify.  Couldn't get authorization code")
            sys.exit(-1)
        token = sp_oauth.get_access_token(authorization_code)
    if not token:
        print("\n[!] Unable to authenticate to Spotify.  Couldn't get access token.")
        sys.exit(-1)
    try:
        with open("token.json", "w+") as fh:
            fh.write(json.dumps(token))
    except:
        print("\n[!] Unable to to write token object to disk.  This is non-fatal.")
    return token


def get_all_playlists():
    playlists_pager = spotify.user_playlists(username)
    playlists = playlists_pager["items"]
    while playlists_pager["next"]:
        playlists_pager = spotify.next(playlists_pager)
        playlists.extend(playlists_pager["items"])
    return playlists


def create_playlist(playlist_name):
    playlist = spotify.user_playlist_create(username, playlist_name)
    return playlist["id"]


def get_playlist_id(playlist_name):
    playlists = get_all_playlists()
    for playlist in playlists:
        if playlist["name"] == playlist_name:
            return playlist["id"]
    return None


def do_durations_match(source_track_duration, found_track_duration):
    if source_track_duration == found_track_duration:
        print("\t\t\t\t[+] Durations match")
        return True
    else:
        print("\t\t\t\t[!] Durations do not match")
        return False


def most_popular_track(tracks):
    # Popularity does not always yield the correct result
    high_score = 0
    winner = None
    for track in tracks:
        if track["popularity"] > high_score:
            winner = track["id"]
            high_score = track["popularity"]
    return winner


def best_of_multiple_matches(source_track, found_tracks):
    counter = 1
    duration_matches = [0, ]
    for track in found_tracks:
        print("\t\t\t[+] Match {}: {}".format(counter, track["id"]))
        if do_durations_match(source_track["duration_ms"], track["duration_ms"]):
            duration_matches[0] += 1
            duration_matches.append(track)
        counter += 1
    if duration_matches[0] == 1:
        best_track = duration_matches.pop()["id"]
        print("\t\t\t[+] Only one exact match with matching duration, going with that one: {}".format(best_track))
        return best_track
    # TODO: Popularity does not always yield the correct result
    best_track = most_popular_track(found_tracks)
    print("\t\t\t[+] Multiple exact matches with matching durations, going with the most popular one: {}".format(best_track))
    return best_track


def search_for_track(track):
    # TODO: This is repetitive, can probably refactor but works for now
    print("\n[+] Searching for track: {}{}by {} on {}".format(track["name"], " " if not track["mix"] else " ({}) ".format(track["mix"]), ", ".join(track["artists"]), track["release"]))
    # Search with Title, Mix, Artists, and Release / Album
    query = "{}{}{} {}".format(track["name"], " " if not track["mix"] else " {} ".format(track["mix"]), " ".join(track["artists"]), track["release"])
    print("\t[+] Search Query: {}".format(query))
    search_results = spotify.search(query)
    if len(search_results["tracks"]["items"]) == 1:
        track_id = search_results["tracks"]["items"][0]["id"]
        print("\t\t[+] Found an exact match on name, mix, artists, and release: {}".format(track_id))
        do_durations_match(track["duration_ms"], search_results["tracks"]["items"][0]["duration_ms"])
        return track_id

    if len(search_results["tracks"]["items"]) > 1:
        print("\t\t[+] Found multiple exact matches ({}) on name, mix, artists, and release.".format(len(search_results["tracks"]["items"])))
        return best_of_multiple_matches(track, search_results["tracks"]["items"])

    # Not enough results, search w/o release
    print("\t\t[+] No exact matches on name, mix, artists, and release.  Trying without release.")
    # Search with Title, Mix, and Artists
    query = "{}{}{}".format(track["name"], " " if not track["mix"] else " {} ".format(track["mix"]), " ".join(track["artists"]))
    print("\t[+] Search Query: {}".format(query))
    search_results = spotify.search(query)
    if len(search_results["tracks"]["items"]) == 1:
        track_id = search_results["tracks"]["items"][0]["id"]
        print("\t\t[+] Found an exact match on name, mix, and artists: {}".format(track_id))
        do_durations_match(track["duration_ms"], search_results["tracks"]["items"][0]["duration_ms"])
        return track_id

    if len(search_results["tracks"]["items"]) > 1:
        print("\t\t[+] Found multiple exact matches ({}) on name, mix, and artists.".format(len(search_results["tracks"]["items"])))
        return best_of_multiple_matches(track, search_results["tracks"]["items"])

    # Not enough results, search w/o mix, but with release
    print("\t\t[+] No exact matches on name, mix, and artists.  Trying without mix, but with release.")
    query = "{} {} {}".format(track["name"], " ".join(track["artists"]), track["release"])
    print("\t[+] Search Query: {}".format(query))
    search_results = spotify.search(query)
    if len(search_results["tracks"]["items"]) == 1:
        track_id = search_results["tracks"]["items"][0]["id"]
        print("\t\t[+] Found an exact match on name, artists, and release: {}".format(track_id))
        do_durations_match(track["duration_ms"], search_results["tracks"]["items"][0]["duration_ms"])
        return track_id

    if len(search_results["tracks"]["items"]) > 1:
        print("\t\t[+] Found multiple exact matches ({}) on name, artists, and release.".format(len(search_results["tracks"]["items"])))
        return best_of_multiple_matches(track, search_results["tracks"]["items"])

    # Not enough results, search w/o mix or release
    print("\t\t[+] No exact matches on name, artists, and release.  Trying with just name and artists.")
    query = "{} {}".format(track["name"], " ".join(track["artists"]))
    print("\t[+] Search Query: {}".format(query))
    search_results = spotify.search(query)
    if len(search_results["tracks"]["items"]) == 1:
        track_id = search_results["tracks"]["items"][0]["id"]
        print("\t\t[+] Found an exact match on name and artists: {}".format(track_id))
        do_durations_match(track["duration_ms"], search_results["tracks"]["items"][0]["duration_ms"])
        return track_id

    if len(search_results["tracks"]["items"]) > 1:
        print("\t\t[+] Found multiple exact matches ({}) on name and artists.".format(len(search_results["tracks"]["items"])))
        return best_of_multiple_matches(track, search_results["tracks"]["items"])

    print("\t\t[+] No exact matches on name and artists.")
    print("\t[!] Could not find this song on Spotify!")
    return None


def track_in_playlist(playlist_id, track_id):
    for track in get_all_tracks_in_playlist(playlist_id):
        if track["track"]["id"] == track_id:
            return True
    return False


def add_tracks_to_playlist(playlist_id, track_ids):
    if track_ids:
        spotify.user_playlist_add_tracks(username, playlist_id, track_ids)


def get_all_tracks_in_playlist(playlist_id):
    if playlist_id in playlist_track_cache:
        return playlist_track_cache[playlist_id]
    playlist_tracks_results = spotify.playlist(playlist_id, fields="tracks")
    playlist_tracks_pager = playlist_tracks_results["tracks"]
    playlist_tracks = playlist_tracks_pager["items"]
    while playlist_tracks_pager["next"]:
        playlist_tracks_pager = spotify.next(playlist_tracks_pager)
        playlist_tracks.extend(playlist_tracks_pager["items"])
    playlist_track_cache[playlist_id] = playlist_tracks
    return playlist_track_cache[playlist_id]


def clear_playlist(playlist_id):
    for track in get_all_tracks_in_playlist(playlist_id):
        spotify.user_playlist_remove_all_occurrences_of_tracks(username, playlist_id, [track["track"]["id"],])


def add_new_tracks_to_playlist(genre, tracks_dict):
    persistent_top_100_playlist_name = "Beatporter: {} - Top 100".format(genre)
    daily_top_10_playlist_name = "Beatporter: {} - Daily Top 10".format(genre)
    print("[+] Identifying new tracks for playlist: \"{}\"".format(persistent_top_100_playlist_name))

    playlists = [{"name": persistent_top_100_playlist_name, "id": get_playlist_id(persistent_top_100_playlist_name)},
                 {"name": daily_top_10_playlist_name, "id": get_playlist_id(daily_top_10_playlist_name)}]

    for playlist in playlists:
        if not playlist["id"]:
            print("\t[!] Playlist \"{}\" does not exist, creating it.".format(playlist["name"]))
            playlist["id"] = create_playlist(playlist["name"])

    # Clear daily playlist
    clear_playlist(playlists[1]["id"])

    persistent_top_100_track_ids = list()
    daily_top_10_track_ids = list()
    track_count = 0
    for track in tracks_dict:
        track_id = search_for_track(track)
        if track_id and not track_in_playlist(playlists[0]["id"], track_id):
            persistent_top_100_track_ids.append(track_id)
        if track_id and track_count < 10:
            daily_top_10_track_ids.append(track_id)
        track_count += 1
    print("\n[+] Adding {} new tracks to the playlist: \"{}\"".format(len(persistent_top_100_track_ids), persistent_top_100_playlist_name))
    add_tracks_to_playlist(playlists[0]["id"], persistent_top_100_track_ids)
    print("\n[+] Adding {} new tracks to the playlist: \"{}\"".format(len(daily_top_10_track_ids), daily_top_10_playlist_name))
    add_tracks_to_playlist(playlists[1]["id"], daily_top_10_track_ids)

playlist_track_cache = dict()

# Get authenticated to Spotify on import
sp_oauth = oauth2.SpotifyOAuth(client_id, client_secret, redirect_uri, username=username, scope=scope)
token_info = do_spotify_oauth()
spotify = spotipy.Spotify(auth=token_info["access_token"], requests_timeout=120)

import sys
import json
import socket
import spotipy
import asyncio
import webbrowser
from time import time
from spotipy import oauth2

from config import *


def listen_for_callback_code():
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind(('localhost', int(redirect_uri.split(":")[-1])))
    s.listen(1)
    while True:
        connection, address = s.accept()
        buf = str(connection.recv(1024))
        if len(buf) > 0:
            break
    start_code = buf.find("?code=") + 6
    end_code = buf.find(" ", start_code)
    if "&" in buf[start_code:end_code]:
        end_code = buf.find("&")
    return buf[start_code:end_code]


async def get_spotify_auth_code():
    auth_url = sp_oauth.get_authorize_url()
    webbrowser.open(auth_url)


async def async_get_auth_code():
    task = asyncio.create_task(get_spotify_auth_code())
    await task
    return listen_for_callback_code()


def do_spotify_oauth():
    try:
        with open("token.json", "r") as fh:
            token = fh.read()
        token = json.loads(token)
    except:
        token = None
    if token:
        if int(time()) > token["expires_at"]:
            token = sp_oauth.refresh_access_token(token["refresh_token"])
    else:
        authorization_code = asyncio.run(async_get_auth_code())
        print(authorization_code)
        if not authorization_code:
            print("\n[!] Unable to authenticate to Spotify.  Couldn't get authorization code")
            sys.exit(-1)
        token = sp_oauth.get_access_token(authorization_code)
    if not token:
        print("\n[!] Unable to authenticate to Spotify.  Couldn't get access token.")
        sys.exit(-1)
    try:
        with open("token.json", "w+") as fh:
            fh.write(json.dumps(token))
    except:
        print("\n[!] Unable to to write token object to disk.  This is non-fatal.")
    return token


def get_all_playlists():
    playlists_pager = spotify.user_playlists(username)
    playlists = playlists_pager["items"]
    while playlists_pager["next"]:
        playlists_pager = spotify.next(playlists_pager)
        playlists.extend(playlists_pager["items"])
    return playlists


def create_playlist(playlist_name):
    playlist = spotify.user_playlist_create(username, playlist_name)
    return playlist["id"]


def get_playlist_id(playlist_name):
    playlists = get_all_playlists()
    for playlist in playlists:
        if playlist["name"] == playlist_name:
            return playlist["id"]
    return None


def do_durations_match(source_track_duration, found_track_duration):
    if source_track_duration == found_track_duration:
        print("\t\t\t\t[+] Durations match")
        return True
    else:
        print("\t\t\t\t[!] Durations do not match")
        return False


def most_popular_track(tracks):
    # Popularity does not always yield the correct result
    high_score = 0
    winner = None
    for track in tracks:
        if track["popularity"] > high_score:
            winner = track["id"]
            high_score = track["popularity"]
    return winner


def best_of_multiple_matches(source_track, found_tracks):
    counter = 1
    duration_matches = [0, ]
    for track in found_tracks:
        print("\t\t\t[+] Match {}: {}".format(counter, track["id"]))
        if do_durations_match(source_track["duration_ms"], track["duration_ms"]):
            duration_matches[0] += 1
            duration_matches.append(track)
        counter += 1
    if duration_matches[0] == 1:
        best_track = duration_matches.pop()["id"]
        print("\t\t\t[+] Only one exact match with matching duration, going with that one: {}".format(best_track))
        return best_track
    # TODO: Popularity does not always yield the correct result
    best_track = most_popular_track(found_tracks)
    print("\t\t\t[+] Multiple exact matches with matching durations, going with the most popular one: {}".format(best_track))
    return best_track


def search_for_track(track):
    # TODO: This is repetitive, can probably refactor but works for now
    print("\n[+] Searching for track: {}{}by {} on {}".format(track["name"], " " if not track["mix"] else " ({}) ".format(track["mix"]), ", ".join(track["artists"]), track["release"]))
    # Search with Title, Mix, Artists, and Release / Album
    query = "{}{}{} {}".format(track["name"], " " if not track["mix"] else " {} ".format(track["mix"]), " ".join(track["artists"]), track["release"])
    print("\t[+] Search Query: {}".format(query))
    search_results = spotify.search(query)
    if len(search_results["tracks"]["items"]) == 1:
        track_id = search_results["tracks"]["items"][0]["id"]
        print("\t\t[+] Found an exact match on name, mix, artists, and release: {}".format(track_id))
        do_durations_match(track["duration_ms"], search_results["tracks"]["items"][0]["duration_ms"])
        return track_id

    if len(search_results["tracks"]["items"]) > 1:
        print("\t\t[+] Found multiple exact matches ({}) on name, mix, artists, and release.".format(len(search_results["tracks"]["items"])))
        return best_of_multiple_matches(track, search_results["tracks"]["items"])

    # Not enough results, search w/o release
    print("\t\t[+] No exact matches on name, mix, artists, and release.  Trying without release.")
    # Search with Title, Mix, and Artists
    query = "{}{}{}".format(track["name"], " " if not track["mix"] else " {} ".format(track["mix"]), " ".join(track["artists"]))
    print("\t[+] Search Query: {}".format(query))
    search_results = spotify.search(query)
    if len(search_results["tracks"]["items"]) == 1:
        track_id = search_results["tracks"]["items"][0]["id"]
        print("\t\t[+] Found an exact match on name, mix, and artists: {}".format(track_id))
        do_durations_match(track["duration_ms"], search_results["tracks"]["items"][0]["duration_ms"])
        return track_id

    if len(search_results["tracks"]["items"]) > 1:
        print("\t\t[+] Found multiple exact matches ({}) on name, mix, and artists.".format(len(search_results["tracks"]["items"])))
        return best_of_multiple_matches(track, search_results["tracks"]["items"])

    # Not enough results, search w/o mix, but with release
    print("\t\t[+] No exact matches on name, mix, and artists.  Trying without mix, but with release.")
    query = "{} {} {}".format(track["name"], " ".join(track["artists"]), track["release"])
    print("\t[+] Search Query: {}".format(query))
    search_results = spotify.search(query)
    if len(search_results["tracks"]["items"]) == 1:
        track_id = search_results["tracks"]["items"][0]["id"]
        print("\t\t[+] Found an exact match on name, artists, and release: {}".format(track_id))
        do_durations_match(track["duration_ms"], search_results["tracks"]["items"][0]["duration_ms"])
        return track_id

    if len(search_results["tracks"]["items"]) > 1:
        print("\t\t[+] Found multiple exact matches ({}) on name, artists, and release.".format(len(search_results["tracks"]["items"])))
        return best_of_multiple_matches(track, search_results["tracks"]["items"])

    # Not enough results, search w/o mix or release
    print("\t\t[+] No exact matches on name, artists, and release.  Trying with just name and artists.")
    query = "{} {}".format(track["name"], " ".join(track["artists"]))
    print("\t[+] Search Query: {}".format(query))
    search_results = spotify.search(query)
    if len(search_results["tracks"]["items"]) == 1:
        track_id = search_results["tracks"]["items"][0]["id"]
        print("\t\t[+] Found an exact match on name and artists: {}".format(track_id))
        do_durations_match(track["duration_ms"], search_results["tracks"]["items"][0]["duration_ms"])
        return track_id

    if len(search_results["tracks"]["items"]) > 1:
        print("\t\t[+] Found multiple exact matches ({}) on name and artists.".format(len(search_results["tracks"]["items"])))
        return best_of_multiple_matches(track, search_results["tracks"]["items"])

    print("\t\t[+] No exact matches on name and artists.")
    print("\t[!] Could not find this song on Spotify!")
    return None


def track_in_playlist(playlist_id, track_id):
    for track in get_all_tracks_in_playlist(playlist_id):
        if track["track"]["id"] == track_id:
            return True
    return False


def add_tracks_to_playlist(playlist_id, track_ids):
    if track_ids:
        spotify.user_playlist_add_tracks(username, playlist_id, track_ids)


def get_all_tracks_in_playlist(playlist_id):
    if playlist_id in playlist_track_cache:
        return playlist_track_cache[playlist_id]
    playlist_tracks_results = spotify.playlist(playlist_id, fields="tracks")
    playlist_tracks_pager = playlist_tracks_results["tracks"]
    playlist_tracks = playlist_tracks_pager["items"]
    while playlist_tracks_pager["next"]:
        playlist_tracks_pager = spotify.next(playlist_tracks_pager)
        playlist_tracks.extend(playlist_tracks_pager["items"])
    playlist_track_cache[playlist_id] = playlist_tracks
    return playlist_track_cache[playlist_id]


def clear_playlist(playlist_id):
    for track in get_all_tracks_in_playlist(playlist_id):
        spotify.user_playlist_remove_all_occurrences_of_tracks(username, playlist_id, [track["track"]["id"],])


def add_new_tracks_to_playlist(genre, tracks_dict):
    persistent_top_100_playlist_name = "Beatporter: {} - Top 100".format(genre)
    daily_top_10_playlist_name = "Beatporter: {} - Daily Top 10".format(genre)
    print("[+] Identifying new tracks for playlist: \"{}\"".format(persistent_top_100_playlist_name))

    playlists = [{"name": persistent_top_100_playlist_name, "id": get_playlist_id(persistent_top_100_playlist_name)},
                 {"name": daily_top_10_playlist_name, "id": get_playlist_id(daily_top_10_playlist_name)}]

    for playlist in playlists:
        if not playlist["id"]:
            print("\t[!] Playlist \"{}\" does not exist, creating it.".format(playlist["name"]))
            playlist["id"] = create_playlist(playlist["name"])

    # Clear daily playlist
    clear_playlist(playlists[1]["id"])

    persistent_top_100_track_ids = list()
    daily_top_10_track_ids = list()
    track_count = 0
    for track in tracks_dict:
        track_id = search_for_track(track)
        if track_id and not track_in_playlist(playlists[0]["id"], track_id):
            persistent_top_100_track_ids.append(track_id)
        if track_id and track_count < 10:
            daily_top_10_track_ids.append(track_id)
        track_count += 1
    print("\n[+] Adding {} new tracks to the playlist: \"{}\"".format(len(persistent_top_100_track_ids), persistent_top_100_playlist_name))
    add_tracks_to_playlist(playlists[0]["id"], persistent_top_100_track_ids)
    print("\n[+] Adding {} new tracks to the playlist: \"{}\"".format(len(daily_top_10_track_ids), daily_top_10_playlist_name))
    add_tracks_to_playlist(playlists[1]["id"], daily_top_10_track_ids)

playlist_track_cache = dict()

# Get authenticated to Spotify on import
sp_oauth = oauth2.SpotifyOAuth(client_id, client_secret, redirect_uri, username=username, scope=scope)
token_info = do_spotify_oauth()
spotify = spotipy.Spotify(auth=token_info["access_token"], requests_timeout=120)
# Popularity does not always yield the correct result # TODO: Popularity does not always yield the correct result # TODO: This is repetitive, can probably refactor but works for now # Search with Title, Mix, Artists, and Release / Album # Not enough results, search w/o release # Search with Title, Mix, and Artists # Not enough results, search w/o mix, but with release # Not enough results, search w/o mix or release # Clear daily playlist # Get authenticated to Spotify on import