import datetime import base64 import smtplib from email.message import EmailMessage import time import sqlite3 import sys import spotipy import subprocess sys.path.insert(1, 'j:\\Spotty\\secrets_and_database') from pkce_auth import Authorize class BigCynthia: def __init__(self): # Configurable attributes self.email_address = "vikingindustries2023@gmail.com" self.email_password = b'dGFvZSBlZnluIHF2dmwgZnZ4eQ==' self.bitpassword = base64.b64decode(self.email_password) self.email_password = self.bitpassword.decode('utf-8') self.queue_id = '7do6UPdCRri7Z0ok0wcYHp' self.queue_name = 'Enormous Queue' self.database = r"j:\Spotty\secrets_and_database\Enormous Feed.db" # State attributes self.queue_list = [] self.queue_artist_name_list = [] self.queue_track_name_list = [] self.queue_track_duration_list = [] self.queue_track_list = [] self.queue_track_pic_list = [] self.queue_track_preview_list = [] self.rowid_list = [] self.played_track = "" self.uri_match = 0 self.name_match = 0 self.email_type = 0 self.queue_total = 0 self.queue_owner = "" self.no_exist = False self.uri_change = False def authorise(self): print("[•] Authorising App....") self.sp_base = Authorize().auth() def create_connection(self): return sqlite3.connect(self.database) def repopulate_the_queue(self): tic = time.perf_counter() now = datetime.datetime.now() print("Current Run Time:= " + now.strftime("%d-%m-%Y %H:%M:%S")) print("Enormous Cynthia v.2.0.0 30/11/2024 \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.2.0.0 Change to Spotipy. ...\033[0m") print("\033[91m################################################################################################\033[0m") self.flex_temp_table() print("Getting Queue Count") self.get_queue_count() print("Getting Queue Items") self.get_the_queue() print("Getting Last 50 Played Tracks") self.new_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() print("Making A Preview Table") self.make_preview_table() print("Compressing Database") self.compress_database() print("Updating Web Pages") subprocess.run(["python", "Enorhtmlbuilder.py"]) print("Remaining Un-Played Tracks in Database = " + str(self.remain_records)) print(f"{self.uri_match} URI Matches") print(f"{self.name_match} URI Alterations") print(f"{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"Enormous Cynthia Took {res}\n") def get_queue_count(self): query = self.sp_base.playlist(self.queue_id, market="GB") self.queue_total = query['tracks']['total'] self.queue_owner = query['owner']['id'] def get_the_queue(self): results = self.sp_base.playlist_items(self.queue_id, market="GB") tracks = results['items'] while results['next']: results = self.sp_base.next(results) tracks.extend(results['items']) for track in tracks: t = track['track'] self.track_uri = t['uri'] self.track_name = t["name"] self.artist_name = t['album']['artists'][0]['name'] if self.artist_name == "Various Artists": self.artist_name = t["artists"][0]["name"] self.track_duration = t["duration_ms"] self.sort_out_the_track_length() self.album_image = t["album"]["images"][1]["url"] if t["album"]["images"] else "" self.track_preview = t["preview_url"] self.sort_out_the_queue_track_name() self.queue_list.append(self.track_uri) self.queue_track_list.append(self.track_name.lower()) self.queue_artist_name_list.append(self.artist_name.lower()) self.queue_track_name_list.append(self.track_artist_name) self.queue_track_duration_list.append(self.track_duration) self.queue_track_pic_list.append(self.album_image) self.queue_track_preview_list.append(self.track_preview) def sort_out_the_queue_track_name(self): self.track_name = self.track_name.replace("'", "\\'") self.artist_name = self.artist_name.replace("'", "\\'") self.track_artist_name = f"{self.track_name} - {self.artist_name}" def shorten_name(self): self.temp_track_name = self.temp_track_name[:12] for sep in [" -", " ("]: idx = self.temp_track_name.find(sep) if idx != -1: self.temp_track_name = self.temp_track_name[:idx] self.temp_track_name = self.temp_track_name.lower().strip() def new_check_played_list(self): last_track = False self.check_last_played_table() results = self.sp_base.current_user_recently_played(limit=50) tracks = results['items'] while results['next']: results = self.sp_base.next(results) tracks.extend(results['items']) for track in tracks: t = track['track'] self.played_track = t['uri'] self.played_name = t['name'].lower() self.played_artist = t['album']['artists'][0]['name'].lower() if self.played_artist == "various artists": self.played_artist = t["artists"][0]["name"].lower() self.played_track_duration = t["duration_ms"] self.played_at = track['played_at'] if not last_track: self.played_artist_name = f"{self.played_name.title()} - {self.played_artist.title()}" print("Saving Last Track Played Which Was:- " + self.played_artist_name) self.last_song_played = self.played_artist_name last_track = True continue # Scenario [1] Are both URIs the same? if self.played_track in self.queue_list: self.no_exist = False self.uri_match += 1 self.temp_uri = self.played_track self.add_last_played_routine() self.mark_played_and_remove() continue # Scenario [2] URIs different. Are both Artists the same? elif self.played_artist in self.queue_artist_name_list: self.uri_change = False indices = [i for i, val in enumerate(self.queue_artist_name_list) if val == self.played_artist] self.found = False for ind in indices: self.temp_track_name = self.queue_track_list[ind] self.shorten_name() if self.played_name.startswith(self.temp_track_name): self.temp_uri = self.queue_list[ind] self.found = True break if not self.found: continue self.add_last_played_routine() self.mark_played_and_remove() self.amend_track_list_and_playlist() self.check_and_add_to_disc_table() self.found = False continue else: # Scenario [3] URIs different. Artists different. Try to match by name self.uri_change = False self.temp_track_name = self.played_name self.shorten_name() r = [s for s in self.queue_track_list if s.startswith(self.temp_track_name)] if r: try: indices = [i for i, val in enumerate(self.queue_artist_name_list) if val == self.played_artist] if not indices: self.no_exist = True print(f"{self.temp_uri} - {self.played_name} - {self.played_artist_name} This may be erroneous!!") self.add_last_played_routine() self.crap = 30 self.email_type = 3 self.email_me() self.email_type = 0 self.no_exist = False continue self.ai = self.queue_artist_name_list.index(self.played_artist) except Exception: self.temp_uri = self.played_track self.played_artist_name = f"{self.played_name.title()} - {self.played_artist.title()}" self.crap = 31 self.email_type = 3 self.email_me() self.email_type = 0 continue self.track_uri = self.queue_list[self.ai] self.temp_uri = self.track_uri self.add_last_played_routine() self.mark_played_and_remove() self.amend_track_list_and_playlist() self.check_and_add_to_disc_table() continue else: self.no_exist = True self.add_last_played_routine() self.crap = 32 self.email_type = 2 self.email_me() self.email_type = 0 self.no_exist = False continue def add_last_played_routine(self): self.track_duration = self.played_track_duration self.sort_out_the_track_length() self.played_track_duration = self.track_duration self.sort_out_the_last_played_track_name() self.sort_out_the_played_at_time() self.add_new_last_played_track() def sort_out_the_last_played_track_name(self): self.played_name = self.played_name.replace("'", "\\'") self.played_artist = self.played_artist.replace("'", "\\'") self.played_artist_name = f"{self.played_name} - {self.played_artist}" def mark_played_and_remove(self): self.mark_played_in_db() self.remove_from_playlist() def mark_played_in_db(self): with self.create_connection() as conn: cur = conn.cursor() try: cur.execute("SELECT * FROM Track_List WHERE URI = ?", (self.temp_uri,)) row = cur.fetchone() if not row: return self.email_track = row[1] self.email_artist = row[2] release_date = row[4] if release_date == "9999": self.email_type = 1 self.email_me() self.email_type = 0 play_count = row[6] + 1 cur.execute("UPDATE Track_List SET PLAY_COUNT = ? WHERE URI = ?", (play_count, self.temp_uri)) conn.commit() except Exception: print("Error with this track") self.crap = 33 self.email_type = 3 self.email_me() self.email_type = 0 self.remove_from_playlist() def remove_from_playlist(self): try: self.sp_base.user_playlist_remove_all_occurrences_of_tracks( user=self.queue_owner, playlist_id=self.queue_id, tracks=[self.temp_uri[-22:]] ) except spotipy.SpotifyException as error: print(error) print(f"RFP Incorrect URI for:- {self.played_name} by {self.played_artist} uri-NOW {self.played_track}") self.played_track = "" def get_new_tracks(self): self.get_queue_count() self.remaining_queue = self.queue_total self.get_tracks() def get_tracks(self): self.needed = 50 - self.remaining_queue self.record_wanted = self.needed print(f"{self.needed} New Tracks Required!") print("Performing Record Check") self.check_played_count() if self.needed <= 0: return with self.create_connection() as conn: 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 def set_queued_flag(self): with self.create_connection() as conn: cur = conn.cursor() cur.execute("SELECT rowid FROM Track_List WHERE URI = ?", (self.newtrack,)) toe = cur.fetchone() if toe: self.rowid_list.append(toe[0]) cur.execute("UPDATE Track_List SET QUEUED = 1 WHERE URI = ?", (self.newtrack,)) conn.commit() def add_track_to_queue(self): self.sp_base.playlist_add_items(playlist_id=self.queue_id, items=[self.newtrack]) def check_played_count(self): with self.create_connection() as conn: 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() def reset_queued_flag(self): print("Reseting Database Enormous Feed") with self.create_connection() as conn: cur = conn.cursor() cur.execute("UPDATE Track_List SET QUEUED = 0") cur.execute("UPDATE Track_List SET PLAY_COUNT = 0") conn.commit() def sort_out_the_played_at_time(self): 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 = f"{pdmh}-{pddy}-{pdyr} {pdtm}" self.disp_played_at = f"{pddy}-{pdmh}-{pdyr} {pdtm}" def sort_out_the_track_length(self): mill_sec = self.track_duration total_sec = mill_sec / 1000 min = int(total_sec // 60) sec = int(total_sec % 60) self.track_duration = f"{min}m {sec}s" def check_last_played_table(self): with self.create_connection() as conn: cur = conn.cursor() cur.execute( "CREATE TABLE IF NOT EXISTS Last_Played_Tracks (URI text PRIMARY KEY, TRACK_NAME text, ARTIST_NAME text, TRACK_LENGTH text, PLAY_COUNT integer, DISPLAY_DATE integer, EMAIL integer, LAST_PLAYED text);" ) conn.commit() def add_new_last_played_track(self): play_count = -100 if self.no_exist else 0 email_sent = 9 if self.no_exist else 0 self.played_name = self.played_name.title() self.played_artist = self.played_artist.title() with self.create_connection() as conn: cur = conn.cursor() tracks = ( self.played_track, self.played_name, self.played_artist, self.played_track_duration, play_count, self.disp_played_at, email_sent, self.played_at ) sql = ''' INSERT OR IGNORE INTO Last_Played_Tracks (URI,TRACK_NAME,ARTIST_NAME,TRACK_LENGTH,PLAY_COUNT,DISPLAY_DATE,EMAIL,LAST_PLAYED) VALUES(?,?,?,?,?,?,?,?) ''' cur.execute(sql, tracks) sql2 = ''' INSERT OR IGNORE INTO tmp_played (URI,TRACK_NAME,ARTIST_NAME,LAST_PLAYED) VALUES(?,?,?,?) ''' cur.execute(sql2, (self.played_track, self.played_name, self.played_artist, self.played_at)) conn.commit() def update_last_played_time(self): with self.create_connection() as conn: cur = conn.cursor() cur.execute("SELECT * FROM tmp_played ORDER BY LAST_PLAYED ASC ") rows = cur.fetchall() for row in rows: played_track2 = row[0] played_at2 = row[3] olddate = played_at2 mth = olddate[:2] dy = olddate[3:5] newdate = dy + "-" + mth + olddate[5:] disp_played_at2 = newdate cur.execute("UPDATE Last_Played_Tracks SET LAST_PLAYED = ? WHERE URI = ?", (played_at2, played_track2)) cur.execute("UPDATE Last_Played_Tracks SET DISPLAY_DATE = ? WHERE URI = ?", (disp_played_at2, played_track2)) conn.commit() def flex_temp_table(self): with self.create_connection() as conn: cur = conn.cursor() cur.execute("DROP TABLE IF EXISTS tmp_played") cur.execute("CREATE TABLE IF NOT EXISTS tmp_played (URI text, TRACK_NAME text, ARTIST_NAME text, LAST_PLAYED text PRIMARY KEY);") conn.commit() def update_play_count_in_last_played(self): with self.create_connection() as conn: cur = conn.cursor() cur.execute("SELECT * FROM tmp_played ORDER BY LAST_PLAYED DESC") rows = cur.fetchall() for row in rows: check_track = row[0] cur.execute("SELECT * FROM Track_List WHERE URI = ?", (check_track,)) tlrow = cur.fetchone() if tlrow is None: played_name = row[1].title().replace("'", "''") played_artist = row[2].replace("'", "''") cur.execute("SELECT * FROM Track_List WHERE TRACK_NAME = ? AND ARTIST_NAME = ?", (played_name, played_artist)) nmrow = cur.fetchone() if nmrow is None: self.crap = 34 self.email_type = 2 self.email_me() self.email_type = 0 plcount = -100 cur.execute("UPDATE Last_Played_Tracks SET PLAY_COUNT = ?, EMAIL = 1 WHERE URI = ?", (plcount, check_track)) conn.commit() continue else: plcount = nmrow[6] else: plcount = tlrow[6] cur.execute("UPDATE Last_Played_Tracks SET PLAY_COUNT = ? WHERE URI = ?", (plcount, check_track)) conn.commit() def update_playlist_info(self): now = datetime.datetime.now() plname3 = self.queue_name pldesc3 = f":- 50 Tracks from a Total of {self.total_records[0]} in the Master Playlist. {self.played_records[0]} played tracks with {self.remain_records} remaining" self.sp_base.playlist_change_details( playlist_id=self.queue_id, name=plname3, description=f"LAST UPDATED {now.strftime('%d-%m-%Y %H:%M:%S')} {pldesc3}", public=True ) 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") self.played_name = self.played_name.title() self.played_artist = self.played_artist.title() with self.create_connection() as conn: 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);" ) track = (newtoday, self.temp_uri, self.played_track, self.played_name, self.played_artist, disptoday) sql = ''' INSERT OR IGNORE INTO URI_Change (DATE_OF_CHANGE,OLD_URI,NEW_URI,TRACK_NAME,ARTIST_NAME,DISPLAY_DATE) VALUES(?,?,?,?,?,?) ''' cur.execute(sql, track) conn.commit() self.uri_change = True def amend_track_list_and_playlist(self): if self.uri_change: return self.name_match += 1 self.played_name = self.played_name.title() self.played_artist = self.played_artist.title() with self.create_connection() as conn: cur = conn.cursor() print(f"(New)SPT= {self.played_track} (Old)STU= {self.track_uri}") print(f"Track {self.played_name} Artist {self.played_artist}") cur.execute("UPDATE Track_List SET URI = ? WHERE URI = ?", (self.played_track, self.track_uri)) conn.commit() try: self.sp_base.user_playlist_remove_all_occurrences_of_tracks( user=self.queue_owner, playlist_id=self.queue_id, tracks=[self.track_uri[-22:]] ) print(f"Changing track details for:- {self.played_name} By {self.played_artist}") except spotipy.SpotifyException as er1: print(er1) print("Well that didn't work did it!! Amend track list and playlist function") print(f"Track {self.played_name} Artist {self.played_artist}") self.uri_change = False def do_a_queue_check(self): with self.create_connection() as conn: cur = conn.cursor() cur.execute("DROP TABLE IF EXISTS Queue_check") cur.execute( "CREATE TABLE IF NOT EXISTS Queue_check (URI text, TRACK_NAME text, ARTIST_NAME text, ALBUM_NAME text);" ) self.get_queue_count() self.reset_the_queue() self.get_the_queue() cur.execute("SELECT * FROM Track_list WHERE QUEUED = 1 AND PLAY_COUNT = 0") rows = cur.fetchall() for row in rows: self.quri, self.qtkn, self.qarn, self.qaln = row[0], row[1], row[2], row[3] if self.quri not in self.queue_list: track = (self.quri, self.qtkn, self.qarn, self.qaln) sql = ''' INSERT INTO Queue_check (URI,TRACK_NAME,ARTIST_NAME,ALBUM_NAME) VALUES(?,?,?,?) ''' cur.execute(sql, track) conn.commit() self.email_type = 6 self.email_me() self.email_type = 0 def make_preview_table(self): with self.create_connection() as conn: cur = conn.cursor() cur.execute("DROP TABLE IF EXISTS Queue_preview") cur.execute( "CREATE TABLE IF NOT EXISTS Queue_preview (URI text, TRACK_NAME text, ALBUM_PIC text, TRACK_PREVIEW text);" ) for qtknum in range(min(50, len(self.queue_list))): self.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 = (self.quri, qtrack, qpic, qpre) sql = ''' INSERT INTO Queue_preview (URI,TRACK_NAME,ALBUM_PIC,TRACK_PREVIEW) VALUES(?,?,?,?) ''' cur.execute(sql, picpre) conn.commit() def reset_the_queue(self): self.queue_list.clear() self.queue_artist_name_list.clear() self.queue_track_name_list.clear() self.queue_track_duration_list.clear() self.queue_track_list.clear() self.queue_track_pic_list.clear() self.queue_track_preview_list.clear() def email_me(self): addressee = "allanlockwoodsmith@gmail.com" msg = EmailMessage() msg['From'] = self.email_address msg['To'] = addressee if self.email_type == 1: with self.create_connection() as conn: cur = conn.cursor() cur.execute("SELECT * FROM Last_Played_Tracks WHERE URI = ?", (self.played_track,)) rows5 = cur.fetchone() if rows5 and rows5[6] == 1: return msg['Subject'] = "Unknown Release Date Found" msg.set_content(f"The following track was found to need a release date - \n{self.email_track} by {self.email_artist}\n\n DO NOT REPLY TO THIS EMAIL IT IS NOT MONITORED!") elif self.email_type == 2: with self.create_connection() as conn: cur = conn.cursor() cur.execute("SELECT * FROM Last_Played_Tracks WHERE URI = ?", (self.played_track,)) rows5 = cur.fetchone() if not rows5 or rows5[4] != -100 or rows5[6] != 0: return msg['Subject'] = "Played Track Mis-match Found" msg.set_content(f"The following track was not found in the Track_list Table - \n{self.crap} - {self.played_track} - {self.played_name} by {self.played_artist}\n\n DO NOT REPLY TO THIS EMAIL IT IS NOT MONITORED!") cur.execute("UPDATE Last_Played_Tracks SET EMAIL = 1 WHERE URI = ?", (self.played_track,)) conn.commit() elif self.email_type == 3: with self.create_connection() as conn: cur = conn.cursor() cur.execute("SELECT * FROM Last_Played_Tracks WHERE URI = ?", (self.played_track,)) rows5 = cur.fetchone() if rows5 and rows5[6] != 0: return msg['Subject'] = "Error with this Track " msg.set_content(f"The following track needs manual intervention - \n{self.crap} - {self.temp_uri} - {self.played_artist_name}\n\n DO NOT REPLY TO THIS EMAIL IT IS NOT MONITORED!") cur.execute("UPDATE Last_Played_Tracks SET EMAIL = 1 WHERE URI = ?", (self.temp_uri,)) conn.commit() elif self.email_type == 6: msg['Subject'] = "GB Error with this Track " msg.set_content(f"The following track has an old, Non-GB, URI - \n{self.quri} - {self.qtkn} - {self.qarn} - {self.qaln}\n\n DO NOT REPLY TO THIS EMAIL IT IS NOT MONITORED!") try: with smtplib.SMTP_SSL('smtp.gmail.com', 465) as smtp: smtp.login(self.email_address, self.email_password) smtp.send_message(msg) except smtplib.SMTPException as e: print(f"SMTP error: {e}") except Exception as e: print(f"General error: {e}") def compress_database(self): with self.create_connection() as conn: cur = conn.cursor() cur.execute("VACUUM") conn.commit() if __name__ == "__main__": cynthia = BigCynthia() cynthia.authorise() cynthia.repopulate_the_queue()