subpy/SubSonic.py

import time
import random
import string
import hashlib
import logging
import urllib.parse

import xml.etree
import xml.etree.ElementTree

import requests


logger = logging.getLogger(__name__)

class Playlist:
    id: str
    name: str
    songCount: int
    duration: str
    owner: str
    coverArtId: str

    def __init__(self, info: dict):
        self.id = info['id']
        self.name = info['name']
        self.songCount = int(info['songCount'])
        self.duration = time.strftime('%H:%M:%S', time.gmtime(int(info['duration'])))
        self.owner = info['owner']
        self.coverArtId = info['coverArt']

class Album:
    id: str
    title: str
    name: str
    album: str
    coverArtId: str
    duration: int
    artistId: str
    songCount: int
    playCount: int
    year: int | None
    duration: str

    def __init__(self, album_info: dict):
        self.id = album_info['id']
        self.title = album_info['title']
        self.name = album_info['name']
        self.album = album_info['album']
        self.coverArtId = album_info['coverArt']
        self.duration = int(album_info['duration'])
        self.artistId = album_info['artistId']
        self.songCount = int(album_info['songCount'])
        self.playCount = int(album_info['playCount']) if 'playCount' in album_info else 0
        self.year = int(album_info['year']) if 'year' in album_info else None
        self.duration = self._secondToMinutes(int(album_info['duration']))

    def _secondToMinutes(self, s: int) -> str:
        return f'{s // 60}:{s % 60 if s % 60 > 9 else "0" + str(s % 60)}'

class Song:
    id: str
    title: str
    album: str
    artist: str
    trackNumber: int
    contentType: str
    suffix: str
    duration: int
    bitRate: int
    path: str
    discNumber: int
    albumId: str
    artistId: str
    comment: str
    channelCount: int
    samplingRate: int
    playCount: int | None

    def __init__(self, song_info: dict):
        self.id = song_info['id']
        self.title = song_info['title']
        self.album = song_info['album']
        self.artist = song_info['artist']
        self.trackNumber = int(song_info['track']) if 'track' in song_info else None
        self.contentType = song_info['contentType']
        self.suffix = song_info['suffix']
        self.duration = int(song_info['duration'])
        self.bitRate = int(song_info['bitRate'])
        self.path = song_info['path']
        self.discNumber = int(song_info['discNumber']) if 'discNumber' in song_info else None
        self.albumId = song_info['albumId']
        self.artistId = song_info['artistId'] if 'artistId' in song_info else None
        self.comment = song_info['comment']
        self.channelCount = int(song_info['channelCount'])
        self.samplingRate = int(song_info['samplingRate'])
        self.playCount = int(song_info['playCount']) if 'playCount' in song_info else None

class Artist:
    id: str
    name: str
    albumCount: int
    coverArt: str
    artistImageUrl: str

    def __init__(self, id: str, name: str, albumCount: str | int, coverArt: str, artistImageUrl: str):
        self.id = id
        self.name = name
        self.albumCount = int(albumCount)
        self.coverArt = coverArt
        self.artistImageUrl = artistImageUrl


class ArtistIndex:
    name: str
    artists: list[Artist]

    def __init__(self, indexInfo: dict, artists: list[dict]):
        self.name = indexInfo['name']

        self.artists = []
        for artist in artists:
            self.artists.append(Artist(
                artist['id'],
                artist['name'],
                artist['albumCount'],
                artist['coverArt'],
                artist['artistImageUrl']
            ))

class ArtistIndexes:
    indexes: list[ArtistIndex]

    def __init__(self, xml_data: bytes):
        self._parseXml(xml_data.decode('utf-8'))

    def _parseXml(self, xml_data: str):
        root = xml.etree.ElementTree.fromstring(xml_data)

        self.indexes: list[ArtistIndex] = []

        if root.get('status') != 'ok':
            return None

        in_index = False
        artists = []
        index_info = None
        for item in root.iter():
            # print(item.tag, dict(item.items()))

            if item.tag == '{http://subsonic.org/restapi}index':

                if in_index:
                    in_index = not in_index

                    self.indexes.append(ArtistIndex(index_info, artists))

                    artists = []
                else:
                    in_index = not in_index
                    index_info = dict(item.items())

            elif item.tag == '{http://subsonic.org/restapi}artist':
                artists.append(dict(item.items()))

