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(
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        **kwargs: Any,
250    ) -> list[Track]:
251        """
252        Get in-database tracks.
253
254        :param favorite: Filter by favorite status.
255        :param search: Filter by search query.
256        :param limit: Maximum number of items to return.
257        :param offset: Number of items to skip.
258        :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
259        :param provider: Filter by provider instance ID (single string or list).
260        :param genre: Filter by genre id(s).
261        :param played_only: Filter to only played tracks.
262        :param explicit: Filter by explicit content (True=only explicit, False=no explicit, None=all).
263        :param summary: When True (default), return slim summary items containing only the
264            fields needed for a list view. Set to False to get fully hydrated items.
265        """
266        extra_query_params: dict[str, Any] = {}
267        extra_query_parts: list[str] = []
268        extra_join_parts: list[str] = []
269
270        # Apply explicit content filter
271        if explicit is not None:
272            if explicit:
273                # Only explicit tracks
274                extra_query_parts.append("json_extract(tracks.metadata, '$.explicit') = 1")
275            else:
276                # No explicit tracks (null or false)
277                extra_query_parts.append(
278                    "(json_extract(tracks.metadata, '$.explicit') IS NULL "
279                    "OR json_extract(tracks.metadata, '$.explicit') = 0)"
280                )
281
282        if (order_by and "track_artist_name" in order_by) or (search and " - " in search):
283            extra_join_parts.append(
284                "JOIN track_artists ON track_artists.track_id = tracks.item_id "
285                "JOIN artists ON artists.item_id = track_artists.artist_id "
286            )
287
288        if search and " - " in search:
289            # handle combined artist + title search
290            artist_str, title_str = search.split(" - ", 1)
291            search = None
292            title_str = create_safe_string(title_str, True, True)
293            artist_str = create_safe_string(artist_str, True, True)
294            extra_query_parts.append(
295                search_name_match_clause("tracks", title_str, "search_title", extra_query_params)
296            )
297            extra_query_parts.append(
298                search_name_match_clause("artists", artist_str, "search_artist", extra_query_params)
299            )
300        result = await self.get_library_items_by_query(
301            favorite=favorite,
302            search=search,
303            genre_ids=genre,
304            limit=limit,
305            offset=offset,
306            order_by=order_by,
307            provider_filter=self._ensure_provider_filter(provider),
308            extra_query_parts=extra_query_parts,
309            extra_query_params=extra_query_params,
310            extra_join_parts=extra_join_parts,
311            played_only=played_only,
312            in_library_only=True,
313            summary=summary,
314        )
315        if search and len(result) < 25 and not offset:
316            # append artist items to result
317            artist_search_str = create_safe_string(search, True, True)
318            if order_by and "track_artist_name" in order_by:
319                # JOIN already exists for sorting, only add WHERE clause
320                extra_query_parts.append(
321                    search_name_match_clause(
322                        "artists", artist_search_str, "search_artist", extra_query_params
323                    )
324                )
325            else:
326                # JOIN not yet added, add it with the search condition
327                extra_join_parts.append(
328                    "JOIN track_artists ON track_artists.track_id = tracks.item_id "
329                    "JOIN artists ON artists.item_id = track_artists.artist_id "
330                    "AND "
331                    + search_name_match_clause(
332                        "artists", artist_search_str, "search_artist", extra_query_params
333                    )
334                )
335            existing_uris = {item.uri for item in result}
336            for _track in await self.get_library_items_by_query(
337                favorite=favorite,
338                search=None,
339                genre_ids=genre,
340                limit=limit,
341                order_by=order_by,
342                provider_filter=self._ensure_provider_filter(provider),
343                extra_query_parts=extra_query_parts,
344                extra_query_params=extra_query_params,
345                extra_join_parts=extra_join_parts,
346                in_library_only=True,
347                summary=summary,
348            ):
349                # prevent duplicates (when artist is also in the title)
350                if _track.uri not in existing_uris:
351                    result.append(_track)
352        return result
353
354    async def versions(
355        self,
356        item_id: str,
357        provider_instance_id_or_domain: str,
358    ) -> UniqueList[Track]:
359        """Return all versions of a track we can find on all providers."""
360        track = await self.get(item_id, provider_instance_id_or_domain)
361        search_query = f"{track.artist_str} - {track.name}"
362        result: UniqueList[Track] = UniqueList()
363        for provider_id in self.mass.music.get_unique_providers():
364            provider = self.mass.get_provider(provider_id)
365            if not isinstance(provider, MusicProvider):
366                continue
367            if not self.mass.music.library_supported(provider, MediaType.TRACK):
368                continue
369            result.extend(
370                prov_item
371                for prov_item in await self.search(search_query, provider_id)
372                if loose_compare_strings(track.name, prov_item.name)
373                and compare_artists(prov_item.artists, track.artists, any_match=True)
374                # make sure that the 'base' version is NOT included
375                and not track.provider_mappings.intersection(prov_item.provider_mappings)
376            )
377        return result
378
379    async def albums(
380        self,
381        item_id: str,
382        provider_instance_id_or_domain: str,
383        in_library_only: bool = False,
384    ) -> UniqueList[Album]:
385        """Return all albums the track appears on."""
386        full_track = await self.get(item_id, provider_instance_id_or_domain)
387        db_items = (
388            await self.get_library_track_albums(full_track.item_id)
389            if full_track.provider == "library"
390            else []
391        )
392        # return all (unique) items from all providers
393        result: UniqueList[Album] = UniqueList(db_items)
394        # use search to get all items on the provider
395        search_query = f"{full_track.artist_str} - {full_track.name}"
396        # TODO: we could use musicbrainz info here to get a list of all releases known
397        unique_ids: set[str] = set()
398        # explicitly search all providers as we want all album versions
399        # of this track, including those already mapped in the library
400        search_providers = ["library", *self.mass.music.get_unique_providers()]
401        search_results = await self.mass.music.search(
402            search_query, [MediaType.TRACK], providers=search_providers
403        )
404        for prov_item in search_results.tracks:
405            if not isinstance(prov_item, Track):  # for type checking
406                continue
407            if not loose_compare_strings(full_track.name, prov_item.name):
408                continue
409            if not prov_item.album:
410                continue
411            if not compare_artists(full_track.artists, prov_item.artists, any_match=True):
412                continue
413            unique_id = f"{prov_item.album.name}.{prov_item.album.version}"
414            if unique_id in unique_ids:
415                continue
416            unique_ids.add(unique_id)
417            # prefer db item
418            if db_item := await self.mass.music.albums.get_library_item_by_prov_id(
419                prov_item.album.item_id, prov_item.album.provider
420            ):
421                result.append(db_item)
422            elif not in_library_only and isinstance(prov_item.album, Album):
423                result.append(prov_item.album)
424        return result
425
426    async def similar_tracks(
427        self,
428        item_id: str,
429        provider_instance_id_or_domain: str,
430        limit: int = 25,
431        allow_lookup: bool = False,
432        preferred_provider_instances: list[str] | None = None,
433    ) -> list[Track]:
434        """
435        Get a list of similar tracks for the given track.
436
437        :param item_id: The item ID of the track.
438        :param provider_instance_id_or_domain: The provider instance ID or domain.
439        :param limit: Maximum number of similar tracks to return.
440        :param allow_lookup: Allow lookup on other providers if not found.
441        :param preferred_provider_instances: List of preferred provider instance IDs to use.
442            When provided, these providers will be tried first before falling back to others.
443        """
444        ref_item = await self.get(item_id, provider_instance_id_or_domain)
445
446        # Sort provider mappings to prefer user's provider instances
447        def sort_key(mapping: ProviderMapping) -> tuple[int, int]:
448            # Primary sort: preferred providers first (0), then others (1)
449            preferred = (
450                0
451                if preferred_provider_instances
452                and mapping.provider_instance in preferred_provider_instances
453                else 1
454            )
455            # Secondary sort: by quality (higher is better, so negate)
456            quality = -(mapping.quality or 0)
457            return (preferred, quality)
458
459        sorted_mappings = sorted(ref_item.provider_mappings, key=sort_key)
460
461        # Try preferred providers first, then fall back to others
462        for allow_other_provider in (False, True):
463            for prov_mapping in sorted_mappings:
464                if (
465                    not allow_other_provider
466                    and preferred_provider_instances
467                    and prov_mapping.provider_instance not in preferred_provider_instances
468                ):
469                    continue
470                prov = self.mass.get_provider(prov_mapping.provider_instance)
471                if prov is None:
472                    continue
473                if not isinstance(prov, MusicProvider):
474                    continue
475                if ProviderFeature.SIMILAR_TRACKS not in prov.supported_features:
476                    continue
477                # Grab similar tracks from the music provider
478                try:
479                    if result := await prov.get_similar_tracks(
480                        prov_track_id=prov_mapping.item_id, limit=limit
481                    ):
482                        return result
483                except NotImplementedError:
484                    continue
485
486        # Fallback: consult metadata/plugin providers that claim SIMILAR_TRACKS
487        for prov in self.mass.get_providers_supporting_feature(
488            ProviderFeature.SIMILAR_TRACKS,
489            priority=(ProviderType.METADATA, ProviderType.PLUGIN),
490        ):
491            try:
492                cross_prov = cast("MetadataProvider | PluginProvider", prov)
493                if result := await cross_prov.get_similar_tracks(ref_item, limit=limit):
494                    return result
495            except NotImplementedError:
496                continue
497
498        if not allow_lookup:
499            return []
500
501        music_prov: MusicProvider | None = None
502        for prov in self.mass.music.providers:
503            if ProviderFeature.SIMILAR_TRACKS in prov.supported_features:
504                music_prov = prov
505                break
506        if music_prov is None:
507            msg = "No Music Provider found that supports requesting similar tracks."
508            raise UnsupportedFeaturedException(msg)
509
510        if mappings := await self.match_provider(ref_item, music_prov):
511            if ref_item.provider == "library":
512                # update database with new provider mappings
513                await self.add_provider_mappings(ref_item.item_id, mappings)
514            ref_item.provider_mappings.update(mappings)
515            return await music_prov.get_similar_tracks(
516                prov_track_id=mappings[0].item_id, limit=limit
517            )
518
519        return []
520
521    async def remove_item_from_library(self, item_id: str | int, recursive: bool = True) -> None:
522        """Delete record from the database."""
523        db_id = int(item_id)  # ensure integer
524        # delete entry(s) from albumtracks table
525        await self.mass.music.database.delete(DB_TABLE_ALBUM_TRACKS, {"track_id": db_id})
526        # delete entry(s) from trackartists table
527        await self.mass.music.database.delete(DB_TABLE_TRACK_ARTISTS, {"track_id": db_id})
528        # delete the track itself from db
529        await super().remove_item_from_library(db_id)
530
531    async def set_identifiers(
532        self,
533        item_id: str,
534        provider_instance_id_or_domain: str,
535        mbid: str | None = None,
536        acoustid: str | None = None,
537        isrcs: list[str] | None = None,
538    ) -> None:
539        """
540        Persist MBID / AcoustID / ISRCs onto the library track row.
541
542        :param item_id: Provider-native track ID.
543        :param provider_instance_id_or_domain: Music provider instance ID or domain.
544        :param mbid: MusicBrainz recording ID.
545        :param acoustid: AcoustID UUID.
546        :param isrcs: ISRC codes.
547        """
548        # MBID is filled only when empty; AcoustID/ISRCs are appended via
549        # external_ids without clobbering tag-sourced values.
550        if not mbid and not acoustid and not isrcs:
551            return
552        try:
553            track = await self.get_library_item_by_prov_id(item_id, provider_instance_id_or_domain)
554        except MusicAssistantError as err:
555            self.logger.debug(
556                "set_identifiers: failed to load library track %s/%s: %s",
557                provider_instance_id_or_domain,
558                item_id,
559                err,
560            )
561            return
562        if track is None:
563            return
564
565        changed = False
566        if mbid and not track.mbid:
567            track.mbid = mbid
568            changed = True
569        if acoustid and not any(
570            ext_id[0] == ExternalID.ACOUSTID and ext_id[1] == acoustid
571            for ext_id in track.external_ids
572        ):
573            track.add_external_id(ExternalID.ACOUSTID, acoustid)
574            changed = True
575        for isrc in isrcs or ():
576            if isrc:
577                track.add_external_id(ExternalID.ISRC, isrc)
578                changed = True
579        if not changed:
580            return
581
582        await self.update_item_in_library(int(track.item_id), track)
583
584    async def get_preview_url(self, provider_instance_id_or_domain: str, item_id: str) -> str:
585        """Return url to short preview sample."""
586        track = await self.get_provider_item(item_id, provider_instance_id_or_domain)
587        # prefer provider-provided preview
588        if preview := track.metadata.preview:
589            return preview
590        # fallback to a preview/sample hosted by our own webserver
591        enc_track_id = urllib.parse.quote(item_id)
592        return (
593            f"{self.mass.webserver.base_url}/preview?"
594            f"provider={provider_instance_id_or_domain}&item_id={enc_track_id}"
595        )
596
597    async def get_library_track_albums(
598        self,
599        item_id: str | int,
600    ) -> list[Album]:
601        """Return all in-library albums for a track."""
602        db_id = int(item_id)  # ensure integer
603        subquery = (
604            f"SELECT album_id FROM {DB_TABLE_ALBUM_TRACKS} "
605            f"WHERE {DB_TABLE_ALBUM_TRACKS}.track_id = :track_id"
606        )
607        query = f"{DB_TABLE_ALBUMS}.item_id in ({subquery})"
608        return await self.mass.music.albums.get_library_items_by_query(
609            extra_query_parts=[query],
610            extra_query_params={"track_id": db_id},
611            in_library_only=True,
612        )
613
614    async def match_provider(
615        self,
616        base_track: Track,
617        provider: MusicProvider,
618        strict: bool = True,
619        ref_albums: list[Album] | None = None,
620    ) -> list[ProviderMapping]:
621        """
622        Try to find match on (streaming) provider for the provided track.
623
624        This is used to link objects of different providers/qualities together.
625        """
626        if ref_albums is None:
627            ref_albums = await self.albums(base_track.item_id, base_track.provider)
628        self.logger.debug("Trying to match track %s on provider %s", base_track.name, provider.name)
629        matches: list[ProviderMapping] = []
630        for artist in base_track.artists:
631            if matches:
632                break
633            search_str = f"{artist.name} - {base_track.name}"
634            search_result = await self.search(search_str, provider.domain)
635            for search_result_item in search_result:
636                if not search_result_item.available:
637                    continue
638                # do a basic compare first
639                if not compare_media_item(base_track, search_result_item, strict=False):
640                    continue
641                # we must fetch the full version, search results can be simplified objects
642                prov_track = await self.get_provider_item(
643                    search_result_item.item_id,
644                    search_result_item.provider,
645                    fallback=search_result_item,
646                )
647                if compare_track(base_track, prov_track, strict=strict, track_albums=ref_albums):
648                    matches.extend(search_result_item.provider_mappings)
649
650        if not matches:
651            self.logger.debug(
652                "Could not find match for Track %s on provider %s",
653                base_track.name,
654                provider.name,
655            )
656        return matches
657
658    async def match_providers(self, db_track: Track) -> None:
659        """
660        Try to find matching track on all providers for the provided (database) track_id.
661
662        This is used to link objects of different providers/qualities together.
663        """
664        if db_track.provider != "library":
665            return  # Matching only supported for database items
666
667        track_albums = await self.albums(db_track.item_id, db_track.provider)
668        # try to find match on all providers
669        processed_domains = set()
670        for provider in self.mass.music.providers:
671            if provider.domain in processed_domains:
672                continue
673            if ProviderFeature.SEARCH not in provider.supported_features:
674                continue
675            if not self.mass.music.library_supported(provider, MediaType.TRACK):
676                continue
677            if not provider.is_streaming_provider:
678                # matching on unique providers is pointless as they push (all) their content to MA
679                continue
680            if match := await self.match_provider(
681                db_track, provider, strict=True, ref_albums=track_albums
682            ):
683                # 100% match, we update the db with the additional provider mapping(s)
684                await self.add_provider_mappings(db_track.item_id, match)
685                processed_domains.add(provider.domain)
686
687    async def _add_library_item(self, item: Track, overwrite_existing: bool = False) -> int:
688        """Add a new item record to the database."""
689        if not isinstance(item, Track):  # TODO: Remove this once the codebase is fully typed
690            msg = "Not a valid Track object (ItemMapping can not be added to db)"  # type: ignore[unreachable]
691            raise InvalidDataError(msg)
692        if not item.artists:
693            msg = "Track is missing artist(s)"
694            raise InvalidDataError(msg)
695        # normalize synced lyrics so clients only need a minimal single-timestamp LRC parser
696        # promoting LRC formatted text stored in the plain lyrics tag
697        item.metadata.lrc_lyrics = normalize_lrc_lyrics(
698            item.metadata.lrc_lyrics or extract_lrc_lyrics(item.metadata.lyrics)
699        )
700        db_id = await self.mass.music.database.insert(
701            self.db_table,
702            {
703                "name": item.name,
704                "sort_name": item.sort_name,
705                "version": item.version,
706                "duration": item.duration,
707                "favorite": item.favorite,
708                "metadata": serialize_to_json(item.metadata),
709                "search_name": create_safe_string(item.name, True, True),
710                "search_sort_name": create_safe_string(item.sort_name or "", True, True),
711                "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
712            },
713        )
714        # update/set external id lookup table
715        await self.set_external_ids(db_id, item.external_ids)
716        # update/set provider_mappings table
717        await self.set_provider_mappings(db_id, item.provider_mappings)
718        # set track artist(s)
719        await self._set_track_artists(db_id, item.artists)
720        # handle track album
721        if item.album:
722            await self._set_track_album(
723                db_id=db_id,
724                album=item.album,
725                disc_number=getattr(item, "disc_number", 0),
726                track_number=getattr(item, "track_number", 0),
727            )
728        self.logger.debug("added %s to database (id: %s)", item.name, db_id)
729        return db_id
730
731    async def _update_library_item(
732        self, item_id: str | int, update: Track, overwrite: bool = False
733    ) -> None:
734        """Update Track record in the database, merging data."""
735        db_id = int(item_id)  # ensure integer
736        cur_item = await self.get_library_item(db_id)
737        metadata = update.metadata if overwrite else cur_item.metadata.update(update.metadata)
738        metadata.lrc_lyrics = normalize_lrc_lyrics(
739            metadata.lrc_lyrics or extract_lrc_lyrics(metadata.lyrics)
740        )
741        cur_item.external_ids.update(update.external_ids)
742        name = update.name if overwrite else cur_item.name
743        sort_name = update.sort_name if overwrite else cur_item.sort_name or update.sort_name
744        await self.mass.music.database.update(
745            self.db_table,
746            {"item_id": db_id},
747            {
748                "name": name,
749                "sort_name": sort_name,
750                "version": update.version if overwrite else cur_item.version or update.version,
751                "duration": update.duration if overwrite else cur_item.duration or update.duration,
752                "metadata": serialize_to_json(metadata),
753                "search_name": create_safe_string(name, True, True),
754                "search_sort_name": create_safe_string(sort_name or "", True, True),
755                "timestamp_added": int(update.date_added.timestamp())
756                if update.date_added
757                else UNSET,
758            },
759        )
760        # update/set external id lookup table
761        await self.set_external_ids(
762            db_id, update.external_ids if overwrite else cur_item.external_ids
763        )
764        # update/set provider_mappings table
765        provider_mappings = (
766            update.provider_mappings
767            if overwrite
768            else {*update.provider_mappings, *cur_item.provider_mappings}
769        )
770        await self.set_provider_mappings(db_id, provider_mappings, overwrite)
771        # set track artist(s)
772        artists = update.artists if overwrite else cur_item.artists + update.artists
773        await self._set_track_artists(db_id, artists, overwrite=overwrite)
774        # update/set track album
775        if update.album:
776            await self._set_track_album(
777                db_id=db_id,
778                album=update.album,
779                disc_number=update.disc_number or cur_item.disc_number,
780                track_number=update.track_number or cur_item.track_number,
781                overwrite=overwrite,
782            )
783        self.logger.debug("updated %s in database: (id %s)", update.name, db_id)
784
785    async def _set_track_album(
786        self,
787        db_id: int,
788        album: Album | ItemMapping,
789        disc_number: int,
790        track_number: int,
791        overwrite: bool = False,
792    ) -> None:
793        """
794        Store Track Album info.
795
796        A track can exist on multiple albums so we have a mapping table between
797        albums and tracks which stores the relation between the two and it also
798        stores the track and disc number of the track within an album.
799        For digital releases, the discnumber will be just 0 or 1.
800        Track number should start counting at 1.
801        """
802        db_album: Album | ItemMapping | None = None
803        if album.provider == "library":
804            db_album = album
805        elif existing := await self.mass.music.albums.get_library_item_by_prov_id(
806            album.item_id, album.provider
807        ):
808            db_album = existing
809
810        if not db_album or overwrite:
811            # ensure we have an actual album object
812            if isinstance(album, ItemMapping):
813                db_album = await self.mass.music.albums.add_item_mapping_as_album_to_library(album)
814            else:
815                db_album = await self.mass.music.albums.add_item_to_library(
816                    album,
817                    overwrite_existing=overwrite,
818                )
819        # write (or update) record in album_tracks table
820        await self.mass.music.database.insert_or_replace(
821            DB_TABLE_ALBUM_TRACKS,
822            {
823                "track_id": db_id,
824                "album_id": int(db_album.item_id),
825                "disc_number": disc_number,
826                "track_number": track_number,
827            },
828        )
829
830    async def _set_track_artists(
831        self,
832        db_id: int,
833        artists: Iterable[Artist | ItemMapping],
834        overwrite: bool = False,
835    ) -> None:
836        """Store Track Artists."""
837        if overwrite:
838            # on overwrite, clear the track_artists table first
839            await self.mass.music.database.delete(
840                DB_TABLE_TRACK_ARTISTS,
841                {
842                    "track_id": db_id,
843                },
844            )
845        artist_mappings: UniqueList[ItemMapping] = UniqueList()
846        for artist in artists:
847            mapping = await self._set_track_artist(db_id, artist=artist, overwrite=overwrite)
848            artist_mappings.append(mapping)
849
850    async def _set_track_artist(
851        self, db_id: int, artist: Artist | ItemMapping, overwrite: bool = False
852    ) -> ItemMapping:
853        """Store Track Artist info."""
854        db_artist: Artist | ItemMapping | None = None
855        if artist.provider == "library":
856            db_artist = artist
857        elif existing := await self.mass.music.artists.get_library_item_by_prov_id(
858            artist.item_id, artist.provider
859        ):
860            db_artist = existing
861
862        if not db_artist or overwrite:
863            # Convert ItemMapping to Artist if needed
864            artist_to_add = (
865                self.mass.music.artists.artist_from_item_mapping(artist)
866                if isinstance(artist, ItemMapping)
867                else artist
868            )
869            db_artist = await self.mass.music.artists.add_item_to_library(
870                artist_to_add, overwrite_existing=overwrite
871            )
872        # write (or update) record in track_artists table
873        await self.mass.music.database.insert_or_replace(
874            DB_TABLE_TRACK_ARTISTS,
875            {
876                "track_id": db_id,
877                "artist_id": int(db_artist.item_id),
878            },
879        )
880        return ItemMapping.from_item(db_artist)
881
882    def _sync_details_query_parts(self) -> tuple[str, str, dict[str, Any]]:
883        """Return extra (columns, joins, params) for the tracks sync-details query."""
884        # the sync loop needs to know if the track has a (valid) album link
885        # to be able to backfill a missing album on existing library tracks
886        extra_columns = """
887            , EXISTS (
888                SELECT 1 FROM album_tracks
889                JOIN albums ON albums.item_id = album_tracks.album_id
890                WHERE album_tracks.track_id = tracks.item_id
891            ) AS has_album
892        """
893        return extra_columns, "", {}
894
895    def _parse_sync_details_row(self, db_row: Mapping[str, Any]) -> TrackSyncDetails:
896        """Parse a raw sync-details db row into a TrackSyncDetails object."""
897        return TrackSyncDetails(
898            item_id=db_row["item_id"],
899            favorite=bool(db_row["favorite"]),
900            date_added=datetime.fromtimestamp(db_row["timestamp_added"], tz=UTC),
901            provider_mappings=self._parse_sync_details_mappings(db_row),
902            has_album=bool(db_row["has_album"]),
903        )
904
905    def _parse_summary_row(self, db_row: Mapping[str, Any]) -> TrackSummary:
906        """Parse a raw summary db row into a TrackSummary object."""
907        item = cast("TrackSummary", super()._parse_summary_row(db_row))
908        item.version = db_row["version"] or ""
909        item.duration = db_row["duration"] or 0
910        item.metadata.explicit = None if db_row["explicit"] is None else bool(db_row["explicit"])
911        if raw_release_date := db_row["release_date"]:
912            item.metadata.release_date = datetime.fromisoformat(raw_release_date)
913        item.artists = self._parse_summary_artist_mappings(db_row)
914        if raw_album := db_row["track_album"]:
915            album: dict[str, Any] = json_loads(raw_album)
916            album_thumb: MediaItemImage | None = None
917            if album_images := album.get("images"):
918                for image in album_images:
919                    if image["type"] != ImageType.THUMB.value:
920                        continue
921                    album_thumb = MediaItemImage(
922                        type=ImageType.THUMB,
923                        path=image["path"],
924                        provider=image["provider"],
925                        remotely_accessible=image.get("remotely_accessible", False),
926                    )
927                    break
928            item.album = ItemMappingSummary(
929                media_type=MediaType.ALBUM,
930                item_id=str(album["item_id"]),
931                provider="library",
932                name=album["name"],
933                sort_name=album["sort_name"],
934                year=album["year"],
935                image=album_thumb,
936            )
937            item.disc_number = album["disc_number"] or 0
938            item.track_number = album["track_number"] or 0
939            if album_thumb:
940                # always prefer album image over track image
941                item.metadata.images = UniqueList([album_thumb])
942        return item
943