/
/
/
1"""Async Lidarr API client for Music Assistant."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import re
8from dataclasses import dataclass
9from typing import Any
10
11import aiohttp
12
13# After adding a new artist, Lidarr needs time to fetch metadata from MusicBrainz.
14# We poll for albums with this interval and max attempts.
15ARTIST_REFRESH_POLL_INTERVAL = 5 # seconds between polls
16ARTIST_REFRESH_MAX_ATTEMPTS = 24 # max 120 seconds (2 minutes) total wait
17
18
19@dataclass
20class LidarrArtist:
21 """Representation of a Lidarr artist."""
22
23 id: int
24 foreign_artist_id: str # MusicBrainz artist ID
25 name: str
26 monitored: bool
27 path: str
28
29
30@dataclass
31class LidarrAlbum:
32 """Representation of a Lidarr album."""
33
34 id: int
35 foreign_album_id: str # MusicBrainz release group ID
36 title: str
37 artist_id: int
38 monitored: bool
39
40
41@dataclass
42class LidarrRequestResult:
43 """Result of a Lidarr request operation."""
44
45 success: bool
46 artist_name: str
47 album_title: str
48 action: str # "added_and_searched", "monitored_and_searched", "already_monitored", "error"
49 message: str
50
51
52class LidarrClient:
53 """Async client for the Lidarr API v1."""
54
55 def __init__(
56 self,
57 base_url: str,
58 api_key: str,
59 quality_profile_id: int,
60 metadata_profile_id: int,
61 root_folder_path: str,
62 logger: logging.Logger,
63 ) -> None:
64 """Initialize the Lidarr client.
65
66 :param base_url: Lidarr base URL (e.g. http://localhost:8686).
67 :param api_key: Lidarr API key.
68 :param quality_profile_id: Quality profile ID for new artists.
69 :param metadata_profile_id: Metadata profile ID for new artists.
70 :param root_folder_path: Root folder path for new artists.
71 :param logger: Logger instance.
72 """
73 self.base_url = base_url.rstrip("/")
74 self.api_key = api_key
75 self.quality_profile_id = quality_profile_id
76 self.metadata_profile_id = metadata_profile_id
77 self.root_folder_path = root_folder_path
78 self.logger = logger
79 self._session: aiohttp.ClientSession | None = None
80
81 @property
82 def _api_url(self) -> str:
83 return f"{self.base_url}/api/v1"
84
85 @property
86 def _headers(self) -> dict[str, str]:
87 return {
88 "X-Api-Key": self.api_key,
89 "Content-Type": "application/json",
90 }
91
92 async def _get_session(self) -> aiohttp.ClientSession:
93 if self._session is None or self._session.closed:
94 self._session = aiohttp.ClientSession(headers=self._headers)
95 return self._session
96
97 async def close(self) -> None:
98 """Close the HTTP session."""
99 if self._session and not self._session.closed:
100 await self._session.close()
101
102 async def _get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any:
103 """Perform a GET request to the Lidarr API."""
104 session = await self._get_session()
105 url = f"{self._api_url}/{endpoint}"
106 async with session.get(url, params=params) as resp:
107 resp.raise_for_status()
108 return await resp.json()
109
110 async def _post(self, endpoint: str, data: dict[str, Any] | None = None) -> Any:
111 """Perform a POST request to the Lidarr API."""
112 session = await self._get_session()
113 url = f"{self._api_url}/{endpoint}"
114 async with session.post(url, json=data) as resp:
115 resp.raise_for_status()
116 return await resp.json()
117
118 async def _put(self, endpoint: str, data: dict[str, Any] | None = None) -> Any:
119 """Perform a PUT request to the Lidarr API."""
120 session = await self._get_session()
121 url = f"{self._api_url}/{endpoint}"
122 async with session.put(url, json=data) as resp:
123 resp.raise_for_status()
124 return await resp.json()
125
126 async def test_connection(self) -> bool:
127 """Test connectivity to the Lidarr instance."""
128 try:
129 result = await self._get("system/status")
130 return "version" in result
131 except Exception:
132 self.logger.exception("Failed to connect to Lidarr at %s", self.base_url)
133 return False
134
135 async def get_artist_by_musicbrainz_id(self, mbid: str) -> LidarrArtist | None:
136 """Look up an artist in the Lidarr library by MusicBrainz ID.
137
138 :param mbid: MusicBrainz artist ID.
139 """
140 try:
141 artists = await self._get("artist", params={"mbId": mbid})
142 if artists:
143 a = artists[0]
144 return LidarrArtist(
145 id=a["id"],
146 foreign_artist_id=a["foreignArtistId"],
147 name=a["artistName"],
148 monitored=a["monitored"],
149 path=a.get("path", ""),
150 )
151 except Exception:
152 self.logger.debug("Artist with mbid %s not found in Lidarr library", mbid)
153 return None
154
155 async def lookup_artist(self, term: str) -> list[dict[str, Any]]:
156 """Search for an artist on MusicBrainz via Lidarr.
157
158 :param term: Search term or MusicBrainz ID prefixed with lidarr:.
159 """
160 try:
161 return await self._get("artist/lookup", params={"term": term})
162 except Exception:
163 self.logger.exception("Failed to lookup artist: %s", term)
164 return []
165
166 async def lookup_album(self, term: str) -> list[dict[str, Any]]:
167 """Search for an album on MusicBrainz via Lidarr.
168
169 :param term: Search term or MusicBrainz ID prefixed with lidarr:.
170 """
171 try:
172 return await self._get("album/lookup", params={"term": term})
173 except Exception:
174 self.logger.exception("Failed to lookup album: %s", term)
175 return []
176
177 async def add_artist(self, foreign_artist_id: str, artist_name: str) -> LidarrArtist | None:
178 """Add an artist to Lidarr with monitoring set to none.
179
180 :param foreign_artist_id: MusicBrainz artist ID.
181 :param artist_name: Artist name for logging.
182 """
183 try:
184 # look up the artist first to get the full payload Lidarr expects
185 results = await self.lookup_artist(f"lidarr:{foreign_artist_id}")
186 if not results:
187 self.logger.warning(
188 "Could not find artist '%s' (mbid: %s) in Lidarr lookup",
189 artist_name,
190 foreign_artist_id,
191 )
192 return None
193
194 artist_data = results[0]
195 artist_data.update(
196 {
197 "qualityProfileId": self.quality_profile_id,
198 "metadataProfileId": self.metadata_profile_id,
199 "rootFolderPath": self.root_folder_path,
200 "monitored": True,
201 "addOptions": {
202 "monitor": "none",
203 "searchForMissingAlbums": False,
204 },
205 }
206 )
207
208 result = await self._post("artist", data=artist_data)
209 self.logger.info(
210 "Added artist '%s' to Lidarr (id: %s)",
211 artist_name,
212 result.get("id"),
213 )
214 return LidarrArtist(
215 id=result["id"],
216 foreign_artist_id=result["foreignArtistId"],
217 name=result["artistName"],
218 monitored=result["monitored"],
219 path=result.get("path", ""),
220 )
221 except aiohttp.ClientResponseError as err:
222 if err.status == 400:
223 # artist likely already exists â try to fetch it
224 self.logger.debug(
225 "Artist '%s' may already exist in Lidarr, attempting fetch",
226 artist_name,
227 )
228 return await self.get_artist_by_musicbrainz_id(foreign_artist_id)
229 self.logger.exception("Failed to add artist '%s' to Lidarr", artist_name)
230 return None
231 except Exception:
232 self.logger.exception("Failed to add artist '%s' to Lidarr", artist_name)
233 return None
234
235 async def refresh_artist(self, artist_id: int) -> bool:
236 """Trigger a metadata refresh for an artist in Lidarr.
237
238 :param artist_id: Lidarr internal artist ID.
239 """
240 try:
241 await self._post("command", data={"name": "RefreshArtist", "artistId": artist_id})
242 return True
243 except Exception:
244 self.logger.exception("Failed to trigger RefreshArtist for id %s", artist_id)
245 return False
246
247 async def _wait_for_album_match(
248 self,
249 artist_id: int,
250 album_title: str,
251 album_mbid: str | None = None,
252 ) -> tuple[LidarrAlbum | None, list[LidarrAlbum]]:
253 """Poll Lidarr until the specific target album appears in the discography.
254
255 Albums populate incrementally after an artist is added/refreshed, so we
256 poll until we find the specific album we're looking for rather than
257 stopping as soon as any albums appear.
258
259 :param artist_id: Lidarr internal artist ID.
260 :param album_title: Album title to match.
261 :param album_mbid: Optional MusicBrainz release group ID for exact match.
262 """
263 prev_count = 0
264 stable_rounds = 0
265
266 for attempt in range(1, ARTIST_REFRESH_MAX_ATTEMPTS + 1):
267 await asyncio.sleep(ARTIST_REFRESH_POLL_INTERVAL)
268 albums = await self.get_albums_for_artist(artist_id)
269
270 # try to match the target album
271 match = self._match_album(albums, album_title, album_mbid)
272 if match:
273 self.logger.info(
274 "Found target album '%s' after %d poll attempts (%d total albums)",
275 match.title,
276 attempt,
277 len(albums),
278 )
279 return match, albums
280
281 # track whether the album count has stabilised to avoid
282 # hanging for artists that genuinely don't have the target album
283 current_count = len(albums)
284 if current_count > 0 and current_count == prev_count:
285 stable_rounds += 1
286 else:
287 stable_rounds = 0
288 prev_count = current_count
289
290 # if the album count hasn't changed for 3 consecutive polls,
291 # the discography is likely fully populated â stop waiting
292 if stable_rounds >= 3:
293 self.logger.info(
294 "Album count stable at %d for %d rounds, target '%s' not found â stopping poll",
295 current_count,
296 stable_rounds,
297 album_title,
298 )
299 return None, albums
300
301 self.logger.debug(
302 "Waiting for '%s' in artist id %s discography "
303 "(attempt %d/%d, %d albums so far)",
304 album_title,
305 artist_id,
306 attempt,
307 ARTIST_REFRESH_MAX_ATTEMPTS,
308 current_count,
309 )
310
311 albums = await self.get_albums_for_artist(artist_id)
312 self.logger.warning(
313 "Timed out waiting for album '%s' for artist id %s after %d attempts (%d albums found)",
314 album_title,
315 artist_id,
316 ARTIST_REFRESH_MAX_ATTEMPTS,
317 len(albums),
318 )
319 return None, albums
320
321 async def get_albums_for_artist(
322 self, artist_id: int, include_all: bool = True
323 ) -> list[LidarrAlbum]:
324 """Get all albums for an artist in the Lidarr library.
325
326 :param artist_id: Lidarr internal artist ID.
327 :param include_all: Include all release types (singles, EPs, etc.)
328 regardless of metadata profile filtering.
329 """
330 try:
331 params: dict[str, Any] = {"artistId": artist_id}
332 if include_all:
333 params["includeAllArtistAlbums"] = "true"
334 albums = await self._get("album", params=params)
335 return [
336 LidarrAlbum(
337 id=a["id"],
338 foreign_album_id=a["foreignAlbumId"],
339 title=a["title"],
340 artist_id=a["artistId"],
341 monitored=a["monitored"],
342 )
343 for a in albums
344 ]
345 except Exception:
346 self.logger.exception("Failed to get albums for artist id %s", artist_id)
347 return []
348
349 async def monitor_album(self, album_id: int) -> bool:
350 """Set an album to monitored status.
351
352 :param album_id: Lidarr internal album ID.
353 """
354 try:
355 await self._put("album/monitor", data={"albumIds": [album_id], "monitored": True})
356 return True
357 except Exception:
358 self.logger.exception("Failed to monitor album id %s", album_id)
359 return False
360
361 async def search_album(self, album_ids: list[int]) -> bool:
362 """Trigger a search for specific albums in Lidarr.
363
364 :param album_ids: List of Lidarr internal album IDs.
365 """
366 try:
367 result = await self._post(
368 "command", data={"name": "AlbumSearch", "albumIds": album_ids}
369 )
370 self.logger.debug("AlbumSearch command queued: %s", result.get("id"))
371 return True
372 except Exception:
373 self.logger.exception("Failed to trigger album search for ids %s", album_ids)
374 return False
375
376 @staticmethod
377 def _match_album(
378 albums: list[LidarrAlbum],
379 album_title: str,
380 album_mbid: str | None = None,
381 ) -> LidarrAlbum | None:
382 """Match an album by MusicBrainz ID or title within a list of albums.
383
384 Handles compound titles like 'Utopia / Porto' by also checking
385 individual parts against the album list.
386
387 :param albums: List of albums to search.
388 :param album_title: Album title to match.
389 :param album_mbid: Optional MusicBrainz release group ID for exact match.
390 """
391 if album_mbid:
392 match = next(
393 (a for a in albums if a.foreign_album_id == album_mbid),
394 None,
395 )
396 if match:
397 return match
398
399 album_title_lower = album_title.lower()
400
401 # exact title match
402 match = next(
403 (a for a in albums if a.title.lower() == album_title_lower),
404 None,
405 )
406 if match:
407 return match
408
409 # partial match: search title contained in album title
410 match = next(
411 (a for a in albums if album_title_lower in a.title.lower()),
412 None,
413 )
414 if match:
415 return match
416
417 # partial match: album title contained in search title
418 # (e.g. album "Porto" matches search "Utopia / Porto")
419 match = next(
420 (a for a in albums if a.title.lower() in album_title_lower),
421 None,
422 )
423 if match:
424 return match
425
426 # split compound title (separated by / & , etc.) and try each part
427 parts = re.split(
428 r"\s*/\s*|\s*&\s*|\s*,\s*", album_title, flags=re.IGNORECASE
429 )
430 if len(parts) > 1:
431 for part in parts:
432 part_lower = part.strip().lower()
433 if not part_lower:
434 continue
435 match = next(
436 (a for a in albums if a.title.lower() == part_lower),
437 None,
438 )
439 if match:
440 return match
441
442 return None
443
444 async def _find_album_via_lookup(
445 self,
446 artist_name: str,
447 album_title: str,
448 albums: list[LidarrAlbum],
449 ) -> LidarrAlbum | None:
450 """Fall back to MusicBrainz search to find an album not matched by title.
451
452 Searches for the album via Lidarr's lookup endpoint (which queries
453 MusicBrainz directly), then matches the result back to the artist's
454 discography by foreignAlbumId.
455
456 :param artist_name: Artist name for search context.
457 :param album_title: Album/single title to search for.
458 :param albums: The artist's existing album list in Lidarr.
459 """
460 search_term = f"{album_title} {artist_name}"
461 self.logger.info("Falling back to album lookup search: '%s'", search_term)
462
463 results = await self.lookup_album(search_term)
464 if not results:
465 return None
466
467 # build a set of known foreign IDs for quick matching
468 known_ids = {a.foreign_album_id: a for a in albums}
469
470 # check if any lookup result matches a known album in the discography
471 for result in results:
472 foreign_id = result.get("foreignAlbumId", "")
473 if foreign_id in known_ids:
474 self.logger.info(
475 "Album lookup matched '%s' (foreignAlbumId: %s) to existing discography entry",
476 result.get("title"),
477 foreign_id,
478 )
479 return known_ids[foreign_id]
480
481 # if no match in existing discography, the lookup result might be a single/EP
482 # that Lidarr knows about but hasn't associated yet â try matching by foreign ID
483 # directly against the album list (which includes all types with includeAll=true)
484 for result in results:
485 foreign_id = result.get("foreignAlbumId", "")
486 if not foreign_id:
487 continue
488 # check if this release belongs to the same artist
489 result_artist = result.get("artist", {})
490 result_artist_name = result_artist.get("artistName", "")
491 if result_artist_name.lower() != artist_name.lower():
492 continue
493 # found a match from the right artist â return a synthetic album
494 # this can happen when Lidarr's includeAll still doesn't list it
495 self.logger.info(
496 "Album lookup found '%s' by '%s' (foreignAlbumId: %s) via MusicBrainz, "
497 "but not in Lidarr discography â cannot monitor directly",
498 result.get("title"),
499 result_artist_name,
500 foreign_id,
501 )
502 # we can't monitor an album that isn't in the discography,
503 # so return None and let the caller handle it
504 break
505
506 return None
507
508 @staticmethod
509 def _split_compound_artist(artist_name: str) -> list[str]:
510 """Split a compound artist string into individual artist names.
511
512 Music Assistant may report multiple artists as a single string
513 separated by '/', '&', ',', ' x ', or ' feat. '.
514
515 :param artist_name: Possibly compound artist name.
516 """
517 pattern = r"\s*/\s*|\s*&\s*|\s*,\s*|\s+x\s+|\s+feat\.?\s+"
518 parts = re.split(pattern, artist_name, flags=re.IGNORECASE)
519 return [p.strip() for p in parts if p.strip()]
520
521 async def _find_or_add_artist(
522 self,
523 artist_name: str,
524 artist_mbid: str | None = None,
525 ) -> LidarrArtist | None:
526 """Find an existing artist in Lidarr or add them.
527
528 Handles compound artist names (e.g. 'Worakls/Apashe/Wasiu') by
529 trying each individual artist name when the full string fails.
530
531 :param artist_name: Artist name (may be compound).
532 :param artist_mbid: MusicBrainz artist ID (preferred lookup method).
533 """
534 if artist_mbid:
535 artist = await self.get_artist_by_musicbrainz_id(artist_mbid)
536 if artist:
537 return artist
538 artist = await self.add_artist(artist_mbid, artist_name)
539 if artist:
540 return artist
541
542 # try the full name first (or as MBID fallback)
543 names_to_try = [artist_name]
544
545 # if compound, add individual names as fallbacks
546 parts = self._split_compound_artist(artist_name)
547 if len(parts) > 1:
548 self.logger.info(
549 "Compound artist detected: '%s' â trying individual artists: %s",
550 artist_name,
551 parts,
552 )
553 names_to_try = parts # skip the compound name, try individuals
554
555 for name in names_to_try:
556 results = await self.lookup_artist(name)
557 if results:
558 mbid = results[0].get("foreignArtistId")
559 if mbid:
560 artist = await self.get_artist_by_musicbrainz_id(mbid)
561 if not artist:
562 artist = await self.add_artist(mbid, name)
563 if artist:
564 self.logger.info("Resolved artist '%s' from input '%s'", name, artist_name)
565 return artist
566
567 return None
568
569 async def request_album(
570 self,
571 artist_name: str,
572 album_title: str,
573 artist_mbid: str | None = None,
574 album_mbid: str | None = None,
575 ) -> LidarrRequestResult:
576 """Request an album in Lidarr â the main entry point.
577
578 Handles the full workflow: ensure artist exists, find the album,
579 monitor it, and trigger a search.
580
581 :param artist_name: Artist name (for display/fallback search).
582 :param album_title: Album title (for display/fallback search).
583 :param artist_mbid: MusicBrainz artist ID (preferred lookup method).
584 :param album_mbid: MusicBrainz release group ID (preferred lookup method).
585 """
586 # Step 1: Find or add the artist
587 artist = await self._find_or_add_artist(artist_name, artist_mbid)
588
589 if not artist:
590 return LidarrRequestResult(
591 success=False,
592 artist_name=artist_name,
593 album_title=album_title,
594 action="error",
595 message=f"Could not find or add artist '{artist_name}' in Lidarr",
596 )
597
598 # Step 2: Find the album in the artist's Lidarr discography.
599 # Try an immediate match first (works for existing artists with full metadata).
600 albums = await self.get_albums_for_artist(artist.id)
601 target_album = self._match_album(albums, album_title, album_mbid)
602
603 # If not found, trigger a refresh and poll until the specific album appears
604 # or the discography stabilises. Albums populate incrementally in Lidarr,
605 # so we must keep polling even if some albums are already present.
606 if not target_album:
607 self.logger.info(
608 "Album '%s' not found in %d current entries for '%s' (id: %s), "
609 "triggering refresh and polling for match",
610 album_title,
611 len(albums),
612 artist.name,
613 artist.id,
614 )
615 await self.refresh_artist(artist.id)
616 target_album, albums = await self._wait_for_album_match(
617 artist.id, album_title, album_mbid
618 )
619
620 # if still not found, fall back to MusicBrainz lookup
621 if not target_album:
622 self.logger.info(
623 "Album '%s' not found after polling %d discography entries for '%s', "
624 "trying MusicBrainz lookup",
625 album_title,
626 len(albums),
627 artist.name,
628 )
629 target_album = await self._find_album_via_lookup(
630 artist.name, album_title, albums
631 )
632
633 if not target_album:
634 return LidarrRequestResult(
635 success=False,
636 artist_name=artist_name,
637 album_title=album_title,
638 action="error",
639 message=f"Album/single '{album_title}' not found in Lidarr's "
640 f"discography for '{artist.name}' ({len(albums)} entries checked, "
641 "including MusicBrainz lookup fallback)",
642 )
643
644 # Step 3: Monitor the album and trigger search.
645 # Always call monitor + search even if the API reports monitored=True,
646 # because includeAllArtistAlbums=true can return stale/default values.
647 monitored = await self.monitor_album(target_album.id)
648 if not monitored:
649 return LidarrRequestResult(
650 success=False,
651 artist_name=artist_name,
652 album_title=target_album.title,
653 action="error",
654 message=f"Failed to set album '{target_album.title}' as monitored in Lidarr",
655 )
656
657 searched = await self.search_album([target_album.id])
658 action = "monitored_and_searched" if searched else "monitored_no_search"
659
660 return LidarrRequestResult(
661 success=True,
662 artist_name=artist_name,
663 album_title=target_album.title,
664 action=action,
665 message=f"Album '{target_album.title}' by '{artist.name}' "
666 f"{'monitored and search triggered' if searched else 'monitored but search failed'}",
667 )
668