/
/
/
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 genre: int | list[int] | None = None,
172 **kwargs: Any,
173 ) -> list[Track]:
174 """Get in-database tracks.
175
176 :param favorite: Filter by favorite status.
177 :param search: Filter by search query.
178 :param limit: Maximum number of items to return.
179 :param offset: Number of items to skip.
180 :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
181 :param provider: Filter by provider instance ID (single string or list).
182 :param genre: Filter by genre id(s).
183 """
184 extra_query_params: dict[str, Any] = {}
185 extra_query_parts: list[str] = []
186 extra_join_parts: list[str] = []
187 if search and " - " in search:
188 # handle combined artist + title search
189 artist_str, title_str = search.split(" - ", 1)
190 search = None
191 title_str = create_safe_string(title_str, True, True)
192 artist_str = create_safe_string(artist_str, True, True)
193 extra_query_parts.append("tracks.search_name LIKE :search_title")
194 extra_query_params["search_title"] = f"%{title_str}%"
195 # use join with artists table to filter on artist name
196 extra_join_parts.append(
197 "JOIN track_artists ON track_artists.track_id = tracks.item_id "
198 "JOIN artists ON artists.item_id = track_artists.artist_id "
199 "AND artists.search_name LIKE :search_artist"
200 )
201 extra_query_params["search_artist"] = f"%{artist_str}%"
202 result = await self.get_library_items_by_query(
203 favorite=favorite,
204 search=search,
205 genre_ids=genre,
206 limit=limit,
207 offset=offset,
208 order_by=order_by,
209 provider_filter=self._ensure_provider_filter(provider),
210 extra_query_parts=extra_query_parts,
211 extra_query_params=extra_query_params,
212 extra_join_parts=extra_join_parts,
213 in_library_only=True,
214 )
215 if search and len(result) < 25 and not offset:
216 # append artist items to result
217 artist_search_str = create_safe_string(search, True, True)
218 extra_join_parts.append(
219 "JOIN track_artists ON track_artists.track_id = tracks.item_id "
220 "JOIN artists ON artists.item_id = track_artists.artist_id "
221 "AND artists.search_name LIKE :search_artist"
222 )
223 extra_query_params["search_artist"] = f"%{artist_search_str}%"
224 existing_uris = {item.uri for item in result}
225 for _track in await self.get_library_items_by_query(
226 favorite=favorite,
227 search=None,
228 genre_ids=genre,
229 limit=limit,
230 order_by=order_by,
231 provider_filter=self._ensure_provider_filter(provider),
232 extra_query_parts=extra_query_parts,
233 extra_query_params=extra_query_params,
234 extra_join_parts=extra_join_parts,
235 in_library_only=True,
236 ):
237 # prevent duplicates (when artist is also in the title)
238 if _track.uri not in existing_uris:
239 result.append(_track)
240 return result
241
242 async def versions(
243 self,
244 item_id: str,
245 provider_instance_id_or_domain: str,
246 ) -> UniqueList[Track]:
247 """Return all versions of a track we can find on all providers."""
248 track = await self.get(item_id, provider_instance_id_or_domain)
249 search_query = f"{track.artist_str} - {track.name}"
250 result: UniqueList[Track] = UniqueList()
251 for provider_id in self.mass.music.get_unique_providers():
252 provider = self.mass.get_provider(provider_id)
253 if not isinstance(provider, MusicProvider):
254 continue
255 if not provider.library_supported(MediaType.TRACK):
256 continue
257 result.extend(
258 prov_item
259 for prov_item in await self.search(search_query, provider_id)
260 if loose_compare_strings(track.name, prov_item.name)
261 and compare_artists(prov_item.artists, track.artists, any_match=True)
262 # make sure that the 'base' version is NOT included
263 and not track.provider_mappings.intersection(prov_item.provider_mappings)
264 )
265 return result
266
267 async def albums(
268 self,
269 item_id: str,
270 provider_instance_id_or_domain: str,
271 in_library_only: bool = False,
272 ) -> UniqueList[Album]:
273 """Return all albums the track appears on."""
274 full_track = await self.get(item_id, provider_instance_id_or_domain)
275 db_items = (
276 await self.get_library_track_albums(full_track.item_id)
277 if full_track.provider == "library"
278 else []
279 )
280 # return all (unique) items from all providers
281 result: UniqueList[Album] = UniqueList(db_items)
282 # use search to get all items on the provider
283 search_query = f"{full_track.artist_str} - {full_track.name}"
284 # TODO: we could use musicbrainz info here to get a list of all releases known
285 unique_ids: set[str] = set()
286 for prov_item in (await self.mass.music.search(search_query, [MediaType.TRACK])).tracks:
287 if not isinstance(prov_item, Track): # for type checking
288 continue
289 if not loose_compare_strings(full_track.name, prov_item.name):
290 continue
291 if not prov_item.album:
292 continue
293 if not compare_artists(full_track.artists, prov_item.artists, any_match=True):
294 continue
295 unique_id = f"{prov_item.album.name}.{prov_item.album.version}"
296 if unique_id in unique_ids:
297 continue
298 unique_ids.add(unique_id)
299 # prefer db item
300 if db_item := await self.mass.music.albums.get_library_item_by_prov_id(
301 prov_item.album.item_id, prov_item.album.provider
302 ):
303 result.append(db_item)
304 elif not in_library_only and isinstance(prov_item.album, Album):
305 result.append(prov_item.album)
306 return result
307
308 async def similar_tracks(
309 self,
310 item_id: str,
311 provider_instance_id_or_domain: str,
312 limit: int = 25,
313 allow_lookup: bool = False,
314 preferred_provider_instances: list[str] | None = None,
315 ) -> list[Track]:
316 """
317 Get a list of similar tracks for the given track.
318
319 :param item_id: The item ID of the track.
320 :param provider_instance_id_or_domain: The provider instance ID or domain.
321 :param limit: Maximum number of similar tracks to return.
322 :param allow_lookup: Allow lookup on other providers if not found.
323 :param preferred_provider_instances: List of preferred provider instance IDs to use.
324 When provided, these providers will be tried first before falling back to others.
325 """
326 ref_item = await self.get(item_id, provider_instance_id_or_domain)
327
328 # Sort provider mappings to prefer user's provider instances
329 def sort_key(mapping: ProviderMapping) -> tuple[int, int]:
330 # Primary sort: preferred providers first (0), then others (1)
331 preferred = (
332 0
333 if preferred_provider_instances
334 and mapping.provider_instance in preferred_provider_instances
335 else 1
336 )
337 # Secondary sort: by quality (higher is better, so negate)
338 quality = -(mapping.quality or 0)
339 return (preferred, quality)
340
341 sorted_mappings = sorted(ref_item.provider_mappings, key=sort_key)
342
343 # Try preferred providers first, then fall back to others
344 for allow_other_provider in (False, True):
345 for prov_mapping in sorted_mappings:
346 if (
347 not allow_other_provider
348 and preferred_provider_instances
349 and prov_mapping.provider_instance not in preferred_provider_instances
350 ):
351 continue
352 prov = self.mass.get_provider(prov_mapping.provider_instance)
353 if prov is None:
354 continue
355 if not isinstance(prov, MusicProvider):
356 continue
357 if ProviderFeature.SIMILAR_TRACKS not in prov.supported_features:
358 continue
359 # Grab similar tracks from the music provider
360 return await prov.get_similar_tracks(
361 prov_track_id=prov_mapping.item_id, limit=limit
362 )
363
364 if not allow_lookup:
365 return []
366
367 # check if we have any provider that supports dynamic tracks
368 # TODO: query metadata provider(s) (such as lastfm?)
369 # to get similar tracks (or tracks from similar artists)
370 music_prov: MusicProvider | None = None
371 for prov in self.mass.music.providers:
372 if ProviderFeature.SIMILAR_TRACKS in prov.supported_features:
373 music_prov = prov
374 break
375 if music_prov is None:
376 msg = "No Music Provider found that supports requesting similar tracks."
377 raise UnsupportedFeaturedException(msg)
378
379 if mappings := await self.match_provider(ref_item, music_prov):
380 if ref_item.provider == "library":
381 # update database with new provider mappings
382 await self.add_provider_mappings(ref_item.item_id, mappings)
383 ref_item.provider_mappings.update(mappings)
384 return await music_prov.get_similar_tracks(
385 prov_track_id=mappings[0].item_id, limit=limit
386 )
387
388 return []
389
390 async def remove_item_from_library(self, item_id: str | int, recursive: bool = True) -> None:
391 """Delete record from the database."""
392 db_id = int(item_id) # ensure integer
393 # delete entry(s) from albumtracks table
394 await self.mass.music.database.delete(DB_TABLE_ALBUM_TRACKS, {"track_id": db_id})
395 # delete entry(s) from trackartists table
396 await self.mass.music.database.delete(DB_TABLE_TRACK_ARTISTS, {"track_id": db_id})
397 # delete the track itself from db
398 await super().remove_item_from_library(db_id)
399
400 async def get_preview_url(self, provider_instance_id_or_domain: str, item_id: str) -> str:
401 """Return url to short preview sample."""
402 track = await self.get_provider_item(item_id, provider_instance_id_or_domain)
403 # prefer provider-provided preview
404 if preview := track.metadata.preview:
405 return preview
406 # fallback to a preview/sample hosted by our own webserver
407 enc_track_id = urllib.parse.quote(item_id)
408 return (
409 f"{self.mass.webserver.base_url}/preview?"
410 f"provider={provider_instance_id_or_domain}&item_id={enc_track_id}"
411 )
412
413 async def get_library_track_albums(
414 self,
415 item_id: str | int,
416 ) -> list[Album]:
417 """Return all in-library albums for a track."""
418 db_id = int(item_id) # ensure integer
419 subquery = (
420 f"SELECT album_id FROM {DB_TABLE_ALBUM_TRACKS} "
421 f"WHERE {DB_TABLE_ALBUM_TRACKS}.track_id = :track_id"
422 )
423 query = f"{DB_TABLE_ALBUMS}.item_id in ({subquery})"
424 return await self.mass.music.albums.get_library_items_by_query(
425 extra_query_parts=[query],
426 extra_query_params={"track_id": db_id},
427 in_library_only=True,
428 )
429
430 async def match_provider(
431 self,
432 base_track: Track,
433 provider: MusicProvider,
434 strict: bool = True,
435 ref_albums: list[Album] | None = None,
436 ) -> list[ProviderMapping]:
437 """
438 Try to find match on (streaming) provider for the provided track.
439
440 This is used to link objects of different providers/qualities together.
441 """
442 if ref_albums is None:
443 ref_albums = await self.albums(base_track.item_id, base_track.provider)
444 self.logger.debug("Trying to match track %s on provider %s", base_track.name, provider.name)
445 matches: list[ProviderMapping] = []
446 for artist in base_track.artists:
447 if matches:
448 break
449 search_str = f"{artist.name} - {base_track.name}"
450 search_result = await self.search(search_str, provider.domain)
451 for search_result_item in search_result:
452 if not search_result_item.available:
453 continue
454 # do a basic compare first
455 if not compare_media_item(base_track, search_result_item, strict=False):
456 continue
457 # we must fetch the full version, search results can be simplified objects
458 prov_track = await self.get_provider_item(
459 search_result_item.item_id,
460 search_result_item.provider,
461 fallback=search_result_item,
462 )
463 if compare_track(base_track, prov_track, strict=strict, track_albums=ref_albums):
464 matches.extend(search_result_item.provider_mappings)
465
466 if not matches:
467 self.logger.debug(
468 "Could not find match for Track %s on provider %s",
469 base_track.name,
470 provider.name,
471 )
472 return matches
473
474 async def match_providers(self, db_track: Track) -> None:
475 """
476 Try to find matching track on all providers for the provided (database) track_id.
477
478 This is used to link objects of different providers/qualities together.
479 """
480 if db_track.provider != "library":
481 return # Matching only supported for database items
482
483 track_albums = await self.albums(db_track.item_id, db_track.provider)
484 # try to find match on all providers
485 processed_domains = set()
486 for provider in self.mass.music.providers:
487 if provider.domain in processed_domains:
488 continue
489 if ProviderFeature.SEARCH not in provider.supported_features:
490 continue
491 if not provider.library_supported(MediaType.TRACK):
492 continue
493 if not provider.is_streaming_provider:
494 # matching on unique providers is pointless as they push (all) their content to MA
495 continue
496 if match := await self.match_provider(
497 db_track, provider, strict=True, ref_albums=track_albums
498 ):
499 # 100% match, we update the db with the additional provider mapping(s)
500 await self.add_provider_mappings(db_track.item_id, match)
501 processed_domains.add(provider.domain)
502
503 async def radio_mode_base_tracks(
504 self,
505 item: Track,
506 preferred_provider_instances: list[str] | None = None,
507 ) -> list[Track]:
508 """
509 Get the list of base tracks from the controller used to calculate the dynamic radio.
510
511 :param item: The Track to get base tracks for.
512 :param preferred_provider_instances: List of preferred provider instance IDs to use.
513 """
514 return [item]
515
516 async def _add_library_item(self, item: Track, overwrite_existing: bool = False) -> int:
517 """Add a new item record to the database."""
518 if not isinstance(item, Track): # TODO: Remove this once the codebase is fully typed
519 msg = "Not a valid Track object (ItemMapping can not be added to db)" # type: ignore[unreachable]
520 raise InvalidDataError(msg)
521 if not item.artists:
522 msg = "Track is missing artist(s)"
523 raise InvalidDataError(msg)
524 db_id = await self.mass.music.database.insert(
525 self.db_table,
526 {
527 "name": item.name,
528 "sort_name": item.sort_name,
529 "version": item.version,
530 "duration": item.duration,
531 "favorite": item.favorite,
532 "external_ids": serialize_to_json(item.external_ids),
533 "metadata": serialize_to_json(item.metadata),
534 "search_name": create_safe_string(item.name, True, True),
535 "search_sort_name": create_safe_string(item.sort_name or "", True, True),
536 "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
537 },
538 )
539 # update/set provider_mappings table
540 await self.set_provider_mappings(db_id, item.provider_mappings)
541 # set track artist(s)
542 await self._set_track_artists(db_id, item.artists)
543 # handle track album
544 if item.album:
545 await self._set_track_album(
546 db_id=db_id,
547 album=item.album,
548 disc_number=getattr(item, "disc_number", 0),
549 track_number=getattr(item, "track_number", 0),
550 )
551 self.logger.debug("added %s to database (id: %s)", item.name, db_id)
552 return db_id
553
554 async def _update_library_item(
555 self, item_id: str | int, update: Track, overwrite: bool = False
556 ) -> None:
557 """Update Track record in the database, merging data."""
558 db_id = int(item_id) # ensure integer
559 cur_item = await self.get_library_item(db_id)
560 metadata = update.metadata if overwrite else cur_item.metadata.update(update.metadata)
561 cur_item.external_ids.update(update.external_ids)
562 name = update.name if overwrite else cur_item.name
563 sort_name = update.sort_name if overwrite else cur_item.sort_name or update.sort_name
564 await self.mass.music.database.update(
565 self.db_table,
566 {"item_id": db_id},
567 {
568 "name": name,
569 "sort_name": sort_name,
570 "version": update.version if overwrite else cur_item.version or update.version,
571 "duration": update.duration if overwrite else cur_item.duration or update.duration,
572 "metadata": serialize_to_json(metadata),
573 "external_ids": serialize_to_json(
574 update.external_ids if overwrite else cur_item.external_ids
575 ),
576 "search_name": create_safe_string(name, True, True),
577 "search_sort_name": create_safe_string(sort_name or "", True, True),
578 "timestamp_added": int(update.date_added.timestamp())
579 if update.date_added
580 else UNSET,
581 },
582 )
583 # update/set provider_mappings table
584 provider_mappings = (
585 update.provider_mappings
586 if overwrite
587 else {*update.provider_mappings, *cur_item.provider_mappings}
588 )
589 await self.set_provider_mappings(db_id, provider_mappings, overwrite)
590 # set track artist(s)
591 artists = update.artists if overwrite else cur_item.artists + update.artists
592 await self._set_track_artists(db_id, artists, overwrite=overwrite)
593 # update/set track album
594 if update.album:
595 await self._set_track_album(
596 db_id=db_id,
597 album=update.album,
598 disc_number=update.disc_number or cur_item.disc_number,
599 track_number=update.track_number or cur_item.track_number,
600 overwrite=overwrite,
601 )
602 self.logger.debug("updated %s in database: (id %s)", update.name, db_id)
603
604 async def _set_track_album(
605 self,
606 db_id: int,
607 album: Album | ItemMapping,
608 disc_number: int,
609 track_number: int,
610 overwrite: bool = False,
611 ) -> None:
612 """
613 Store Track Album info.
614
615 A track can exist on multiple albums so we have a mapping table between
616 albums and tracks which stores the relation between the two and it also
617 stores the track and disc number of the track within an album.
618 For digital releases, the discnumber will be just 0 or 1.
619 Track number should start counting at 1.
620 """
621 db_album: Album | ItemMapping | None = None
622 if album.provider == "library":
623 db_album = album
624 elif existing := await self.mass.music.albums.get_library_item_by_prov_id(
625 album.item_id, album.provider
626 ):
627 db_album = existing
628
629 if not db_album or overwrite:
630 # ensure we have an actual album object
631 if isinstance(album, ItemMapping):
632 db_album = await self.mass.music.albums.add_item_mapping_as_album_to_library(album)
633 else:
634 db_album = await self.mass.music.albums.add_item_to_library(
635 album,
636 overwrite_existing=overwrite,
637 )
638 # write (or update) record in album_tracks table
639 await self.mass.music.database.insert_or_replace(
640 DB_TABLE_ALBUM_TRACKS,
641 {
642 "track_id": db_id,
643 "album_id": int(db_album.item_id),
644 "disc_number": disc_number,
645 "track_number": track_number,
646 },
647 )
648
649 async def _set_track_artists(
650 self,
651 db_id: int,
652 artists: Iterable[Artist | ItemMapping],
653 overwrite: bool = False,
654 ) -> None:
655 """Store Track Artists."""
656 if overwrite:
657 # on overwrite, clear the track_artists table first
658 await self.mass.music.database.delete(
659 DB_TABLE_TRACK_ARTISTS,
660 {
661 "track_id": db_id,
662 },
663 )
664 artist_mappings: UniqueList[ItemMapping] = UniqueList()
665 for artist in artists:
666 mapping = await self._set_track_artist(db_id, artist=artist, overwrite=overwrite)
667 artist_mappings.append(mapping)
668
669 async def _set_track_artist(
670 self, db_id: int, artist: Artist | ItemMapping, overwrite: bool = False
671 ) -> ItemMapping:
672 """Store Track Artist info."""
673 db_artist: Artist | ItemMapping | None = None
674 if artist.provider == "library":
675 db_artist = artist
676 elif existing := await self.mass.music.artists.get_library_item_by_prov_id(
677 artist.item_id, artist.provider
678 ):
679 db_artist = existing
680
681 if not db_artist or overwrite:
682 # Convert ItemMapping to Artist if needed
683 artist_to_add = (
684 self.mass.music.artists.artist_from_item_mapping(artist)
685 if isinstance(artist, ItemMapping)
686 else artist
687 )
688 db_artist = await self.mass.music.artists.add_item_to_library(
689 artist_to_add, overwrite_existing=overwrite
690 )
691 # write (or update) record in track_artists table
692 await self.mass.music.database.insert_or_replace(
693 DB_TABLE_TRACK_ARTISTS,
694 {
695 "track_id": db_id,
696 "artist_id": int(db_artist.item_id),
697 },
698 )
699 return ItemMapping.from_item(db_artist)
700