/
/
/
1"""
2The LRCLIB Metadata provider for Music Assistant.
3
4Used for retrieval of synchronized lyrics.
5"""
6
7from __future__ import annotations
8
9from json import JSONDecodeError
10from typing import TYPE_CHECKING, Any, cast
11
12from aiohttp import ClientError
13from music_assistant_models.config_entries import ConfigEntry
14from music_assistant_models.enums import ConfigEntryType, ProviderFeature
15from music_assistant_models.errors import ResourceTemporarilyUnavailable
16from music_assistant_models.media_items import MediaItemMetadata, Track
17
18from music_assistant.controllers.cache import use_cache
19from music_assistant.helpers.throttle_retry import ThrottlerManager, throttle_with_retries
20from music_assistant.models.metadata_provider import MetadataProvider
21
22if TYPE_CHECKING:
23 from music_assistant_models.config_entries import ProviderConfig
24 from music_assistant_models.provider import ProviderManifest
25
26 from music_assistant.mass import MusicAssistant
27 from music_assistant.models import ProviderInstanceType
28
29SUPPORTED_FEATURES = {
30 ProviderFeature.TRACK_METADATA,
31 ProviderFeature.LYRICS,
32}
33
34CONF_API_URL = "api_url"
35DEFAULT_API_URL = "https://lrclib.net/api"
36USER_AGENT = "MusicAssistant (https://github.com/music-assistant/server)"
37
38
39async def setup(
40 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
41) -> ProviderInstanceType:
42 """Initialize provider(instance) with given configuration."""
43 return LrclibProvider(mass, manifest, config, SUPPORTED_FEATURES)
44
45
46class LrclibProvider(MetadataProvider):
47 """LRCLIB provider for handling synchronized lyrics."""
48
49 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
50 """Return Config entries to setup this provider."""
51 return (
52 ConfigEntry(
53 key=CONF_API_URL,
54 type=ConfigEntryType.STRING,
55 default_value=DEFAULT_API_URL,
56 required=False,
57 ),
58 )
59
60 async def handle_async_init(self) -> None:
61 """Handle async initialization of the provider."""
62 # Get the API URL from config
63 self.api_url = self.config.get_value(CONF_API_URL)
64
65 # Only use strict throttling if using the default API
66 if self.api_url == DEFAULT_API_URL:
67 self.throttler = ThrottlerManager(rate_limit=1, period=30)
68 self.logger.debug("Using default API with standard throttling (1 request per 30s)")
69 else:
70 # Less strict throttling for custom API endpoint
71 self.throttler = ThrottlerManager(rate_limit=1, period=1)
72 self.logger.debug("Using custom API endpoint: %s (throttling disabled)", self.api_url)
73
74 async def get_track_metadata(self, track: Track) -> MediaItemMetadata | None:
75 """Retrieve synchronized lyrics for a track."""
76 if track.metadata and (track.metadata.lyrics or track.metadata.lrc_lyrics):
77 self.logger.debug(
78 "Lyrics already exist for %s, skipping LRCLIB lookup for this track.",
79 track.name,
80 )
81 return None
82
83 if not track.artists:
84 self.logger.info("Skipping lyrics lookup for %s: No artist information", track.name)
85 return None
86
87 artist_name = track.artists[0].name
88 album_name = track.album.name if track.album else ""
89
90 duration = track.duration or 0
91
92 if not duration:
93 self.logger.info("Skipping lyrics lookup for %s: No duration information", track.name)
94 return None
95
96 self.logger.debug(
97 "Fetching synchronized lyrics for %s by %s (%s) on lrclib.net",
98 track.name,
99 artist_name,
100 album_name,
101 )
102
103 search_params = {
104 "track_name": track.name,
105 "artist_name": artist_name,
106 "album_name": album_name,
107 "duration": duration,
108 }
109
110 self.logger.debug("Searching lyrics (sync-ed preferred) with params: %s", search_params)
111
112 if data := await self._get_data(**search_params):
113 synced_lyrics = data.get("syncedLyrics")
114
115 if synced_lyrics:
116 metadata = MediaItemMetadata()
117 metadata.lrc_lyrics = synced_lyrics
118
119 self.logger.debug("Found synchronized lyrics for %s by %s", track.name, artist_name)
120 return metadata
121
122 self.logger.debug(
123 "No synchronized lyrics found for %s by %s with album name %s and with a "
124 "duration within 2 secs of %s",
125 track.name,
126 artist_name,
127 album_name,
128 duration,
129 )
130
131 plain_lyrics = data.get("plainLyrics")
132
133 if plain_lyrics:
134 metadata = MediaItemMetadata()
135 metadata.lyrics = plain_lyrics
136
137 self.logger.debug("Found plain lyrics for %s by %s", track.name, artist_name)
138 return metadata
139 self.logger.info(
140 "No lyrics found for %s by %s with album name %s and with a "
141 "duration within 2 secs of %s",
142 track.name,
143 artist_name,
144 album_name,
145 duration,
146 )
147 return None
148
149 @use_cache(3600 * 24 * 14) # Cache for 14 days
150 @throttle_with_retries
151 async def _get_data(self, **params: Any) -> dict[str, Any] | None:
152 """Get data from LRCLib API with throttling and retries."""
153 headers = {"User-Agent": USER_AGENT}
154 try:
155 async with self.mass.http_session.get(
156 f"{self.api_url}/get", params=params, headers=headers
157 ) as response:
158 # 204/404 mean there are genuinely no lyrics for this track
159 if response.status in (204, 404):
160 return None
161 response.raise_for_status()
162 return cast("dict[str, Any]", await response.json())
163 except (ClientError, TimeoutError, JSONDecodeError) as err:
164 # any other failure (5xx, network, malformed json) is transient â surface it as
165 # ResourceTemporarilyUnavailable so callers degrade instead of caching "no lyrics"
166 raise ResourceTemporarilyUnavailable("LRCLIB request failed") from err
167