music-assistant-server

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