music-assistant-server

1"""Async Lidarr API client for Music Assistant."""
2
3from __future__ import annotations
4
5import logging
6from dataclasses import dataclass
7from typing import Any
8
9import aiohttp
10
11
12@dataclass
13class LidarrArtist:
14    """Representation of a Lidarr artist."""
15
16    id: int
17    foreign_artist_id: str  # MusicBrainz artist ID
18    name: str
19    monitored: bool
20    path: str
21
22
23@dataclass
24class LidarrAlbum:
25    """Representation of a Lidarr album."""
26
27    id: int
28    foreign_album_id: str  # MusicBrainz release group ID
29    title: str
30    artist_id: int
31    monitored: bool
32
33
34@dataclass
35class LidarrRequestResult:
36    """Result of a Lidarr request operation."""
37
38    success: bool
39    artist_name: str
40    album_title: str
41    action: str  # "added_and_searched", "monitored_and_searched", "already_monitored", "error"
42    message: str
43
44
45class LidarrClient:
46    """Async client for the Lidarr API v1."""
47
48    def __init__(
49        self,
50        base_url: str,
51        api_key: str,
52        quality_profile_id: int,
53        metadata_profile_id: int,
54        root_folder_path: str,
55        logger: logging.Logger,
56    ) -> None:
57        """Initialize the Lidarr client.
58
59        :param base_url: Lidarr base URL (e.g. http://localhost:8686).
60        :param api_key: Lidarr API key.
61        :param quality_profile_id: Quality profile ID for new artists.
62        :param metadata_profile_id: Metadata profile ID for new artists.
63        :param root_folder_path: Root folder path for new artists.
64        :param logger: Logger instance.
65        """
66        self.base_url = base_url.rstrip("/")
67        self.api_key = api_key
68        self.quality_profile_id = quality_profile_id
69        self.metadata_profile_id = metadata_profile_id
70        self.root_folder_path = root_folder_path
71        self.logger = logger
72        self._session: aiohttp.ClientSession | None = None
73
74    @property
75    def _api_url(self) -> str:
76        return f"{self.base_url}/api/v1"
77
78    @property
79    def _headers(self) -> dict[str, str]:
80        return {
81            "X-Api-Key": self.api_key,
82            "Content-Type": "application/json",
83        }
84
85    async def _get_session(self) -> aiohttp.ClientSession:
86        if self._session is None or self._session.closed:
87            self._session = aiohttp.ClientSession(headers=self._headers)
88        return self._session
89
90    async def close(self) -> None:
91        """Close the HTTP session."""
92        if self._session and not self._session.closed:
93            await self._session.close()
94
95    async def _get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any:
96        """Perform a GET request to the Lidarr API."""
97        session = await self._get_session()
98        url = f"{self._api_url}/{endpoint}"
99        async with session.get(url, params=params) as resp:
100            resp.raise_for_status()
101            return await resp.json()
102
103    async def _post(self, endpoint: str, data: dict[str, Any] | None = None) -> Any:
104        """Perform a POST request to the Lidarr API."""
105        session = await self._get_session()
106        url = f"{self._api_url}/{endpoint}"
107        async with session.post(url, json=data) as resp:
108            resp.raise_for_status()
109            return await resp.json()
110
111    async def _put(self, endpoint: str, data: dict[str, Any] | None = None) -> Any:
112        """Perform a PUT request to the Lidarr API."""
113        session = await self._get_session()
114        url = f"{self._api_url}/{endpoint}"
115        async with session.put(url, json=data) as resp:
116            resp.raise_for_status()
117            return await resp.json()
118
119    async def test_connection(self) -> bool:
120        """Test connectivity to the Lidarr instance."""
121        try:
122            result = await self._get("system/status")
123            return "version" in result
124        except Exception:
125            self.logger.exception("Failed to connect to Lidarr at %s", self.base_url)
126            return False
127
128    async def get_artist_by_musicbrainz_id(self, mbid: str) -> LidarrArtist | None:
129        """Look up an artist in the Lidarr library by MusicBrainz ID.
130
131        :param mbid: MusicBrainz artist ID.
132        """
133        try:
134            artists = await self._get("artist", params={"mbId": mbid})
135            if artists:
136                a = artists[0]
137                return LidarrArtist(
138                    id=a["id"],
139                    foreign_artist_id=a["foreignArtistId"],
140                    name=a["artistName"],
141                    monitored=a["monitored"],
142                    path=a.get("path", ""),
143                )
144        except Exception:
145            self.logger.debug("Artist with mbid %s not found in Lidarr library", mbid)
146        return None
147
148    async def lookup_artist(self, term: str) -> list[dict[str, Any]]:
149        """Search for an artist on MusicBrainz via Lidarr.
150
151        :param term: Search term or MusicBrainz ID prefixed with lidarr:.
152        """
153        try:
154            return await self._get("artist/lookup", params={"term": term})
155        except Exception:
156            self.logger.exception("Failed to lookup artist: %s", term)
157            return []
158
159    async def lookup_album(self, term: str) -> list[dict[str, Any]]:
160        """Search for an album on MusicBrainz via Lidarr.
161
162        :param term: Search term or MusicBrainz ID prefixed with lidarr:.
163        """
164        try:
165            return await self._get("album/lookup", params={"term": term})
166        except Exception:
167            self.logger.exception("Failed to lookup album: %s", term)
168            return []
169
170    async def add_artist(self, foreign_artist_id: str, artist_name: str) -> LidarrArtist | None:
171        """Add an artist to Lidarr with monitoring set to none.
172
173        :param foreign_artist_id: MusicBrainz artist ID.
174        :param artist_name: Artist name for logging.
175        """
176        try:
177            # look up the artist first to get the full payload Lidarr expects
178            results = await self.lookup_artist(f"lidarr:{foreign_artist_id}")
179            if not results:
180                self.logger.warning(
181                    "Could not find artist '%s' (mbid: %s) in Lidarr lookup",
182                    artist_name,
183                    foreign_artist_id,
184                )
185                return None
186
187            artist_data = results[0]
188            artist_data.update(
189                {
190                    "qualityProfileId": self.quality_profile_id,
191                    "metadataProfileId": self.metadata_profile_id,
192                    "rootFolderPath": self.root_folder_path,
193                    "monitored": True,
194                    "addOptions": {
195                        "monitor": "none",
196                        "searchForMissingAlbums": False,
197                    },
198                }
199            )
200
201            result = await self._post("artist", data=artist_data)
202            self.logger.info(
203                "Added artist '%s' to Lidarr (id: %s)",
204                artist_name,
205                result.get("id"),
206            )
207            return LidarrArtist(
208                id=result["id"],
209                foreign_artist_id=result["foreignArtistId"],
210                name=result["artistName"],
211                monitored=result["monitored"],
212                path=result.get("path", ""),
213            )
214        except aiohttp.ClientResponseError as err:
215            if err.status == 400:
216                # artist likely already exists — try to fetch it
217                self.logger.debug(
218                    "Artist '%s' may already exist in Lidarr, attempting fetch",
219                    artist_name,
220                )
221                return await self.get_artist_by_musicbrainz_id(foreign_artist_id)
222            self.logger.exception("Failed to add artist '%s' to Lidarr", artist_name)
223            return None
224        except Exception:
225            self.logger.exception("Failed to add artist '%s' to Lidarr", artist_name)
226            return None
227
228    async def get_albums_for_artist(self, artist_id: int) -> list[LidarrAlbum]:
229        """Get all albums for an artist in the Lidarr library.
230
231        :param artist_id: Lidarr internal artist ID.
232        """
233        try:
234            albums = await self._get("album", params={"artistId": artist_id})
235            return [
236                LidarrAlbum(
237                    id=a["id"],
238                    foreign_album_id=a["foreignAlbumId"],
239                    title=a["title"],
240                    artist_id=a["artistId"],
241                    monitored=a["monitored"],
242                )
243                for a in albums
244            ]
245        except Exception:
246            self.logger.exception("Failed to get albums for artist id %s", artist_id)
247            return []
248
249    async def monitor_album(self, album_id: int) -> bool:
250        """Set an album to monitored status.
251
252        :param album_id: Lidarr internal album ID.
253        """
254        try:
255            await self._put("album/monitor", data={"albumIds": [album_id], "monitored": True})
256            return True
257        except Exception:
258            self.logger.exception("Failed to monitor album id %s", album_id)
259            return False
260
261    async def search_album(self, album_ids: list[int]) -> bool:
262        """Trigger a search for specific albums in Lidarr.
263
264        :param album_ids: List of Lidarr internal album IDs.
265        """
266        try:
267            result = await self._post(
268                "command", data={"name": "AlbumSearch", "albumIds": album_ids}
269            )
270            self.logger.debug("AlbumSearch command queued: %s", result.get("id"))
271            return True
272        except Exception:
273            self.logger.exception("Failed to trigger album search for ids %s", album_ids)
274            return False
275
276    async def request_album(
277        self,
278        artist_name: str,
279        album_title: str,
280        artist_mbid: str | None = None,
281        album_mbid: str | None = None,
282    ) -> LidarrRequestResult:
283        """Request an album in Lidarr — the main entry point.
284
285        Handles the full workflow: ensure artist exists, find the album,
286        monitor it, and trigger a search.
287
288        :param artist_name: Artist name (for display/fallback search).
289        :param album_title: Album title (for display/fallback search).
290        :param artist_mbid: MusicBrainz artist ID (preferred lookup method).
291        :param album_mbid: MusicBrainz release group ID (preferred lookup method).
292        """
293        # Step 1: Find or add the artist
294        artist: LidarrArtist | None = None
295
296        if artist_mbid:
297            artist = await self.get_artist_by_musicbrainz_id(artist_mbid)
298            if not artist:
299                artist = await self.add_artist(artist_mbid, artist_name)
300        else:
301            # fallback: search by name
302            results = await self.lookup_artist(artist_name)
303            if results:
304                mbid = results[0].get("foreignArtistId")
305                if mbid:
306                    artist = await self.get_artist_by_musicbrainz_id(mbid)
307                    if not artist:
308                        artist = await self.add_artist(mbid, artist_name)
309
310        if not artist:
311            return LidarrRequestResult(
312                success=False,
313                artist_name=artist_name,
314                album_title=album_title,
315                action="error",
316                message=f"Could not find or add artist '{artist_name}' in Lidarr",
317            )
318
319        # Step 2: Find the album in the artist's Lidarr discography
320        albums = await self.get_albums_for_artist(artist.id)
321        target_album: LidarrAlbum | None = None
322
323        if album_mbid:
324            # try exact match by MusicBrainz ID
325            target_album = next(
326                (a for a in albums if a.foreign_album_id == album_mbid),
327                None,
328            )
329
330        if not target_album:
331            # fallback: fuzzy match by title
332            album_title_lower = album_title.lower()
333            target_album = next(
334                (a for a in albums if a.title.lower() == album_title_lower),
335                None,
336            )
337
338        if not target_album:
339            # still not found — try partial match
340            target_album = next(
341                (a for a in albums if album_title_lower in a.title.lower()),
342                None,
343            )
344
345        if not target_album:
346            return LidarrRequestResult(
347                success=False,
348                artist_name=artist_name,
349                album_title=album_title,
350                action="error",
351                message=f"Album '{album_title}' not found in Lidarr's "
352                f"discography for '{artist_name}' ({len(albums)} albums checked)",
353            )
354
355        # Step 3: Check if already monitored
356        if target_album.monitored:
357            return LidarrRequestResult(
358                success=True,
359                artist_name=artist_name,
360                album_title=target_album.title,
361                action="already_monitored",
362                message=f"Album '{target_album.title}' by '{artist_name}' "
363                "is already monitored in Lidarr",
364            )
365
366        # Step 4: Monitor the album and trigger search
367        monitored = await self.monitor_album(target_album.id)
368        if not monitored:
369            return LidarrRequestResult(
370                success=False,
371                artist_name=artist_name,
372                album_title=target_album.title,
373                action="error",
374                message=f"Failed to set album '{target_album.title}' as monitored in Lidarr",
375            )
376
377        searched = await self.search_album([target_album.id])
378        action = "monitored_and_searched" if searched else "monitored_no_search"
379
380        return LidarrRequestResult(
381            success=True,
382            artist_name=artist_name,
383            album_title=target_album.title,
384            action=action,
385            message=f"Album '{target_album.title}' by '{artist_name}' "
386            f"{'monitored and search triggered' if searched else 'monitored but search failed'}",
387        )
388