import datetime import json, os, requests, time, sqlite3, sys sys.path.insert(1, 'j:\\Spotty\secrets_and_database') from alive_progress import alive_bar from endless_playlist_secrets import queue_id, spotify_user_id, master_id, queue_nm from refresh import Refresh refreshCaller = Refresh() class cynthia: def __init__(self): self.user_id = spotify_user_id self.spotify_token = "" self.here_i_am = os.path.dirname(os.path.abspath(__file__)) self.uridisc_id = "" self.dbomissions_id = "" self.playlist_id = master_id self.queue_id = queue_id self.queue_name = queue_nm self.queue_list = [] self.queue_track_name_list = [] self.queue_track_list = [] self.queue_track_pic_list = [] self.queue_track_preview_list = [] self.rowid_list = [] self.plid_list = [] self.dbom_list = [] self.uridis_list = [] self.played_track = "" self.dbom = False self.uri_match = 0 self.name_match = 0 def call_refresh(self): # NOTE Euan is king 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 repopulate_the_queue(self): # NOTE List of functions tic = time.perf_counter() now = datetime.datetime.now() print("Current Run Time:= " + now.strftime("%d-%m-%Y %H:%M:%S")) print("Cynthia (Len Goodman Version) v.7.9.4 30/01/2023 \U00002694 Viking Utilities \U00002694") print("Amazing Things Happen When You Least Expect Them \U0001F9D0 \U0001F4A9 \U0001F6BD") print("\033[91m####################################### Latest Update ########################################\033[0m") print("\033[33mv.7.9.4 New queue playlist Information...\033[0m") print("\033[91m################################################################################################\033[0m") self.get_endless_playlist() #self.database = self.here_i_am + r"\\" + self.playlist_name + ".db" self.database = r"j:\Spotty\secrets_and_database\\" + self.playlist_name + ".db" self.flex_temp_table() self.check_error_playlists() print("Getting Queue Count") self.get_queue_count() self.queue_head = "Getting Queue Items" self.get_the_queue() print("Getting Last 50 Played Tracks") self.check_played_list() print("Updating Last Played Date & Time") self.update_last_played_time() print("Updating Last Played Play Count") self.update_play_count_in_last_played() print("Getting New Tracks For Queue") self.get_new_tracks() print("Updating Playlist Information") self.update_playlist_info() print("Checking Queued Items in Database") self.do_a_queue_check() self.make_preview_table() print("Compressing Database") self.compress_database() print("Updating Web Pages") import htmlbuilder print("Remaining Un-Played Tracks in Database = " + str(self.remain_records)) print(str(self.uri_match) + " URI Matches") print(str(self.name_match) + " URI Alterations") print(str(self.record_wanted) + " Tracks added from database records:- ") print(*self.rowid_list, sep=', ') toc = time.perf_counter() ty_res = time.gmtime(toc - tic) res = time.strftime(" %H Hours : %M Minutes : %S Seconds", ty_res) print(f"Cynthia Took {res}") print("") def get_queue_count(self): # NOTE Get how many tracks in Endless Queue query = "https://api.spotify.com/v1/playlists/{}".format(self.queue_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.queue_total = (json_data["tracks"]["total"]) return def get_the_queue_again(self): # NOTE Get Individual tracks in Endless Queue Again q = 0 while q < self.queue_total: for q in range(self.queue_total): self.offset = q try: query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.queue_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) 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.sort_out_the_queue_track_name() if self.played_artist_name.upper() == self.track_artist_name.upper(): self.check_and_add_to_disc_table() self.amend_track_list_and_playlist() self.check_track = self.played_track self.add_to_error_playlist("U") self.temp_uri = self.played_track self.played_track = self.track_uri return if response.status_code == 200: q+=1 continue except: q+=1 continue def get_the_queue(self): # NOTE Get Individual tracks in Endless Queue i = 0 while i < self.queue_total: with alive_bar(self.queue_total, title=self.queue_head) as bar: for i in range(self.queue_total): self.offset = i query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.queue_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) 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_image = json_data["items"][0]["track"]["album"]["images"][1]["url"] self.track_preview = json_data["items"][0]["track"]["preview_url"] self.sort_out_the_queue_track_name() self.queue_list.append(self.track_uri) self.queue_track_list.append(self.track_name) self.queue_track_name_list.append(self.track_artist_name.upper()) self.queue_track_pic_list.append(self.album_image) self.queue_track_preview_list.append(self.track_preview) if response.status_code == 200: i+=1 bar() continue def sort_out_the_queue_track_name(self): # NOTE Format The Track Name for Duplicate Checks 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() self.track_artist_name = self.track_name + " - " + self.artist_name def check_played_list(self): # NOTE Get Last 50 played tracks. Checking to see if each one is in the queue list last_track = False self.check_last_played_table() query = "https://api.spotify.com/v1/me/player/recently-played" response = requests.request("get", url=query, headers={"Content-type": "application/json","Authorization": "Bearer {}".format(self.spotify_token)},params={"limit": 50}) response_json = response.json() print("Adding Tracks to Last Played table") for k in response_json["items"]: found = False self.played_track = (k["track"]["uri"]) self.played_name = (k["track"]["name"]) self.played_artist = (k["track"]["artists"][0]["name"]) self.played_at = (k["played_at"]) self.sort_out_the_last_played_track_name() self.sort_out_the_played_at_time() self.add_new_last_played_track() if not last_track: print("Saving Last Track Played Which Was:- " + self.played_artist_name) self.last_song_played = self.played_artist_name last_track = True continue if self.played_track in self.queue_list: self.uri_match+=1 self.temp_uri = self.played_track self.mark_played_and_remove() found = True continue if not found: if self.played_artist_name.upper() in self.queue_track_name_list: if self.played_artist_name == self.last_song_played: continue self.get_the_queue_again() self.name_match+=1 self.mark_played_and_remove() continue else: self.remove_from_playlist() def sort_out_the_last_played_track_name(self): # NOTE Format The Track Name for Duplicate Checks self.played_name = self.played_name.replace("'", "") self.played_artist = self.played_artist.replace("'", "") self.hyphen = str(self.played_name).find(" - ") if self.hyphen != -1: self.played_name = self.played_name[:self.hyphen] self.sqbrack = str(self.played_name).find("[") if self.sqbrack != -1: self.played_name = self.played_name[:self.sqbrack] self.feat = str(self.played_name).find(" (feat. ") if self.feat != -1: self.played_name = self.played_name[:self.feat] self.remaster = str(self.played_name).find("(Remastered)") if self.remaster != -1: self.played_name = self.played_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.played_name).find("(Digitally") if self.digrem != -1: self.played_name = self.played_name[:self.digrem] self.dubstep = str(self.played_name).find("(Dubstep") if self.dubstep != -1: self.played_name = self.played_name[:self.dubstep] self.played_name = self.played_name.strip() self.played_artist = self.played_artist.strip() self.played_artist_name = self.played_name + " - " + self.played_artist def mark_played_and_remove(self): # NOTE If it is in the queue and has been played then deal with it self.mark_played_in_db() self.remove_from_playlist() return def get_endless_playlist(self): # NOTE Get the Endless Playlist Name 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"]) return def mark_played_in_db(self): # NOTE Mark the played tracks in the database global conn conn = self.create_connection(self.database) cur = conn.cursor() with conn: try: cur.execute("SELECT * FROM Track_List WHERE URI = '" + self.temp_uri + "'") row = cur.fetchone() play_count = row[5] play_count+=1 cur.execute("UPDATE Track_List SET PLAY_COUNT = '" + str(play_count) + "' WHERE URI = '" + self.temp_uri + "'") conn.commit() except: print("Error with this track") self.remove_from_playlist() def create_connection(self, db_file): # NOTE Connect to the database conn = None conn = sqlite3.connect(db_file) return conn def remove_from_playlist(self): # NOTE Remove played tracks from the Endless Queue playlist query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.queue_id) response = requests.request("delete", url=query, headers={"Content-type": "application/json","Authorization": "Bearer {}".format(self.spotify_token)},data=json.dumps({"tracks": [{"uri": self.played_track}]})) if response.status_code != 200: print("Incorrect URI for:- " +self.played_name+ " by " +self.played_artist) self.played_track = "" return def get_new_tracks(self): # NOTE Get how many new tracks are needed self.get_queue_count() self.remaining_queue = self.queue_total self.get_tracks() return def get_tracks(self): # NOTE Get the new tracks from the Endless Playlist self.needed = 50 - self.remaining_queue self.record_wanted = self.needed print(str(self.needed) + " New Tracks Required!") print("Performing Record Check") self.check_played_count() if self.needed <= 0: return conn = self.create_connection(self.database) cur = conn.cursor() cur.execute("SELECT * FROM Track_List WHERE QUEUED = 0 order by RANDOM()") rows = cur.fetchall() for row in rows: self.newtrack = row[0] self.add_track_to_queue() self.set_queued_flag() self.needed-=1 if self.needed == 0: return return def set_queued_flag(self): # NOTE Mark Track As Queued conn = self.create_connection(self.database) cur = conn.cursor() cur.execute("SELECT * FROM Track_List WHERE URI = '" + self.newtrack + "'") cur.execute("SELECT rowid, * FROM Track_List WHERE URI = '" + self.newtrack + "'") toe = cur.fetchone() self.rowid_list.append(toe[0]) cur.execute("UPDATE Track_List SET QUEUED = True WHERE URI = '" + self.newtrack + "'") conn.commit() return def add_track_to_queue(self): # NOTE Add the new tracks to the Endless Queue Playlist query = "https://api.spotify.com/v1/playlists/{}/tracks?uris={}".format(self.queue_id, self.newtrack) response = requests.request("post", url=query, headers={"Content-type": "application/json","Authorization": "Bearer {}".format(self.spotify_token)}) return def check_played_count(self): # NOTE See how many more tracks are un-played conn = self.create_connection(self.database) cur = conn.cursor() cur.execute("SELECT COUNT (*) FROM Track_List") self.total_records = cur.fetchone() cur.execute("SELECT COUNT (*) FROM Track_List WHERE QUEUED = 1") self.played_records = cur.fetchone() self.remain_records = self.total_records[0] - self.played_records[0] if self.remain_records < self.needed: self.reset_queued_flag() return def reset_queued_flag(self): # NOTE If no tracks left to play then reset the whole database so they are all available print("Reseting Database '" + self.playlist_name + "'") conn = self.create_connection(self.database) with conn: cur = conn.cursor() cur.execute("SELECT QUEUED FROM Track_List") rows = cur.fetchall() for row in rows: cur.execute("UPDATE Track_List SET QUEUED = 0") cur.execute("UPDATE Track_List SET PLAY_COUNT = 0") conn.commit() return def sort_out_the_played_at_time(self): # NOTE Format Last Played Date and Time pdyr = self.played_at[0:4] pdmh = self.played_at[5:7] pddy = self.played_at[8:10] pdtm = self.played_at[11:19] self.played_at = pdmh + "-" + pddy + "-" + pdyr + " " + pdtm self.disp_played_at = pddy + "-" + pdmh + "-" + pdyr + " " + pdtm def check_last_played_table(self): # NOTE Make Sure Last Played Table Is Present conn = self.create_connection(self.database) cur = conn.cursor() with conn: cur.execute("CREATE TABLE IF NOT EXISTS Last_Played_Tracks (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, LAST_PLAYED text, PLAY_COUNT integer, DISPLAY_DATE integer);") conn.commit() def add_new_last_played_track(self): # NOTE Add Last Played Tracks To Last Played Table and a Temporary Table self.play_count = 0 conn = self.create_connection(self.database) cur = conn.cursor() track2 = (self.played_track, self.played_name, self.played_artist, self.played_at) # NOTE Set the SQL Values tracks = (self.played_track, self.played_name, self.played_artist, self.played_at, self.play_count, self.disp_played_at) # NOTE Set the SQL Values sql = ''' INSERT OR IGNORE INTO Last_Played_Tracks (URI,TRACK_NAME,ARTIST_NAME,LAST_PLAYED,PLAY_COUNT,DISPLAY_DATE) VALUES(?,?,?,?,?,?) ''' sql2 = ''' INSERT OR IGNORE INTO tmp_played (URI,TRACK_NAME,ARTIST_NAME,LAST_PLAYED) VALUES(?,?,?,?) ''' cur = conn.cursor() cur.execute(sql, tracks) cur.execute(sql2, track2) conn.commit() def update_last_played_time(self): # NOTE Update Last Played Date and Time to Last Time the Track was Played conn = self.create_connection(self.database) cur = conn.cursor() cur.execute("SELECT * FROM tmp_played ORDER BY LAST_PLAYED ASC ") rows = cur.fetchall() for row in rows: self.played_track2 = row[0] self.played_at2 = row[3] olddate = self.played_at2 mth = olddate[:2] dy = olddate[3:5] newdate = dy + "-" + mth + olddate[5:] self.disp_played_at2 = newdate cur.execute("UPDATE Last_Played_Tracks SET LAST_PLAYED = '" + self.played_at2 + "' WHERE URI = '" + self.played_track2 + "'") conn.commit() cur.execute("UPDATE Last_Played_Tracks SET DISPLAY_DATE = '" + self.disp_played_at2 + "' WHERE URI = '" + self.played_track2 + "'") conn.commit() def flex_temp_table(self): # NOTE Delete and Create a Temporary Table conn = self.create_connection(self.database) cur = conn.cursor() cur.execute("DROP TABLE IF EXISTS tmp_played") conn.commit() cur.execute("CREATE TABLE IF NOT EXISTS tmp_played (URI text, TRACK_NAME text, ARTIST_NAME text, LAST_PLAYED text PRIMARY KEY);") conn.commit() cur.execute("CREATE TABLE IF NOT EXISTS Erroneous_Data (URI_PLAYLIST text, DB_OMISSIONS_PLAYLIST text);") conn.commit() def update_play_count_in_last_played(self): # NOTE Update The Play Count in the Last Played Table conn = self.create_connection(self.database) cur = conn.cursor() cur.execute("SELECT * FROM Erroneous_Data") erow = cur.fetchone() self.plid_list.append(erow[0]) self.plid_list.append(erow[1]) plid = 0 while plid < 2: self.put_playlist_in_list(self.plid_list[plid]) self.dbom = True plid+=1 cur.execute("SELECT * FROM tmp_played ORDER BY LAST_PLAYED DESC") rows = cur.fetchall() for row in rows: self.check_track = row[0] self.check_name = row[1] self.check_artist = row[2] cur.execute("SELECT * FROM Track_List WHERE URI = '" + self.check_track + "'") tlrow = cur.fetchone() if tlrow is None: self.played_name = row[1] self.played_artist = row[2] query = "SELECT * FROM Track_List WHERE TRACK_NAME = '" + self.played_name + "' AND ARTIST_NAME = '" + self.played_artist + "'" cur.execute(query) nmrow = cur.fetchone() if nmrow is None: vik = "O" if self.check_track in self.dbom_list: plcount = -100 else: self.add_to_error_playlist(vik) plcount = -100 cur.execute("UPDATE Last_Played_Tracks SET PLAY_COUNT = '" + str(plcount) + "' WHERE URI = '" + self.check_track + "'") conn.commit() continue else: vik = "U" if self.check_track in self.uridis_list: plcount = nmrow[5] else: self.add_to_error_playlist(vik) plcount = nmrow[5] else: plcount = tlrow[5] cur.execute("UPDATE Last_Played_Tracks SET PLAY_COUNT = '" + str(plcount) + "' WHERE URI = '" + self.check_track + "'") conn.commit() def check_error_playlists(self): conn = self.create_connection(self.database) cur = conn.cursor() cur.execute("SELECT URI_PLAYLIST FROM Erroneous_Data") erow = cur.fetchone() if erow is None: errtype = "URI" self.add_error_playlist(errtype) self.uridisc_id = self.error_playlist_id sql = "INSERT INTO Erroneous_Data (URI_PLAYLIST) VALUES (?)" cur = conn.cursor() cur.execute(sql, [self.uridisc_id]) conn.commit() else: self.uridisc_id = erow[0] cur.execute("SELECT DB_OMISSIONS_PLAYLIST FROM Erroneous_Data") erow = cur.fetchone() if erow[0] is None: errtype = "OMS" self.add_error_playlist(errtype) self.dbomissions_id = self.error_playlist_id cur.execute("UPDATE Erroneous_Data SET DB_OMISSIONS_PLAYLIST = '" + self.dbomissions_id + "'") conn.commit() else: self.dbomissions_id = erow[0] def add_error_playlist(self, errtype): if errtype == "URI": pltype = "Endless URI/Artist Discrepancies" pldesc = "Discrepancies between the URI of These Tracks and the Database. Information Only! Tracks have been replaced in the Database and Main Feed. Tracks will be removed after 7 days." if errtype == "OMS": pltype = "Endless Database Omissions/Discrepancies" pldesc = "Tracks not found in the Database OR with The Same Title BUT Different Artists. Tracks will be removed after 7 days." now = datetime.datetime.now() query = "https://api.spotify.com/v1/users/{}/playlists".format(self.user_id) request_body = json.dumps({"name": "" + pltype + "", "description": "LAST UPDATED: " + now.strftime("%d-%m-%Y %H:%M:%S") + " " + pldesc + "", "public": False}) response = requests.post(query, data=request_body, headers={"Content-Type": "application/json","Authorization": "Bearer {}".format(self.spotify_token)}) response_json = response.json() self.error_playlist_id = response_json["id"] return self.error_playlist_id def add_to_error_playlist(self, vik): if vik == "O": self.err_text = "No Database Entry" self.error_playlist_id = self.dbomissions_id if self.check_name in self.queue_track_list: name_index = self.queue_track_list.index(self.check_name) name_uri = self.queue_list[name_index] self.played_track = name_uri self.remove_from_playlist() if vik == "U": self.err_text = "Incorrect URI" self.error_playlist_id = self.uridisc_id query = "https://api.spotify.com/v1/playlists/{}/tracks?uris={}".format(self.error_playlist_id,self.check_track) response = requests.request("post", url=query, headers={"Content-type": "application/json","Authorization": "Bearer {}".format(self.spotify_token)}) def put_playlist_in_list(self, plid): self.error_playlist_id = plid nowtoday = datetime.date.today() query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.error_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.total = (json_data["total"]) for i in range(self.total): self.offset = i query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.error_playlist_id) response = requests.request("get", url=query, headers={"Content-type": "application/json","Authorization": "Bearer {}".format(self.spotify_token)},params={"limit": 100, "offset": self.offset}) json_data = json.loads(response.text) try: self.added_at = json_data["items"][0]["added_at"] except: continue oldconv1 = self.added_at[0:4] oldconv2 = self.added_at[5:7] oldconv3 = self.added_at[8:10] gap = nowtoday - datetime.date(int(oldconv1), int(oldconv2), int(oldconv3)) if gap.days > 7: self.remove_uri = json_data["items"][0]["track"]["uri"] self.remove_from_error_playlist(plid) i-=1 continue if self.dbom: self.dbom_list.append(json_data["items"][0]["track"]["uri"]) else: self.uridis_list.append(json_data["items"][0]["track"]["uri"]) def remove_from_error_playlist(self, plid): query = "https://api.spotify.com/v1/playlists/{}/tracks".format(plid) headers={"Content-type": "application/json", "Authorization": "Bearer {}".format(self.spotify_token)} params = {"tracks": [{"uri": self.remove_uri}]} response = requests.request("delete", url=query, headers=headers, data=json.dumps(params)) if response.status_code != 200: print("Incorrect URI for:- " + self.played_name + " by " + self.played_artist) self.played_track = "" return def update_playlist_info(self): now = datetime.datetime.now() plname1 = "Endless URI/Artist Discrepancies" pldesc1 = ":- Discrepancies between the URI of These Tracks and the Database. Information Only! Tracks have been replaced in the Database and Main Feed. Tracks will be removed after 7 days. " plname2 = "Endless Database Omissions/Discrepancies" pldesc2 = ":- Tracks not found in the Database OR with The Same Title BUT Different Artists. Tracks will be removed after 7 days." plname3 = self.queue_name pldesc3 = ":- 50 Tracks from a Total of " + str(self.total_records[0]) + " in the Master Playlist. " + str(self.played_records[0]) + " played tracks with " + str(self.remain_records) + " remaining" query = "https://api.spotify.com/v1/playlists/{}".format(self.uridisc_id) request_body = json.dumps({"name": "" + plname1 + "", "description": "LAST UPDATED: " + now.strftime("%d-%m-%Y %H:%M:%S") + " " + pldesc1 + "", "public": False}) response = requests.put(query, data=request_body, headers={"Content-Type": "application/json","Authorization": "Bearer {}".format(self.spotify_token)}) query = "https://api.spotify.com/v1/playlists/{}".format(self.dbomissions_id) request_body = json.dumps({"name": "" + plname2 + "", "description": "LAST UPDATED: " + now.strftime("%d-%m-%Y %H:%M:%S") + " " + pldesc2 + "", "public": False}) response = requests.put(query, data=request_body, headers={"Content-Type": "application/json","Authorization": "Bearer {}".format(self.spotify_token)}) query = "https://api.spotify.com/v1/playlists/{}".format(self.queue_id) request_body = json.dumps({"name": "" + plname3 + "", "description": "LAST UPDATED: " + now.strftime("%d-%m-%Y %H:%M:%S") + " " + pldesc3 + "", "public": True}) response = requests.put(query, data=request_body, headers={"Content-Type": "application/json","Authorization": "Bearer {}".format(self.spotify_token)}) def check_and_add_to_disc_table(self): from datetime import date today = date.today() newtoday = today.strftime("%m-%d-%Y") disptoday = today.strftime("%d-%m-%Y") conn = self.create_connection(self.database) cur = conn.cursor() cur.execute("CREATE TABLE IF NOT EXISTS URI_Change (DATE_OF_CHANGE integer, OLD_URI text, NEW_URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, DISPLAY_DATE integer);") conn.commit() track = (newtoday, self.track_uri, self.played_track, self.played_name, self.played_artist, disptoday) # NOTE Set the SQL Values sql = ''' INSERT OR IGNORE INTO URI_Change (DATE_OF_CHANGE,OLD_URI,NEW_URI,TRACK_NAME,ARTIST_NAME,DISPLAY_DATE) VALUES(?,?,?,?,?,?) ''' cur = conn.cursor() cur.execute(sql, track) conn.commit() return def amend_track_list_and_playlist(self): conn = self.create_connection(self.database) cur = conn.cursor() cur.execute("UPDATE Track_list SET URI = '" + self.played_track + "' WHERE URI = '" + self.track_uri + "'") conn.commit() query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.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}]})) if response.status_code != 200: print("Changing track details for:- " + str(self.track_name) + " By " + str(self.track_artist_name)) query = "https://api.spotify.com/v1/playlists/{}/tracks?uris={}".format(self.playlist_id, self.played_track) response = requests.request("post", url=query, headers={"Content-type": "application/json", "Authorization": "Bearer {}".format(self.spotify_token)}) return def do_a_queue_check(self): conn = self.create_connection(self.database) cur = conn.cursor() cur.execute("DROP TABLE IF EXISTS Queue_check") conn.commit() cur.execute( "CREATE TABLE IF NOT EXISTS Queue_check (URI text, TRACK_NAME text, ARTIST_NAME text, ALBUM_NAME text);") conn.commit() self.queue_head = "Checking Queue Contents" self.get_queue_count() self.queue_list = [] self.queue_track_pic_list = [] self.queue_track_preview_list = [] self.get_the_queue() cur.execute("SELECT * FROM Track_list WHERE QUEUED = 1 AND PLAY_COUNT = 0") rows = cur.fetchall() for row in rows: quri = row[0] if quri not in self.queue_list: qtkn = row[1] qarn = row[2] qaln = row[3] track = (quri, qtkn, qarn, qaln) sql = ''' INSERT INTO Queue_check (URI,TRACK_NAME,ARTIST_NAME,ALBUM_NAME) VALUES(?,?,?,?) ''' cur.execute(sql, track) conn.commit() cur.execute("UPDATE Track_List SET QUEUED = 0 WHERE URI = '" + quri + "'") conn.commit() def make_preview_table(self): qtknum = 0 conn = self.create_connection(self.database) cur = conn.cursor() cur.execute("DROP TABLE IF EXISTS Queue_preview") conn.commit() cur.execute( "CREATE TABLE IF NOT EXISTS Queue_preview (URI text, TRACK_NAME text, ALBUM_PIC text, TRACK_PREVIEW text);") conn.commit() while qtknum < 50: quri = self.queue_list[qtknum] qtrack = self.queue_track_list[qtknum] qpic = self.queue_track_pic_list[qtknum] qpre = self.queue_track_preview_list[qtknum] picpre = (quri, qtrack, qpic, qpre) sql = ''' INSERT INTO Queue_preview (URI,TRACK_NAME,ALBUM_PIC,TRACK_PREVIEW) VALUES(?,?,?,?) ''' cur.execute(sql, picpre) conn.commit() qtknum+=1 def compress_database(self): conn = self.create_connection(self.database) cur = conn.cursor() cur.execute("VACUUM") conn.commit() a = cynthia() a.call_refresh() a.repopulate_the_queue()