from datetime import datetime import shutil import sys import os import json import requests import time from alive_progress import alive_bar import sqlite3 from sqlite3 import Error from secrets import familyfeed_id from secrets import spotify_user_id from refresh import Refresh refreshCaller = Refresh() job_done = False class violet: def __init__(self): self.user_id = spotify_user_id self.spotify_token = "" self.more_than_fifty = 0 self.here_i_am = os.path.dirname(os.path.abspath(__file__)) self.play_count = 0 self.present = False self.played = False self.playlist_id = familyfeed_id self.deleted_records = 0 self.added_records = 0 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: pass File_object = open(self.here_i_am + r"\\Changes.txt", "w") File_object.writelines(["Latest Additions \n\n"]) File_object.close() self.get_the_playlist() 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("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.here_i_am + 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") os.system("Notepad.exe Changes.txt") 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("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() for row in rows: 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.call_refresh() os.system('cls') print("Violet Carson v.3.1.1 21/06/2022 Viking Utilities") 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): 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() 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: self.check_database() t = t + 1 bar() continue except: 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\n"]) File_object.close() t = t - 2 continue self.clear_db_deleted_tracks() job_done = True return def check_database(self): # NOTE See if the track is already in the database # NOTE If it is, set PRESENT flag. If not then add it. global conn conn = self.create_connection(self.database) cur = conn.cursor() try: with conn: 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() cur.execute("SELECT * FROM Track_List WHERE URI = '" + self.track_uri + "'") data = cur.fetchall() if len(data)== 0: File_object = open(self.here_i_am + r"\\Changes.txt", "a") File_object.writelines([self.track_uri + " " + self.track_name + " by " + self.artist_name + "\n"]) File_object.close() self.added_records = self.added_records + 1 self.add_to_db(conn) else: cur.execute("UPDATE Track_List SET PRESENT = True WHERE URI = '" + self.track_uri + "'") conn.commit() return except Error as e2: 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, self.track_name, self.artist_name, self.album_name,self.release_date, 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, self.track_name, self.artist_name, self.album_name) 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("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) File_object = open(self.here_i_am + r"\\Changes.txt", "a") File_object.writelines(["\n\nDeletions \n\n"]) File_object.close() 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]) File_object = open(self.here_i_am + r"\\Changes.txt", "a") File_object.writelines([row[0] + " " + row[1] + " by " + row[2] + "\n"]) File_object.close() 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("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("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.here_i_am + r"\\" + self.playlist_name + ".db" target = self.here_i_am + 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 = str(self.track_name).upper() self.track_name = self.track_name.replace("'", "") self.artist_name = str(self.artist_name).upper() self.artist_name = self.artist_name.replace("'", "") self.album_name = str(self.album_name).upper() self.hyphen = str(self.track_name).find(" - ") if self.hyphen != -1: self.track_name = self.track_name[:self.hyphen] self.track_name = self.track_name.strip() self.artist_name = self.artist_name.strip() self.album_name = self.album_name.strip() a = violet() a.call_refresh() a.violate_the_database()