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