from alive_progress import alive_bar import os import json import requests import time import sqlite3 from sqlite3 import Error from secrets import familyfeed_id from secrets import familyqueue_id from secrets import myfeed_id from secrets import myqueue_id from secrets import 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 = "" self.queue_id = "" self.queue_list = [] self.played_track = "" self.feed_count = 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 os.system('cls') print("Cynthia Yih Shih v.1.2.2 03/03/2021 Viking Utilities") while self.feed_count < 2: if self.feed_count == 0: self.playlist_id = familyfeed_id self.queue_id = familyqueue_id else: self.playlist_id = myfeed_id self.queue_id = myqueue_id self.get_endless_playlist() self.database = self.here_i_am + r"\\" + self.playlist_name + ".db" print("Getting Queue Count") self.get_queue_count() print("Getting Queue Items") 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("Performing Record Check") self.check_played_count() self.feed_count = self.feed_count + 1 continue return 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(self): # NOTE Get Details of each Track in Endless Queue and put it in a list #for i in range(self.queue_total): i = 0 while i < self.queue_total: with alive_bar(self.queue_total, title= self.queue_name) 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.queue_list.append(self.track_uri) if response.status_code == 200: i = i + 1 bar() continue 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 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() for i in response_json["items"]: self.played_track = (i["track"]["uri"]) if last_track == False: print("Saving Last Track Played Which Was:- " + (i["track"]["name"])) last_track = True continue if self.played_track in self.queue_list: self.mark_played_and_remove() continue 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: 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 PLAYED = True WHERE URI = '" + self.played_track + "'") cur.execute( "UPDATE Track_List SET PLAY_COUNT = '" + str(play_count) + "' WHERE URI = '" + self.played_track + "'") conn.commit() 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}]})) 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 if self.remaining_queue < 50: self.get_tracks() return def get_tracks(self): # NOTE Get the new tracks from the Endless Playlist self.needed = 50 - self.remaining_queue print(str(self.needed) +" new tracks required!") conn = self.create_connection(self.database) cur = conn.cursor() try: cur.execute("SELECT * FROM Track_List WHERE PLAY_COUNT = 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.needed = self.needed - 1 if self.needed == 0: return 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 PLAYED = 1") played_records = cur.fetchone() remain_records = total_records[0] - played_records[0] print("Remaining Un-Played Tracks in Database = " + str(remain_records)) if remain_records == 0: self.reset_played_flag() except Error as e: print(e) return def reset_played_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 PLAYED FROM Track_List") except Error as e: print(e) return rows = cur.fetchall() for row in rows: cur.execute("UPDATE Track_List SET PLAYED = 0") return a = cynthia() a.call_refresh() a.repopulate_the_queue()