storm/navipy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Couple bugs fixed. Shuffle shuffles from beginning.

Browse Source
This commit is contained in:
Storm Dragon
2025-12-16 03:54:17 -05:00
parent 1658321a4b
commit dd2d9a890a
3 changed files with 330 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

158
src/api/client.py
View File

@@ -2,6 +2,8 @@
Subsonic API client for Navidrome communication
"""
import hashlib
import json
from typing import Optional, List, Dict, Any
from urllib.parse import urlencode
import requests
@@ -53,6 +55,27 @@ class SubsonicClient:
self.session.headers.update({
'User-Agent': f'{self.CLIENT_NAME}/1.0'
})
self._cache: Dict[str, Any] = {}
def _cacheKey(self, endpoint: str, params: Optional[dict] = None) -> str:
"""Generate a cache key from endpoint and parameters."""
keyData = {"endpoint": endpoint, "params": params or {}}
keyString = json.dumps(keyData, sort_keys=True)
return hashlib.md5(keyString.encode()).hexdigest()
def _getCached(self, endpoint: str, params: Optional[dict] = None) -> Optional[Any]:
"""Retrieve a cached response if available."""
key = self._cacheKey(endpoint, params)
return self._cache.get(key)
def _setCache(self, endpoint: str, params: Optional[dict], value: Any) -> None:
"""Store a response in the cache."""
key = self._cacheKey(endpoint, params)
self._cache[key] = value
def clearCache(self) -> None:
"""Clear all cached responses. Call this on library refresh."""
self._cache.clear()
def _makeRequest(self, endpoint: str, params: dict = None) -> dict:
"""
@@ -187,6 +210,10 @@ class SubsonicClient:
if musicFolderId:
params['musicFolderId'] = musicFolderId
cached = self._getCached('getArtists', params)
if cached is not None:
return cached
response = self._makeRequest('getArtists', params)
artists = []
@@ -195,6 +222,7 @@ class SubsonicClient:
for artistData in index.get('artist', []):
artists.append(Artist.fromDict(artistData))
self._setCache('getArtists', params, artists)
return artists
def getArtist(self, artistId: str) -> Dict[str, Any]:
@@ -207,13 +235,20 @@ class SubsonicClient:
Returns:
dict with 'artist' (Artist) and 'albums' (List[Album])
"""
response = self._makeRequest('getArtist', {'id': artistId})
params = {'id': artistId}
cached = self._getCached('getArtist', params)
if cached is not None:
return cached
response = self._makeRequest('getArtist', params)
artistData = response.get('artist', {})
artist = Artist.fromDict(artistData)
albums = [Album.fromDict(a) for a in artistData.get('album', [])]
return {'artist': artist, 'albums': albums}
result = {'artist': artist, 'albums': albums}
self._setCache('getArtist', params, result)
return result
def getAlbum(self, albumId: str) -> Dict[str, Any]:
"""
@@ -225,13 +260,20 @@ class SubsonicClient:
Returns:
dict with 'album' (Album) and 'songs' (List[Song])
"""
response = self._makeRequest('getAlbum', {'id': albumId})
params = {'id': albumId}
cached = self._getCached('getAlbum', params)
if cached is not None:
return cached
response = self._makeRequest('getAlbum', params)
albumData = response.get('album', {})
album = Album.fromDict(albumData)
songs = [Song.fromDict(s) for s in albumData.get('song', [])]
return {'album': album, 'songs': songs}
result = {'album': album, 'songs': songs}
self._setCache('getAlbum', params, result)
return result
def getSong(self, songId: str) -> Song:
"""
@@ -243,8 +285,15 @@ class SubsonicClient:
Returns:
Song object
"""
response = self._makeRequest('getSong', {'id': songId})
return Song.fromDict(response.get('song', {}))
params = {'id': songId}
cached = self._getCached('getSong', params)
if cached is not None:
return cached
response = self._makeRequest('getSong', params)
result = Song.fromDict(response.get('song', {}))
self._setCache('getSong', params, result)
return result
def getGenres(self) -> List[Genre]:
"""
@@ -253,9 +302,15 @@ class SubsonicClient:
Returns:
List of Genre objects
"""
cached = self._getCached('getGenres', None)
if cached is not None:
return cached
response = self._makeRequest('getGenres')
genresData = response.get('genres', {})
return [Genre.fromDict(g) for g in genresData.get('genre', [])]
result = [Genre.fromDict(g) for g in genresData.get('genre', [])]
self._setCache('getGenres', None, result)
return result
def getAlbumList(self, listType: str, size: int = 20, offset: int = 0,
musicFolderId: str = None, genre: str = None,
@@ -289,9 +344,15 @@ class SubsonicClient:
if toYear:
params['toYear'] = toYear
cached = self._getCached('getAlbumList2', params)
if cached is not None:
return cached
response = self._makeRequest('getAlbumList2', params)
albumsData = response.get('albumList2', {})
return [Album.fromDict(a) for a in albumsData.get('album', [])]
result = [Album.fromDict(a) for a in albumsData.get('album', [])]
self._setCache('getAlbumList2', params, result)
return result
def getSongsByGenre(self, genre: str, count: int = 100, offset: int = 0) -> List[Song]:
"""
@@ -310,9 +371,16 @@ class SubsonicClient:
'count': count,
'offset': offset
}
cached = self._getCached('getSongsByGenre', params)
if cached is not None:
return cached
response = self._makeRequest('getSongsByGenre', params)
songsData = response.get('songsByGenre', {})
return [Song.fromDict(s) for s in songsData.get('song', [])]
result = [Song.fromDict(s) for s in songsData.get('song', [])]
self._setCache('getSongsByGenre', params, result)
return result
def getRandomSongs(self, size: int = 10, genre: str = None,
fromYear: int = None, toYear: int = None) -> List[Song]:
@@ -362,14 +430,21 @@ class SubsonicClient:
'albumCount': albumCount,
'songCount': songCount
}
cached = self._getCached('search3', params)
if cached is not None:
return cached
response = self._makeRequest('search3', params)
searchResult = response.get('searchResult3', {})
return {
result = {
'artists': [Artist.fromDict(a) for a in searchResult.get('artist', [])],
'albums': [Album.fromDict(a) for a in searchResult.get('album', [])],
'songs': [Song.fromDict(s) for s in searchResult.get('song', [])]
}
self._setCache('search3', params, result)
return result
# ==================== Playlists ====================
@@ -380,9 +455,15 @@ class SubsonicClient:
Returns:
List of Playlist objects
"""
cached = self._getCached('getPlaylists', None)
if cached is not None:
return cached
response = self._makeRequest('getPlaylists')
playlistsData = response.get('playlists', {})
return [Playlist.fromDict(p) for p in playlistsData.get('playlist', [])]
result = [Playlist.fromDict(p) for p in playlistsData.get('playlist', [])]
self._setCache('getPlaylists', None, result)
return result
def getPlaylist(self, playlistId: str) -> Playlist:
"""
@@ -394,8 +475,15 @@ class SubsonicClient:
Returns:
Playlist object with songs populated
"""
response = self._makeRequest('getPlaylist', {'id': playlistId})
return Playlist.fromDict(response.get('playlist', {}))
params = {'id': playlistId}
cached = self._getCached('getPlaylist', params)
if cached is not None:
return cached
response = self._makeRequest('getPlaylist', params)
result = Playlist.fromDict(response.get('playlist', {}))
self._setCache('getPlaylist', params, result)
return result
def createPlaylist(self, name: str, songIds: List[str] = None) -> Playlist:
"""
@@ -413,6 +501,7 @@ class SubsonicClient:
params['songId'] = songIds
response = self._makeRequest('createPlaylist', params)
self._invalidatePlaylistCache()
return Playlist.fromDict(response.get('playlist', {}))
def updatePlaylist(self, playlistId: str, name: str = None,
@@ -436,6 +525,7 @@ class SubsonicClient:
params['songIndexToRemove'] = songIndexesToRemove
self._makeRequest('updatePlaylist', params)
self._invalidatePlaylistCache(playlistId)
def deletePlaylist(self, playlistId: str) -> None:
"""
@@ -445,6 +535,15 @@ class SubsonicClient:
playlistId: ID of the playlist to delete
"""
self._makeRequest('deletePlaylist', {'id': playlistId})
self._invalidatePlaylistCache(playlistId)
def _invalidatePlaylistCache(self, playlistId: str = None) -> None:
"""Invalidate playlist-related cache entries."""
listKey = self._cacheKey('getPlaylists', None)
self._cache.pop(listKey, None)
if playlistId:
detailKey = self._cacheKey('getPlaylist', {'id': playlistId})
self._cache.pop(detailKey, None)
# ==================== User Actions ====================
@@ -463,6 +562,7 @@ class SubsonicClient:
}.get(itemType, 'id')
self._makeRequest('star', {paramName: itemId})
self._invalidateStarredCache()
def unstar(self, itemId: str, itemType: str = 'song') -> None:
"""
@@ -479,6 +579,12 @@ class SubsonicClient:
}.get(itemType, 'id')
self._makeRequest('unstar', {paramName: itemId})
self._invalidateStarredCache()
def _invalidateStarredCache(self) -> None:
"""Invalidate the starred/favorites cache."""
key = self._cacheKey('getStarred2', None)
self._cache.pop(key, None)
def setRating(self, itemId: str, rating: int) -> None:
"""
@@ -545,14 +651,20 @@ class SubsonicClient:
Returns:
dict with 'artists', 'albums', 'songs' lists
"""
cached = self._getCached('getStarred2', None)
if cached is not None:
return cached
response = self._makeRequest('getStarred2')
starredData = response.get('starred2', {})
return {
result = {
'artists': [Artist.fromDict(a) for a in starredData.get('artist', [])],
'albums': [Album.fromDict(a) for a in starredData.get('album', [])],
'songs': [Song.fromDict(s) for s in starredData.get('song', [])]
}
self._setCache('getStarred2', None, result)
return result
# ==================== Recommendations ====================
@@ -570,9 +682,16 @@ class SubsonicClient:
if not artist:
return []
params = {'artist': artist, 'count': count}
cached = self._getCached('getTopSongs', params)
if cached is not None:
return cached
response = self._makeRequest('getTopSongs', params)
topData = response.get('topSongs', {})
return [Song.fromDict(s) for s in topData.get('song', [])]
result = [Song.fromDict(s) for s in topData.get('song', [])]
self._setCache('getTopSongs', params, result)
return result
def getSimilarSongs(self, songId: str, count: int = 50) -> List[Song]:
"""
@@ -585,9 +704,16 @@ class SubsonicClient:
if not songId:
return []
params = {'id': songId, 'count': count}
cached = self._getCached('getSimilarSongs2', params)
if cached is not None:
return cached
response = self._makeRequest('getSimilarSongs2', params)
data = response.get('similarSongs2', {})
return [Song.fromDict(s) for s in data.get('song', [])]
result = [Song.fromDict(s) for s in data.get('song', [])]
self._setCache('getSimilarSongs2', params, result)
return result
def getNowPlaying(self) -> List[Song]:
"""