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 05-11-2020 @ 17.16 - © 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.getartist = "" self.getname = "" self.played = "" def get_now_playing_track(self): # NOTE: Get Spotify NOW PLAYING track. global nowtrack, getartist, getname, played 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. nowtrack = (json_data["item"]["uri"])[-22:] getartist = json_data["item"]["album"]["artists"][0]["name"] getname = json_data["item"]["name"] played = date_time global temp if temp != self.nowtrack: # NOTE: Check for a NEW Track Playing. prevtrack = temp # print("----------------------------------------------------------------------") # print("[<] 'prevtrack' " + prevtrack) temp = self.nowtrack # print("[>] 'now-track' " + temp) 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. print("Adding to Playlist Database.........") self.add_to_db() # NOTE: Add to Playlist.db time.sleep(3) # NOTE: Delay to prevent contact bounce. # print("[1] Refresh Access Token.") 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. # print("[2] Remove PREVIOUS track from QUEUE.") 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. # print("[-] Remove PREVIOUS track from start of ROTATION.") 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. # print("[3] Recycle PREVIOUS track into ROTATION.") 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 < 3: time.sleep(3) self.get_rotation_tracks() def get_rotation_tracks(self): # NOTE: Get additional tracks from your ROTATION playlist. # print("[X] Get fresh tracks from ROTATION.") query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.rot_playlist_id) params={"limit":20} 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. # print("[Y] Add 20 fresh tracks 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): # create a database connection to the SQLite database specified by db_file #:param db_file: database file # :return: Connection object or None conn = None try: conn = sqlite3.connect(db_file) except Error as e: print(e) return conn def create_table(self, conn, create_table_sql): # create a table from the create_table_sql statement #:param conn: Connection object # :param create_table_sql: a CREATE TABLE statement # :return: try: c = conn.cursor() c.execute(create_table_sql) except Error as e: print(e) def create_track(self, conn, trackS): # Create a new track into the tracks table # :param conn: # :param track: # :return: track id try: sql = ''' INSERT INTO tracks(URI,NAME,ARTIST,PLAY_DATE) VALUES(?,?,?,?) ''' cur = conn.cursor() cur.execute(sql, trackS) conn.commit() return cur.lastrowid except: print("track already present..... Ignoring!!") ### NEED TO UPDATE PLAYED TIME HERE!!! return cur.lastrowid def add_to_db(self): global track_id, database, conn database = r"C:\spotty\db\Play_History.db" sql_create_tracks_table = """ CREATE TABLE IF NOT EXISTS tracks (URI text PRIMARY KEY, NAME text, ARTIST text, PLAY_DATE text); """ # create a database connection conn = self.create_connection(database) # create tables if conn is not None: # create tracks table self.create_table(conn, sql_create_tracks_table) else: print("Error! cannot create the database connection.") # create a database connection conn = self.create_connection(database) with conn: # create a new track instance track = (nowtrack, getname, getartist, played) track_id = self.create_track(conn, track) 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()