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