import sys, os, json, requests, time, sqlite3, datetime from alive_progress import alive_bar from sqlite3 import Error sys.path.insert(1, 'j:\\HGMS') #sys.path.insert(1, 'C:\\Spotty Main\MASTER to BOOM Dupe finder') from refresh import Refresh refreshCaller = Refresh() class boom: def __init__(self): self.spotify_token = "" self.here_i_am = os.path.dirname(os.path.abspath(__file__)) self.database = self.here_i_am + r"\\Boom Tracks Database.db" ######################################################### ########## ADD SOURCE PLAYLISTS HERE ########## self.source_playlist_id_0 = '7kf6QvyFIabfVRoirKr5Ke' # NOTE: "Live Radio 00:10" self.source_playlist_id_1 = '5XTFRkvck5OMuRWDkS3tpR' # NOTE: "Live Radio KX Classics" self.source_playlist_id_2 = '036smgHlPEJYvCCPqIl9wv' # NOTE: "BOOM Radio - 🔊 live 📻 playlist by ohrenweide" self.source_playlist_id_3 = '' self.source_playlist_id_4 = '' self.source_playlist_id_5 = '' self.source_playlist_id_6 = '' self.source_playlist_id_7 = '' self.source_playlist_id_8 = '' self.source_playlist_id_9 = '' ########## LIMITED TO 10 (theory infinite?) ############# ######################################################### ######################################################### ######## BOOM Playlists URIs and Names Here ######### self.playlist_id_5 = '1E1nHgaO8Cz8dFUCFlT5MD' self.playlist_id_6 = '5TBhX6wlbLDXJq2dtUwYQu' self.playlist_id_7 = '4t5lyX9BbAuyRRtIP15ZC6' self.playlist_id_8 = '3LdCKR1n9teAFoc8LTXeH6' self.playlist_id_9 = '1FXcdHnqA3zt6FDWN4xHe0' self.playlist_id_0 = '3GdHSS5yr8l6VMWkjB9vuD' self.playlist_id_1 = '5UyaP511gEHzpyViMZlIGB' self.playlist_id_2 = '0lKB6Wndcaz1iovqxM2L76' self.playlist_id_3 = '2TUlBeESUyEg6qZ2HYFsIP' self.playlist_id_4 = '2TUlBeESUyEg6qZ2HYFsIP' self.playlist_name_5 = 'BOOM_1950s' self.playlist_name_6 = 'BOOM_1960s' self.playlist_name_7 = 'BOOM_1970s' self.playlist_name_8 = 'BOOM_1980s' self.playlist_name_9 = 'BOOM_1990s' self.playlist_name_0 = 'BOOM_2000s' self.playlist_name_1 = 'BOOM_2010s' self.playlist_name_2 = 'BOOM_2020s' self.playlist_name_3 = 'BOOM_Others' self.playlist_name_4 = 'BOOM_Others' ############## Nine Required ######################## ######################################################### self.playlist_id_list = [] self.playlist_name_list = [] self.source_playlist_list =[] self.srcpl_count = 0 def boom_bang_a_bang(self): self.job_start() self.database_config() while self.srcpl_count < 10: self.nopl = False self.get_boom_playlist_count() if not self.nopl: self.get_boom_playlist_tracks() self.srcpl_count+=1 self.clear_decade_playlists() self.add_new_tracks_to_playlist() self.update_playlist_info() import megaViolet import mega_dupe import Mega_Rebuilder self.job_end() def job_start(self): os.system('cls') self.tic = time.perf_counter() now = datetime.datetime.now() print("Current Run Time:= " + now.strftime("%d-%m-%Y %H:%M:%S")) print("Boom Shackalacka v.1.2.0. 28/12/2022 \U00002694 Viking Utilities \U00002694") print("Extracting Oldies with Aplomb \U0001F9D0 \U0001F4A9 \U0001F6BD") print("\033[91m####################################### Latest Update ########################################\033[0m") print("\033[33mv.1.2.0. BOOM and OTHER Live Playlists...\033[0m") print("\033[91m################################################################################################\033[0m") self.source_playlist_list.extend([self.source_playlist_id_0, self.source_playlist_id_1, self.source_playlist_id_2, self.source_playlist_id_3, self.source_playlist_id_4, self.source_playlist_id_5, self.source_playlist_id_6, self.source_playlist_id_7, self.source_playlist_id_8, self.source_playlist_id_9]) self.playlist_id_list.extend([self.playlist_id_0, self.playlist_id_1, self.playlist_id_2, self.playlist_id_3, self.playlist_id_4, self.playlist_id_5, self.playlist_id_6, self.playlist_id_7, self.playlist_id_8, self.playlist_id_9]) self.playlist_name_list.extend([self.playlist_name_0, self.playlist_name_1, self.playlist_name_2, self.playlist_name_3, self.playlist_name_4, self.playlist_name_5, self.playlist_name_6, self.playlist_name_7, self.playlist_name_8, self.playlist_name_9]) return def database_config(self): plid = 0 conn = self.create_connection(self.database) cur = conn.cursor() while plid < 10: try: with conn: cur.execute(""" CREATE TABLE IF NOT EXISTS """ + self.playlist_name_list[plid] + """ (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, RELEASE_DATE text, ADDED_AT text, DECADE integer, HISTORIC integer, SOURCE_PLAYLIST text); """) conn.commit() except Error as e2: print("Database Config Error with playlist " + str(plid)) print(e2) plid+=1 return def get_boom_playlist_count(self): try: query = "https://api.spotify.com/v1/playlists/{}".format(self.source_playlist_list[self.srcpl_count]) 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.playlist_name = (json_data["name"]) return except: self.nopl = True return def get_boom_playlist_tracks(self): with alive_bar(self.queue_total, title="Getting Playlist " + self.playlist_name) as bar: t = 0 while t < self.queue_total: try: if t == 1000: self.call_refresh() self.offset = t query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.source_playlist_list[self.srcpl_count]) 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.added_at = json_data["items"][0]["added_at"] self.release_date = json_data["items"][0]["track"]["album"]["release_date"] self.decade_num = self.release_date[2] if response.status_code == 200: self.check_decade_database(self.decade_num) bar() t = t + 1 continue except: print("error" + str(t)) t =t - 2 continue return def check_decade_database(self, decade): historic = False conn = self.create_connection(self.database) cur = conn.cursor() try: tracks = (self.track_uri, self.track_name, self.artist_name, self.release_date, self.added_at, self.decade_num, historic, self.playlist_name) sql = ''' INSERT OR IGNORE INTO ''' + self.playlist_name_list[int(decade)] + '''(URI,TRACK_NAME,ARTIST_NAME,RELEASE_DATE,ADDED_AT,DECADE,HISTORIC,SOURCE_PLAYLIST) VALUES(?,?,?,?,?,?,?,?) ''' cur = conn.cursor() cur.execute(sql, tracks) conn.commit() return cur.lastrowid except Error as e3: print("Error adding new track to database") print(e3) return cur.lastrowid # NOTE return with the last rowid def clear_decade_playlists(self): plid = 0 while plid < 10: query = "https://api.spotify.com/v1/playlists/{}/tracks?uris={}".format(self.playlist_id_list[plid], "spotify:track:3lpDrxUkr0tIe1kmJvdK7d") response = requests.request("put", url=query, headers={"Content-type": "application/json", "Authorization": "Bearer {}".format(self.spotify_token)}) query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.playlist_id_list[plid]) response = requests.request("delete", url=query, headers={"Content-type": "application/json", "Authorization": "Bearer {}".format( self.spotify_token)}, data=json.dumps({"tracks": [{"uri": "spotify:track:3lpDrxUkr0tIe1kmJvdK7d"}]})) plid+=1 return def add_new_tracks_to_playlist(self): self.call_refresh() conn = self.create_connection( self.database) cur = conn.cursor() try: plid = 0 while plid < 10: print("Working on playlist:- " + self.playlist_name_list[plid]) cur.execute("SELECT * FROM '" + self.playlist_name_list[plid] + "' WHERE DECADE = '" + str(plid) + "'") rows = cur.fetchall() for row in rows: self.newtrack = row[0] self.add_to_playlist(plid) cur.execute("UPDATE " + self.playlist_name_list[plid] + " SET DECADE = 99 WHERE URI = '" + self.newtrack + "'") conn.commit() plid+=1 return except Error as e4: print("Error adding new track to playlist") print(e4) return cur.lastrowid def add_to_playlist(self, plid): query = "https://api.spotify.com/v1/playlists/{}/tracks?uris={}".format(self.playlist_id_list[plid], self.newtrack) response = requests.request("post", url=query, headers={"Content-type": "application/json", "Authorization": "Bearer {}".format(self.spotify_token)}) return def update_playlist_info(self): now = datetime.datetime.now() plid = 0 while plid < 10: plname = self.playlist_name_list[plid] pldesc = "........Tracks Gathered from BOOM Live and put in the decade of their release OR NOT :-(" query = "https://api.spotify.com/v1/playlists/{}".format(self.playlist_id_list[plid]) request_body = json.dumps({"name": "" + plname + "", "description": "LAST UPDATED: " + now.strftime("%d-%m-%Y %H:%M:%S") + " " + pldesc + "", "public": True}) response = requests.put(query, data=request_body, headers={"Content-Type": "application/json", "Authorization": "Bearer {}".format(self.spotify_token)}) plid+=1 return def job_end(self): self.toc = time.perf_counter() ty_res = time.gmtime(self.toc - self.tic) res = time.strftime("%H Hours :%M Minutes :%S Seconds",ty_res) print(f"Boom Took {res} Seconds To Run" ) return def create_connection(self, db_file): conn = None try: conn = sqlite3.connect(db_file) return conn except Error as e1: print("Connection Error") print(e1) return conn def call_refresh(self): 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 a = boom() a.call_refresh() a.boom_bang_a_bang()