embycovers/services/navidrome.py

"""Navidrome integration via the Subsonic API.

Navidrome speaks the Subsonic/OpenSubsonic REST API. Authentication uses the
salted-token scheme (``t = md5(password + salt)``) so the password never travels
in the clear. All functions take an ``httpx.AsyncClient`` so they share the
app-wide client and stay easy to test.
"""

from __future__ import annotations

import hashlib
import os
import secrets
from datetime import datetime

import httpx

NAVIDROME_URL = os.environ.get("NAVIDROME_URL", "http://10.0.0.2:4533")
NAVIDROME_USER = os.environ.get("NAVIDROME_USER", "")
NAVIDROME_PASSWORD = os.environ.get("NAVIDROME_PASSWORD", "")
NAVIDROME_CLIENT = "HomelabToolkit"
SUBSONIC_API_VERSION = "1.16.1"


class NavidromeError(Exception):
    def __init__(self, message: str, status: int = 502):
        self.message = message
        self.status = status
        super().__init__(message)


def is_configured() -> bool:
    return bool(NAVIDROME_URL and NAVIDROME_USER and NAVIDROME_PASSWORD)


def _require_configured() -> None:
    if not is_configured():
        raise NavidromeError(
            "Navidrome is not configured. Set NAVIDROME_URL, NAVIDROME_USER and "
            "NAVIDROME_PASSWORD.",
            status=503,
        )


def _auth_params() -> dict:
    salt = secrets.token_hex(8)
    token = hashlib.md5((NAVIDROME_PASSWORD + salt).encode("utf-8")).hexdigest()
    return {
        "u": NAVIDROME_USER,
        "t": token,
        "s": salt,
        "v": SUBSONIC_API_VERSION,
        "c": NAVIDROME_CLIENT,
        "f": "json",
    }


def _base_url() -> str:
    return NAVIDROME_URL.rstrip("/")


async def _call(client: httpx.AsyncClient, method: str, params: dict | None = None) -> dict:
    _require_configured()
    url = f"{_base_url()}/rest/{method}.view"
    request_params = {**_auth_params(), **(params or {})}
    try:
        response = await client.get(url, params=request_params)
    except httpx.RequestError as exc:
        raise NavidromeError(f"Could not reach Navidrome at {NAVIDROME_URL}: {exc}") from exc

    try:
        response.raise_for_status()
    except httpx.HTTPStatusError as exc:
        raise NavidromeError(
            f"Navidrome returned HTTP {response.status_code} for {method}.", status=502
        ) from exc

    try:
        payload = response.json().get("subsonic-response", {})
    except ValueError as exc:
        raise NavidromeError("Navidrome returned an invalid response.", status=502) from exc

    if payload.get("status") != "ok":
        error = payload.get("error") or {}
        message = error.get("message") or "Navidrome request failed."
        code = error.get("code")
        # Subsonic auth failures (codes 40/41) are the user's credentials, not a
        # server error — surface as 401 so the UI can prompt for setup.
        status = 401 if code in (40, 41, 44) else 502
        raise NavidromeError(message, status=status)

    return payload


def _cover_url(cover_art: str | None) -> str | None:
    if not cover_art:
        return None
    return f"/api/navidrome/cover/{cover_art}"


def _map_album(raw: dict) -> dict:
    return {
        "id": raw.get("id"),
        "name": raw.get("name") or raw.get("album") or "",
        "artist": raw.get("artist") or "",
        "artist_id": raw.get("artistId"),
        "year": raw.get("year"),
        "genre": raw.get("genre"),
        "song_count": raw.get("songCount") or 0,
        "duration": raw.get("duration") or 0,
        "cover_art": raw.get("coverArt"),
        "cover_url": _cover_url(raw.get("coverArt")),
        "created": raw.get("created"),
        "starred": bool(raw.get("starred")),
    }


