music-assistant-server

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(search_result_item.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, item_id: str | int, update: Track, overwrite: bool = False
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:
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 _set_track_album(
797        self,
798        db_id: int,
799        album: Album | ItemMapping,
800        disc_number: int,
801        track_number: int,
802        overwrite: bool = False,
803    ) -> None:
804        """
805        Store Track Album info.
806
807        A track can exist on multiple albums so we have a mapping table between
808        albums and tracks which stores the relation between the two and it also
809        stores the track and disc number of the track within an album.
810        For digital releases, the discnumber will be just 0 or 1.
811        Track number should start counting at 1.
812        """
813        db_album: Album | ItemMapping | None = None
814        if album.provider == "library":
815            db_album = album
816        elif existing := await self.mass.music.albums.get_library_item_by_prov_id(
817            album.item_id, album.provider
818        ):
819            db_album = existing
820
821        if not db_album or overwrite:
822            # ensure we have an actual album object
823            if isinstance(album, ItemMapping):
824                db_album = await self.mass.music.albums.add_item_mapping_as_album_to_library(album)
825            else:
826                db_album = await self.mass.music.albums.add_item_to_library(
827                    album,
828                    overwrite_existing=overwrite,
829                )
830        # write (or update) record in album_tracks table
831        await self.mass.music.database.insert_or_replace(
832            DB_TABLE_ALBUM_TRACKS,
833            {
834                "track_id": db_id,
835                "album_id": int(db_album.item_id),
836                "disc_number": disc_number,
837                "track_number": track_number,
838            },
839        )
840
841    async def _set_track_artists(
842        self,
843        db_id: int,
844        artists: Iterable[Artist | ItemMapping],
845        overwrite: bool = False,
846    ) -> None:
847        """Store Track Artists."""
848        if overwrite:
849            # on overwrite, clear the track_artists table first
850            await self.mass.music.database.delete(
851                DB_TABLE_TRACK_ARTISTS,
852                {
853                    "track_id": db_id,
854                },
855            )
856        artist_mappings: UniqueList[ItemMapping] = UniqueList()
857        for artist in artists:
858            mapping = await self._set_track_artist(db_id, artist=artist, overwrite=overwrite)
859            artist_mappings.append(mapping)
860
861    async def _set_track_artist(
862        self, db_id: int, artist: Artist | ItemMapping, overwrite: bool = False
863    ) -> ItemMapping:
864        """Store Track Artist info."""
865        db_artist: Artist | ItemMapping | None = None
866        if artist.provider == "library":
867            db_artist = artist
868        elif existing := await self.mass.music.artists.get_library_item_by_prov_id(
869            artist.item_id, artist.provider
870        ):
871            db_artist = existing
872
873        if not db_artist or overwrite:
874            # Convert ItemMapping to Artist if needed
875            artist_to_add = (
876                self.mass.music.artists.artist_from_item_mapping(artist)
877                if isinstance(artist, ItemMapping)
878                else artist
879            )
880            db_artist = await self.mass.music.artists.add_item_to_library(
881                artist_to_add, overwrite_existing=overwrite
882            )
883        # write (or update) record in track_artists table
884        await self.mass.music.database.insert_or_replace(
885            DB_TABLE_TRACK_ARTISTS,
886            {
887                "track_id": db_id,
888                "artist_id": int(db_artist.item_id),
889            },
890        )
891        return ItemMapping.from_item(db_artist)
892
893    def _sync_details_query_parts(self) -> tuple[str, str, dict[str, Any]]:
894        """Return extra (columns, joins, params) for the tracks sync-details query."""
895        # the sync loop needs to know if the track has a (valid) album link
896        # to be able to backfill a missing album on existing library tracks
897        extra_columns = """
898            , EXISTS (
899                SELECT 1 FROM album_tracks
900                JOIN albums ON albums.item_id = album_tracks.album_id
901                WHERE album_tracks.track_id = tracks.item_id
902            ) AS has_album
903        """
904        return extra_columns, "", {}
905
906    def _parse_sync_details_row(self, db_row: Mapping[str, Any]) -> TrackSyncDetails:
907        """Parse a raw sync-details db row into a TrackSyncDetails object."""
908        return TrackSyncDetails(
909            item_id=db_row["item_id"],
910            favorite=bool(db_row["favorite"]),
911            date_added=datetime.fromtimestamp(db_row["timestamp_added"], tz=UTC),
912            provider_mappings=self._parse_sync_details_mappings(db_row),
913            has_album=bool(db_row["has_album"]),
914        )
915
916    def _parse_summary_row(self, db_row: Mapping[str, Any]) -> TrackSummary:
917        """Parse a raw summary db row into a TrackSummary object."""
918        item = cast("TrackSummary", super()._parse_summary_row(db_row))
919        item.version = db_row["version"] or ""
920        item.duration = db_row["duration"] or 0
921        item.metadata.explicit = None if db_row["explicit"] is None else bool(db_row["explicit"])
922        if raw_release_date := db_row["release_date"]:
923            item.metadata.release_date = datetime.fromisoformat(raw_release_date)
924        item.artists = self._parse_summary_artist_mappings(db_row)
925        if raw_album := db_row["track_album"]:
926            album: dict[str, Any] = json_loads(raw_album)
927            album_thumb: MediaItemImage | None = None
928            if album_images := album.get("images"):
929                for image in album_images:
930                    if image["type"] != ImageType.THUMB.value:
931                        continue
932                    album_thumb = MediaItemImage(
933                        type=ImageType.THUMB,
934                        path=image["path"],
935                        provider=image["provider"],
936                        remotely_accessible=image.get("remotely_accessible", False),
937                    )
938                    break
939            item.album = ItemMappingSummary(
940                media_type=MediaType.ALBUM,
941                item_id=str(album["item_id"]),
942                provider="library",
943                name=album["name"],
944                sort_name=album["sort_name"],
945                year=album["year"],
946                image=album_thumb,
947            )
948            item.disc_number = album["disc_number"] or 0
949            item.track_number = album["track_number"] or 0
950            if album_thumb:
951                # always prefer album image over track image
952                item.metadata.images = UniqueList([album_thumb])
953        return item
954