import os import json from sqlite3.dbapi2 import OperationalError import requests import datetime import time import math import sqlite3 from sqlite3 import Error from secrets import spotify_user_id from refresh import Refresh refreshCaller = Refresh() os.system('cls') print("[•_•] Decades Playlist Balancer and Shuffler") print("Version 2.15.0: 28-JUL-17:10 (c) Viking Design.") print("Creates and Maintains 10 Playlists of Shuffled Tracks based on Decade") print("The number of Tracks from each Decade depends on the total nunber of tracks in that Decade") print("Repeat occurences are prevented by marking the tracks selected as played in the database") print("When 99% of tracks have been selected a full database reset is carried out") print("User: " + spotify_user_id) class PlaylistGrab: def __init__(self): self.user_id = spotify_user_id self.spotify_token = "" self.here_i_am = os.path.dirname(os.path.abspath(__file__)) self.more_than_fifty = 0 self.total_tracks = 0 self.tt = [0, 0, 0, 0, 0, 0, 0, 0, 0, 0,] self.pt = [0, 0, 0, 0, 0, 0, 0, 0, 0, 0,] self.pl_name = ["", "", "", "", "", "", "", "", "", "",] self.date_time_ = datetime.datetime.now() self.total_played_percentage = 0 self.database = self.here_i_am + r"\\All Playlists Tracks for " + spotify_user_id + ".db" #=================ENTER QUEUE SIZE================================= "This will vary by a few (always more!) depending on the the maths carried out" "If this number is large, significant delays will occur but the app is still running" self.queue_size = 25 #=================ENTER DAILY PLAYLIST I.D.======================== "Enter the chosen name of the listening playlists you have chosen" "This can be anything of 4 or more characters. It has been tested to 35 characters." self.user_pl_id = "HERE" #=================ENTER DECADES PLAYLISTS IDENTIFIERS============== "Enter the last character of your Decades Playlists" self.user_dec_id = "9" #================================================================== def call_refresh(self): print("[•] Refreshing Token....") global refreshCaller while True: self.spotify_token = refreshCaller.refresh() if self.spotify_token == -1: print("[X] Error Refreshing Token - Retrying in 30 Secs [X]") time.sleep(30) else: break def reset_listening_database(self): conn = self.create_connection(self.database) with conn: cur = conn.cursor() try: cur.execute("SELECT * FROM your_listening_playlists ") except OperationalError as e: if 'no such table' in str(e): print("ERROR your_listening_playlists table is missing \n Please run Sauron.py to rectify this problem") exit() else: print(e) exit() rows = cur.fetchall() for row in rows: upl_num = row[0] cur.execute("UPDATE your_listening_playlists SET EXIST = False ") cur.execute("UPDATE your_listening_playlists SET NAME = '" + self.user_pl_id + " Playlist No. " + upl_num + "' WHERE NUMBER = '" + upl_num + "'") cur.execute("DROP TABLE IF EXISTS tmp_shuffle") conn.commit() return def create_connection(self, db_file): conn = None try: conn = sqlite3.connect(db_file) return conn except Error as e: print(e) return conn def get_the_playlists(self): params={"limit":50,"offset":self.more_than_fifty} query = "https://api.spotify.com/v1/users/{}/playlists".format(self.user_id) response = requests.request("get", url=query, headers={"Content-type": "application/json","Authorization":"Bearer {}".format(self.spotify_token)},params=params) response_json = response.json() for i in response_json["items"]: listening2 = False self.name = (i["name"]) self.total = (i["tracks"]["total"]) self.uri = (i["uri"]) pid = (self.name[-2:-1]) check = (self.name[-1:]) listening = self.name listening2 = "Playlist" in self.name if listening[:4] == self.user_pl_id: listening_num = self.name[-2:] listening_num = listening_num.replace(" ", "") self.check_listening(listening_num) continue if check == self.user_dec_id and not listening2:#needs a bit heree self.tt[int(pid)] = self.total self.pl_name[int(pid)] = self.name if self.more_than_fifty == 0: self.more_than_fifty = 50 self.get_the_playlists() self.check_all_listenings_exist() self.get_playlist_number_and_id() self.clear_the_playlist() self.do_the_sums() self.shuffle_tracks_in_database() exit() def check_listening(self, listening_num): conn = self.create_connection(self.database) with conn: cur = conn.cursor() try: cur.execute("SELECT NUMBER FROM your_listening_playlists") except Error as e: print(e) return rows = cur.fetchall() for row in rows: if row[0] == listening_num: cur.execute("UPDATE your_listening_playlists SET EXIST = True WHERE NUMBER = '" + listening_num + "'") conn.commit() return def check_all_listenings_exist(self): self.this_pl = False conn = self.create_connection(self.database) with conn: cur = conn.cursor() try: cur.execute("SELECT * FROM your_listening_playlists") except Error as e: print(e) return rows = cur.fetchall() for row in rows: if row[2] == "" or row[3] == 0: self.listening_pl_name = row[1] self.listening_pl_num = row[0] self.create_new_playlist() self.listening_update_pl_name() return def get_playlist_number_and_id(self): conn = self.create_connection(self.database) with conn: cur = conn.cursor() try: cur.execute("SELECT PLAYLIST FROM playlist_number ") except OperationalError as e: if 'no such table' in str(e): print("ERROR playlist_number table is missing \n Please run Sauron.py to rectify this problem") exit() else: print(e) exit() rows = cur.fetchall() for row in rows: self.listening_plnum = (row[0]) self.get_listening_id(conn) new_pln = int(self.listening_plnum) new_pln = new_pln + 1 if new_pln == 11: new_pln = 1 cur.execute("UPDATE playlist_number SET PLAYLIST = '" + str(new_pln) + "'") return def get_listening_id(self, conn): cur = conn.cursor() try: cur.execute("SELECT * FROM your_listening_playlists WHERE NUMBER = '" + self.listening_plnum + "'") except Error as e: print(e) return rows = cur.fetchall() if rows == None: self.norow = True return for row in rows: self.listening_playlist_id = row[2] self.build_name = row[1] def clear_the_playlist(self): query = "https://api.spotify.com/v1/playlists/{}/tracks?uris={}".format(self.listening_playlist_id, "spotify:track:3lpDrxUkr0tIe1kmJvdK7d") response = requests.request("put", url=query, headers={"Content-type": "application/json","Authorization":"Bearer {}".format(self.spotify_token)}) query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.listening_playlist_id) response = requests.request("delete", url=query, headers={"Content-type": "application/json","Authorization":"Bearer {}".format(self.spotify_token)}, data=json.dumps({"tracks": [{"uri": "spotify:track:3lpDrxUkr0tIe1kmJvdK7d"}]})) self.track_uri = "" return def do_the_sums(self): self.total_tracks = self.tt[4] + self.tt[5] + self.tt[6] + self.tt[7] + self.tt[8] + self.tt[9] + self.tt[0] + self.tt[1] + self.tt[2] self.pt[0] = math.ceil(self.queue_size*self.tt[0]/self.total_tracks) self.pt[1] = math.ceil(self.queue_size*self.tt[1]/self.total_tracks) self.pt[2] = math.ceil(self.queue_size*self.tt[2]/self.total_tracks) self.pt[3] = math.ceil(self.queue_size*self.tt[3]/self.total_tracks) self.pt[4] = math.ceil(self.queue_size*self.tt[4]/self.total_tracks) self.pt[5] = math.ceil(self.queue_size*self.tt[5]/self.total_tracks) self.pt[6] = math.ceil(self.queue_size*self.tt[6]/self.total_tracks) self.pt[7] = math.ceil(self.queue_size*self.tt[7]/self.total_tracks) self.pt[8] = math.ceil(self.queue_size*self.tt[8]/self.total_tracks) self.pt[9] = math.ceil(self.queue_size*self.tt[9]/self.total_tracks) for x in range(10): if self.pl_name[x] != "": print("Working On Decade Playlist.. '" + self.pl_name[x] + "'") self.get_tracks_from_each_database(self.pt[x], self.pl_name[x]) self.clear_the_playlist() return def get_tracks_from_each_database(self, tracks, list_name): list_name = list_name.replace(" ", "_") list_name = list_name.replace("-", "_") list_name = list_name.replace("&", "and") list_name = list_name.replace(":", "_") list_name = ("Decade_" + str(list_name)) if list_name == "Decade_": return conn = self.create_connection(self.database) with conn: self.select_all_rows(conn, tracks, list_name) self.check_played_count(conn, list_name) return def select_all_rows(self, conn, add_tracks, list_name): cur = conn.cursor() try: cur.execute("SELECT * FROM '" + list_name + "' WHERE PLAY_COUNT = 0 order by RANDOM()") except Error as e: print(e) return rows = cur.fetchall() for row in rows: self.temp_newtrack = row[0] self.temp_name = row[1] cur.execute("UPDATE '" + list_name + "' SET PLAY_COUNT = 1 WHERE URI = '"+ self.temp_newtrack + "'") cur.execute(""" CREATE TABLE IF NOT EXISTS tmp_shuffle (URI text PRIMARY KEY, NAME text); """)## conn.commit() tracks = (self.temp_newtrack, self.temp_name)## sql = ''' INSERT INTO tmp_shuffle(URI,NAME) VALUES(?,?) ''' cur.execute(sql, tracks) conn.commit() add_tracks = add_tracks - 1 if add_tracks == 0: conn.commit() return return def check_played_count(self, conn, list_name): cur = conn.cursor() try: cur.execute("SELECT COUNT (*) FROM '" + list_name + "'") total_records = cur.fetchone() cur.execute("SELECT COUNT (*) FROM '" + list_name + "' WHERE PLAY_COUNT = 1") played_records = cur.fetchone() played_percentage = played_records[0]/total_records[0]*100 self.total_played_percentage = self.total_played_percentage + played_percentage #================================================ if played_percentage > 99: #================================================ self.decade_database_reset(list_name) except Error as e: print(e) return def shuffle_tracks_in_database(self): print("Building the Playlist.." + str(self.build_name)) conn = self.create_connection(self.database) with conn: cur = conn.cursor() try: cur.execute("SELECT * FROM tmp_shuffle order by RANDOM()") except Error as e: print(e) return rows = cur.fetchall() for row in rows: newtrack = row[0] self.add_track_to_playlist(newtrack) cur.execute("DROP TABLE IF EXISTS tmp_shuffle") conn.execute("VACUUM") conn.commit() self.total_played_percentage = self.total_played_percentage/9 print("{:.3}".format(self.total_played_percentage) + "% of All Tracks used") return def add_track_to_playlist(self, newtrack): query = "https://api.spotify.com/v1/playlists/{}/tracks?uris={}".format(self.listening_playlist_id, newtrack) response = requests.request("post", url=query, headers={"Content-type": "application/json","Authorization":"Bearer {}".format(self.spotify_token)}) return def create_new_playlist(self): query = "https://api.spotify.com/v1/users/{}/playlists".format(self.user_id) request_body = json.dumps({"name": "" + self.listening_pl_name + "", "description": "1 of 10 Random Decades Playlists. Created: " + self.date_time_.strftime("%b %d %Y"), "public": True}) response = requests.post(query, data=request_body, headers={"Content-Type": "application/json","Authorization": "Bearer {}".format(self.spotify_token)}) response_json = response.json() self.listening_playlist_id = response_json["id"] return def listening_update_pl_name(self): conn = self.create_connection(self.database) with conn: cur = conn.cursor() try: cur.execute("SELECT * FROM your_listening_playlists") except Error as e: print(e) return rows = cur.fetchall() for row in rows: if row[0] == self.listening_pl_num: cur.execute("UPDATE your_listening_playlists SET EXIST = True, URI = '" + self.listening_playlist_id + "' WHERE NUMBER = '" + self.listening_pl_num + "'") if not self.this_pl: cur.execute("UPDATE playlist_number SET PLAYLIST = '" + self.listening_pl_num + "'") self.this_pl = True conn.commit() return def decade_database_reset(self, list_name): print("Reseting Database '" + list_name + "'") conn = self.create_connection(self.database) with conn: cur = conn.cursor() try: cur.execute("SELECT PLAY_COUNT FROM '" + list_name + "'") except Error as e: print(e) return rows = cur.fetchall() for row in rows: cur.execute("UPDATE '" + list_name + "' SET PLAY_COUNT = 0") return a = PlaylistGrab() a.call_refresh() a.reset_listening_database() a.get_the_playlists()