def _map_song(raw: dict) -> dict:
    play_count = raw.get("playCount")
    try:
        play_count = int(play_count) if play_count is not None else 0
    except (TypeError, ValueError):
        play_count = 0
    return {
        "id": raw.get("id"),
        "title": raw.get("title") or "",
        "track": raw.get("track"),
        "disc": raw.get("discNumber"),
        "artist": raw.get("artist") or "",
        "artist_id": raw.get("artistId"),
        "album": raw.get("album") or "",
        "album_id": raw.get("albumId") or raw.get("parent"),
        "year": raw.get("year"),
        "genre": raw.get("genre"),
        "duration": raw.get("duration") or 0,
        "bitrate": raw.get("bitRate"),
        "suffix": raw.get("suffix"),
        "size": raw.get("size"),
        "path": raw.get("path"),
        "cover_art": raw.get("coverArt"),
        "cover_url": _cover_url(raw.get("coverArt")),
        "play_count": play_count,
        "created": raw.get("created"),
        "starred": raw.get("starred"),
    }


def _parse_dt(value: str | None) -> datetime | None:
    if not value:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None


def _dt_timestamp(value: str | None) -> float:
    parsed = _parse_dt(value)
    return parsed.timestamp() if parsed else 0.0


async def ping(client: httpx.AsyncClient) -> dict:
    if not is_configured():
        return {"connected": False, "configured": False, "url": NAVIDROME_URL}
    try:
        payload = await _call(client, "ping")
        return {
            "connected": True,
            "configured": True,
            "url": NAVIDROME_URL,
            "version": payload.get("version"),
            "server": payload.get("type") or payload.get("serverVersion"),
        }
    except NavidromeError as exc:
        return {
            "connected": False,
            "configured": True,
            "url": NAVIDROME_URL,
            "error": exc.message,
        }


async def get_artists(client: httpx.AsyncClient) -> list[dict]:
    payload = await _call(client, "getArtists")
    indexes = ((payload.get("artists") or {}).get("index")) or []
    artists: list[dict] = []
    for index in indexes:
        for artist in index.get("artist") or []:
            artists.append(
                {
                    "id": artist.get("id"),
                    "name": artist.get("name") or "",
                    "album_count": artist.get("albumCount") or 0,
                    "cover_art": artist.get("coverArt"),
                    "cover_url": _cover_url(artist.get("coverArt")),
                }
            )
    artists.sort(key=lambda entry: entry["name"].lower())
    return artists


async def get_albums(
    client: httpx.AsyncClient,
    *,
    list_type: str = "alphabeticalByName",
    size: int = 100,
    offset: int = 0,
) -> list[dict]:
    payload = await _call(
        client,
        "getAlbumList2",
        {"type": list_type, "size": size, "offset": offset},
    )
    raw_albums = ((payload.get("albumList2") or {}).get("album")) or []
    return [_map_album(album) for album in raw_albums]


async def search_albums(client: httpx.AsyncClient, query: str, *, count: int = 60) -> list[dict]:
    payload = await _call(
        client,
        "search3",
        {"query": query, "albumCount": count, "artistCount": 0, "songCount": 0},
    )
    raw_albums = ((payload.get("searchResult3") or {}).get("album")) or []
    return [_map_album(album) for album in raw_albums]


async def get_album(client: httpx.AsyncClient, album_id: str) -> dict:
    payload = await _call(client, "getAlbum", {"id": album_id})
    raw = payload.get("album") or {}
    album = _map_album(raw)
    album["songs"] = [_map_song(song) for song in (raw.get("song") or [])]
    return album


async def search_songs(client: httpx.AsyncClient, query: str, *, count: int = 100, offset: int = 0) -> list[dict]:
    payload = await _call(
        client,
        "search3",
        {"query": query, "artistCount": 0, "albumCount": 0, "songCount": count, "songOffset": offset},
    )
    raw_songs = ((payload.get("searchResult3") or {}).get("song")) or []
    return [_map_song(song) for song in raw_songs]


