music-assistant-server

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