Harvey
297 lines
8.5 KiB
Python
297 lines
8.5 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncIterator
|
|
from typing import Any
|
|
|
|
import msgspec
|
|
|
|
from core.exceptions import NavidromeApiError, NavidromeAuthError, NavidromeSubsonicError
|
|
|
|
|
|
class SubsonicArtist(msgspec.Struct):
|
|
id: str
|
|
name: str
|
|
albumCount: int = 0
|
|
coverArt: str = ""
|
|
musicBrainzId: str = ""
|
|
|
|
|
|
class SubsonicSong(msgspec.Struct):
|
|
id: str
|
|
title: str
|
|
album: str = ""
|
|
albumId: str = ""
|
|
artist: str = ""
|
|
artistId: str = ""
|
|
track: int = 0
|
|
discNumber: int = 1
|
|
year: int = 0
|
|
duration: int = 0
|
|
bitRate: int = 0
|
|
suffix: str = ""
|
|
contentType: str = ""
|
|
musicBrainzId: str = ""
|
|
|
|
|
|
class SubsonicAlbum(msgspec.Struct):
|
|
id: str
|
|
name: str
|
|
artist: str = ""
|
|
artistId: str = ""
|
|
year: int = 0
|
|
genre: str = ""
|
|
songCount: int = 0
|
|
duration: int = 0
|
|
coverArt: str = ""
|
|
musicBrainzId: str = ""
|
|
song: list[SubsonicSong] | None = None
|
|
|
|
|
|
class SubsonicPlaylist(msgspec.Struct):
|
|
id: str
|
|
name: str
|
|
songCount: int = 0
|
|
duration: int = 0
|
|
owner: str = ""
|
|
public: bool = False
|
|
created: str = ""
|
|
changed: str = ""
|
|
coverArt: str = ""
|
|
entry: list[SubsonicSong] | None = None
|
|
|
|
|
|
class SubsonicGenre(msgspec.Struct):
|
|
name: str = ""
|
|
songCount: int = 0
|
|
albumCount: int = 0
|
|
|
|
|
|
class SubsonicArtistIndex(msgspec.Struct):
|
|
name: str = ""
|
|
artists: list[SubsonicArtist] = msgspec.field(default_factory=list)
|
|
|
|
|
|
class SubsonicMusicFolder(msgspec.Struct):
|
|
id: str = ""
|
|
name: str = ""
|
|
|
|
|
|
class SubsonicSearchResult(msgspec.Struct):
|
|
artist: list[SubsonicArtist] = msgspec.field(default_factory=list)
|
|
album: list[SubsonicAlbum] = msgspec.field(default_factory=list)
|
|
song: list[SubsonicSong] = msgspec.field(default_factory=list)
|
|
|
|
|
|
class StreamProxyResult(msgspec.Struct):
|
|
status_code: int
|
|
headers: dict[str, str]
|
|
media_type: str
|
|
body_chunks: AsyncIterator[bytes] | None = None
|
|
|
|
|
|
def parse_subsonic_response(data: dict[str, Any]) -> dict[str, Any]:
|
|
resp = data.get("subsonic-response")
|
|
if resp is None:
|
|
raise NavidromeApiError("Missing subsonic-response envelope")
|
|
status = resp.get("status", "")
|
|
if status != "ok":
|
|
error = resp.get("error", {})
|
|
code = error.get("code", 0)
|
|
message = error.get("message", "Unknown Subsonic API error")
|
|
if code in (40, 41):
|
|
raise NavidromeAuthError(message, code=code)
|
|
raise NavidromeSubsonicError(message, code=code)
|
|
return resp
|
|
|
|
|
|
def parse_artist(data: dict[str, Any]) -> SubsonicArtist:
|
|
return SubsonicArtist(
|
|
id=data.get("id", ""),
|
|
name=data.get("name", "Unknown"),
|
|
albumCount=data.get("albumCount", 0),
|
|
coverArt=data.get("coverArt", ""),
|
|
musicBrainzId=data.get("musicBrainzId", ""),
|
|
)
|
|
|
|
|
|
def parse_song(data: dict[str, Any]) -> SubsonicSong:
|
|
return SubsonicSong(
|
|
id=data.get("id", ""),
|
|
title=data.get("title", "Unknown"),
|
|
album=data.get("album", ""),
|
|
albumId=data.get("albumId", ""),
|
|
artist=data.get("artist", ""),
|
|
artistId=data.get("artistId", ""),
|
|
track=data.get("track", 0),
|
|
discNumber=data.get("discNumber", 1),
|
|
year=data.get("year", 0),
|
|
duration=data.get("duration", 0),
|
|
bitRate=data.get("bitRate", 0),
|
|
suffix=data.get("suffix", ""),
|
|
contentType=data.get("contentType", ""),
|
|
musicBrainzId=data.get("musicBrainzId", ""),
|
|
)
|
|
|
|
|
|
def parse_album(data: dict[str, Any]) -> SubsonicAlbum:
|
|
songs: list[SubsonicSong] | None = None
|
|
raw_songs = data.get("song")
|
|
if raw_songs is not None:
|
|
songs = [parse_song(s) for s in raw_songs]
|
|
|
|
return SubsonicAlbum(
|
|
id=data.get("id", ""),
|
|
name=data.get("name", data.get("title", "Unknown")),
|
|
artist=data.get("artist", ""),
|
|
artistId=data.get("artistId", ""),
|
|
year=data.get("year", 0),
|
|
genre=data.get("genre", ""),
|
|
songCount=data.get("songCount", 0),
|
|
duration=data.get("duration", 0),
|
|
coverArt=data.get("coverArt", ""),
|
|
musicBrainzId=data.get("musicBrainzId", ""),
|
|
song=songs,
|
|
)
|
|
|
|
|
|
def parse_genre(data: dict[str, Any]) -> SubsonicGenre:
|
|
return SubsonicGenre(
|
|
name=data.get("value", data.get("name", "")),
|
|
songCount=data.get("songCount", 0),
|
|
albumCount=data.get("albumCount", 0),
|
|
)
|
|
|
|
|
|
class SubsonicArtistInfo(msgspec.Struct):
|
|
biography: str = ""
|
|
musicBrainzId: str = ""
|
|
smallImageUrl: str = ""
|
|
mediumImageUrl: str = ""
|
|
largeImageUrl: str = ""
|
|
similarArtist: list[SubsonicArtist] = msgspec.field(default_factory=list)
|
|
|
|
|
|
class SubsonicNowPlayingEntry(msgspec.Struct):
|
|
id: str = ""
|
|
title: str = ""
|
|
artist: str = ""
|
|
album: str = ""
|
|
albumId: str = ""
|
|
artistId: str = ""
|
|
coverArt: str = ""
|
|
duration: int = 0
|
|
bitRate: int = 0
|
|
suffix: str = ""
|
|
username: str = ""
|
|
minutesAgo: int = 0
|
|
playerId: int = 0
|
|
playerName: str = ""
|
|
|
|
|
|
def parse_now_playing_entries(data: dict[str, Any]) -> list[SubsonicNowPlayingEntry]:
|
|
"""Extract now-playing entries from a Subsonic getNowPlaying response."""
|
|
np = data.get("nowPlaying", {})
|
|
entries_raw = np.get("entry", [])
|
|
if not entries_raw:
|
|
return []
|
|
result: list[SubsonicNowPlayingEntry] = []
|
|
for e in entries_raw:
|
|
result.append(SubsonicNowPlayingEntry(
|
|
id=e.get("id", ""),
|
|
title=e.get("title", ""),
|
|
artist=e.get("artist", ""),
|
|
album=e.get("album", ""),
|
|
albumId=e.get("albumId", ""),
|
|
artistId=e.get("artistId", ""),
|
|
coverArt=e.get("coverArt", ""),
|
|
duration=e.get("duration", 0),
|
|
bitRate=e.get("bitRate", 0),
|
|
suffix=e.get("suffix", ""),
|
|
username=e.get("username", ""),
|
|
minutesAgo=e.get("minutesAgo", 0),
|
|
playerId=e.get("playerId", 0),
|
|
playerName=e.get("playerName", ""),
|
|
))
|
|
return result
|
|
|
|
|
|
def parse_artist_info(data: dict[str, Any]) -> SubsonicArtistInfo:
|
|
"""Extract artist info from a Subsonic getArtistInfo2 response."""
|
|
info = data.get("artistInfo2", {})
|
|
if not info:
|
|
return SubsonicArtistInfo()
|
|
similar_raw = info.get("similarArtist", [])
|
|
similar = [parse_artist(a) for a in similar_raw] if similar_raw else []
|
|
return SubsonicArtistInfo(
|
|
biography=info.get("biography", ""),
|
|
musicBrainzId=info.get("musicBrainzId", ""),
|
|
smallImageUrl=info.get("smallImageUrl", ""),
|
|
mediumImageUrl=info.get("mediumImageUrl", ""),
|
|
largeImageUrl=info.get("largeImageUrl", ""),
|
|
similarArtist=similar,
|
|
)
|
|
|
|
|
|
class SubsonicAlbumInfo(msgspec.Struct):
|
|
notes: str = ""
|
|
musicBrainzId: str = ""
|
|
lastFmUrl: str = ""
|
|
smallImageUrl: str = ""
|
|
mediumImageUrl: str = ""
|
|
largeImageUrl: str = ""
|
|
|
|
|
|
class SubsonicLyricLine(msgspec.Struct):
|
|
value: str = ""
|
|
start: int | None = None # milliseconds from OpenSubsonic structuredLyrics
|
|
|
|
|
|
class SubsonicLyrics(msgspec.Struct):
|
|
value: str = ""
|
|
artist: str = ""
|
|
title: str = ""
|
|
lines: list[SubsonicLyricLine] = []
|
|
is_synced: bool = False
|
|
|
|
|
|
def parse_album_info(data: dict[str, Any]) -> SubsonicAlbumInfo:
|
|
"""Extract album info from a Subsonic getAlbumInfo2 response."""
|
|
info = data.get("albumInfo", {})
|
|
if not info:
|
|
return SubsonicAlbumInfo()
|
|
return SubsonicAlbumInfo(
|
|
notes=info.get("notes", ""),
|
|
musicBrainzId=info.get("musicBrainzId", ""),
|
|
lastFmUrl=info.get("lastFmUrl", ""),
|
|
smallImageUrl=info.get("smallImageUrl", ""),
|
|
mediumImageUrl=info.get("mediumImageUrl", ""),
|
|
largeImageUrl=info.get("largeImageUrl", ""),
|
|
)
|
|
|
|
|
|
def parse_lyrics(data: dict[str, Any]) -> SubsonicLyrics | None:
|
|
"""Extract lyrics from a Subsonic getLyrics response."""
|
|
lyrics = data.get("lyrics", {})
|
|
if not lyrics:
|
|
return None
|
|
value = lyrics.get("value", "")
|
|
if not value:
|
|
return None
|
|
return SubsonicLyrics(
|
|
value=value,
|
|
artist=lyrics.get("artist", ""),
|
|
title=lyrics.get("title", ""),
|
|
)
|
|
|
|
|
|
def parse_top_songs(data: dict[str, Any]) -> list[SubsonicSong]:
|
|
"""Extract songs from a Subsonic getTopSongs response."""
|
|
raw = data.get("topSongs", {}).get("song", [])
|
|
return [parse_song(s) for s in raw] if raw else []
|
|
|
|
|
|
def parse_similar_songs(data: dict[str, Any]) -> list[SubsonicSong]:
|
|
"""Extract songs from a Subsonic getSimilarSongs2 response."""
|
|
raw = data.get("similarSongs2", {}).get("song", [])
|
|
return [parse_song(s) for s in raw] if raw else []
|