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()))