import json import requests import time import random import sqlite3 from sqlite3 import Error from datetime import datetime from secrets import spotify_user_id from secrets import queue_id from secrets import rotation_id from refresh import Refresh # NOTE: Use a real track e.g. (Introduction by Scooter), rather than empty URI. [DON'T CHANGE - see Line: 68] temp = "spotify:track:5vWAGutZGHHfGZejH119Fg" print(" ") print("[•_•] DABOTIFY An 'Auto DJ' Play Queue Manager with Database App for Spotify.") print("v2.0.2 17-11-2020 @ 11.36 - © Vikings Design Studios.") print(" ") class RunBotify: def __init__(self): # NOTE: Top of the Class. self.user_id = spotify_user_id self.spotify_token = "" self.nowtrack = "" self.newtracks = "" self.que_playlist_id = queue_id self.rot_playlist_id = rotation_id self.dbcheck = 0 self.qcheck = 0 def get_now_playing_track(self): # NOTE: Get Spotify NOW PLAYING track. global nowtrack, getartist, getname, played, get_album, cover_art, get_year if self.dbcheck == 0: # NOTE: Make sure it only checks the Database Once. self.check_db() # NOTE: Check the Database and Table exists. if self.qcheck == 0: # Get tracks from QUEUE and top-up if required on first pass only. print("[?] Does QUEUE contain tracks?") self.get_track_total_from_queue() self.qcheck = 1 query = "https://api.spotify.com/v1/me/player/currently-playing" headers={"Content-type": "application/json","Authorization": "Bearer {}".format(self.spotify_token)} response = requests.request("get", url=query, headers=headers) if response.status_code == 204: print("No Track Playing! - [Re-check in 45 seconds]") time.sleep(45) print("[0] Refresh Access Token.") self.call_refresh() query = "https://api.spotify.com/v1/me/player/currently-playing" headers={"Content-type": "application/json","Authorization": "Bearer {}".format(self.spotify_token)} response = requests.request("get", url=query, headers=headers) json_data = json.loads(response.text) # NOTE: Get NOW PLAYING Track ITEMS. now = datetime.now() date_time = now.strftime("%d/%m/%Y at %H:%M:%S") self.nowtrack = json_data["item"]["uri"] # NOTE: Get NOW PLAYING Track URI. if ((json_data["item"]["uri"])[8:13]) == "local": nowtrack = (json_data["item"]["uri"])[14:] getartist = json_data["item"]["artists"][0]["name"] getname = json_data["item"]["name"] get_album = json_data["item"]["album"]["name"] get_year = ("XXXX") cover_art = ("LOCAL FILE") played = date_time else: nowtrack = (json_data["item"]["uri"])[-22:] getartist = json_data["item"]["album"]["artists"][0]["name"] getname = json_data["item"]["name"] get_album = json_data["item"]["album"]["name"] get_year = (json_data["item"]["album"]["release_date"])[:4] cover_art = json_data["item"]["album"]["images"][0]["url"] played = date_time global temp if temp != self.nowtrack: # NOTE: Check for a NEW Track Playing. prevtrack = temp temp = self.nowtrack now = datetime.now() date_time = now.strftime("%d/%m/%y at %H:%M:%S") logEntry = "• " + date_time + " [" + spotify_user_id + "] started playing '" + json_data["item"]["name"] + "'.\r" print(logEntry) # NOTE: Details of NOW PLAYING track. self.update_existing_track() # NOTE: Update Track if exists in Database self.add_to_db() # NOTE: Add to Playlist.db time.sleep(3) # NOTE: Delay to prevent contact bounce. self.call_refresh() # NOTE: Additional Refresh to beat the 60 minute timeout? (Acts on PREVIOUS tracks). self.nowtrack = prevtrack # NOTE: Set NOW PLAYING track as PREVIOUS track. self.played_track_manager() # NOTE: Manager for the PREVIOUS NOW PLAYING track. def played_track_manager(self): # NOTE: Manager for the PREVIOUS NOW PLAYING track. if self.nowtrack != "spotify:track:5vWAGutZGHHfGZejH119Fg": self.remove_track_from_queue() # NOTE: Remove track from QUEUE playlist. self.get_track_total_from_queue() # NOTE: Ensure the number of tracks in QUEUE is at least 3. self.remove_track_from_rotation() # NOTE: Remove track from ROTATION playlist. time.sleep(5) # NOTE: Delay to prevent contact bounce. self.add_track_to_rotation() # NOTE: Now add track BACK to the ROTATION playlist. while True: time.sleep(12) self.get_now_playing_track() def remove_track_from_queue(self): # NOTE: Remove track from QUEUE playlist. query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.que_playlist_id) headers={"Content-type": "application/json","Authorization":"Bearer {}".format(self.spotify_token)} params = {"tracks": [{"uri": self.nowtrack}]} response = requests.request("delete", url=query, headers=headers, data=json.dumps(params)) def remove_track_from_rotation(self): # NOTE: Remove track from ROTATION playlist. query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.rot_playlist_id) headers={"Content-type": "application/json","Authorization":"Bearer {}".format(self.spotify_token)} params = {"tracks": [{"uri": self.nowtrack}]} response = requests.request("delete", url=query, headers=headers, data=json.dumps(params)) def add_track_to_rotation(self): # NOTE: Add track to the ROTATION playlist. query = "https://api.spotify.com/v1/playlists/{}/tracks?uris={}".format(self.rot_playlist_id, self.nowtrack) headers={"Content-type": "application/json","Authorization":"Bearer {}".format(self.spotify_token)} response = requests.request("post", url=query, headers=headers) def get_track_total_from_queue(self): # NOTE: Get number of tracks in QUEUE playlist. query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.que_playlist_id) headers={"Content-type": "application/json","Authorization":"Bearer {}".format(self.spotify_token)} response = requests.request("get", url=query, headers=headers) response_json = response.json() tots_queue = 0 for i in response_json["items"]: tots_queue = tots_queue + 1 # NOTE: When QUEUE track numbers below set limit, add more from your ROTATION playlist. print("At the last count there were '" + str(tots_queue) + "' tracks in the QUEUE.") if tots_queue < 2: time.sleep(3) self.get_rotation_tracks() def get_rotation_tracks(self): # NOTE: Get additional tracks from your ROTATION playlist. query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.rot_playlist_id) params={"limit":30} headers={"Content-type": "application/json","Authorization":"Bearer {}".format(self.spotify_token)} response = requests.request("get", url=query, headers=headers, params=params) response_json = response.json() self.newtracks = "" for i in response_json["items"]: self.newtracks += i["track"]["uri"] + "," self.newtracks = self.newtracks[:-1] time.sleep(2) # NOTE: Delay to prevent contact bounce. self.add_to_queue() # NOTE: Add designated number of tracks from the ROTATION playlist to QUEUE playlist. def add_to_queue(self): # NOTE: Add tracks from your ROTATION to QUEUE. query = "https://api.spotify.com/v1/playlists/{}/tracks?uris={}".format(self.que_playlist_id,self.newtracks) headers={"Content-type": "application/json","Authorization": "Bearer {}".format(self.spotify_token)} response = requests.request("post", url=query, headers=headers) def create_connection(self, db_file): # NOTE: Create ConnectionTo Database conn = None try: conn = sqlite3.connect(db_file) return conn except Error as e: print(e) return conn def create_table(self, conn, create_table_sql): # NOTE: Create DAtabase Table try: c = conn.cursor() c.execute(create_table_sql) c.close() except Error as e: print(e) def check_db(self): # NOTE: Create Database and Table If Not Exist global database, conn self.dbcheck = 1 database = r"C:\spotty\db\Play_History - " + spotify_user_id + ".db" sql_create_tracks_table = """ CREATE TABLE IF NOT EXISTS tracks (URI text PRIMARY KEY, NAME text, ALBUM text, ARTIST text, DATE text, PLAY_DATE text, COVER_ART text); """ conn = self.create_connection(database) if conn is not None: self.create_table(conn, sql_create_tracks_table) else: print("Error! cannot create the database connection.") def create_track(self, conn, trackS): # NOTE: Insert The Track Into Database try: sql = ''' INSERT INTO tracks(URI,NAME,ALBUM,ARTIST,DATE,PLAY_DATE,COVER_ART) VALUES(?,?,?,?,?,?,?) ''' cur = conn.cursor() cur.execute(sql, trackS) conn.commit() return cur.lastrowid except: return cur.lastrowid def update_existing_track(self): # NOTE: Update Track If Exists database = r"C:\spotty\db\Play_History - " + spotify_user_id + ".db" conn = self.create_connection(database) with conn: self.update_track(conn, (getname, get_album, getartist, get_year, played, cover_art, nowtrack)) def update_track(self, conn, task): # NOTE: Execute the Update """ update priority, begin_date, and end date of a task :param conn: :param task: :return: project id """ sql = ''' UPDATE tracks SET NAME = ? , ALBUM = ? , ARTIST = ? , DATE = ? , PLAY_DATE = ? , COVER_ART = ? WHERE URI = ?''' cur = conn.cursor() cur.execute(sql, task) conn.commit() return cur.lastrowid def add_to_db(self): #NOTE: Add New Track To Database global track_id database = r"C:\spotty\db\Play_History - " + spotify_user_id + ".db" conn = self.create_connection(database) with conn: tracks = (nowtrack, getname, get_album, getartist, get_year, played, cover_art) track_id = self.create_track(conn, tracks) def call_refresh(self): # NOTE: Refresh the Access Token. refreshCaller = Refresh() self.spotify_token = refreshCaller.refresh() self.get_now_playing_track() a = RunBotify() # NOTE: This actually starts the program! a.call_refresh()