obschatbot/services/SpotifyService.py

from obspython import *
from services.Service import Service
from utils.OAuth2 import OAuth2
import base64
import requests

KNOWN_COMMANDS = ["addsong", "dellast", "playlist"]
USER_KEY = "last_song_added"

AUTH_URL = "https://accounts.spotify.com"
API_URL = "https://api.spotify.com/v1"
SPOTIFY_SCOPES = "playlist-modify-public playlist-modify-private user-read-private"

class SpotifyService(OAuth2, Service):

    title = "Spotify Service"
    enabled = False

    msg_missing_args = ""
    msg_not_found = ""
    msg_not_added = ""
    msg_no_song = ""
    msg_deleted = ""

    code_url = f"{AUTH_URL}/authorize"
    authorization_url = f"{AUTH_URL}/api/token"
    refresh_url = f"{AUTH_URL}/api/token"
    scope = SPOTIFY_SCOPES
    playlist = ""
    msg_not_deleted = ""

    # OAuth -----

    def get_client_headers(self):
        return base64.b64encode(bytes(f"{self.client_id}:{self.client_secret}", "ascii")).decode("ascii")

    # Service -----

    def knows(self, command):
        return command in KNOWN_COMMANDS

    def eval(self, command, response, users):
        if command == "addsong":
            user = response["username"]
            components = response["message"].split(" ", 1)
            if len(components) != 2:
                self.queue.append(f"🎵 {self.msg_missing_args}, @{user}🎵")
                return
            query = components[1]
            self.queue.append(self.add_song(user, query, users))
        elif command == "dellast":
            self.queue.append(self.del_last(response["username"], users))
        elif command == "playlist":
            self.queue.append(self.get_playlist_url())

    def list_commands(self):
        return KNOWN_COMMANDS

    def start(self):
        if self.refresh_token == "":
            self.get_authorization()

    # OBS subset -----

    def load(self, settings):
        self.refresh_token = obs_data_get_string(settings, "spotify_token")

    def create_properties(self):
        props = obs_properties_create()
        obs_properties_add_text(props, "spotify_playlist_id", "Playlist ID", OBS_TEXT_DEFAULT)
        obs_properties_add_text(props, "spotify_client_id", "Client ID", OBS_TEXT_DEFAULT)
        obs_properties_add_text(props, "spotify_client_secret", "Client secret", OBS_TEXT_PASSWORD)
        token = obs_properties_add_text(props, "spotify_token", "Refresh token", OBS_TEXT_DEFAULT)
        obs_property_set_visible(token, False)
        obs_properties_add_text(props, "spotify_msg_missing_args", "Message: Missing arguments", OBS_TEXT_DEFAULT)
        obs_properties_add_text(props, "spotify_msg_not_found", "Message: Song not found", OBS_TEXT_DEFAULT)
        obs_properties_add_text(props, "spotify_msg_not_added", "Message: Song not added", OBS_TEXT_DEFAULT)
        obs_properties_add_text(props, "spotify_msg_no_song", "Message: No song added", OBS_TEXT_DEFAULT)
        obs_properties_add_text(props, "spotify_msg_deleted", "Message: Song deleted", OBS_TEXT_DEFAULT)
        obs_properties_add_text(props, "spotify_msg_not_deleted", "Message: Song not deleted", OBS_TEXT_DEFAULT)
        return props

    def update(self, settings):
        self.playlist = obs_data_get_string(settings, "spotify_playlist_id")
        self.set_credentials(
            obs_data_get_string(settings, "spotify_client_id"),
            obs_data_get_string(settings, "spotify_client_secret")
        )
        if obs_data_get_string(settings, "spotify_token") != "":
            self.set_refresh_token(obs_data_get_string(settings, "spotify_token"))
        self.msg_missing_args = obs_data_get_string(settings, "spotify_msg_missing_args")
        self.msg_not_found = obs_data_get_string(settings, "spotify_msg_not_found")
        self.msg_not_added = obs_data_get_string(settings, "spotify_msg_not_added")
        self.msg_no_song = obs_data_get_string(settings, "spotify_msg_no_song")
        self.msg_deleted = obs_data_get_string(settings, "spotify_msg_deleted")
        self.msg_not_deleted = obs_data_get_string(settings, "spotify_msg_not_deleted")

    def save(self, settings):
        obs_data_set_string(settings, "spotify_token", self.refresh_token)

    # Commands -----

    def add_song(self, user, query, users):
        if self.has_expired():
            self.refresh()
        track = self.search(query)
        if track is None:
            return f"🎵 {self.msg_not_found} @{user} 🎵"
        if self.add_to_playlist(track):
            artist = track["artists"][0]["name"]
            title = track["name"]
            if user not in users:
                users[user] = {}
            users[user][USER_KEY] = track["uri"]
            return f"🎵 Le morceau '{title}' de {artist} a été ajouté 🎵"
        else:
            return f"🎵 {self.msg_not_added}, @{user} 🎵"

    def search(self, query):
        track = None
        response = requests.get(
            f"{API_URL}/search",
            params={
                "q": query,
                "type": "track",
                "limit": 1
            },
            headers={"Authorization": f"Bearer {self.access_token}"}
        )
        if response.ok and response.json()["tracks"]["total"] > 0:
            track = response.json()["tracks"]["items"][0]
        return track

    def add_to_playlist(self, track):
        response = requests.post(
            f"{API_URL}/playlists/{self.playlist}/tracks",
            json={"uris": [track["uri"]]},
            headers={
                "Authorization": f"Bearer {self.access_token}",
                "Content-Type": "application/json"
            }
        )
        return response.ok

    def del_last(self, user, users):
        if user not in users or USER_KEY not in users[user]:
            return f"🎵 {self.msg_no_song}, @{user} 🎵"
        if self.has_expired():
            self.refresh()
        response = requests.delete(
            f"{API_URL}/playlists/{self.playlist}/tracks",
            json={"tracks": [{"uri": users[user][USER_KEY]}]},
            headers={
                "Authorization": f"Bearer {self.access_token}",
                "Content-Type": "application/json"
            }
        )
        if response.ok:
            users[user].pop(USER_KEY)
        return f"🎵 {self.msg_deleted if response.ok else self.msg_not_deleted}, @{user} 🎵"

    def get_playlist_url(self):
        return f"🎵 https://open.spotify.com/playlist/{self.playlist} 🎵"