from datetime import datetime import json, os, requests, time, sqlite3 from alive_progress import alive_bar from sqlite3 import Error from secrets import familyfeed_id, familyqueue_id, spotify_user_id 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.playlist_id = familyfeed_id self.queue_id = familyqueue_id self.queue_list = [] self.queue_track_name_list = [] self.queue_track_list = [] self.rowid_list = [] self.urimatch_list = [] self.namematch_list = [] self.played_track = "" self.uri_match = 0 self.name_match = 0 self.nodb = False self.dbluri = False 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 os.system('cls') current_datetime = datetime.now().strftime("%d-%m-%Y_%I-%M_%p") self.str_current_datetime = str(current_datetime) print("Cynthia (The CAPITAL A' Version) v.7.0.0 18/06/2022 Viking Utilities") print("Includes The Sweat of my Brow") File_object = open(self.here_i_am + r"\\Queue Errors.txt", "a") File_object.writelines("\nThe Following Tracks Had Problems at " + self.str_current_datetime + "\n\n") File_object.close() self.get_endless_playlist() self.database = self.here_i_am + r"\\" + self.playlist_name + ".db" print("Getting Queue Count") self.get_queue_count() self.get_the_queue() print("Getting Last 50 Played Tracks") self.check_played_list() print("Getting New Tracks For Queue") self.get_new_tracks() print("Remaining Un-Played Tracks in Database = " + str(self.remain_records)) print(str(self.uri_match) + " URI Matches") print(str(self.name_match) + " Name Matches") print(str(self.record_wanted) + " Tracks added from database records:- ") print(*self.rowid_list, sep=', ') 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"]) self.queue_name = (json_data["name"]) return def get_the_queue_again(self): # NOTE Get Individual tracks in Endless Queue Again q = 0 while q < self.queue_total: try: for q in range(self.queue_total): self.offset = q 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 response.status_code == 200: q = q + 1 continue except: File_object = open(self.here_i_am + r"\\Queue Errors.txt", "a") File_object.writelines("Track " + self.played_name + " Not Present in " + self.queue_name + " Playlist! Please Investigate Playlist & Database!!\n") File_object.close() return 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="Getting Queue Items") 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.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) if response.status_code == 200: i = i + 1 bar() continue def sort_out_the_queue_track_name(self): # NOTE Format Thee Track Name for Duplicate Checks 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.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.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.flex_temp_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"]["album"]["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.check_last_played_table() self.add_new_last_played_track() if last_track == False: print("Saving Last Track Played Which Was:- " + (k["track"]["name"])) last_track = True continue if self.played_track in self.queue_list: self.uri_match = self.uri_match + 1 self.urimatch_list.append(self.played_name) self.mark_played_and_remove() found = True continue if found == False: if self.played_artist_name in self.queue_track_name_list or self.played_name in self.queue_track_list: self.get_the_queue_again() self.name_match = self.name_match + 1 self.namematch_list.append(self.played_name) self.mark_played_and_remove() continue print("Updating Last Played Date & Time") self.update_last_played_time() print("Updating Last Played Play Count") self.update_play_count_in_last_played() def sort_out_the_last_played_track_name(self): # NOTE Format The Track Name for Duplicate Checks self.played_name = str(self.played_name).upper() self.played_name = self.played_name.replace("'", "") self.played_artist = str(self.played_artist).upper() 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.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() try: with conn: try: cur.execute("SELECT * FROM Track_List WHERE URI = '" + self.played_track + "'") row = cur.fetchone() play_count = row[5] play_count = play_count + 1 cur.execute("UPDATE Track_List SET PLAY_COUNT = '" + str(play_count) + "' WHERE URI = '" + self.played_track + "'") conn.commit() return except: File_object = open(self.here_i_am + r"\\Queue Errors.txt", "a") File_object.writelines("Track " + str(self.played_name) + " MAY Not Be Present in Database. Please Investigate Playlist & Database!!\n") File_object.close() return except Error as e2: print("E2") print(e2) return def create_connection(self, db_file): # NOTE Connect to the database conn = None try: conn = sqlite3.connect(db_file) return conn except Error as e1: print("E1") print(e1) 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(response.status_code) 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() try: cur.execute("SELECT * FROM Track_List WHERE QUEUED = 0 order by RANDOM()") except Error as e: print(e) return rows = cur.fetchall() for row in rows: self.newtrack = row[0] self.add_track_to_queue() self.set_queued_flag() self.needed = 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() try: cur.execute("SELECT COUNT (*) FROM Track_List") total_records = cur.fetchone() cur.execute("SELECT COUNT (*) FROM Track_List WHERE QUEUED = 1") played_records = cur.fetchone() self.remain_records = total_records[0] - played_records[0] if self.remain_records < self.needed: self.reset_queued_flag() return except Error as e: print(e) 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() try: cur.execute("SELECT QUEUED FROM Track_List") except Error as e: print(e) return rows = cur.fetchall() for row in rows: cur.execute("UPDATE Track_List SET QUEUED = 0") 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:23] self.played_at = pddy + "-" + pdmh + "-" + pdyr + " " + pdtm return def check_last_played_table(self): # NOTE Make Sure Last Played Table Is Present conn = self.create_connection(self.database) cur = conn.cursor() try: 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);") conn.commit() except Error as e2: print("Last Played Table") print(e2) return 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() try: 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) # NOTE Set the SQL Values sql = ''' INSERT OR IGNORE INTO Last_Played_Tracks (URI,TRACK_NAME,ARTIST_NAME,LAST_PLAYED,PLAY_COUNT) 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() except sqlite3.IntegrityError as e3: print("Last Played Table 2") print(e3) return 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() try: 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] cur.execute("UPDATE Last_Played_Tracks SET LAST_PLAYED = '" + self.played_at2 + "' WHERE URI = '" + self.played_track2 + "'") conn.commit() return except sqlite3.Error as e4: print("Last Played Table 3") print(e4) return 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() return def update_play_count_in_last_played(self): #NOTE Update The Play Count in the Last Played Table #NOTE Also Sort Out Those Tracks with Differing URI's AND Tracks #NOTE Played that are Not In The Main Database Table conn = self.create_connection(self.database) cur = conn.cursor() try: cur.execute("SELECT * FROM Last_Played_Tracks") rows = cur.fetchall() for row in rows: track = row[0] cur.execute("SELECT * FROM Track_List WHERE URI = '" + track + "'") tlrow = cur.fetchone() if tlrow == 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 == None: if not self.nodb: File_object = open(self.here_i_am + r"\\Database Omissions.txt", "a") File_object.writelines("\nThe Following Tracks Are Not In The Database at: " + self.str_current_datetime + "\n\n") File_object.close() self.nodb = True File_object = open(self.here_i_am + r"\\Database Omissions.txt", "a") File_object.writelines(row[0] + " " + row[1] + " By " + row[2] + "\n") File_object.close() plcount = -100 cur.execute("UPDATE Last_Played_Tracks SET PLAY_COUNT = '" + str(plcount) + "' WHERE URI = '" + track + "'") conn.commit() continue if not self.dbluri: File_object = open(self.here_i_am + r"\\URI Issues.txt", "a") File_object.writelines("\nThe Following Tracks Have Differing URI's In The Database at: " + self.str_current_datetime + "\n\n") File_object.close() self.dbluri = True File_object = open(self.here_i_am + r"\\URI Issues.txt", "a") File_object.writelines(row[1] + " By " + row[2] + "\n") File_object.close() plcount = nmrow[5] else: plcount = tlrow[5] cur.execute("UPDATE Last_Played_Tracks SET PLAY_COUNT = '" + str(plcount) + "' WHERE URI = '" + track + "'") conn.commit() return except sqlite3.Error as e5: print("Play Count Update") print(e5) return a = cynthia() a.call_refresh() a.repopulate_the_queue()