librespot-python/librespot/dealer/ApiClient.py

import logging
import typing

import requests

from librespot.core.ApResolver import ApResolver
from librespot.metadata import AlbumId, ArtistId, EpisodeId, ShowId, TrackId
from librespot.proto import Connect, Metadata
from librespot.standard import Closeable


class ApiClient(Closeable):
    _LOGGER: logging = logging.getLogger(__name__)
    _session = None
    _baseUrl: str = None

    def __init__(self, session):
        self._session = session
        self._baseUrl = "https://{}".format(ApResolver.get_random_spclient())

    def build_request(
        self,
        method: str,
        suffix: str,
        headers: typing.Union[None, typing.Dict[str, str]],
        body: typing.Union[None, bytes],
    ) -> requests.PreparedRequest:
        request = requests.PreparedRequest()
        request.method = method
        request.data = body
        request.headers = {}
        if headers is not None:
            request.headers = headers
        request.headers["Authorization"] = "Bearer {}".format(
            self._session.tokens().get("playlist-read")
        )
        request.url = self._baseUrl + suffix
        return request

    def send(
        self,
        method: str,
        suffix: str,
        headers: typing.Union[None, typing.Dict[str, str]],
        body: typing.Union[None, bytes],
    ) -> requests.Response:
        resp = self._session.client().send(
            self.build_request(method, suffix, headers, body)
        )
        return resp

    def put_connect_state(
        self, connection_id: str, proto: Connect.PutStateRequest
    ) -> None:
        resp = self.send(
            "PUT",
            "/connect-state/v1/devices/{}".format(self._session.device_id()),
            {
                "Content-Type": "application/protobuf",
                "X-Spotify-Connection-Id": connection_id,
            },
            proto.SerializeToString(),
        )

        if resp.status_code == 413:
            self._LOGGER.warning(
                "PUT state payload is too large: {} bytes uncompressed.".format(
                    len(proto.SerializeToString())
                )
            )
        elif resp.status_code != 200:
            self._LOGGER.warning(
                "PUT state returned {}. headers: {}".format(
                    resp.status_code, resp.headers
                )
            )

    def get_metadata_4_track(self, track: TrackId) -> Metadata.Track:
        resp = self.send(
            "GET", "/metadata/4/track/{}".format(track.hex_id()), None, None
        )
        ApiClient.StatusCodeException.check_status(resp)

        body = resp.content
        if body is None:
            raise RuntimeError()
        proto = Metadata.Track()
        proto.ParseFromString(body)
        return proto

    def get_metadata_4_episode(self, episode: EpisodeId) -> Metadata.Episode:
        resp = self.send(
            "GET", "/metadata/4/episode/{}".format(episode.hex_id()), None, None
        )
        ApiClient.StatusCodeException.check_status(resp)

        body = resp.content
        if body is None:
            raise IOError()
        proto = Metadata.Episode()
        proto.ParseFromString(body)
        return proto

    def get_metadata_4_album(self, album: AlbumId) -> Metadata.Album:
        resp = self.send(
            "GET", "/metadata/4/album/{}".format(album.hex_id()), None, None
        )
        ApiClient.StatusCodeException.check_status(resp)

        body = resp.content
        if body is None:
            raise IOError()
        proto = Metadata.Album()
        proto.ParseFromString(body)
        return proto

    def get_metadata_4_artist(self, artist: ArtistId) -> Metadata.Artist:
        resp = self.send(
            "GET", "/metadata/4/artist/{}".format(artist.hex_id()), None, None
        )
        ApiClient.StatusCodeException.check_status(resp)

        body = resp.content
        if body is None:
            raise IOError()
        proto = Metadata.Artist()
        proto.ParseFromString(body)
        return proto

    def get_metadata_4_show(self, show: ShowId) -> Metadata.Show:
        resp = self.send("GET", "/metadata/4/show/{}".format(show.hex_id()), None, None)
        ApiClient.StatusCodeException.check_status(resp)

        body = resp.content
        if body is None:
            raise IOError()
        proto = Metadata.Show()
        proto.ParseFromString(body)
        return proto

    class StatusCodeException(IOError):
        code: int

        def __init__(self, resp: requests.Response):
            super().__init__(resp.status_code)
            self.code = resp.status_code

        @staticmethod
        def check_status(resp: requests.Response) -> None:
            if resp.status_code != 200:
                raise ApiClient.StatusCodeException(resp)