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