/
/
/
1"""Manage MediaItems of type Track."""
2
3from __future__ import annotations
4
5import urllib.parse
6from collections.abc import Iterable
7from typing import TYPE_CHECKING, Any
8
9from music_assistant_models.enums import MediaType, ProviderFeature
10from music_assistant_models.errors import (
11 InvalidDataError,
12 MusicAssistantError,
13 UnsupportedFeaturedException,
14)
15from music_assistant_models.media_items import (
16 Album,
17 Artist,
18 ItemMapping,
19 ProviderMapping,
20 Track,
21 UniqueList,
22)
23
24from music_assistant.constants import (
25 DB_TABLE_ALBUM_TRACKS,
26 DB_TABLE_ALBUMS,
27 DB_TABLE_TRACK_ARTISTS,
28 DB_TABLE_TRACKS,
29)
30from music_assistant.helpers.compare import (
31 compare_artists,
32 compare_media_item,
33 compare_track,
34 create_safe_string,
35 loose_compare_strings,
36)
37from music_assistant.helpers.database import UNSET
38from music_assistant.helpers.json import serialize_to_json
39from music_assistant.models.music_provider import MusicProvider
40
41from .base import MediaControllerBase
42
43if TYPE_CHECKING:
44 from music_assistant import MusicAssistant
45
46
47class TracksController(MediaControllerBase[Track]):
48 """Controller managing MediaItems of type Track."""
49
50 db_table = DB_TABLE_TRACKS
51 media_type = MediaType.TRACK
52 item_cls = Track
53
54 def __init__(self, mass: MusicAssistant) -> None:
55 """Initialize class."""
56 super().__init__(mass)
57 self.base_query = """
58 SELECT
59 tracks.*,
60 (SELECT JSON_GROUP_ARRAY(
61 json_object(
62 'item_id', track_pm.provider_item_id,
63 'provider_domain', track_pm.provider_domain,
64 'provider_instance', track_pm.provider_instance,
65 'available', track_pm.available,
66 'audio_format', json(track_pm.audio_format),
67 'url', track_pm.url,
68 'details', track_pm.details,
69 'in_library', track_pm.in_library,
70 'is_unique', track_pm.is_unique
71 )) FROM provider_mappings track_pm WHERE track_pm.item_id = tracks.item_id AND track_pm.media_type = 'track') AS provider_mappings,
72
73 (SELECT JSON_GROUP_ARRAY(
74 json_object(
75 'item_id', artists.item_id,
76 'provider', 'library',
77 'name', artists.name,
78 'sort_name', artists.sort_name,
79 'media_type', 'artist'
80 )) FROM artists JOIN track_artists on track_artists.track_id = tracks.item_id WHERE artists.item_id = track_artists.artist_id) AS artists,
81 (SELECT
82 json_object(
83 'item_id', albums.item_id,
84 'provider', 'library',
85 'name', albums.name,
86 'sort_name', albums.sort_name,
87 'media_type', 'album',
88 'year', albums.year,
89 'disc_number', album_tracks.disc_number,
90 'track_number', album_tracks.track_number,
91 'images', json_extract(albums.metadata, '$.images')
92 ) FROM albums WHERE albums.item_id = album_tracks.album_id) AS track_album
93 FROM tracks
94 LEFT JOIN album_tracks on album_tracks.track_id = tracks.item_id
95 """ # noqa: E501
96 # register (extra) api handlers
97 api_base = self.api_base
98 self.mass.register_api_command(f"music/{api_base}/track_versions", self.versions)
99 self.mass.register_api_command(f"music/{api_base}/track_albums", self.albums)
100 self.mass.register_api_command(f"music/{api_base}/preview", self.get_preview_url)
101 self.mass.register_api_command(f"music/{api_base}/similar_tracks", self.similar_tracks)
102
103 async def get(
104 self,
105 item_id: str,
106 provider_instance_id_or_domain: str,
107 recursive: bool = True,
108 album_uri: str | None = None,
109 ) -> Track:
110 """Return (full) details for a single media item."""
111 track = await super().get(
112 item_id,
113 provider_instance_id_or_domain,
114 )
115 if not recursive and album_uri is None:
116 # return early if we do not want recursive full details and no album uri is provided
117 return track
118
119 # append full album details to full track item (resolve ItemMappings)
120 try:
121 if album_uri:
122 item = await self.mass.music.get_item_by_uri(album_uri)
123 if isinstance(item, Album):
124 track.album = item
125 elif provider_instance_id_or_domain == "library":
126 # grab the first album this track is attached to
127 for album_track_row in await self.mass.music.database.get_rows(
128 DB_TABLE_ALBUM_TRACKS, {"track_id": int(item_id)}, limit=1
129 ):
130 track.album = await self.mass.music.albums.get_library_item(
131 album_track_row["album_id"]
132 )
133 elif isinstance(track.album, ItemMapping) or (track.album and not track.album.image):
134 track.album = await self.mass.music.albums.get(
135 track.album.item_id, track.album.provider, recursive=False
136 )
137 except MusicAssistantError as err:
138 # edge case where playlist track has invalid albumdetails
139 self.logger.warning("Unable to fetch album details for %s - %s", track.uri, str(err))
140
141 if not recursive:
142 return track
143
144 # append artist details to full track item (resolve ItemMappings)
145 track_artists = []
146 for artist in track.artists:
147 if not isinstance(artist, ItemMapping):
148 track_artists.append(artist)
149 continue
150 try:
151 track_artists.append(
152 await self.mass.music.artists.get(
153 artist.item_id,
154 artist.provider,
155 )
156 )
157 except MusicAssistantError as err:
158 # edge case where playlist track has invalid artistdetails
159 self.logger.warning("Unable to fetch artist details %s - %s", artist.uri, str(err))
160 track.artists = UniqueList(track_artists)
161 return track
162
163 async def library_items(
164 self,
165 favorite: bool | None = None,
166 search: str | None = None,
167 limit: int = 500,
168 offset: int = 0,
169 order_by: str = "sort_name",
170 provider: str | list[str] | None = None,
171 **kwargs: Any,
172 ) -> list[Track]:
173 """Get in-database tracks.
174
175 :param favorite: Filter by favorite status.
176 :param search: Filter by search query.
177 :param limit: Maximum number of items to return.
178 :param offset: Number of items to skip.
179 :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
180 :param provider: Filter by provider instance ID (single string or list).
181 """
182 extra_query_params: dict[str, Any] = {}
183 extra_query_parts: list[str] = []
184 extra_join_parts: list[str] = []
185 if search and " - " in search:
186 # handle combined artist + title search
187 artist_str, title_str = search.split(" - ", 1)
188 search = None
189 title_str = create_safe_string(title_str, True, True)
190 artist_str = create_safe_string(artist_str, True, True)
191 extra_query_parts.append("tracks.search_name LIKE :search_title")
192 extra_query_params["search_title"] = f"%{title_str}%"
193 # use join with artists table to filter on artist name
194 extra_join_parts.append(
195 "JOIN track_artists ON track_artists.track_id = tracks.item_id "
196 "JOIN artists ON artists.item_id = track_artists.artist_id "
197 "AND artists.search_name LIKE :search_artist"
198 )
199 extra_query_params["search_artist"] = f"%{artist_str}%"
200 result = await self.get_library_items_by_query(
201 favorite=favorite,
202 search=search,
203 limit=limit,
204 offset=offset,
205 order_by=order_by,
206 provider_filter=self._ensure_provider_filter(provider),
207 extra_query_parts=extra_query_parts,
208 extra_query_params=extra_query_params,
209 extra_join_parts=extra_join_parts,
210 in_library_only=True,
211 )
212 if search and len(result) < 25 and not offset:
213 # append artist items to result
214 artist_search_str = create_safe_string(search, True, True)
215 extra_join_parts.append(
216 "JOIN track_artists ON track_artists.track_id = tracks.item_id "
217 "JOIN artists ON artists.item_id = track_artists.artist_id "
218 "AND artists.search_name LIKE :search_artist"
219 )
220 extra_query_params["search_artist"] = f"%{artist_search_str}%"
221 existing_uris = {item.uri for item in result}
222 for _track in await self.get_library_items_by_query(
223 favorite=favorite,
224 search=None,
225 limit=limit,
226 order_by=order_by,
227 provider_filter=self._ensure_provider_filter(provider),
228 extra_query_parts=extra_query_parts,
229 extra_query_params=extra_query_params,
230 extra_join_parts=extra_join_parts,
231 in_library_only=True,
232 ):
233 # prevent duplicates (when artist is also in the title)
234 if _track.uri not in existing_uris:
235 result.append(_track)
236 return result
237
238 async def versions(
239 self,
240 item_id: str,
241 provider_instance_id_or_domain: str,
242 ) -> UniqueList[Track]:
243 """Return all versions of a track we can find on all providers."""
244 track = await self.get(item_id, provider_instance_id_or_domain)
245 search_query = f"{track.artist_str} - {track.name}"
246 result: UniqueList[Track] = UniqueList()
247 for provider_id in self.mass.music.get_unique_providers():
248 provider = self.mass.get_provider(provider_id)
249 if not isinstance(provider, MusicProvider):
250 continue
251 if not provider.library_supported(MediaType.TRACK):
252 continue
253 result.extend(
254 prov_item
255 for prov_item in await self.search(search_query, provider_id)
256 if loose_compare_strings(track.name, prov_item.name)
257 and compare_artists(prov_item.artists, track.artists, any_match=True)
258 # make sure that the 'base' version is NOT included
259 and not track.provider_mappings.intersection(prov_item.provider_mappings)
260 )
261 return result
262
263 async def albums(
264 self,
265 item_id: str,
266 provider_instance_id_or_domain: str,
267 in_library_only: bool = False,
268 ) -> UniqueList[Album]:
269 """Return all albums the track appears on."""
270 full_track = await self.get(item_id, provider_instance_id_or_domain)
271 db_items = (
272 await self.get_library_track_albums(full_track.item_id)
273 if full_track.provider == "library"
274 else []
275 )
276 # return all (unique) items from all providers
277 result: UniqueList[Album] = UniqueList(db_items)
278 # use search to get all items on the provider
279 search_query = f"{full_track.artist_str} - {full_track.name}"
280 # TODO: we could use musicbrainz info here to get a list of all releases known
281 unique_ids: set[str] = set()
282 for prov_item in (await self.mass.music.search(search_query, [MediaType.TRACK])).tracks:
283 if not isinstance(prov_item, Track): # for type checking
284 continue
285 if not loose_compare_strings(full_track.name, prov_item.name):
286 continue
287 if not prov_item.album:
288 continue
289 if not compare_artists(full_track.artists, prov_item.artists, any_match=True):
290 continue
291 unique_id = f"{prov_item.album.name}.{prov_item.album.version}"
292 if unique_id in unique_ids:
293 continue
294 unique_ids.add(unique_id)
295 # prefer db item
296 if db_item := await self.mass.music.albums.get_library_item_by_prov_id(
297 prov_item.album.item_id, prov_item.album.provider
298 ):
299 result.append(db_item)
300 elif not in_library_only and isinstance(prov_item.album, Album):
301 result.append(prov_item.album)
302 return result
303
304 async def similar_tracks(
305 self,
306 item_id: str,
307 provider_instance_id_or_domain: str,
308 limit: int = 25,
309 allow_lookup: bool = False,
310 preferred_provider_instances: list[str] | None = None,
311 ) -> list[Track]:
312 """
313 Get a list of similar tracks for the given track.
314
315 :param item_id: The item ID of the track.
316 :param provider_instance_id_or_domain: The provider instance ID or domain.
317 :param limit: Maximum number of similar tracks to return.
318 :param allow_lookup: Allow lookup on other providers if not found.
319 :param preferred_provider_instances: List of preferred provider instance IDs to use.
320 When provided, these providers will be tried first before falling back to others.
321 """
322 ref_item = await self.get(item_id, provider_instance_id_or_domain)
323
324 # Sort provider mappings to prefer user's provider instances
325 def sort_key(mapping: ProviderMapping) -> tuple[int, int]:
326 # Primary sort: preferred providers first (0), then others (1)
327 preferred = (
328 0
329 if preferred_provider_instances
330 and mapping.provider_instance in preferred_provider_instances
331 else 1
332 )
333 # Secondary sort: by quality (higher is better, so negate)
334 quality = -(mapping.quality or 0)
335 return (preferred, quality)
336
337 sorted_mappings = sorted(ref_item.provider_mappings, key=sort_key)
338
339 # Try preferred providers first, then fall back to others
340 for allow_other_provider in (False, True):
341 for prov_mapping in sorted_mappings:
342 if (
343 not allow_other_provider
344 and preferred_provider_instances
345 and prov_mapping.provider_instance not in preferred_provider_instances
346 ):
347 continue
348 prov = self.mass.get_provider(prov_mapping.provider_instance)
349 if prov is None:
350 continue
351 if not isinstance(prov, MusicProvider):
352 continue
353 if ProviderFeature.SIMILAR_TRACKS not in prov.supported_features:
354 continue
355 # Grab similar tracks from the music provider
356 return await prov.get_similar_tracks(
357 prov_track_id=prov_mapping.item_id, limit=limit
358 )
359
360 if not allow_lookup:
361 return []
362
363 # check if we have any provider that supports dynamic tracks
364 # TODO: query metadata provider(s) (such as lastfm?)
365 # to get similar tracks (or tracks from similar artists)
366 music_prov: MusicProvider | None = None
367 for prov in self.mass.music.providers:
368 if ProviderFeature.SIMILAR_TRACKS in prov.supported_features:
369 music_prov = prov
370 break
371 if music_prov is None:
372 msg = "No Music Provider found that supports requesting similar tracks."
373 raise UnsupportedFeaturedException(msg)
374
375 if mappings := await self.match_provider(ref_item, music_prov):
376 if ref_item.provider == "library":
377 # update database with new provider mappings
378 await self.add_provider_mappings(ref_item.item_id, mappings)
379 ref_item.provider_mappings.update(mappings)
380 return await music_prov.get_similar_tracks(
381 prov_track_id=mappings[0].item_id, limit=limit
382 )
383
384 return []
385
386 async def remove_item_from_library(self, item_id: str | int, recursive: bool = True) -> None:
387 """Delete record from the database."""
388 db_id = int(item_id) # ensure integer
389 # delete entry(s) from albumtracks table
390 await self.mass.music.database.delete(DB_TABLE_ALBUM_TRACKS, {"track_id": db_id})
391 # delete entry(s) from trackartists table
392 await self.mass.music.database.delete(DB_TABLE_TRACK_ARTISTS, {"track_id": db_id})
393 # delete the track itself from db
394 await super().remove_item_from_library(db_id)
395
396 async def get_preview_url(self, provider_instance_id_or_domain: str, item_id: str) -> str:
397 """Return url to short preview sample."""
398 track = await self.get_provider_item(item_id, provider_instance_id_or_domain)
399 # prefer provider-provided preview
400 if preview := track.metadata.preview:
401 return preview
402 # fallback to a preview/sample hosted by our own webserver
403 enc_track_id = urllib.parse.quote(item_id)
404 return (
405 f"{self.mass.webserver.base_url}/preview?"
406 f"provider={provider_instance_id_or_domain}&item_id={enc_track_id}"
407 )
408
409 async def get_library_track_albums(
410 self,
411 item_id: str | int,
412 ) -> list[Album]:
413 """Return all in-library albums for a track."""
414 db_id = int(item_id) # ensure integer
415 subquery = (
416 f"SELECT album_id FROM {DB_TABLE_ALBUM_TRACKS} "
417 f"WHERE {DB_TABLE_ALBUM_TRACKS}.track_id = :track_id"
418 )
419 query = f"{DB_TABLE_ALBUMS}.item_id in ({subquery})"
420 return await self.mass.music.albums.get_library_items_by_query(
421 extra_query_parts=[query],
422 extra_query_params={"track_id": db_id},
423 in_library_only=True,
424 )
425
426 async def match_provider(
427 self,
428 base_track: Track,
429 provider: MusicProvider,
430 strict: bool = True,
431 ref_albums: list[Album] | None = None,
432 ) -> list[ProviderMapping]:
433 """
434 Try to find match on (streaming) provider for the provided track.
435
436 This is used to link objects of different providers/qualities together.
437 """
438 if ref_albums is None:
439 ref_albums = await self.albums(base_track.item_id, base_track.provider)
440 self.logger.debug("Trying to match track %s on provider %s", base_track.name, provider.name)
441 matches: list[ProviderMapping] = []
442 for artist in base_track.artists:
443 if matches:
444 break
445 search_str = f"{artist.name} - {base_track.name}"
446 search_result = await self.search(search_str, provider.domain)
447 for search_result_item in search_result:
448 if not search_result_item.available:
449 continue
450 # do a basic compare first
451 if not compare_media_item(base_track, search_result_item, strict=False):
452 continue
453 # we must fetch the full version, search results can be simplified objects
454 prov_track = await self.get_provider_item(
455 search_result_item.item_id,
456 search_result_item.provider,
457 fallback=search_result_item,
458 )
459 if compare_track(base_track, prov_track, strict=strict, track_albums=ref_albums):
460 matches.extend(search_result_item.provider_mappings)
461
462 if not matches:
463 self.logger.debug(
464 "Could not find match for Track %s on provider %s",
465 base_track.name,
466 provider.name,
467 )
468 return matches
469
470 async def match_providers(self, db_track: Track) -> None:
471 """
472 Try to find matching track on all providers for the provided (database) track_id.
473
474 This is used to link objects of different providers/qualities together.
475 """
476 if db_track.provider != "library":
477 return # Matching only supported for database items
478
479 track_albums = await self.albums(db_track.item_id, db_track.provider)
480 # try to find match on all providers
481 processed_domains = set()
482 for provider in self.mass.music.providers:
483 if provider.domain in processed_domains:
484 continue
485 if ProviderFeature.SEARCH not in provider.supported_features:
486 continue
487 if not provider.library_supported(MediaType.TRACK):
488 continue
489 if not provider.is_streaming_provider:
490 # matching on unique providers is pointless as they push (all) their content to MA
491 continue
492 if match := await self.match_provider(
493 db_track, provider, strict=True, ref_albums=track_albums
494 ):
495 # 100% match, we update the db with the additional provider mapping(s)
496 await self.add_provider_mappings(db_track.item_id, match)
497 processed_domains.add(provider.domain)
498
499 async def radio_mode_base_tracks(
500 self,
501 item: Track,
502 preferred_provider_instances: list[str] | None = None,
503 ) -> list[Track]:
504 """
505 Get the list of base tracks from the controller used to calculate the dynamic radio.
506
507 :param item: The Track to get base tracks for.
508 :param preferred_provider_instances: List of preferred provider instance IDs to use.
509 """
510 return [item]
511
512 async def _add_library_item(self, item: Track, overwrite_existing: bool = False) -> int:
513 """Add a new item record to the database."""
514 if not isinstance(item, Track): # TODO: Remove this once the codebase is fully typed
515 msg = "Not a valid Track object (ItemMapping can not be added to db)" # type: ignore[unreachable]
516 raise InvalidDataError(msg)
517 if not item.artists:
518 msg = "Track is missing artist(s)"
519 raise InvalidDataError(msg)
520 db_id = await self.mass.music.database.insert(
521 self.db_table,
522 {
523 "name": item.name,
524 "sort_name": item.sort_name,
525 "version": item.version,
526 "duration": item.duration,
527 "favorite": item.favorite,
528 "external_ids": serialize_to_json(item.external_ids),
529 "metadata": serialize_to_json(item.metadata),
530 "search_name": create_safe_string(item.name, True, True),
531 "search_sort_name": create_safe_string(item.sort_name or "", True, True),
532 "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
533 },
534 )
535 # update/set provider_mappings table
536 await self.set_provider_mappings(db_id, item.provider_mappings)
537 # set track artist(s)
538 await self._set_track_artists(db_id, item.artists)
539 # handle track album
540 if item.album:
541 await self._set_track_album(
542 db_id=db_id,
543 album=item.album,
544 disc_number=getattr(item, "disc_number", 0),
545 track_number=getattr(item, "track_number", 0),
546 )
547 self.logger.debug("added %s to database (id: %s)", item.name, db_id)
548 return db_id
549
550 async def _update_library_item(
551 self, item_id: str | int, update: Track, overwrite: bool = False
552 ) -> None:
553 """Update Track record in the database, merging data."""
554 db_id = int(item_id) # ensure integer
555 cur_item = await self.get_library_item(db_id)
556 metadata = update.metadata if overwrite else cur_item.metadata.update(update.metadata)
557 cur_item.external_ids.update(update.external_ids)
558 name = update.name if overwrite else cur_item.name
559 sort_name = update.sort_name if overwrite else cur_item.sort_name or update.sort_name
560 await self.mass.music.database.update(
561 self.db_table,
562 {"item_id": db_id},
563 {
564 "name": name,
565 "sort_name": sort_name,
566 "version": update.version if overwrite else cur_item.version or update.version,
567 "duration": update.duration if overwrite else cur_item.duration or update.duration,
568 "metadata": serialize_to_json(metadata),
569 "external_ids": serialize_to_json(
570 update.external_ids if overwrite else cur_item.external_ids
571 ),
572 "search_name": create_safe_string(name, True, True),
573 "search_sort_name": create_safe_string(sort_name or "", True, True),
574 "timestamp_added": int(update.date_added.timestamp())
575 if update.date_added
576 else UNSET,
577 },
578 )
579 # update/set provider_mappings table
580 provider_mappings = (
581 update.provider_mappings
582 if overwrite
583 else {*update.provider_mappings, *cur_item.provider_mappings}
584 )
585 await self.set_provider_mappings(db_id, provider_mappings, overwrite)
586 # set track artist(s)
587 artists = update.artists if overwrite else cur_item.artists + update.artists
588 await self._set_track_artists(db_id, artists, overwrite=overwrite)
589 # update/set track album
590 if update.album:
591 await self._set_track_album(
592 db_id=db_id,
593 album=update.album,
594 disc_number=update.disc_number or cur_item.disc_number,
595 track_number=update.track_number or cur_item.track_number,
596 overwrite=overwrite,
597 )
598 self.logger.debug("updated %s in database: (id %s)", update.name, db_id)
599
600 async def _set_track_album(
601 self,
602 db_id: int,
603 album: Album | ItemMapping,
604 disc_number: int,
605 track_number: int,
606 overwrite: bool = False,
607 ) -> None:
608 """
609 Store Track Album info.
610
611 A track can exist on multiple albums so we have a mapping table between
612 albums and tracks which stores the relation between the two and it also
613 stores the track and disc number of the track within an album.
614 For digital releases, the discnumber will be just 0 or 1.
615 Track number should start counting at 1.
616 """
617 db_album: Album | ItemMapping | None = None
618 if album.provider == "library":
619 db_album = album
620 elif existing := await self.mass.music.albums.get_library_item_by_prov_id(
621 album.item_id, album.provider
622 ):
623 db_album = existing
624
625 if not db_album or overwrite:
626 # ensure we have an actual album object
627 if isinstance(album, ItemMapping):
628 db_album = await self.mass.music.albums.add_item_mapping_as_album_to_library(album)
629 else:
630 db_album = await self.mass.music.albums.add_item_to_library(
631 album,
632 overwrite_existing=overwrite,
633 )
634 # write (or update) record in album_tracks table
635 await self.mass.music.database.insert_or_replace(
636 DB_TABLE_ALBUM_TRACKS,
637 {
638 "track_id": db_id,
639 "album_id": int(db_album.item_id),
640 "disc_number": disc_number,
641 "track_number": track_number,
642 },
643 )
644
645 async def _set_track_artists(
646 self,
647 db_id: int,
648 artists: Iterable[Artist | ItemMapping],
649 overwrite: bool = False,
650 ) -> None:
651 """Store Track Artists."""
652 if overwrite:
653 # on overwrite, clear the track_artists table first
654 await self.mass.music.database.delete(
655 DB_TABLE_TRACK_ARTISTS,
656 {
657 "track_id": db_id,
658 },
659 )
660 artist_mappings: UniqueList[ItemMapping] = UniqueList()
661 for artist in artists:
662 mapping = await self._set_track_artist(db_id, artist=artist, overwrite=overwrite)
663 artist_mappings.append(mapping)
664
665 async def _set_track_artist(
666 self, db_id: int, artist: Artist | ItemMapping, overwrite: bool = False
667 ) -> ItemMapping:
668 """Store Track Artist info."""
669 db_artist: Artist | ItemMapping | None = None
670 if artist.provider == "library":
671 db_artist = artist
672 elif existing := await self.mass.music.artists.get_library_item_by_prov_id(
673 artist.item_id, artist.provider
674 ):
675 db_artist = existing
676
677 if not db_artist or overwrite:
678 # Convert ItemMapping to Artist if needed
679 artist_to_add = (
680 self.mass.music.artists.artist_from_item_mapping(artist)
681 if isinstance(artist, ItemMapping)
682 else artist
683 )
684 db_artist = await self.mass.music.artists.add_item_to_library(
685 artist_to_add, overwrite_existing=overwrite
686 )
687 # write (or update) record in track_artists table
688 await self.mass.music.database.insert_or_replace(
689 DB_TABLE_TRACK_ARTISTS,
690 {
691 "track_id": db_id,
692 "artist_id": int(db_artist.item_id),
693 },
694 )
695 return ItemMapping.from_item(db_artist)
696