import os 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 precheck = 0 temp = "xxx" queue_max = 10 # NOTE: <== Your preffered QUEUE size. queue_count = 0 queue_dif = 0 timeout = 0 print(" ") print("[•_•] BOTIFY maintains a QUEUE playlist, fed by the") print("Least Recently Played tracks from a CYCLIC playlist.") print("with database and Webpage support") print("Version 3.0.0 09-Jan-21 @ 12:45 (c) Viking Designs.") print("Current User: " + spotify_user_id) print(" ") #os.system('python server.py') class RunBotify: def __init__(self): # NOTE: Top of the Class. global websong, webalbum, webcover 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.get_artist = "" self.get_album = "" self.get_year = "" self.dbcheck = 0 def get_now_playing_track(self): # NOTE: Get Spotify NOW PLAYING track. global nowtrack, getartist, getname, played, get_album, cover_art, get_year, websong, webalbum, webcover, precheck if self.dbcheck == 0: # NOTE: Make sure it only checks the Database Once. self.check_db() # NOTE: Check the Database and Table exists. if precheck == 0: # Get tracks from QUEUE and top-up if required on first pass only. self.get_track_total_from_queue() precheck = 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: self.nothing_playing() print("No Track Playing! - [Re-check in 45 seconds]") time.sleep(60) timeout = timeout + 1 if timeout == 50: 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"] self.get_artist = json_data["item"]["album"]["artists"][0]["name"] self.get_album = json_data["item"]["album"]["name"] self.get_year = json_data["item"]["album"]["release_date"][:4] websong = json_data["item"]["name"] webalbum = json_data["item"]["album"]["name"] webcover = json_data["item"]["album"]["images"][0]["url"] File_object = open(r"Web-data (DO NOT DELETE!).txt","w") File_object.writelines([websong + "\n", webalbum + "\n", webcover + "\n"]) File_object.close() 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") print("--------------------------------------------------------") print(date_time) print("Track " + json_data["item"]["name"] + ".") print("Artist " + self.get_artist + ".") print("Album " + self.get_album + ".") print("Release " + self.get_year + ".") 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). print("[1] Refresh Access Token.") 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 != "xxx": self.remove_track_from_queue() # NOTE: Remove track from QUEUE playlist. self.remove_track_from_rotation() # NOTE: Remove track from ROTATION playlist. self.add_track_to_rotation() # NOTE: Now add track BACK to the ROTATION playlist. self.get_track_total_from_queue() 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("[3] Remove PREVIOUS track from start of RECYCLE.") 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("[4] Replace PREVIOUS track to the end of RECYCLE.") 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() queue_tot = 0 for i in response_json["items"]: queue_tot = queue_tot + 1 print("[=] Current total '" + str(queue_tot) + "' tracks in the QUEUE.") if queue_tot < queue_max: self.get_rotation_tracks(queue_tot) def get_rotation_tracks(self, queue_tot): # NOTE: Get additional tracks from your ROTATION playlist. global queue_count, queue_dif # NOTE: QUEUE DIFF is the difference between the QUEUE MAX and current QUEUE TOTAL. queue_count = queue_max queue_dif = queue_count - queue_tot print("[<] Getting '" + str(queue_dif) + "' fresh tracks from RECYCLE.") query = "https://api.spotify.com/v1/playlists/{}/tracks".format(self.rot_playlist_id) params={"limit":queue_count} 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"] + "," # NOTE: The variable 'add_pos' is the position in the 'newtracks' data where the # NOTE: 'addition' point should start, thus skipping those tracks already in QUEUE. add_pos = (queue_dif * 37) # NOTE: 37 bits of data per track URI. self.newtracks = self.newtracks[-(add_pos):-1] self.add_to_queue() def add_to_queue(self): global queue_dif print("[+] Adding '" + str(queue_dif) + "' fresh tracks to the 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)} # NOTE: Keep 'POST' (append) as 'PUT' (replace) removes any remaining tracks from QUEUE. 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 Main\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 Main\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 Main\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()