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