import os import json import requests import time import sqlite3 from sqlite3 import Error from secrets import spotify_user_id, shuffle_source_id, shuffle_target_id from refresh import Refresh refreshCaller = Refresh() job_done = False os.system('cls') print("[•_•] Shuffle - Playlist Shuffler.") print("Version 3.0.0: 08-APR-2022 @ 14:01 (c) Viking Design.") print("User: " + spotify_user_id) class ShuffleFeed: def __init__(self): self.user_id = spotify_user_id self.shuffle_source_playlist_id = shuffle_source_id self.shuffle_target_playlist_id = shuffle_target_id self.spotify_token = "" self.track = "" self.trk = 1 self.getname = "" self.here_i_am = os.path.dirname(os.path.abspath(__file__)) self.database = self.here_i_am + r"\Playlist Shuffle Tracks - " + spotify_user_id + ".db" def the_shuffle(self): accepted = '' while accepted.lower() == '': accepted = input(f'\n\nDO YOU WANT TO UPDATE THE DATABASE [y/n] :') #NOTE Make sure the focus is in the right place before hitting y or n if accepted.lower() == 'n': database_status = os.path.exists(self.here_i_am + r"\Playlist Shuffle Tracks - " + spotify_user_id + ".db") if not database_status: print("No Database Present Therefore...") print("Database Update Required...") self.tic = time.perf_counter() db_refresh = True self.database_refresh(db_refresh) exit() else: print("No Database Update Required...") self.tic = time.perf_counter() db_refresh = False self.shuffle_source_playlist_id = self.shuffle_target_playlist_id #self.clear_the_playlist() #self.database_refresh(db_refresh) self.shuffle_tracks_in_database() exit() if accepted.lower() == 'y': print("Database Update Required...") self.tic = time.perf_counter() db_refresh = True self.database_refresh(db_refresh) exit() else: print("Incorrect Key Pressed!!") self.the_shuffle() def clear_the_playlist(self): #NOTE Empties the playlist query = "https://api.spotify.com/v1/playlists/{}/tracks?uris={}".format(self.shuffle_target_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.shuffle_target_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 database_refresh(self, db_refresh): # NOTE Rebuilds the database from the playlist if db_refresh == True: try: os.remove(self.here_i_am + "\Playlist Shuffle Tracks - " + spotify_user_id + ".db") except: pass while True: self.get_track_from_playlist(db_refresh) if job_done: toc = time.perf_counter() ty_res = time.gmtime(toc - self.tic) res = time.strftime("%H Hours :%M Minutes :%S Seconds",ty_res) print(f"Total Shuffle Process Took {res} To Complete" ) return if db_refresh == True: self.add_to_db() self.remove_track_from_playlist() def check_database(self): conn = self.create_connection(self.database) cur = conn.cursor() with conn: cur.execute("CREATE TABLE IF NOT EXISTS tracks (URI text PRIMARY KEY, NAME text);") 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 add_to_db(self): #NOTE: Adds New Track To Database print("Adding track " + str(self.trk) + " to Database", end='\r') self.trk+=1 conn = self.create_connection(self.database) with conn: tracks = (self.track_uri, self.getname) sql = ''' INSERT INTO tracks(URI,NAME) VALUES(?,?) ''' cur = conn.cursor() cur.execute(sql, tracks) return cur.lastrowid def get_track_from_playlist(self, db_refresh): # NOTE Gets each track from the playlist global job_done query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.shuffle_source_playlist_id) response = requests.request("get", url=query, headers={"Content-type": "application/json","Authorization":"Bearer {}".format(self.spotify_token)}, params={"limit":1}) response_json = response.json() json_data = json.loads(response.text) try: self.track_uri = json_data["items"][0]["track"]["uri"] self.track = json_data["items"][0]["track"]["id"] self.getname = json_data["items"][0]["track"]["name"] except: toc = time.perf_counter() ty_res = time.gmtime(toc - self.tic) res = time.strftime("%H Hours :%M Minutes :%S Seconds",ty_res) if db_refresh == True: print() print(f"Adding Tracks to Database & Removing From Playlist Took {res} To Complete") else: print(f"Removing Tracks From Playlist Took {res} To Complete") self.shuffle_tracks_in_database() job_done = True return if response.status_code == 200: self.check_database() return def shuffle_tracks_in_database(self): # NOTE Shuffles the database tracks print("Shuffling Tracks back to playlist") conn = self.create_connection(self.database) cur = conn.cursor() try: cur.execute("SELECT * FROM tracks order by RANDOM()") except: print("No Tracks In Source Playlist......") return rows = cur.fetchall() for row in rows: newtrack = str(row[0]) self.add_track_to_playlist(newtrack) return def remove_track_from_playlist(self): # NOTE Removes track from source playlist query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.shuffle_source_playlist_id) response = requests.request("delete", url=query, headers={"Content-type": "application/json","Authorization":"Bearer {}".format(self.spotify_token)}, data=json.dumps({"tracks": [{"uri": self.track_uri}]})) self.track_uri = "" def add_track_to_playlist(self, newtrack): # NOTE Adds track to target playlist query = "https://api.spotify.com/v1/playlists/{}/tracks?uris={}".format(self.shuffle_target_playlist_id, newtrack) response = requests.request("post", url=query, headers={"Content-type": "application/json","Authorization":"Bearer {}".format(self.spotify_token)}) def call_refresh(self): #NOTE Refreshes Spotify Token global refreshCaller while True: self.spotify_token = refreshCaller.refresh() if self.spotify_token == -1: print( "==============================\n[!] Error refreshing token, retrying in 30 Seconds\n==============================") time.sleep(30) else: break a = ShuffleFeed() a.call_refresh() a.the_shuffle()