music-assistant-server

1"""Parsers to convert Last.fm API responses to Music Assistant media items."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from typing import TYPE_CHECKING, Any, cast
8
9from music_assistant_models.enums import ExternalID, MediaType, ProviderFeature
10from music_assistant_models.errors import MusicAssistantError
11from music_assistant_models.media_items import Album, Artist, ItemMapping, Track
12
13from music_assistant.constants import MASS_LOGGER_NAME
14from music_assistant.helpers.compare import compare_strings
15from music_assistant.providers.lastfm_recommendations.constants import (
16    PROVIDER_SEARCH_LIMIT,
17    SEARCH_CONCURRENCY_LIMIT,
18)
19
20if TYPE_CHECKING:
21    from music_assistant import MusicAssistant
22    from music_assistant.controllers.music.media.albums import AlbumsController
23    from music_assistant.controllers.music.media.artists import ArtistsController
24    from music_assistant.controllers.music.media.tracks import TracksController
25
26LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.lastfm_recommendations")
27
28# Limit concurrent provider searches to avoid overwhelming their APIs.
29_SEARCH_SEMAPHORE = asyncio.Semaphore(SEARCH_CONCURRENCY_LIMIT)
30
31
32def _is_matching_result(
33    item_mapping: ItemMapping, result: Artist | Album | Track, artist_name: str | None
34) -> bool:
35    """
36    Return True if a search result matches the searched item by name (and artist, if known).
37
38    :param item_mapping: ItemMapping that was searched for.
39    :param result: Search result to verify.
40    :param artist_name: Artist name to verify against the result's artists, if known.
41    """
42    # Album/track search names are "Artist - Title" while results usually expose only the title.
43    searched_name = item_mapping.name
44    title_part = (
45        searched_name[len(artist_name) + 3 :]
46        if artist_name and searched_name.startswith(f"{artist_name} - ")
47        else searched_name
48    )
49    if not compare_strings(title_part, result.name, strict=False) and not compare_strings(
50        searched_name, result.name, strict=False
51    ):
52        return False
53
54    # Verify the artist too: a matching title from a different artist (cover or karaoke
55    # version) must not pass. Results without artist info can't contradict, accept those.
56    result_artists = getattr(result, "artists", None)
57    if artist_name and result_artists:
58        return any(
59            compare_strings(artist_name, result_artist.name, strict=False)
60            for result_artist in result_artists
61        )
62    return True
63
64
65def _get_streaming_providers(
66    mass: MusicAssistant, item_mapping: ItemMapping, provider_instance_to_skip: str
67) -> list[Any]:
68    """
69    Return streaming providers that support the ItemMapping's media type.
70
71    :param mass: MusicAssistant instance.
72    :param item_mapping: ItemMapping with the media type to search for.
73    :param provider_instance_to_skip: Provider instance to skip (ourselves).
74    """
75    streaming_providers = []
76    for p in mass.music.providers:
77        if p.instance_id == provider_instance_to_skip:
78            continue
79        if not p.is_streaming_provider:
80            continue
81
82        if item_mapping.media_type == MediaType.ARTIST:
83            if ProviderFeature.LIBRARY_ARTISTS not in p.supported_features:
84                continue
85        elif item_mapping.media_type == MediaType.ALBUM:
86            if ProviderFeature.LIBRARY_ALBUMS not in p.supported_features:
87                continue
88        elif item_mapping.media_type == MediaType.TRACK:
89            if ProviderFeature.LIBRARY_TRACKS not in p.supported_features:
90                continue
91
92        streaming_providers.append(p)
93    return streaming_providers
94
95
96async def _search_provider(
97    ctrl: ArtistsController | AlbumsController | TracksController,
98    item_mapping: ItemMapping,
99    provider: Any,
100) -> Artist | Album | Track | None:
101    """
102    Search a single provider for a matching item.
103
104    :param ctrl: Controller for the media type.
105    :param item_mapping: ItemMapping to search for.
106    :param provider: Provider instance to search.
107    """
108    async with _SEARCH_SEMAPHORE:
109        try:
110            LOGGER.debug(
111                "Searching %s on %s for: %s",
112                item_mapping.media_type.value,
113                provider.name,
114                item_mapping.name,
115            )
116            # Use a higher limit to work around provider bugs (e.g. Spotify misbehaves at limit=1).
117            search_results = await ctrl.search(
118                item_mapping.name, provider.instance_id, limit=PROVIDER_SEARCH_LIMIT
119            )
120            if not search_results:
121                return None
122
123            return search_results[0]
124        except MusicAssistantError as err:
125            LOGGER.debug("Provider %s search failed: %s", provider.name, type(err).__name__)
126            return None
127
128
129async def _search_providers_concurrent(
130    ctrl: ArtistsController | AlbumsController | TracksController,
131    item_mapping: ItemMapping,
132    providers: list[Any],
133    artist_name: str | None,
134) -> Artist | Album | Track | None:
135    """
136    Search multiple providers concurrently and return the first verified match.
137
138    :param ctrl: Controller for the media type.
139    :param item_mapping: ItemMapping to search for.
140    :param providers: List of providers to search.
141    :param artist_name: Artist name to verify candidate matches against, if known.
142    """
143    tasks = [
144        asyncio.create_task(_search_provider(ctrl, item_mapping, provider))
145        for provider in providers
146    ]
147
148    for task in asyncio.as_completed(tasks):
149        result = await task
150        if result is None:
151            continue
152
153        if _is_matching_result(item_mapping, result, artist_name):
154            LOGGER.debug(
155                "Match on %s: %s (searched: %s)",
156                result.provider,
157                result.name,
158                item_mapping.name,
159            )
160            for t in tasks:
161                if not t.done():
162                    t.cancel()
163            await asyncio.gather(*tasks, return_exceptions=True)
164            return result
165
166        LOGGER.debug(
167            "Rejecting %s from %s: name mismatch (searched: %s)",
168            result.name,
169            result.provider,
170            item_mapping.name,
171        )
172
173    return None
174
175
176async def _resolve_item(
177    item_mapping: ItemMapping,
178    mass: MusicAssistant,
179    provider_instance_to_skip: str,
180    artist_name: str | None = None,
181) -> Artist | Album | Track | None:
182    """
183    Resolve an ItemMapping to a library or provider item.
184
185    :param item_mapping: ItemMapping with metadata and external IDs from Last.fm.
186    :param mass: MusicAssistant instance.
187    :param provider_instance_to_skip: Provider instance to skip (ourselves).
188    :param artist_name: Artist name to verify candidate matches against, if known.
189    """
190    ctrl: ArtistsController | AlbumsController | TracksController
191    if item_mapping.media_type == MediaType.ARTIST:
192        ctrl = mass.music.artists
193    elif item_mapping.media_type == MediaType.ALBUM:
194        ctrl = mass.music.albums
195    elif item_mapping.media_type == MediaType.TRACK:
196        ctrl = mass.music.tracks
197    else:
198        return None
199
200    LOGGER.debug(
201        "Resolving %s: %s (external IDs: %s)",
202        item_mapping.media_type.value,
203        item_mapping.name,
204        item_mapping.external_ids or "none",
205    )
206
207    if library_item := await ctrl.get_library_item_by_external_ids(item_mapping.external_ids):
208        LOGGER.debug("Found %s in library: %s", item_mapping.media_type.value, library_item.name)
209        return library_item
210
211    streaming_providers = _get_streaming_providers(mass, item_mapping, provider_instance_to_skip)
212    if not streaming_providers:
213        LOGGER.debug("No streaming providers available for resolution")
214        return None
215
216    result = await _search_providers_concurrent(
217        ctrl, item_mapping, streaming_providers, artist_name
218    )
219    if result is None:
220        LOGGER.debug("Could not resolve %s: %s", item_mapping.media_type.value, item_mapping.name)
221        return None
222
223    # Streaming providers expose ISRCs (and sometimes MBIDs) that Last.fm doesn't always
224    # provide; re-check the library against the resolved item's external IDs so we prefer
225    # the user's own copy when it exists.
226    if result.external_ids:
227        if library_item := await ctrl.get_library_item_by_external_ids(result.external_ids):
228            LOGGER.debug(
229                "Found %s in library via resolved external IDs: %s",
230                item_mapping.media_type.value,
231                library_item.name,
232            )
233            return library_item
234
235    return result
236
237
238async def parse_artist(
239    lastfm_artist: dict[str, Any], mass: MusicAssistant, provider_instance: str
240) -> Artist | None:
241    """
242    Parse a Last.fm artist and resolve it to a library or provider Artist.
243
244    :param lastfm_artist: Raw Last.fm artist dict with 'name' and 'mbid' fields.
245    :param mass: MusicAssistant instance for accessing library and providers.
246    :param provider_instance: Provider instance ID to skip when searching.
247    """
248    name = lastfm_artist.get("name", "Unknown Artist")
249    mbid = lastfm_artist.get("mbid")
250
251    external_ids = set()
252    if mbid:
253        external_ids.add((ExternalID.MB_ARTIST, mbid))
254
255    item_mapping = ItemMapping(
256        media_type=MediaType.ARTIST,
257        item_id="temp",
258        provider="lastfm_recommendations",
259        name=name,
260        external_ids=external_ids,
261    )
262
263    return cast("Artist | None", await _resolve_item(item_mapping, mass, provider_instance))
264
265
266async def parse_track(
267    lastfm_track: dict[str, Any],
268    mass: MusicAssistant,
269    provider_instance: str,
270) -> Track | None:
271    """
272    Parse a Last.fm track and resolve it to a library or provider Track.
273
274    :param lastfm_track: Raw Last.fm track dict with 'name', 'artist', 'mbid', 'duration'.
275    :param mass: MusicAssistant instance for accessing library and providers.
276    :param provider_instance: Provider instance ID to skip when searching.
277    """
278    name = lastfm_track.get("name", "Unknown Track")
279    mbid = lastfm_track.get("mbid")
280
281    artist_data = lastfm_track.get("artist", {})
282    if isinstance(artist_data, str):
283        artist_name = artist_data
284    else:
285        artist_name = artist_data.get("name", "Unknown Artist")
286
287    external_ids = set()
288    if mbid:
289        external_ids.add((ExternalID.MB_RECORDING, mbid))
290
291    item_mapping = ItemMapping(
292        media_type=MediaType.TRACK,
293        item_id="temp",
294        provider="lastfm_recommendations",
295        name=f"{artist_name} - {name}",
296        external_ids=external_ids,
297    )
298
299    return cast(
300        "Track | None", await _resolve_item(item_mapping, mass, provider_instance, artist_name)
301    )
302
303
304async def parse_album(
305    lastfm_album: dict[str, Any], mass: MusicAssistant, provider_instance: str
306) -> Album | None:
307    """
308    Parse a Last.fm album and resolve it to a library or provider Album.
309
310    :param lastfm_album: Raw Last.fm album dict with 'name', 'artist', 'mbid'.
311    :param mass: MusicAssistant instance for accessing library and providers.
312    :param provider_instance: Provider instance ID to skip when searching.
313    """
314    name = lastfm_album.get("name", "Unknown Album")
315    mbid = lastfm_album.get("mbid")
316
317    artist_data = lastfm_album.get("artist", {})
318    if isinstance(artist_data, str):
319        artist_name = artist_data
320    else:
321        artist_name = artist_data.get("name", "Unknown Artist")
322
323    external_ids = set()
324    if mbid:
325        external_ids.add((ExternalID.MB_ALBUM, mbid))
326
327    item_mapping = ItemMapping(
328        media_type=MediaType.ALBUM,
329        item_id="temp",
330        provider="lastfm_recommendations",
331        name=f"{artist_name} - {name}",
332        external_ids=external_ids,
333    )
334
335    return cast(
336        "Album | None", await _resolve_item(item_mapping, mass, provider_instance, artist_name)
337    )
338