plugin.audio.librespot/resources/lib/main_service.py

"""
    plugin.audio.librespot
    Spotify player for Kodi
    main_service.py
    Background service which launches the http service.
"""

import sys, os
sys.path.insert(1, os.path.join(os.path.dirname(__file__), "deps"))

import time
from typing import Dict

import xbmc
import xbmcaddon
import xbmcgui
from xbmc import LOGDEBUG, LOGWARNING

from bottle_manager import LibrespotServer
import utils
from http_video_player_setter import HttpVideoPlayerSetter
from save_recently_played import SaveRecentlyPlayed
from string_ids import HTTP_VIDEO_RULE_ADDED_STR_ID
from utils import PROXY_PORT, log_msg, ADDON_ID

from librespot_auth import LibrespotAuth

from librespot.core import Session

SAVE_TO_RECENTLY_PLAYED_FILE = True


SPOTIFY_ADDON = xbmcaddon.Addon(id=ADDON_ID)


def abort_app(timeout_in_secs: int) -> bool:
    return xbmc.Monitor().waitForAbort(timeout_in_secs)


def add_http_video_rule() -> None:
    video_player_setter = HttpVideoPlayerSetter()

    if not video_player_setter.set_http_rule():
        return

    msg = SPOTIFY_ADDON.getLocalizedString(HTTP_VIDEO_RULE_ADDED_STR_ID)
    dialog = xbmcgui.Dialog()
    header = SPOTIFY_ADDON.getAddonInfo("name")
    dialog.ok(header, msg)

def get_username() -> str:
    addon = xbmcaddon.Addon(id=ADDON_ID)
    spotify_username = addon.getSetting("username")
    if not spotify_username:
        raise Exception("Could not get spotify username.")
    return spotify_username

def get_password() -> str:
    addon = xbmcaddon.Addon(id=ADDON_ID)
    spotify_password = addon.getSetting("password")
    if not spotify_password:
        raise Exception("Could not get spotify password.")
    return spotify_password

class MainService:
    def __init__(self):
        log_msg(f"Spotify plugin version: {xbmcaddon.Addon(id=ADDON_ID).getAddonInfo('version')}.")

        self.__librespot_session: Session = Session.Builder(
            conf=Session.Configuration(
                store_credentials=False
            ).Builder().build()).user_pass(
                get_username(), get_password()
            ).create()

        add_http_video_rule()

        self.__librespot_auth: LibrespotAuth = LibrespotAuth(self.__librespot_session)
        self.__auth_token: Dict[str, str] = dict()

        self.__save_recently_played: SaveRecentlyPlayed = SaveRecentlyPlayed()

    def __save_track_to_recently_played(self, track_id: str) -> None:
        if SAVE_TO_RECENTLY_PLAYED_FILE:
            self.__save_recently_played.save_track(track_id)

    def run(self) -> None:
        log_msg("Starting main service loop.")
        self.__renew_token()

        librespot_server = LibrespotServer(self.__librespot_session)
        librespot_server.run(host='127.0.0.1', port=PROXY_PORT)
        log_msg(f"Started bottle with port {PROXY_PORT}.")


        loop_counter = 0
        loop_wait_in_secs = 6
        while True:
            loop_counter += 1
            if (loop_counter % 10) == 0:
                log_msg(f"Main loop continuing. Loop counter: {loop_counter}.")

            # Monitor authorization.
            if (int(self.__auth_token["expires_at"]) - 60) <= (int(time.time())):
                expire_time = int(self.__auth_token["expires_at"])
                time_now = int(time.time())
                log_msg(
                    f"Spotify token expired."
                    f" Expire time: {self.__get_time_str(expire_time)} ({expire_time});"
                    f" time now: {self.__get_time_str(time_now)} ({time_now})."
                )
                log_msg("Refreshing auth token now.")
                self.__renew_token()

            if abort_app(loop_wait_in_secs):
                break
        
        librespot_server.close()

    def __renew_token(self) -> None:
        log_msg("Retrieving auth token....", LOGDEBUG)

        self.__auth_token = self.__get_retry_auth_token()
        if not self.__auth_token:
            utils.cache_auth_token("")
            raise Exception(
                f"Could not get Spotify auth token for"
                f" user '{self.__spotty_helper.get_username()}'."
            )

        log_msg(
            f"Retrieved Spotify auth token."
            f" Expires at {self.__get_time_str(int(self.__auth_token['expires_at']))}."
        )

        # Cache auth token for easy access by the plugin.
        utils.cache_auth_token(self.__auth_token["access_token"])

    def __get_retry_auth_token(self) -> Dict[str, str]:
        auth_token = None
        max_retries = 20
        count = 0
        while count < max_retries:
            auth_token = self.__get_token()
            if auth_token:
                break
            time.sleep(1)
            count += 1

        if count > 0:
            log_msg(f"Took {count} retries to get authorization token.", LOGWARNING)

        return auth_token

    def __get_token(self) -> Dict[str, str]:
        return self.__librespot_auth.get_token()

    @staticmethod
    def __get_time_str(raw_time: int) -> str:
        return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(float(raw_time)))