from datetime import datetime import shutil, sys, os, sqlite3 sys.path.insert(1, 'j:\\Spotty\\secrets_and_database') from alive_progress import alive_bar from sqlite3 import Error from endless_playlist_secrets import master_id, spotify_user_id from pkce_auth import Authorize authcase = Authorize() import spotipy job_done = False class violet: def __init__(self): self.here_i_am = os.path.dirname(os.path.abspath(__file__)) self.dbpath = 'j:\\Spotty\\secrets_and_database' self.play_count = 0 self.present = False self.played = False self.playlist_id = master_id self.deleted_records = 0 self.added_records = 0 self.track_uri_list = [] self.track_name_list= [] self.artist_name_list = [] self.album_name_list = [] self.release_date_list = [] def violate_the_database(self): # NOTE Delete error file and start the Party! self.get_the_playlist() if self.added_records == 0 and self.deleted_records[0] == 0: print("\033[34m######################################################\033[0m") print("\033[32mNo records changed. Dupefinder not required.\033[0m") print("\033[34m######################################################\033[0m") print() print() exit() import DupeFinder import htmlbuilder def create_connection(self, db_file): # NOTE Create a database connection conn = None try: conn = sqlite3.connect(db_file) return conn except Error as e1: print() print("Create_Connection") print(e1) return conn def get_the_playlist(self): # NOTE Set-up a few things and start the main job self.get_track_count_from_playlist() self.database = self.dbpath + r"\\" + self.playlist_name + ".db" self.backup_database() self.backup_Enomous_database() self.database_setup() self.reset_playlist_database() while True: self.get_track_from_playlist() if job_done: print (str(self.added_records) + " Tracks added to database") print (str(self.deleted_records[0]) + " Tracks removed from database") print("") return def get_track_count_from_playlist(self): #NOTE Get how many tracks in Endless Feed query = self.sp_base.playlist(self.playlist_id) #NOTE Get the Endless Feed Playlist details self.total = query['tracks']['total'] #NOTE Pick out the total tracks from the details self.queue_owner = query['owner']['id'] #NOTE Get the Username of the Queue Owner (You or me!) self.playlist_name = query['name'] #NOTE Get the Playlist name return def database_setup(self): global conn conn = self.create_connection(self.database) cur = conn.cursor() try: with conn: cur.execute("DROP TABLE IF EXISTS Last_Run_Additions") cur.execute("DROP TABLE IF EXISTS Last_Run_Deletions") cur.execute("CREATE TABLE IF NOT EXISTS Last_Run_Additions (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, ALBUM_NAME text);") cur.execute("CREATE TABLE IF NOT EXISTS Last_Run_Deletions (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, ALBUM_NAME text);") conn.commit() except Error as e2: print() print("database_setup") print(e2) return def reset_playlist_database(self): # NOTE Reset the "present" flag for every track. These will be set if the track is still there conn = self.create_connection(self.database) with conn: cur = conn.cursor() try: cur.execute("SELECT PRESENT FROM Track_List") except Error as e4: return rows = cur.fetchall() cur.execute("UPDATE Track_List SET PRESENT = 0") return def get_track_from_playlist(self): # NOTE Step through every track in playlist getting info about it # NOTE Error trap for loss of connection global job_done self.tl = False os.system('cls') print("Violet v.3.4.2 01/07/2023 Viking Utilities") print("\033[91m########################### Latest Update ###########################\033[0m") print("\033[33mv.3.4.2 Usage of Lists to speed up operation....\033[0m") print("\033[91m#####################################################################\033[0m") print(str(self.total) + " Tracks To Check In The " + self.playlist_name + " playlist" ) t = 1 results = self.sp_base.playlist_items(self.playlist_id) #NOTE Get the playlists track details tracks = results['items'] #NOTE Set theTracks to a variable track_number = 1 #NOTE Start a Track count while results['next']: #NOTE While there is another track to do results = self.sp_base.next(results) #NOTE Get the next track details print("Working on Track "+str(t)+ " of " + str(self.total), end="\r") tracks.extend(results['items']) t+=100 print() #NOTE Add the next tracks details to a list for track in tracks: print("Working on Track "+str(track_number)+ " of " + str(self.total), end="\r") try: self.track_uri = track['track']['uri'] self.track_name = track["track"]["name"] self.artist_name = track["track"]["artists"][0]["name"] self.album_name = track["track"]["album"]["name"] self.release_date = track["track"]["album"]["release_date"] self.configure_the_information() self.track_uri_list.append(self.track_uri) self.track_name_list.append(self.track_name) self.artist_name_list.append(self.artist_name) self.album_name_list.append(self.album_name) self.release_date_list.append(self.release_date) flag = False if flag == False: self.last_uri = self.track_uri self.last_track = self.track_name self.last_artist = self.artist_name flag = True track_number+=1 except spotipy.SpotifyException as er1: print(er1) print("Well that didn't work did it!! Amend track list and playlist function") print("Track " +self.track_name+ " Artist " +self.artist_name) continue print() self.check_database() self.clear_db_deleted_tracks() job_done = True return def check_database(self): global conn self.item = 0 self.list_len = len(self.track_uri_list) t=0 conn = self.create_connection(self.database) cur = conn.cursor() try: with conn: if not self.tl: cur.execute("CREATE TABLE IF NOT EXISTS Track_List (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, ALBUM_NAME text, RELEASE_DATE text, PLAY_COUNT integer, PRESENT integer, QUEUED integer);") conn.commit() self.tl = True while self.item < self.list_len: with alive_bar(self.list_len, title="Checking Database Presence") as bar: for self.item in range(self.list_len): if self.item == self.list_len-2: print() cur.execute("SELECT * FROM Track_List WHERE URI = '" + self.track_uri_list[self.item] + "'") data = cur.fetchone() if data is None: self.added_records += 1 self.add_to_db(conn) else: cur.execute("UPDATE Track_List SET PRESENT = True WHERE URI = '" + self.track_uri_list[self.item] + "'") conn.commit() self.item +=1 bar() continue except Error as e2: print() print("Check_Database") print(e2) return def add_to_db(self, conn): # NOTE Add the track to the database self.present = True with conn: try: tracks = (self.track_uri_list[self.item], self.track_name_list[self.item], self.artist_name_list[self.item], self.album_name_list[self.item], self.release_date_list[self.item], self.play_count, self.present, self.played) # NOTE Set the SQL Values sql = ''' INSERT OR IGNORE INTO Track_List (URI,TRACK_NAME,ARTIST_NAME,ALBUM_NAME,RELEASE_DATE,PLAY_COUNT,PRESENT,QUEUED) VALUES(?,?,?,?,?,?,?,?) ''' cur = conn.cursor() cur.execute(sql, tracks) conn.commit() tracks2 = (self.track_uri_list[self.item], self.track_name_list[self.item], self.artist_name_list[self.item], self.album_name_list[self.item]) sql2 = ''' INSERT OR IGNORE INTO Last_Run_Additions (URI,TRACK_NAME,ARTIST_NAME,ALBUM_NAME) VALUES(?,?,?,?) ''' cur = conn.cursor() cur.execute(sql2, tracks2) conn.commit() return cur.lastrowid except sqlite3.IntegrityError as e3: print() print("Add_To_DB") print(e3) return cur.lastrowid def clear_db_deleted_tracks(self): # NOTE Get rid of any tracks not in the playlist but still in the database conn = self.create_connection(self.database) with conn: cur = conn.cursor() try: cur.execute("SELECT * from Track_List WHERE PRESENT = 0") rows = cur.fetchall() for row in rows: tracks2 = (row[0], row[1],row[2], row[3]) sql2 = ''' INSERT OR IGNORE INTO Last_Run_Deletions (URI,TRACK_NAME,ARTIST_NAME,ALBUM_NAME) VALUES(?,?,?,?) ''' cur = conn.cursor() cur.execute(sql2, tracks2) conn.commit() continue cur.execute("SELECT COUNT (*) from Track_List WHERE PRESENT = 0") self.deleted_records = cur.fetchone() cur.execute("DELETE from Track_List WHERE PRESENT = 0") conn.commit() except Error as e5: print() print("Clear_db_deleted_tracks") print(e5) return cur.execute("SELECT COUNT (*) FROM Track_List") total_records = cur.fetchone() if total_records[0] == 0: cur.execute("DROP TABLE IF EXISTS Track_List") conn.commit() cur.execute("PRAGMA auto_vacuum = FULL") conn.commit() cur.execute("PRAGMA optimize") conn.commit() cur.execute("VACUUM") conn.commit() return def backup_database(self): print("Backing Up Database. Don't Panic unless nothing happens for 2 minutes \U0001F633") current_datetime = datetime.now().strftime("%d-%m-%Y_%I-%M_%p") str_current_datetime = str(current_datetime) source = self.dbpath + r"\\" + self.playlist_name + ".db" target = self.dbpath + r"\\" + self.playlist_name + \ "_Database_Backup_" + str_current_datetime+".db" try: shutil.copy(source, target) except IOError as e: print("Unable to copy file. %s" % e) except: print("Unexpected error:", sys.exc_info()) def backup_Enomous_database(self): print("Backing Up Enormous Database. Don't Panic unless nothing happens for 2 minutes \U0001F633") current_datetime = datetime.now().strftime("%d-%m-%Y_%I-%M_%p") str_current_datetime = str(current_datetime) source = self.dbpath + r"\\Enormous Feed.db" target = self.dbpath + r"\\Enormous Feed_Database_Backup_" + str_current_datetime+".db" try: shutil.copy(source, target) except IOError as e: print("Unable to copy Enormous file. %s" % e) except: print("Unexpected Enormous error:", sys.exc_info()) def configure_the_information(self): #self.track_name = self.track_name.replace("'", "") #self.artist_name = self.artist_name.replace("'", "") self.hyphen = str(self.track_name).find(" - ") if self.hyphen != -1: self.track_name = self.track_name[:self.hyphen] self.sqbrack = str(self.track_name).find("[") if self.sqbrack != -1: self.track_name = self.track_name[:self.sqbrack] self.feat = str(self.track_name).find(" (feat. ") if self.feat != -1: self.track_name = self.track_name[:self.feat] self.remaster = str(self.track_name).find("(Remastered)") if self.remaster != -1: self.track_name = self.track_name[:self.remaster] self.rerecord = str(self.track_name).find("(Rerecorded)") if self.rerecord != -1: self.track_name = self.track_name[:self.rerecord] self.digrem = str(self.track_name).find("(Digitally") if self.digrem != -1: self.track_name = self.track_name[:self.digrem] self.dubstep = str(self.track_name).find("(Dubstep") if self.dubstep != -1: self.track_name = self.track_name[:self.dubstep] self.track_name = self.track_name.strip() self.artist_name = self.artist_name.strip() if "á" in self.artist_name: self.artist_name = self.artist_name.replace("á", "a") if "ř" in self.artist_name: self.artist_name = self.artist_name.replace("ř", "r") self.album_name = self.album_name.strip() def author(self): print("[•] Authorising App....") global authcase while True: self.sp_base = authcase.auth() break a = violet() a.author() a.violate_the_database()