class SubpyConn:
    username: str
    password: str
    base_url: str
    salt: str
    salt_len: int = 8
    version: str = "1.16.1"
    client_name: str = "testing"

    def __init__(self, u: str, p: str, base_url: str, salt_len: int | None = 8):
        self.username = u
        self.password = p
        self.base_url = base_url

        self.salt_len = salt_len

    def _saltGenerator(self) -> None:
        self.salt =  ''.join(random.SystemRandom().choice(string.ascii_uppercase + string.digits) for _ in range(self.salt_len))

    def _hashPass(self) -> str:
        self._saltGenerator()
        return hashlib.md5((self.password + self.salt).encode('utf-8')).hexdigest()

    def _getBasePayload(self) -> dict:
        return {
            "u" : self.username,
            "t" : self._hashPass(),
            "s" : self.salt,
            "v" : self.version,
            "c" : self.client_name
        }

    def _encodePayload(self, payload: dict) -> str:
        return urllib.parse.urlencode(payload, quote_via=urllib.parse.quote_plus)

    def getIndexes(self) -> ArtistIndexes:
        url = self.base_url + '/rest/getIndexes?'

        return ArtistIndexes(requests.get(url + self._encodePayload(self._getBasePayload())).content)

    def _getMusicDirectory(self, id: str):
        url = self.base_url + '/rest/getMusicDirectory?'

        payload = self._getBasePayload()
        payload['id'] = id

        return requests.get(url + self._encodePayload(payload))

    def getArtistAlbums(self, artist_id: str) -> list[Album]:
        r: requests.Response = self._getMusicDirectory(artist_id)
        xml_data: str = r.content.decode('utf-8')

        root: xml.etree.ElementTree.Element = xml.etree.ElementTree.fromstring(xml_data)
        albums: list[Album] = []

        for item in root.iter():
            if item.tag == '{http://subsonic.org/restapi}child':
                albums.append(Album(dict(item.items())))

        return albums

    def getAlbumSongs(self, album_id: str) -> list[Song]:
        r: requests.Response = self._getMusicDirectory(album_id)
        logger.debug(f"Requesting all songs for album {album_id} using url {self.base_url}")
        logger.debug(f'Got response with status code {r.status_code}')

        xml_data: str = r.content.decode('utf-8')

        if r.status_code != 200:
            return []

        root: xml.etree.ElementTree.Element = xml.etree.ElementTree.fromstring(xml_data)
        songs: list[Song] = []

        logger.debug(f"Request Status: {root.get('status')}")
        if root.get('status') != 'ok':
            return []

        for item in root.iter():

            if item.tag == '{http://subsonic.org/restapi}child':
                songs.append(Song(dict(item.items())))

        return songs

    def getArtists(self) -> list[Artist]:
        artistIndexes: ArtistIndexes = self.getIndexes()

        artists: list[Artist] = []
        for artistIndex in artistIndexes.indexes:
            artists.extend(artistIndex.artists)

        return artists

    def getAlbums(self, size: int | None = 25, offset: int | None = 0) -> list[Album]:
        url = self.base_url + '/rest/getAlbumList?'

        payload = self._getBasePayload()
        payload['type'] = 'alphabeticalByName'
        payload['size'] = size
        payload['offset'] = offset
        payload['f'] = 'json'

        r = requests.get(url + self._encodePayload(payload))
        json_r = r.json()

        albums: list[Album] = []

        if 'album' not in json_r['subsonic-response']['albumList']:
            return []

        for album in json_r['subsonic-response']['albumList']['album']:
            albums.append(Album(album))

        return albums

    def getAllAlbums(self) -> list[Album]:
        all_albums: list[Album] = []

        offset = 0
        while albums := self.getAlbums(50, offset):
            all_albums.extend(albums)
            offset += 50

        return all_albums

    def getPlaylists(self) -> list[Playlist]:
        url = self.base_url + "/rest/getPlaylists?"

        payload = self._getBasePayload()
        payload['f'] = 'json'

        r = requests.get(url + self._encodePayload(payload))

        json_r = r.json()

        playlists: list[Playlist] = []
        for playlist in json_r['subsonic-response']['playlists']['playlist']:
            playlists.append(Playlist(playlist))

        return playlists

    def getPlaylist(self, playlist_id: str) -> list[Song]:
        url = self.base_url + "/rest/getPlaylist?"

        payload = self._getBasePayload()
        payload['f'] = 'json'
        payload['id'] = playlist_id

        r = requests.get(url + self._encodePayload(payload))

        json_r = r.json()

        songs: list[Song] = []
        for song in json_r['subsonic-response']['playlist']['entry']:
            songs.append(Song(song))

        return songs

    def getStarred(self) -> list[Album | Song | Artist]:
        url = self.base_url + "/rest/getStarred?"

        payload = self._getBasePayload()
        payload['f'] = 'json'

        r = requests.get(url + self._encodePayload(payload))
        json_r = r.json()

        starred: list[Album | Song | Artist] = []
        for album in json_r['subsonic-response']['starred']['artist']:
            starred.append(Album(album))

        for artist in json_r['subsonic-response']['starred']['artist']:
            starred.append(Artist(
                artist['id'],
                artist['name'],
                artist['albumCount'],
                artist['coverArt'],
                artist['artistImageUrl']
            ))

        for song in json_r['subsonic-response']['starred']['song']:
            starred.append(Song(song))

        return starred

    def getStarredSongs(self) -> list[Song]:
        url = self.base_url + "/rest/getStarred?"

        payload = self._getBasePayload()
        payload['f'] = 'json'

        r = requests.get(url + self._encodePayload(payload))
        json_r = r.json()

        starred: list[Song] = []
        for song in json_r['subsonic-response']['starred']['song']:
            starred.append(Song(song))

        return starred

    # Returns a requests Response object, the data stream can be found in Response.content
    def stream(self, song_id: str) -> requests.Response:
        url = self.base_url + '/rest/stream?'

        payload = self._getBasePayload()

        payload['id'] = song_id

        return requests.get(url + self._encodePayload(payload), stream=True)

    def getSong(self, song_id: str) -> Song:
        url = self.base_url + '/rest/getSong?'

        payload = self._getBasePayload()
        payload['id'] = song_id

        r = requests.get(url + self._encodePayload(payload))

        xml_data = r.content.decode('utf-8')

        root = xml.etree.ElementTree.fromstring(xml_data)

        for item in root.iter():
            if item.tag == '{http://subsonic.org/restapi}song':
                return Song(dict(item.items()))