from datetime import datetime import shutil, sys, os, json, requests, time, 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 secretsFRED import master_id, spotify_user_id from refresh import Refresh refreshCaller = Refresh() job_done = False class violet: def __init__(self): self.spotify_token = "" 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 call_refresh(self): # NOTE Euan's Mk.2 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 violate_the_database(self): # NOTE Delete error file and start the Party! try: os.remove(self.here_i_am + r"\\Violet Errors - " + spotify_user_id + ".txt") except: print() pass 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.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 the playlist name and track count. query = "https://api.spotify.com/v1/playlists/{}".format(self.playlist_id) response = requests.request("get", url=query, headers={"Content-type": "application/json", "Authorization": "Bearer {}".format(self.spotify_token)}) json_data = json.loads(response.text) self.playlist_name = (json_data["name"]) self.total = (json_data["tracks"]["total"]) 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 self.call_refresh() 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 = 0 while t < self.total: with alive_bar(self.total, title= self.playlist_name) as bar: for t in range(self.total): if t > 99: if str(t)[-2:] == "00": self.call_refresh() self.offset = t query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.playlist_id) response = requests.request("get", url=query, headers={"Content-type": "application/json","Authorization":"Bearer {}".format(self.spotify_token)}, params={"limit":1,"offset":self.offset}) json_data = json.loads(response.text) try: self.track_uri = json_data["items"][0]["track"]["uri"] self.track_name = json_data["items"][0]["track"]["name"] self.artist_name = json_data["items"][0]["track"]["artists"][0]["name"] self.album_name = json_data["items"][0]["track"]["album"]["name"] self.release_date = json_data["items"][0]["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 if response.status_code == 200: t+=1 bar() continue except: current_datetime = datetime.now().strftime("%d-%m-%Y_%I-%M_%p") str_current_datetime = str(current_datetime) File_object = open(self.here_i_am + r"\\Violet Errors - " + spotify_user_id + ".txt", "a") File_object.writelines(["Playlist Name - " + str(self.playlist_name) + " 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"]) File_object.writelines(["At - " + str_current_datetime + "\n\n"]) File_object.close() t-=2 continue 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 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() a = violet() a.call_refresh() a.violate_the_database()