import os import json import requests import time import sqlite3 from progress.bar import ShadyBar from sqlite3 import Error from datetime import datetime from secrets import spotify_user_id from refresh import Refresh refreshCaller = Refresh() job_done = False class sauron: def __init__(self): self.user_id = spotify_user_id #NOTE User Name self.spotify_token = "" #NOTE Access Token self.more_than_fifty = 0 #NOTE If you have more than 50 playlists this will deal with it self.this_pl_count = 0 #NOTE Sets the number of playlsits to 0 self.here_i_am = os.path.dirname(os.path.abspath(__file__)) #NOTE The current path self.total_tracks = 0 #NOTE Sets the number of tracks in any playlist to 0 self.play_count = 0 #NOTE Sets track play count to 0 self.check = False #NOTE Sets the track "HERE" value to False self.plt = False #NOTE Sets the playlist table check to False self.listening_pl = False #NOTE Sets the listening playlist check to false self.database = self.here_i_am + r"\\All Playlists Tracks for " + spotify_user_id + ".db" #NOTE The Database path #NOTE The following procedures are arranged in the order they are called. # This is not necessarily the same for every call as variable and database status will vary. # This is to satisfy the NOU Boffins requirements :-) def call_refresh(self): #NOTE Euans 2nd Fix 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 def do_the_business(self): #NOTE Iniate the App tic = time.perf_counter() #NOTE Set the Timer self.pl_count = False try: os.remove(self.here_i_am + r"\\Sauron Errors - " + spotify_user_id + ".txt") #NOTE Delete the Error file if it exists except: pass #NOTE Ignore the error if it doesn't while True: self.db_newpl() #NOTE Check The "Grab and Go" Table is present self.check_your_db() #NOTE Check the "listening playlists" self.get_the_playlists() #NOTE Get all the playlists and tracks into the Database if job_done: toc = time.perf_counter() #NOTE Stop the timer ty_res = time.gmtime(toc - tic) #NOTE Work out the Time Taken res = time.strftime("%H Hours :%M Minutes :%S Seconds",ty_res) #NOTE Convert it into Hours, Minutes and Seconds print(f"Sauron Took {res} Seconds To Add {self.playlist_count} Playlists and a Total of {self.total_tracks} Records To The Database" ) return def db_newpl(self): #NOTE Refresh the "listening or 'grab and go' playlists" database table so it remembers where it was up to. global database, conn conn = self.create_connection(self.database) #NOTE Connect to The Database cur = conn.cursor() #NOTE Set The Cursor with conn: #NOTE With the connection try: cur.execute("CREATE TABLE IF NOT EXISTS playlist_number (PLAYLIST text PRIMARY KEY);") #NOTE Check if the playlist_number Table exists and create it if it doesn't cur.execute("SELECT PLAYLIST FROM playlist_number") #NOTE Select the latest Playlist Number from the table rows = cur.fetchone() #NOTE Get the current Playlist Number if rows == None: #NOTE If the record doesn't exist rows = (1,) #NOTE Set it to 1 self.playlist_number = rows[0] #NOTE Put the read value in a variable cur.execute("DROP TABLE IF EXISTS playlist_number") #NOTE Get rid of the playlist_numbeer table cur.execute("CREATE TABLE IF NOT EXISTS playlist_number (PLAYLIST text PRIMARY KEY);") #NOTE Create a new playlist_number table cur.execute("INSERT INTO playlist_number VALUES('" + str(self.playlist_number) + "')") #NOTE Put the value back in the table conn.commit() #NOTE Commit the changes except Error as e: print("E") #NOTE Error print(e) #NOTE Handling return def check_your_db(self): #NOTE Check "Your Listening Playlists" plnum = 1 #NOTE Set A Few Dummy Variables.... pluri = "" #NOTE In Case The Table..... plexist = 0 #NOTE Doesn't Exist..... conn = self.create_connection(self.database) #NOTE Conn to The Database cur = conn.cursor() #NOTE Set the Cursor with conn: #NOTE With the connection try: cur.execute("CREATE TABLE IF NOT EXISTS your_listening_playlists (NUMBER text PRIMARY KEY, NAME text, URI text, EXIST integer); ") #NOTE Create the Table if it doesn't exist while plnum < 11: #NOTE Make sure there is no more than 10 playlists in the table try: plname = "Your Playlist No. " + str(plnum) #NOTE Set the DUmmy Playlist name playrec = (plnum, plname, pluri, plexist) #NOTE Set the sql Values for adding Dummy playlists sql = "INSERT INTO your_listening_playlists (NUMBER,NAME,URI,EXIST) VALUES(?,?,?,?)" #NOTE Put the Dummy data into the table plnum = plnum + 1 #NOTE Increment the playlist Number cur.execute(sql, playrec) #NOTE Execute the sql conn.commit() #NOTE #NOTE Commit rthe changes except: continue #NOTE If the table and records already exist it will cause an error. This ignores it and carries on.... except Error as w: #NOTE Other.... print("W") #NOTE Error.... print(w) #NOTE Handling.... def create_connection(self, db_file): #NOTE Create a connection to the database conn = None #NOTE Clear any existing connections try: conn = sqlite3.connect(db_file) #NOTE Set the connection to the Database return conn #NOTE Send trhe connection details back whence it came except Error as e1: print("E1") #NOTE Error print(e1) #NOTE Handling return conn def get_the_playlists(self): #NOTE Get "every" playlist from the logged on user. A straight lift from 'Playlist Grabber' with modifications global job_done params={"limit":50,"offset":self.more_than_fifty} #NOTE API Call Stuff 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() if not self.pl_count: #NOTE If this hasn't been checked self.playlist_count = response_json["total"] #NOTE Get the Total Number of playlists self.pl_count = True #NOTE Mark it as Done for i in response_json["items"]: #NOTE For each playlist self.decade_pl = False #NOTE Set the Decade Playlist Flag to off self.plrst = False #NOTE Set the Playlist Reset Flag to off self.playlist_uri = (i["uri"]) #NOTE Get the playlist URI self.playlist_id = (i["id"]) #NOTE Get the playlist ID self.playlist_name = (i["name"]) #NOTE Get the playlist name self.total = (i["tracks"]["total"]) #NOTE Get the playlist track count self.listening = self.playlist_name[5:13] #NOTE check for grab and go playlist check = (self.playlist_name[-1:]) #NOTE check for decade playlist if self.listening == "Playlist": #NOTE It's a 'grab and go' playlist self.listening_pl = True #NOTE Ignore these playlists in Progress Bar continue #NOTE Don't do anything with these just move on (next playlist) self.add_to_playlist_table() #NOTE Add the playlist to the playlist_name table if check == "9": #NOTE It's a 'decade' playlist self.decade_pl = True #NOTE Set the Decade Playlist Flag to on self.date_decade = (self.playlist_name[-2:-1]) #NOTE Get the decade from the playlist name else: self.date_decade = " " #NOTE If its not a decade playlist set the decade to null self.dbtable = self.playlist_name #NOTE Prepare to modify the playlist name to suit the Database requirements self.dbtable = self.dbtable.replace(" ", "_") #NOTE Replace any spaces in the playlist name with an underscore self.dbtable = self.dbtable.replace("-", "_") #NOTE Replace any hyphens in the playlist name with an underscore self.dbtable = self.dbtable.replace("&", "and") #NOTE Replace any ampersands in the playlist name with "and" self.dbtable = self.dbtable.replace(":", "_") #NOTE Replace any colons in the playlist name with an underscore #NOTE This may not be the full list but is all I have come across so far if self.decade_pl == True: #NOTE If it is a decade playlist self.dbtable = ("Decade_" + str(self.dbtable)) #NOTE Rename the playlist because the Database does not like things starting with a number if self.plrst == False: #NOTE If the playlist tracks have not been reset self.plrst = True #NOTE Set the flag to on self.reset_playlist_database() #NOTE Reset the playlist tracks (This sorts out additions and deletions from the playlist) self.this_pl_count = self.this_pl_count + 1 #NOTE Increment the playlist count self.get_track_from_playlist() #NOTE Get the playlists tracks self.update_playlist_track_count() #NOTE Put the database track count in the playlist table if self.more_than_fifty == 0: #NOTE If there are more than 50 playlists self.more_than_fifty = 50 #NOTE Set the starting point at number 50 self.get_the_playlists() #NOTE Start again at number 50 job_done = True #NOTE When all playlists are done go to the finish def add_to_playlist_table(self): #NOTE Add the playlist name to the database playlist_name table conn = self.create_connection(self.database) #NOTE Connect to the Database cur = conn.cursor() #NOTE Set the cursor try: with conn: #NOTE With the connection if self.plt == False: #NOTE If it is the first playlist cur.execute("DROP TABLE IF EXISTS playlist_list") #NOTE Delete the old playlist_name table conn.commit() #NOTE Commit the deletion self.plt = True #NOTE Set this flag so that the table isn't deleted again cur.execute(" CREATE TABLE IF NOT EXISTS playlist_list (URI text PRIMARY KEY, NAME text, SPOTTY_COUNT text, DB_COUNT text); ") #NOTE Create a new table conn.commit() #NOTE Commit the Addition playlist = (self.playlist_uri, self.playlist_name, self.total, "0") #NOTE Set the SQL Values sql = ' INSERT INTO playlist_list (URI,NAME,SPOTTY_COUNT,DB_COUNT) VALUES (?,?,?,?) ' #NOTE Formulate the SQL statement to add the playlist record cur.execute(sql, playlist) #NOTE Execute the SQL statement conn.commit() #NOTE Commit the Addition return cur.lastrowid #NOTE Go back with the last row known except Error as e6: #NOTE \ print("E6") #NOTE \\ Error print(e6) #NOTE // Handling return cur.lastrowid #NOTE / def reset_playlist_database(self): #NOTE Reset all playlist tracks to not here conn = self.create_connection(self.database) #NOTE Connect to the database with conn: #NOTE With the connection cur = conn.cursor() #NOTE Set the cursor try: cur.execute("SELECT HERE FROM '" + self.dbtable + "'") #NOTE Get the "HERE" state from all records except Error as e4: return #NOTE Error rows = cur.fetchall() #NOTE Handling for row in rows: #NOTE For each track record cur.execute("UPDATE '" + self.dbtable + "' SET HERE = 0") #NOTE Set the "HERE" state to 0 return #NOTE Go back def get_track_from_playlist(self): #NOTE Get all the tracks from the playlist global job_done self.call_refresh() #NOTE Refresh the Spotify Token self.clear_the_screen() #NOTE Clear the screen to make the Progress Bar look better now = datetime.now() #NOTE Get the Date and TIme current_time = now.strftime("%H:%M:%S") #NOTE Set the time format self.total_tracks = self.total + self.total_tracks #NOTE Update the total tracks to be dealt with print(str(self.total) + " Tracks to add to database from " + self.playlist_name + " playlist. Started " + current_time + " That makes " + str(self.total_tracks) + " in Total. ") #NOTE Print some data if self.total != 0: #NOTE If the playlist isn't empty if self.listening_pl: self.playlist_count = self.playlist_count - 10 self.listening_pl = False with ShadyBar('Processing Playlist ' + self.playlist_name + '. Playlist ' + str(self.this_pl_count) + ' of ' + str(self.playlist_count) + ' ', max=self.total) as bar: #NOTE Set up the Progress Bar data t = 0 #NOTE Set a counter variable for each track while t < self.total: #NOTE For each track self.offset = t #NOTE Setup the API offset query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.playlist_id) #NOTE API Call response = requests.request("get", url=query, headers={"Content-type": "application/json","Authorization":"Bearer {}".format(self.spotify_token)}, params={"limit":1,"offset":self.offset}) response_json = response.json() json_data = json.loads(response.text) try: self.track_uri = json_data["items"][0]["track"]["uri"] #NOTE Get the track URI self.track_name = json_data["items"][0]["track"]["name"] #NOTE Get the track Name self.artist_name = json_data["items"][0]["track"]["artists"][0]["name"] #NOTE Get the Track Artist flag = False #NOTE Set a pre-error flag to off if flag == False: #NOTE If it is False self.last_uri = self.track_uri #NOTE Set the last URI self.last_track = self.track_name #NOTE Set the last track name self.last_artist = self.artist_name #NOTE Set the last track artist flag = True #NOTE Set the flag to on if response.status_code == 200: #NOTE If the data valid self.check_database() #NOTE See if the track exists in the Database bar.next() #NOTE Move the Progress Bar on a bit t = t + 1 #NOTE Increment the track count continue #NOTE next track except: File_object = open(self.here_i_am + r"\\Sauron Errors - " + spotify_user_id + ".txt","a") #NOTE Next 4 lines Write some data to the error file. File_object.writelines(["Playlist Name - " + str(self.dbtable) + " Track - " + str(t+1) + " Total Tracks - " + str(self.total) + "\n"]) File_object.writelines(["Last URI - " + str(self.last_uri) + " Last Track - " + str(self.last_track) + " Last Artist - " + str(self.last_artist) + "\n\n"]) File_object.close() t = t - 2 #NOTE If the data fails add relevent stuff to a file and set the track back 2 to prevent omissions continue #NOTE next track self.clear_db_deleted_tracks() #NOTE Delete any tracks not in the playlist now from the Database return #NOTE Go Back def clear_the_screen(self): #NOTE Self Explanatory os.system('cls') print("[•_•] Sauron Playlist Database Manager - Get All Users Playlists Tracks (NOT Temporary Playlists) and") print("put them in a single Database with Tables Named as the Playlist Name.") print("Remove any Playlist Tables where the Playlist is Empty. Maintain accuracy of the") print("Database tables by Adding or Removing Tracks as in the Playlist") print("Also creates 3 Database Tables:- 1. playlist_list - A full list of all user playlists containing") print("The playlist URI, Name and track counts from both the Database and Spotify. Any differences between the 2 counts indicating") print("'Double Entries in the Playlist on Spotify. 2. your_listening_playlists and 3. playlist_number. These are used in the") print("production of 10 playlists generated from the Decades playlists for your listening pleasure") print("This app should be run weekly OR when any 'Major' changes have been made to user playlists.") print("Version 1.383.02 : 28-Jul-2021 @ 17:15 (c) Viking Design. (^o.o^)") print("User: " + spotify_user_id) return def check_database(self): #NOTE Check if track exists in database global conn conn = self.create_connection(self.database) #NOTE Connect to Database cur = conn.cursor() #NOTE Set the cursor try: with conn: #NOTE With the connection cur.execute(""" CREATE TABLE IF NOT EXISTS """ + self.dbtable + """ (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, DECADE integer, PLAY_COUNT integer, HERE integer); """) #NOTE Create the database table if it doesn't exist conn.commit() #NOTE Commkit the Creation cur.execute("SELECT * FROM '" + self.dbtable + "' WHERE URI = '" + self.track_uri + "'") #NOTE Check if the track exists data = cur.fetchall() #NOTE Get the data fro the track if len(data)== 0: #NOTE If it doesn't exist (i.e. New Track) self.add_to_db(conn) #NOTE Add it to the Database else: #NOTE If it does exist cur.execute("UPDATE '" + self.dbtable + "' SET HERE = True WHERE URI = '" + self.track_uri + "'") #NOTE Set the "HERE" flag to 1 conn.commit() #NOTE Commit the change return #NOTE Go Back except Error as e2: print("E2") #NOTE Error print(e2) #NOTE Handling return #NOTE Go Back def add_to_db(self, conn): #NOTE Add new track to Database self.check = True #NOTE Set the check flag to True with conn: #NOTE With the connection try: tracks = (self.track_uri, self.track_name, self.artist_name, self.date_decade, self.play_count, self.check) #NOTE Set the SQL Values sql = ''' INSERT OR IGNORE INTO ''' + self.dbtable + '''(URI,TRACK_NAME,ARTIST_NAME,DECADE,PLAY_COUNT,HERE) VALUES(?,?,?,?,?,?) ''' #NOTE Formulate the SQL statement to add the playlist record cur = conn.cursor() #NOTE Set the cursor cur.execute(sql, tracks) #NOTE Execute the SQL statement conn.commit() #NOTE Commit the SQL return cur.lastrowid #NOTE return with the last row id except sqlite3.IntegrityError as e3: print("E3") #NOTE Error print(e3) #NOTE Handling return cur.lastrowid #NOTE return with the last rowid def clear_db_deleted_tracks(self): #NOTE Delete any tracks not in playlist from Database conn = self.create_connection(self.database) #NOTE Connect to the database with conn: #NOTE With the connection cur = conn.cursor() #NOTE Set the cursor try: cur.execute("DELETE from '" + self.dbtable + "' WHERE HERE = 0") #NOTE Delete any tracks wher "HERE" is 0 conn.commit() #NOTE Commit the deletion except Error as e5: print("E5") #NOTE Error print(e5) #NOTE Handling return #NOTE Go Back cur.execute("SELECT COUNT (*) FROM '" + self.dbtable + "'") #NOTE Get how many records in the database and therefore playlist total_records = cur.fetchone() #NOTE Get the total records value if total_records[0] == 0: #NOTE If there are no records in table cur.execute("DROP TABLE IF EXISTS '" + self.dbtable + "'") #NOTE Delete the table (NOT THE PLAYLIST) conn.commit() #NOTE Commit the deletion return #NOTE Go Back def update_playlist_track_count(self): conn = self.create_connection(self.database) #NOTE Connect to the database with conn: #NOTE With the connection cur = conn.cursor() #NOTE Set the cursor try: cur.execute("SELECT COUNT (*) FROM '" + self.dbtable + "'") #NOTE Get how many records in the database for playlist total_records = cur.fetchone() #NOTE Get the count result total_records = total_records[0] #NOTE Get only the total records value cur.execute("UPDATE playlist_list SET DB_COUNT = '" + str(total_records) + "' WHERE NAME = '" + self.playlist_name + "'") #NOTE Update the database conn.commit() #NOTE Commit the update return #NOTE Go Back except Error as e8: print("E8") #NOTE Error print(e8) #NOTE Handling return #NOTE Go Back a = sauron() #NOTE Set the initial Variables a.call_refresh() #NOTE Refresh the Access Token a.do_the_business() #NOTE Start the main program