/
/
/
1"""Manage MediaItems of type Track."""
2
3from __future__ import annotations
4
5from collections.abc import Iterable
6from datetime import UTC, datetime
7from typing import TYPE_CHECKING, Any, Never, cast
8
9from aiohttp import ClientError
10from music_assistant_models.auth import Scope
11from music_assistant_models.enums import (
12 ExternalID,
13 ImageType,
14 MediaType,
15 ProviderFeature,
16 ProviderType,
17)
18from music_assistant_models.errors import (
19 InvalidDataError,
20 MusicAssistantError,
21 ProviderUnavailableError,
22 UnsupportedFeaturedException,
23)
24from music_assistant_models.helpers import create_safe_string
25from music_assistant_models.media_items import (
26 Album,
27 Artist,
28 ItemMapping,
29 ItemMappingSummary,
30 MediaItemImage,
31 ProviderMapping,
32 Track,
33 TrackSummary,
34 UniqueList,
35)
36
37from music_assistant.constants import (
38 DB_TABLE_ALBUM_TRACKS,
39 DB_TABLE_ALBUMS,
40 DB_TABLE_TRACK_ARTISTS,
41 DB_TABLE_TRACKS,
42)
43from music_assistant.controllers.music.helpers import (
44 provider_mappings_for_update,
45 search_name_match_clause,
46)
47from music_assistant.helpers.compare import (
48 compare_artists,
49 compare_media_item,
50 compare_track,
51 loose_compare_strings,
52)
53from music_assistant.helpers.database import UNSET
54from music_assistant.helpers.json import json_loads, serialize_to_json
55from music_assistant.helpers.lyrics import extract_lrc_lyrics, normalize_lrc_lyrics
56from music_assistant.models.music_provider import MusicProvider
57
58from .base import MediaControllerBase, TrackSyncDetails
59
60if TYPE_CHECKING:
61 from collections.abc import Mapping
62
63 from music_assistant import MusicAssistant
64 from music_assistant.models.metadata_provider import MetadataProvider
65 from music_assistant.models.plugin import PluginProvider
66
67
68class TracksController(MediaControllerBase[Track]):
69 """Controller managing MediaItems of type Track."""
70
71 db_table = DB_TABLE_TRACKS
72 media_type = MediaType.TRACK
73 item_cls = Track
74 summary_item_cls = TrackSummary
75
76 def __init__(self, mass: MusicAssistant) -> None:
77 """Initialize class."""
78 super().__init__(mass)
79 # register (extra) api handlers
80 api_base = self.api_base
81 self.mass.register_api_command(
82 f"music/{api_base}/track_versions", self.versions, required_scope=Scope.LIBRARY_READ
83 )
84 self.mass.register_api_command(
85 f"music/{api_base}/track_albums", self.albums, required_scope=Scope.LIBRARY_READ
86 )
87 self.mass.register_api_command(
88 f"music/{api_base}/preview", self.get_preview_url, required_scope=Scope.LIBRARY_READ
89 )
90 self.mass.register_api_command(
91 f"music/{api_base}/similar_tracks",
92 self.similar_tracks,
93 required_scope=Scope.LIBRARY_READ,
94 )
95
96 @property
97 def base_query(self) -> tuple[str, dict[str, Any]]:
98 """Return the base SELECT query for tracks and its bound query params."""
99 # NOTE: the track_album subquery is fully self-contained (correlated) so the
100 # outer query needs no join with album_tracks (which would fan out rows for
101 # tracks that appear on multiple albums and force a GROUP BY). For tracks on
102 # multiple albums it prefers :preferred_album_id (used for album track
103 # listings) and otherwise deterministically picks the lowest album id.
104 query = f"""
105 SELECT
106 tracks.*,
107 {self._external_ids_query()} AS external_ids,
108 {self._provider_mappings_query()} AS provider_mappings,
109
110 (SELECT JSON_GROUP_ARRAY(
111 json_object(
112 'item_id', artists.item_id,
113 'provider', 'library',
114 'name', artists.name,
115 'sort_name', artists.sort_name,
116 'media_type', 'artist',
117 'external_ids', json({self._external_ids_query(MediaType.ARTIST, "artists")})
118 )) FROM artists JOIN track_artists on track_artists.track_id = tracks.item_id WHERE artists.item_id = track_artists.artist_id) AS artists,
119 (SELECT
120 json_object(
121 'item_id', albums.item_id,
122 'provider', 'library',
123 'name', albums.name,
124 'sort_name', albums.sort_name,
125 'media_type', 'album',
126 'year', albums.year,
127 'disc_number', album_tracks.disc_number,
128 'track_number', album_tracks.track_number,
129 'images', json_extract(albums.metadata, '$.images')
130 ) FROM album_tracks
131 JOIN albums ON albums.item_id = album_tracks.album_id
132 WHERE album_tracks.track_id = tracks.item_id
133 ORDER BY (album_tracks.album_id IS :preferred_album_id) DESC, album_tracks.album_id
134 LIMIT 1) AS track_album
135 FROM tracks
136 """
137 return query, {"preferred_album_id": None}
138
139 @property
140 def summary_query(self) -> tuple[str, dict[str, Any]]:
141 """Return the slim SELECT query used for track summary listings."""
142 # the track_album subquery follows the same correlated pattern as in base_query
143 # (see the NOTE there), just with the few fields a list row needs
144 query = f"""
145 SELECT
146 {self._summary_base_columns()},
147 tracks.version,
148 tracks.duration,
149 json_extract(tracks.metadata, '$.explicit') AS explicit,
150 json_extract(tracks.metadata, '$.release_date') AS release_date,
151 {self._provider_mappings_query()} AS provider_mappings,
152 {self._artist_mappings_summary_query(DB_TABLE_TRACK_ARTISTS, "track_id")} AS artists,
153 (SELECT
154 json_object(
155 'item_id', albums.item_id,
156 'name', albums.name,
157 'sort_name', albums.sort_name,
158 'year', albums.year,
159 'disc_number', album_tracks.disc_number,
160 'track_number', album_tracks.track_number,
161 'images', json_extract(albums.metadata, '$.images')
162 ) FROM album_tracks
163 JOIN albums ON albums.item_id = album_tracks.album_id
164 WHERE album_tracks.track_id = tracks.item_id
165 ORDER BY (album_tracks.album_id IS :preferred_album_id) DESC, album_tracks.album_id
166 LIMIT 1) AS track_album
167 FROM tracks
168 """
169 return query, {"preferred_album_id": None}
170
171 async def get(
172 self,
173 item_id: str,
174 provider_instance_id_or_domain: str,
175 allow_update_metadata: bool = True,
176 recursive: bool = True,
177 album_uri: str | None = None,
178 ) -> Track:
179 """Return (full) details for a single media item."""
180 track = await super().get(
181 item_id,
182 provider_instance_id_or_domain,
183 allow_update_metadata=allow_update_metadata,
184 )
185 track.audio_metadata = await self.mass.streams.audio_analysis.get_track_audio_metadata(
186 track
187 )
188 if not recursive and album_uri is None:
189 # return early if we do not want recursive full details and no album uri is provided
190 return track
191
192 # append full album details to full track item (resolve ItemMappings)
193 try:
194 if album_uri:
195 item = await self.mass.music.get_item_by_uri(album_uri, allow_update_metadata=False)
196 if isinstance(item, Album):
197 track.album = item
198 elif provider_instance_id_or_domain == "library":
199 # grab the first album this track is attached to
200 for album_track_row in await self.mass.music.database.get_rows(
201 DB_TABLE_ALBUM_TRACKS, {"track_id": int(item_id)}, limit=1
202 ):
203 track.album = await self.mass.music.albums.get_library_item(
204 album_track_row["album_id"]
205 )
206 elif isinstance(track.album, ItemMapping) or (track.album and not track.album.image):
207 track.album = await self.mass.music.albums.get(
208 track.album.item_id,
209 track.album.provider,
210 allow_update_metadata=False,
211 recursive=False,
212 )
213 except MusicAssistantError as err:
214 # edge case where playlist track has invalid albumdetails
215 self.logger.warning("Unable to fetch album details for %s - %s", track.uri, str(err))
216
217 if not recursive:
218 return track
219
220 # append artist details to full track item (resolve ItemMappings)
221 track_artists = []
222 for artist in track.artists:
223 if not isinstance(artist, ItemMapping):
224 track_artists.append(artist)
225 continue
226 try:
227 track_artists.append(
228 await self.mass.music.artists.get(
229 artist.item_id,
230 artist.provider,
231 allow_update_metadata=False,
232 )
233 )
234 except MusicAssistantError as err:
235 # edge case where playlist track has invalid artistdetails
236 self.logger.warning("Unable to fetch artist details %s - %s", artist.uri, str(err))
237 track.artists = UniqueList(track_artists)
238 return track
239
240 async def library_items( # noqa: PLR0913
241 self,
242 favorite: bool | None = None,
243 search: str | None = None,
244 limit: int = 500,
245 offset: int = 0,
246 order_by: str = "sort_name",
247 provider: str | list[str] | None = None,
248 genre: int | list[int] | None = None,
249 played_only: bool = False,
250 explicit: bool | None = None,
251 *,
252 summary: bool = True,
253 reachable_via: list[str] | None = None,
254 **kwargs: Any,
255 ) -> list[Track]:
256 """
257 Get in-database tracks.
258
259 :param favorite: Filter by favorite status.
260 :param search: Filter by search query.
261 :param limit: Maximum number of items to return.
262 :param offset: Number of items to skip.
263 :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
264 :param provider: Filter by provider instance ID (single string or list).
265 :param genre: Filter by genre id(s).
266 :param played_only: Filter to only played tracks.
267 :param explicit: Filter by explicit content (True=only explicit, False=no explicit, None=all).
268 :param summary: When True (default), return slim summary items containing only the
269 fields needed for a list view. Set to False to get fully hydrated items.
270 :param reachable_via: Restrict results to items with a provider mapping reachable
271 through one of these provider instance ids (OR semantics). See
272 `MediaControllerBase.library_items` for the full semantics.
273 """
274 reachable_via = self._resolve_reachable_via(reachable_via)
275 if reachable_via is not None and not reachable_via:
276 return []
277 extra_query_params: dict[str, Any] = {}
278 extra_query_parts: list[str] = []
279 extra_join_parts: list[str] = []
280
281 # Apply explicit content filter
282 if explicit is not None:
283 if explicit:
284 # Only explicit tracks
285 extra_query_parts.append("json_extract(tracks.metadata, '$.explicit') = 1")
286 else:
287 # No explicit tracks (null or false)
288 extra_query_parts.append(
289 "(json_extract(tracks.metadata, '$.explicit') IS NULL "
290 "OR json_extract(tracks.metadata, '$.explicit') = 0)"
291 )
292
293 if (order_by and "track_artist_name" in order_by) or (search and " - " in search):
294 extra_join_parts.append(
295 "JOIN track_artists ON track_artists.track_id = tracks.item_id "
296 "JOIN artists ON artists.item_id = track_artists.artist_id "
297 )
298
299 if search and " - " in search:
300 # handle combined artist + title search
301 artist_str, title_str = search.split(" - ", 1)
302 search = None
303 title_str = create_safe_string(title_str, True, True)
304 artist_str = create_safe_string(artist_str, True, True)
305 extra_query_parts.append(
306 search_name_match_clause("tracks", title_str, "search_title", extra_query_params)
307 )
308 extra_query_parts.append(
309 search_name_match_clause("artists", artist_str, "search_artist", extra_query_params)
310 )
311 result = await self.get_library_items_by_query(
312 favorite=favorite,
313 search=search,
314 genre_ids=genre,
315 limit=limit,
316 offset=offset,
317 order_by=order_by,
318 provider_filter=self._provider_filter_considering_reachability(provider, reachable_via),
319 extra_query_parts=extra_query_parts,
320 extra_query_params=extra_query_params,
321 extra_join_parts=extra_join_parts,
322 played_only=played_only,
323 in_library_only=True,
324 summary=summary,
325 reachable_via=reachable_via,
326 )
327 if search and len(result) < 25 and not offset:
328 # append artist items to result
329 artist_search_str = create_safe_string(search, True, True)
330 if order_by and "track_artist_name" in order_by:
331 # JOIN already exists for sorting, only add WHERE clause
332 extra_query_parts.append(
333 search_name_match_clause(
334 "artists", artist_search_str, "search_artist", extra_query_params
335 )
336 )
337 else:
338 # JOIN not yet added, add it with the search condition
339 extra_join_parts.append(
340 "JOIN track_artists ON track_artists.track_id = tracks.item_id "
341 "JOIN artists ON artists.item_id = track_artists.artist_id "
342 "AND "
343 + search_name_match_clause(
344 "artists", artist_search_str, "search_artist", extra_query_params
345 )
346 )
347 existing_uris = {item.uri for item in result}
348 for _track in await self.get_library_items_by_query(
349 favorite=favorite,
350 search=None,
351 genre_ids=genre,
352 limit=limit,
353 order_by=order_by,
354 provider_filter=self._provider_filter_considering_reachability(
355 provider, reachable_via
356 ),
357 extra_query_parts=extra_query_parts,
358 extra_query_params=extra_query_params,
359 extra_join_parts=extra_join_parts,
360 in_library_only=True,
361 summary=summary,
362 reachable_via=reachable_via,
363 ):
364 # prevent duplicates (when artist is also in the title)
365 if _track.uri not in existing_uris:
366 result.append(_track)
367 return result
368
369 async def versions(
370 self,
371 item_id: str,
372 provider_instance_id_or_domain: str,
373 ) -> UniqueList[Track]:
374 """Return all versions of a track we can find on all providers."""
375 track = await self.get(item_id, provider_instance_id_or_domain)
376 search_query = f"{track.artist_str} - {track.name}"
377 result: UniqueList[Track] = UniqueList()
378 for provider_id in self.mass.music.get_unique_providers():
379 provider = self.mass.get_provider(provider_id)
380 if not isinstance(provider, MusicProvider):
381 continue
382 if MediaType.TRACK not in provider.supported_media_types:
383 continue
384 result.extend(
385 prov_item
386 for prov_item in await self.search(search_query, provider_id)
387 if loose_compare_strings(track.name, prov_item.name)
388 and compare_artists(prov_item.artists, track.artists, any_match=True)
389 # make sure that the 'base' version is NOT included
390 and not track.provider_mappings.intersection(prov_item.provider_mappings)
391 )
392 return result
393
394 async def albums(
395 self,
396 item_id: str,
397 provider_instance_id_or_domain: str,
398 in_library_only: bool = False,
399 ) -> UniqueList[Album]:
400 """Return all albums the track appears on."""
401 full_track = await self.get(item_id, provider_instance_id_or_domain)
402 db_items = (
403 await self.get_library_track_albums(full_track.item_id)
404 if full_track.provider == "library"
405 else []
406 )
407 # return all (unique) items from all providers
408 result: UniqueList[Album] = UniqueList(db_items)
409 # use search to get all items on the provider
410 search_query = f"{full_track.artist_str} - {full_track.name}"
411 # TODO: we could use musicbrainz info here to get a list of all releases known
412 unique_ids: set[str] = set()
413 # explicitly search all providers as we want all album versions
414 # of this track, including those already mapped in the library
415 search_providers = ["library", *self.mass.music.get_unique_providers()]
416 search_results = await self.mass.music.search(
417 search_query, [MediaType.TRACK], providers=search_providers
418 )
419 for prov_item in search_results.tracks:
420 if not isinstance(prov_item, Track): # for type checking
421 continue
422 if not loose_compare_strings(full_track.name, prov_item.name):
423 continue
424 if not prov_item.album:
425 continue
426 if not compare_artists(full_track.artists, prov_item.artists, any_match=True):
427 continue
428 unique_id = f"{prov_item.album.name}.{prov_item.album.version}"
429 if unique_id in unique_ids:
430 continue
431 unique_ids.add(unique_id)
432 # prefer db item
433 if db_item := await self.mass.music.albums.get_library_item_by_prov_id(
434 prov_item.album.item_id, prov_item.album.provider
435 ):
436 result.append(db_item)
437 elif not in_library_only and isinstance(prov_item.album, Album):
438 result.append(prov_item.album)
439 return result
440
441 async def similar_tracks(
442 self,
443 item_id: str,
444 provider_instance_id_or_domain: str,
445 limit: int = 25,
446 allow_lookup: bool = False,
447 preferred_provider_instances: list[str] | None = None,
448 ) -> list[Track]:
449 """
450 Get a list of similar tracks for the given track.
451
452 :param item_id: The item ID of the track.
453 :param provider_instance_id_or_domain: The provider instance ID or domain.
454 :param limit: Maximum number of similar tracks to return.
455 :param allow_lookup: Allow lookup on other providers if not found.
456 :param preferred_provider_instances: List of preferred provider instance IDs to use.
457 When provided, these providers will be tried first before falling back to others.
458 :raises MusicAssistantError: When no provider can complete the request.
459 """
460 ref_item = await self.get(item_id, provider_instance_id_or_domain)
461
462 # Sort provider mappings to prefer user's provider instances
463 def sort_key(mapping: ProviderMapping) -> tuple[int, int]:
464 # Primary sort: preferred providers first (0), then others (1)
465 preferred = (
466 0
467 if preferred_provider_instances
468 and mapping.provider_instance in preferred_provider_instances
469 else 1
470 )
471 # Secondary sort: by quality (higher is better, so negate)
472 quality = -(mapping.quality or 0)
473 return (preferred, quality)
474
475 sorted_mappings = sorted(ref_item.provider_mappings, key=sort_key)
476 last_provider_error: MusicAssistantError | ClientError | OSError | TimeoutError | None = (
477 None
478 )
479 provider_responded = False
480
481 # Try preferred providers first, then fall back to others
482 for prov_mapping in sorted_mappings:
483 prov = self.mass.get_provider(prov_mapping.provider_instance)
484 if (
485 not isinstance(prov, MusicProvider)
486 or ProviderFeature.SIMILAR_TRACKS not in prov.supported_features
487 ):
488 continue
489 result, error = await self._get_similar_tracks_from_provider(
490 prov, ref_item, limit, provider_track_id=prov_mapping.item_id
491 )
492 if error is not None:
493 last_provider_error = error
494 continue
495 if result is None:
496 continue
497 provider_responded = True
498 if result:
499 return result
500
501 # Fallback: consult metadata/plugin providers that claim SIMILAR_TRACKS
502 for prov in self.mass.get_providers_supporting_feature(
503 ProviderFeature.SIMILAR_TRACKS,
504 priority=(ProviderType.METADATA, ProviderType.PLUGIN),
505 ):
506 cross_prov = cast("MetadataProvider | PluginProvider", prov)
507 result, error = await self._get_similar_tracks_from_provider(
508 cross_prov, ref_item, limit
509 )
510 if error is not None:
511 last_provider_error = error
512 continue
513 if result is None:
514 continue
515 provider_responded = True
516 if result:
517 return result
518
519 if not allow_lookup:
520 if not provider_responded and last_provider_error is not None:
521 self._raise_similar_tracks_provider_error(ref_item, last_provider_error)
522 return []
523
524 try:
525 result, error = await self._lookup_similar_tracks_provider(ref_item, limit)
526 except UnsupportedFeaturedException:
527 if provider_responded:
528 return []
529 if last_provider_error is not None:
530 self._raise_similar_tracks_provider_error(ref_item, last_provider_error)
531 raise
532 if error is not None:
533 last_provider_error = error
534 if result is not None:
535 provider_responded = True
536 if result:
537 return result
538
539 if not provider_responded and last_provider_error is not None:
540 self._raise_similar_tracks_provider_error(ref_item, last_provider_error)
541 return []
542
543 async def remove_item_from_library(self, item_id: str | int, recursive: bool = True) -> None:
544 """Delete record from the database."""
545 db_id = int(item_id) # ensure integer
546 # delete entry(s) from albumtracks table
547 await self.mass.music.database.delete(DB_TABLE_ALBUM_TRACKS, {"track_id": db_id})
548 # delete entry(s) from trackartists table
549 await self.mass.music.database.delete(DB_TABLE_TRACK_ARTISTS, {"track_id": db_id})
550 # delete the track itself from db
551 await super().remove_item_from_library(db_id)
552
553 async def set_identifiers(
554 self,
555 item_id: str,
556 provider_instance_id_or_domain: str,
557 mbid: str | None = None,
558 acoustid: str | None = None,
559 isrcs: list[str] | None = None,
560 ) -> None:
561 """
562 Persist MBID / AcoustID / ISRCs onto the library track row.
563
564 :param item_id: Provider-native track ID.
565 :param provider_instance_id_or_domain: Music provider instance ID or domain.
566 :param mbid: MusicBrainz recording ID.
567 :param acoustid: AcoustID UUID.
568 :param isrcs: ISRC codes.
569 """
570 # MBID is filled only when empty; AcoustID/ISRCs are appended via
571 # external_ids without clobbering tag-sourced values.
572 if not mbid and not acoustid and not isrcs:
573 return
574 try:
575 track = await self.get_library_item_by_prov_id(item_id, provider_instance_id_or_domain)
576 except MusicAssistantError as err:
577 self.logger.debug(
578 "set_identifiers: failed to load library track %s/%s: %s",
579 provider_instance_id_or_domain,
580 item_id,
581 err,
582 )
583 return
584 if track is None:
585 return
586
587 changed = False
588 if mbid and not track.mbid:
589 track.mbid = mbid
590 changed = True
591 if acoustid and not any(
592 ext_id[0] == ExternalID.ACOUSTID and ext_id[1] == acoustid
593 for ext_id in track.external_ids
594 ):
595 track.add_external_id(ExternalID.ACOUSTID, acoustid)
596 changed = True
597 for isrc in isrcs or ():
598 if isrc:
599 track.add_external_id(ExternalID.ISRC, isrc)
600 changed = True
601 if not changed:
602 return
603
604 await self.update_item_in_library(int(track.item_id), track)
605
606 async def get_preview_url(self, provider_instance_id_or_domain: str, item_id: str) -> str:
607 """Return url to short preview sample."""
608 track = await self.get_provider_item(item_id, provider_instance_id_or_domain)
609 # prefer provider-provided preview
610 if preview := track.metadata.preview:
611 return preview
612 # fallback to a preview/sample hosted by our own webserver
613 return self.mass.webserver.create_preview_url(provider_instance_id_or_domain, item_id)
614
615 async def get_library_track_albums(
616 self,
617 item_id: str | int,
618 ) -> list[Album]:
619 """Return all in-library albums for a track."""
620 db_id = int(item_id) # ensure integer
621 subquery = (
622 f"SELECT album_id FROM {DB_TABLE_ALBUM_TRACKS} "
623 f"WHERE {DB_TABLE_ALBUM_TRACKS}.track_id = :track_id"
624 )
625 query = f"{DB_TABLE_ALBUMS}.item_id in ({subquery})"
626 return await self.mass.music.albums.get_library_items_by_query(
627 extra_query_parts=[query],
628 extra_query_params={"track_id": db_id},
629 in_library_only=True,
630 )
631
632 async def match_provider(
633 self,
634 base_track: Track,
635 provider: MusicProvider,
636 strict: bool = True,
637 ref_albums: list[Album] | None = None,
638 ) -> list[ProviderMapping]:
639 """
640 Try to find match on (streaming) provider for the provided track.
641
642 This is used to link objects of different providers/qualities together.
643 """
644 if ref_albums is None:
645 ref_albums = await self.albums(base_track.item_id, base_track.provider)
646 self.logger.debug("Trying to match track %s on provider %s", base_track.name, provider.name)
647 matches: list[ProviderMapping] = []
648 for artist in base_track.artists:
649 if matches:
650 break
651 search_str = f"{artist.name} - {base_track.name}"
652 search_result = await self.search(search_str, provider.domain)
653 for search_result_item in search_result:
654 if not search_result_item.available:
655 continue
656 # do a basic compare first
657 if not compare_media_item(base_track, search_result_item, strict=False):
658 continue
659 # we must fetch the full version, search results can be simplified objects
660 prov_track = await self.get_provider_item(
661 search_result_item.item_id,
662 search_result_item.provider,
663 fallback=search_result_item,
664 )
665 if compare_track(base_track, prov_track, strict=strict, track_albums=ref_albums):
666 matches.extend(prov_track.provider_mappings)
667
668 if not matches:
669 self.logger.debug(
670 "Could not find match for Track %s on provider %s",
671 base_track.name,
672 provider.name,
673 )
674 return matches
675
676 async def match_providers(self, db_track: Track) -> None:
677 """
678 Try to find matching track on all providers for the provided (database) track_id.
679
680 This is used to link objects of different providers/qualities together.
681 """
682 if db_track.provider != "library":
683 return # Matching only supported for database items
684
685 track_albums = await self.albums(db_track.item_id, db_track.provider)
686 # try to find match on all providers
687 processed_domains = set()
688 for provider in self.mass.music.providers:
689 if provider.domain in processed_domains:
690 continue
691 if ProviderFeature.SEARCH not in provider.supported_features:
692 continue
693 if MediaType.TRACK not in provider.supported_media_types:
694 continue
695 if not provider.is_streaming_provider:
696 # matching on unique providers is pointless as they push (all) their content to MA
697 continue
698 if match := await self.match_provider(
699 db_track, provider, strict=True, ref_albums=track_albums
700 ):
701 # 100% match, we update the db with the additional provider mapping(s)
702 await self.add_provider_mappings(db_track.item_id, match)
703 processed_domains.add(provider.domain)
704
705 async def _add_library_item(self, item: Track, overwrite_existing: bool = False) -> int:
706 """Add a new item record to the database."""
707 if not isinstance(item, Track): # TODO: Remove this once the codebase is fully typed
708 msg = "Not a valid Track object (ItemMapping can not be added to db)" # type: ignore[unreachable]
709 raise InvalidDataError(msg)
710 if not item.artists:
711 msg = "Track is missing artist(s)"
712 raise InvalidDataError(msg)
713 # normalize synced lyrics so clients only need a minimal single-timestamp LRC parser
714 # promoting LRC formatted text stored in the plain lyrics tag
715 item.metadata.lrc_lyrics = normalize_lrc_lyrics(
716 item.metadata.lrc_lyrics or extract_lrc_lyrics(item.metadata.lyrics)
717 )
718 db_id = await self.mass.music.database.insert(
719 self.db_table,
720 {
721 "name": item.name,
722 "sort_name": item.sort_name,
723 "version": item.version,
724 "duration": item.duration,
725 "favorite": item.favorite,
726 "metadata": serialize_to_json(item.metadata),
727 "search_name": create_safe_string(item.name, True, True),
728 "search_sort_name": create_safe_string(item.sort_name or "", True, True),
729 "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
730 },
731 )
732 # update/set external id lookup table
733 await self.set_external_ids(db_id, item.external_ids)
734 # update/set provider_mappings table
735 await self.set_provider_mappings(db_id, item.provider_mappings)
736 # set track artist(s)
737 await self._set_track_artists(db_id, item.artists)
738 # handle track album
739 if item.album:
740 await self._set_track_album(
741 db_id=db_id,
742 album=item.album,
743 disc_number=getattr(item, "disc_number", 0),
744 track_number=getattr(item, "track_number", 0),
745 )
746 self.logger.debug("added %s to database (id: %s)", item.name, db_id)
747 return db_id
748
749 async def _update_library_item(
750 self,
751 item_id: str | int,
752 update: Track,
753 overwrite: bool = False,
754 *,
755 set_album: bool = True,
756 ) -> None:
757 """Update Track record in the database, merging data."""
758 db_id = int(item_id) # ensure integer
759 cur_item = await self.get_library_item(db_id)
760 metadata = update.metadata if overwrite else cur_item.metadata.update(update.metadata)
761 metadata.lrc_lyrics = normalize_lrc_lyrics(
762 metadata.lrc_lyrics or extract_lrc_lyrics(metadata.lyrics)
763 )
764 cur_item.external_ids.update(update.external_ids)
765 name = update.name if overwrite else cur_item.name
766 sort_name = update.sort_name if overwrite else cur_item.sort_name or update.sort_name
767 await self.mass.music.database.update(
768 self.db_table,
769 {"item_id": db_id},
770 {
771 "name": name,
772 "sort_name": sort_name,
773 "version": update.version if overwrite else cur_item.version or update.version,
774 "duration": update.duration if overwrite else cur_item.duration or update.duration,
775 "metadata": serialize_to_json(metadata),
776 "search_name": create_safe_string(name, True, True),
777 "search_sort_name": create_safe_string(sort_name or "", True, True),
778 "timestamp_added": int(update.date_added.timestamp())
779 if update.date_added
780 else UNSET,
781 },
782 )
783 # update/set external id lookup table
784 await self.set_external_ids(
785 db_id, update.external_ids if overwrite else cur_item.external_ids
786 )
787 # update/set provider_mappings table
788 provider_mappings = provider_mappings_for_update(
789 cur_item.provider_mappings, update.provider_mappings, overwrite
790 )
791 await self.set_provider_mappings(db_id, provider_mappings, overwrite)
792 # set track artist(s)
793 artists = update.artists if overwrite else cur_item.artists + update.artists
794 await self._set_track_artists(db_id, artists, overwrite=overwrite)
795 # update/set track album
796 if update.album and set_album:
797 await self._set_track_album(
798 db_id=db_id,
799 album=update.album,
800 disc_number=update.disc_number or cur_item.disc_number,
801 track_number=update.track_number or cur_item.track_number,
802 overwrite=overwrite,
803 )
804 self.logger.debug("updated %s in database: (id %s)", update.name, db_id)
805
806 async def _update_library_item_for_merge(self, item_id: int, update: Track) -> None:
807 """Merge track model state without replacing existing album relations."""
808 await self._update_library_item(item_id, update, set_album=False)
809
810 async def _set_track_album(
811 self,
812 db_id: int,
813 album: Album | ItemMapping,
814 disc_number: int,
815 track_number: int,
816 overwrite: bool = False,
817 ) -> None:
818 """
819 Store Track Album info.
820
821 A track can exist on multiple albums so we have a mapping table between
822 albums and tracks which stores the relation between the two and it also
823 stores the track and disc number of the track within an album.
824 For digital releases, the discnumber will be just 0 or 1.
825 Track number should start counting at 1.
826 """
827 db_album: Album | ItemMapping | None = None
828 if album.provider == "library":
829 db_album = album
830 elif existing := await self.mass.music.albums.get_library_item_by_prov_id(
831 album.item_id, album.provider
832 ):
833 db_album = existing
834
835 if not db_album or overwrite:
836 # ensure we have an actual album object
837 if isinstance(album, ItemMapping):
838 db_album = await self.mass.music.albums.add_item_mapping_as_album_to_library(album)
839 else:
840 db_album = await self.mass.music.albums.add_item_to_library(
841 album,
842 overwrite_existing=overwrite,
843 )
844 # write (or update) record in album_tracks table
845 await self.mass.music.database.insert_or_replace(
846 DB_TABLE_ALBUM_TRACKS,
847 {
848 "track_id": db_id,
849 "album_id": int(db_album.item_id),
850 "disc_number": disc_number,
851 "track_number": track_number,
852 },
853 )
854
855 async def _set_track_artists(
856 self,
857 db_id: int,
858 artists: Iterable[Artist | ItemMapping],
859 overwrite: bool = False,
860 ) -> None:
861 """
862 Store Track Artists.
863
864 An empty set of artists never clears the stored rows: a track without any
865 artist can not be played or resolved.
866 """
867 all_artists = list(artists)
868 if not all_artists:
869 if overwrite:
870 # a caller asking to replace all artists with none is a bug,
871 # so keep the stored rows and make the attempt visible
872 self.logger.warning("Ignoring request to clear all artists of track id %s", db_id)
873 return
874 if overwrite:
875 # on overwrite, clear the track_artists table first
876 await self.mass.music.database.delete(
877 DB_TABLE_TRACK_ARTISTS,
878 {
879 "track_id": db_id,
880 },
881 )
882 for artist in all_artists:
883 await self._set_track_artist(db_id, artist=artist, overwrite=overwrite)
884
885 async def _set_track_artist(
886 self, db_id: int, artist: Artist | ItemMapping, overwrite: bool = False
887 ) -> ItemMapping:
888 """Store Track Artist info."""
889 db_artist: Artist | ItemMapping | None = None
890 if artist.provider == "library":
891 db_artist = artist
892 elif existing := await self.mass.music.artists.get_library_item_by_prov_id(
893 artist.item_id, artist.provider
894 ):
895 db_artist = existing
896
897 if not db_artist or overwrite:
898 # Convert ItemMapping to Artist if needed
899 artist_to_add = (
900 self.mass.music.artists.artist_from_item_mapping(artist)
901 if isinstance(artist, ItemMapping)
902 else artist
903 )
904 db_artist = await self.mass.music.artists.add_item_to_library(
905 artist_to_add, overwrite_existing=overwrite
906 )
907 # write (or update) record in track_artists table
908 await self.mass.music.database.insert_or_replace(
909 DB_TABLE_TRACK_ARTISTS,
910 {
911 "track_id": db_id,
912 "artist_id": int(db_artist.item_id),
913 },
914 )
915 return ItemMapping.from_item(db_artist)
916
917 def _sync_details_query_parts(self) -> tuple[str, str, dict[str, Any]]:
918 """Return extra (columns, joins, params) for the tracks sync-details query."""
919 # the sync loop needs to know if the track has (valid) album and artist links
920 # to be able to backfill missing ones on existing library tracks
921 extra_columns = """
922 , EXISTS (
923 SELECT 1 FROM album_tracks
924 JOIN albums ON albums.item_id = album_tracks.album_id
925 WHERE album_tracks.track_id = tracks.item_id
926 ) AS has_album
927 , EXISTS (
928 SELECT 1 FROM track_artists
929 JOIN artists ON artists.item_id = track_artists.artist_id
930 WHERE track_artists.track_id = tracks.item_id
931 ) AS has_artists
932 """
933 return extra_columns, "", {}
934
935 def _parse_sync_details_row(self, db_row: Mapping[str, Any]) -> TrackSyncDetails:
936 """Parse a raw sync-details db row into a TrackSyncDetails object."""
937 return TrackSyncDetails(
938 item_id=db_row["item_id"],
939 favorite=bool(db_row["favorite"]),
940 date_added=datetime.fromtimestamp(db_row["timestamp_added"], tz=UTC),
941 provider_mappings=self._parse_sync_details_mappings(db_row),
942 has_album=bool(db_row["has_album"]),
943 has_artists=bool(db_row["has_artists"]),
944 )
945
946 def _parse_summary_row(self, db_row: Mapping[str, Any]) -> TrackSummary:
947 """Parse a raw summary db row into a TrackSummary object."""
948 item = cast("TrackSummary", super()._parse_summary_row(db_row))
949 item.version = db_row["version"] or ""
950 item.duration = db_row["duration"] or 0
951 item.metadata.explicit = None if db_row["explicit"] is None else bool(db_row["explicit"])
952 if raw_release_date := db_row["release_date"]:
953 item.metadata.release_date = datetime.fromisoformat(raw_release_date)
954 item.artists = self._parse_summary_artist_mappings(db_row)
955 if raw_album := db_row["track_album"]:
956 album: dict[str, Any] = json_loads(raw_album)
957 album_thumb: MediaItemImage | None = None
958 if album_images := album.get("images"):
959 for image in album_images:
960 if image["type"] != ImageType.THUMB.value:
961 continue
962 album_thumb = MediaItemImage(
963 type=ImageType.THUMB,
964 path=image["path"],
965 provider=image["provider"],
966 remotely_accessible=image.get("remotely_accessible", False),
967 )
968 break
969 item.album = ItemMappingSummary(
970 media_type=MediaType.ALBUM,
971 item_id=str(album["item_id"]),
972 provider="library",
973 name=album["name"],
974 sort_name=album["sort_name"],
975 year=album["year"],
976 image=album_thumb,
977 )
978 item.disc_number = album["disc_number"] or 0
979 item.track_number = album["track_number"] or 0
980 if album_thumb:
981 # always prefer album image over track image
982 item.metadata.images = UniqueList([album_thumb])
983 return item
984
985 async def _get_similar_tracks_from_provider(
986 self,
987 provider: MusicProvider | MetadataProvider | PluginProvider,
988 ref_item: Track,
989 limit: int,
990 provider_track_id: str | None = None,
991 ) -> tuple[
992 list[Track] | None,
993 MusicAssistantError | ClientError | OSError | TimeoutError | None,
994 ]:
995 """
996 Request similar tracks from a provider.
997
998 :param provider: Provider to request similar tracks from.
999 :param ref_item: Full track supplied to metadata and plugin providers.
1000 :param limit: Maximum number of tracks to return.
1001 :param provider_track_id: Provider track ID supplied to music providers.
1002 """
1003 if isinstance(provider, MusicProvider):
1004 if provider_track_id is None:
1005 raise InvalidDataError("Music provider track ID is required")
1006 request = provider.get_similar_tracks(provider_track_id, limit=limit)
1007 else:
1008 request = provider.get_similar_tracks(ref_item, limit=limit)
1009 try:
1010 result = await request
1011 except NotImplementedError:
1012 return None, None
1013 except (MusicAssistantError, ClientError, OSError, TimeoutError) as err:
1014 self.logger.warning(
1015 "Failed to fetch similar tracks for %s from provider %s: %s",
1016 ref_item.name,
1017 provider.name,
1018 err,
1019 )
1020 return None, err
1021 return result, None
1022
1023 async def _match_similar_tracks_provider(
1024 self, ref_item: Track, provider: MusicProvider
1025 ) -> tuple[
1026 list[ProviderMapping] | None,
1027 MusicAssistantError | ClientError | OSError | TimeoutError | None,
1028 ]:
1029 """
1030 Find a matching track on a provider for a similar-tracks lookup.
1031
1032 :param ref_item: Track to match.
1033 :param provider: Provider to search for a matching track.
1034 """
1035 try:
1036 return await self.match_provider(ref_item, provider), None
1037 except (MusicAssistantError, ClientError, OSError, TimeoutError) as err:
1038 self.logger.warning(
1039 "Failed to match %s on provider %s for similar tracks: %s",
1040 ref_item.name,
1041 provider.name,
1042 err,
1043 )
1044 return None, err
1045
1046 async def _lookup_similar_tracks_provider(
1047 self, ref_item: Track, limit: int
1048 ) -> tuple[
1049 list[Track] | None,
1050 MusicAssistantError | ClientError | OSError | TimeoutError | None,
1051 ]:
1052 """
1053 Find a provider match and request similar tracks from it.
1054
1055 :param ref_item: Track to match.
1056 :param limit: Maximum number of tracks to return.
1057 :raises UnsupportedFeaturedException: When no music provider supports similar tracks.
1058 """
1059 supported_providers = [
1060 prov
1061 for prov in self.mass.music.providers
1062 if ProviderFeature.SIMILAR_TRACKS in prov.supported_features
1063 ]
1064 if not supported_providers:
1065 msg = "No Music Provider found that supports requesting similar tracks."
1066 raise UnsupportedFeaturedException(msg)
1067 mapped_instances = {mapping.provider_instance for mapping in ref_item.provider_mappings}
1068 providers = [
1069 prov for prov in supported_providers if prov.instance_id not in mapped_instances
1070 ]
1071
1072 last_error: MusicAssistantError | ClientError | OSError | TimeoutError | None = None
1073 provider_responded = False
1074 for provider in providers:
1075 mappings, error = await self._match_similar_tracks_provider(ref_item, provider)
1076 if error is not None:
1077 last_error = error
1078 continue
1079 if not mappings:
1080 continue
1081 if ref_item.provider == "library":
1082 await self.add_provider_mappings(ref_item.item_id, mappings)
1083 ref_item.provider_mappings.update(mappings)
1084 result, error = await self._get_similar_tracks_from_provider(
1085 provider, ref_item, limit, provider_track_id=mappings[0].item_id
1086 )
1087 if error is not None:
1088 last_error = error
1089 continue
1090 if result is None:
1091 continue
1092 provider_responded = True
1093 if result:
1094 return result, None
1095 return ([] if provider_responded else None), last_error
1096
1097 @staticmethod
1098 def _raise_similar_tracks_provider_error(
1099 ref_item: Track,
1100 err: MusicAssistantError | ClientError | OSError | TimeoutError,
1101 ) -> Never:
1102 """
1103 Raise a provider error from a similar-tracks lookup using MA's typed error hierarchy.
1104
1105 :param ref_item: The track whose similar tracks were requested.
1106 :param err: The provider error to raise or normalize.
1107 """
1108 if isinstance(err, MusicAssistantError):
1109 raise err
1110 raise ProviderUnavailableError(
1111 f"Failed to fetch similar tracks for {ref_item.name}"
1112 ) from err
1113