271 lines
7.7 KiB
Python
271 lines
7.7 KiB
Python
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 Album:
|
|
id: str
|
|
title: str
|
|
name: str
|
|
album: str
|
|
coverArtId: str
|
|
duration: int
|
|
artistId: str
|
|
songCount: int
|
|
|
|
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'])
|
|
|
|
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
|
|
|
|
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'])
|
|
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']
|
|
self.comment = song_info['comment']
|
|
self.channelCount = int(song_info['channelCount'])
|
|
self.samplingRate = int(song_info['samplingRate'])
|
|
|
|
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) -> list[Album]:
|
|
url = self.base_url + '/rest/getAlbumList?'
|
|
|
|
payload = self._getBasePayload()
|
|
payload['type'] = 'alphabeticalByName'
|
|
payload['size'] = 20
|
|
|
|
r = requests.get(url + self._encodePayload(payload))
|
|
|
|
with open("albumList.xml" , 'wb') as file:
|
|
file.write(r.content)
|
|
|
|
# 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()))
|
|
|