async def get_now_playing(client: httpx.AsyncClient) -> list[dict]:
    payload = await _call(client, "getNowPlaying")
    raw = ((payload.get("nowPlaying") or {}).get("entry")) or []
    return [_map_song(song) for song in raw]


async def get_starred(client: httpx.AsyncClient) -> dict:
    payload = await _call(client, "getStarred2")
    raw = payload.get("starred2") or {}
    return {
        "songs": [_map_song(song) for song in (raw.get("song") or [])],
        "albums": [_map_album(album) for album in (raw.get("album") or [])],
        "artists": [
            {
                "id": artist.get("id"),
                "name": artist.get("name") or "",
                "cover_art": artist.get("coverArt"),
                "cover_url": _cover_url(artist.get("coverArt")),
                "starred": artist.get("starred"),
            }
            for artist in (raw.get("artist") or [])
        ],
    }


async def get_playlists(client: httpx.AsyncClient) -> list[dict]:
    payload = await _call(client, "getPlaylists")
    raw = ((payload.get("playlists") or {}).get("playlist")) or []
    playlists = [
        {
            "id": p.get("id"),
            "name": p.get("name") or "",
            "song_count": p.get("songCount") or 0,
            "owner": p.get("owner") or "",
            "public": bool(p.get("public")),
            "duration": p.get("duration") or 0,
            "changed": p.get("changed"),
        }
        for p in raw
    ]
    playlists.sort(key=lambda entry: entry["name"].lower())
    return playlists


async def delete_playlist(client: httpx.AsyncClient, playlist_id: str) -> None:
    await _call(client, "deletePlaylist", {"id": playlist_id})


async def get_cover_art(
    client: httpx.AsyncClient, cover_id: str, size: int | None = None
) -> tuple[bytes, str]:
    _require_configured()
    url = f"{_base_url()}/rest/getCoverArt.view"
    params = {**_auth_params(), "id": cover_id}
    if size:
        params["size"] = size
    try:
        response = await client.get(url, params=params)
    except httpx.RequestError as exc:
        raise NavidromeError(f"Could not reach Navidrome at {NAVIDROME_URL}: {exc}") from exc
    if response.status_code != 200:
        raise NavidromeError("Cover art not found.", status=404)
    content_type = (response.headers.get("content-type") or "image/jpeg").split(";")[0]
    if not content_type.startswith("image/"):
        # Subsonic returns a JSON error document on failure.
        raise NavidromeError("Cover art not found.", status=404)
    return response.content, content_type


async def start_scan(client: httpx.AsyncClient, *, full: bool = False) -> dict:
    """Trigger a Navidrome library scan (Subsonic ``startScan`` extension)."""
    payload = await _call(client, "startScan", {"fullScan": "true" if full else "false"})
    status = payload.get("scanStatus") or {}
    return {"scanning": bool(status.get("scanning")), "count": status.get("count")}


async def get_genres(client: httpx.AsyncClient) -> list[dict]:
    payload = await _call(client, "getGenres")
    raw = ((payload.get("genres") or {}).get("genre")) or []
    genres = [
        {
            "name": g.get("value") or g.get("name") or "Unknown",
            "song_count": g.get("songCount") or 0,
            "album_count": g.get("albumCount") or 0,
        }
        for g in raw
    ]
    genres.sort(key=lambda g: g["song_count"], reverse=True)
    return genres


async def get_format_breakdown(
    client: httpx.AsyncClient, *, page_size: int = 500, max_pages: int = 400
) -> dict:
    """Count tracks by file format (flac, mp3, m4a, …) by paging all songs.

    Subsonic has no aggregate format endpoint, so we walk ``search3`` with an
    empty query (Navidrome returns the whole library) and tally each song's
    ``suffix``. Cap the page count so a runaway library can't loop forever.
    """
    counts: dict[str, int] = {}
    offset = 0
    for _ in range(max_pages):
        payload = await _call(
            client,
            "search3",
            {
                "query": "",
                "artistCount": 0,
                "albumCount": 0,
                "songCount": page_size,
                "songOffset": offset,
            },
        )
        songs = ((payload.get("searchResult3") or {}).get("song")) or []
        if not songs:
            break
        for song in songs:
            suffix = (song.get("suffix") or "").lower() or "other"
            counts[suffix] = counts.get(suffix, 0) + 1
        if len(songs) < page_size:
            break
        offset += page_size

    total = sum(counts.values())
    formats = sorted(
        ({"format": fmt, "count": count} for fmt, count in counts.items()),
        key=lambda entry: entry["count"],
        reverse=True,
    )
    return {"total": total, "formats": formats}


async def get_stats(client: httpx.AsyncClient) -> dict:
    """Library stats for the dashboard: artists, albums, tracks and genres."""
    artists = await get_artists(client)
    album_count = sum(artist["album_count"] for artist in artists)
    try:
        genres = await get_genres(client)
    except NavidromeError:
        genres = []
    song_count = sum(g["song_count"] for g in genres)
    return {
        "artist_count": len(artists),
        "album_count": album_count,
        "song_count": song_count,
        "genre_count": len(genres),
        "top_genres": genres[:8],
    }


async def get_reporting_snapshot(
    client: httpx.AsyncClient, *, page_size: int = 500, max_pages: int = 400
) -> dict:
    stats = await get_stats(client)
    # Keep the API calls explicit so they remain easy to debug.
    top_albums = await get_albums(client, list_type="frequent", size=12)
    recent_albums = await get_albums(client, list_type="recent", size=12)
    newest_albums = await get_albums(client, list_type="newest", size=12)
    starred = await get_starred(client)
    now_playing = await get_now_playing(client)

    all_songs: list[dict] = []
    offset = 0
    scanned_pages = 0
    truncated = False
    for _ in range(max_pages):
        scanned_pages += 1
        songs = await search_songs(client, "", count=page_size, offset=offset)
        if not songs:
            break
        all_songs.extend(songs)
        if len(songs) < page_size:
            break
        offset += page_size
    else:
        truncated = True

    songs_with_plays = [song for song in all_songs if (song.get("play_count") or 0) > 0]
    total_play_count = sum(song.get("play_count") or 0 for song in songs_with_plays)
    top_tracks = sorted(
        songs_with_plays,
        key=lambda song: ((song.get("play_count") or 0), song.get("title") or "", song.get("artist") or ""),
        reverse=True,
    )[:25]
    recently_added_tracks = sorted(
        [song for song in all_songs if song.get("created")],
        key=lambda song: _dt_timestamp(song.get("created")),
        reverse=True,
    )[:25]
    favorite_tracks = sorted(
        starred["songs"],
        key=lambda song: ((song.get("play_count") or 0), _dt_timestamp(song.get("starred"))),
        reverse=True,
    )[:25]

    return {
        "summary": {
            **stats,
            "library_tracks_scanned": len(all_songs),
            "tracks_with_plays": len(songs_with_plays),
            "total_play_count": total_play_count,
            "favorite_song_count": len(starred["songs"]),
            "favorite_album_count": len(starred["albums"]),
            "favorite_artist_count": len(starred["artists"]),
            "now_playing_count": len(now_playing),
            "scan_pages": scanned_pages,
            "truncated": truncated,
        },
        "top_tracks": top_tracks,
        "favorite_tracks": favorite_tracks,
        "recently_added_tracks": recently_added_tracks,
        "top_albums": top_albums,
        "recent_albums": recent_albums,
        "newest_albums": newest_albums,
        "top_genres": stats["top_genres"],
        "now_playing": now_playing,
        "favorite_albums": starred["albums"][:20],
        "favorite_artists": starred["artists"][:20],
    }