music-assistant-server

1"""All logic for metadata retrieval."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import os
8import random
9import sqlite3
10import threading
11from collections import OrderedDict
12from time import time
13from typing import TYPE_CHECKING, cast
14from uuid import NAMESPACE_URL, uuid5
15
16import aiohttp
17from music_assistant_models.auth import Scope
18from music_assistant_models.background_task import TaskSchedule
19from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
20from music_assistant_models.enums import (
21    AlbumType,
22    ConfigEntryType,
23    MediaType,
24    ProviderFeature,
25    ProviderType,
26)
27from music_assistant_models.errors import MediaNotFoundError, MusicAssistantError
28from music_assistant_models.media_items import BrowseFolder
29
30from music_assistant.constants import (
31    CONF_LANGUAGE,
32    DB_TABLE_ALBUM_ARTISTS,
33    DB_TABLE_ALBUMS,
34    DB_TABLE_ARTISTS,
35    DB_TABLE_PLAYLISTS,
36    VERBOSE_LOG_LEVEL,
37)
38from music_assistant.controllers.tasks.context import (
39    report_current_task_failure,
40    update_current_task_progress,
41    update_current_task_progress_from_index,
42    update_current_task_progress_text,
43)
44from music_assistant.helpers.api import api_command
45from music_assistant.helpers.images import cleanup_thumb_cache
46from music_assistant.helpers.lyrics import extract_lrc_lyrics, normalize_lrc_lyrics
47from music_assistant.helpers.throttle_retry import Throttler
48from music_assistant.helpers.util import try_parse_int
49from music_assistant.models.core_controller import CoreController
50from music_assistant.models.music_provider import MusicProvider
51
52from .constants import (
53    ALBUM_RECONCILIATION_TASK_ID,
54    CONF_ENABLE_ONLINE_METADATA,
55    CONF_ENABLE_RADIO_METADATA_LOOKUP,
56    CONF_PREFER_LOCAL_GENRES,
57    CONF_THUMB_CACHE_MAX_SIZE,
58    DEFAULT_LANGUAGE,
59    DEFAULT_THUMB_CACHE_MAX_SIZE_MB,
60    LOCALES,
61    METADATA_LOOKUP_TASK_ID_PREFIX,
62    METADATA_SCAN_BATCH_SIZE,
63    MISSING_ARTIST_METADATA_SCAN_TASK_ID,
64    PLAYLIST_METADATA_SCAN_TASK_ID,
65    REFRESH_INTERVAL,
66    THUMB_CACHE_CLEANUP_TASK_ID,
67)
68from .enrichment import MetadataEnrichmentMixin
69from .images import ImageProxyMixin
70from .radio import RadioArtworkMixin
71
72if TYPE_CHECKING:
73    from music_assistant_models.config_entries import CoreConfig
74    from music_assistant_models.media_items import (
75        Album,
76        Artist,
77        Audiobook,
78        MediaItemType,
79        Playlist,
80        Podcast,
81        Track,
82    )
83
84    from music_assistant import MusicAssistant
85    from music_assistant.controllers.music.media.base import MediaControllerBase
86    from music_assistant.helpers.json import SerializableType
87    from music_assistant.models.metadata_provider import MetadataProvider
88
89
90class MetaDataController(
91    ImageProxyMixin, RadioArtworkMixin, MetadataEnrichmentMixin, CoreController
92):
93    """Controller that handles metadata retrieval and management for media items."""
94
95    domain: str = "metadata"
96    config: CoreConfig
97
98    def __init__(self, mass: MusicAssistant) -> None:
99        """Initialize class."""
100        super().__init__(mass)
101        self.cache = self.mass.cache
102        self._pref_lang: str | None = None
103        self.manifest.name = "Metadata controller"
104        self.manifest.description = (
105            "Music Assistant's core controller which handles all metadata for music."
106        )
107        self.manifest.icon = "book-information-variant"
108        self._throttler = Throttler(1, 30)
109        # image-id bookkeeping, all bounded by _IMAGE_ID_LRU_MAX and sharing the
110        # same key/id string objects so the combined footprint stays small:
111        # - _image_id_forward: (provider, path) -> image_id memo so serializing a
112        #   known image skips the sha256 and the lock entirely. Read lock-free
113        #   (single dict lookup is atomic), mutated only while holding the lock.
114        # - _image_id_lru: image_id -> (provider, path). Write-through hot cache
115        #   in front of the cache controller so that resolving an image by id
116        #   never blocks on sqlite if the URL was generated recently.
117        # - _image_id_persisted: image_id -> epoch of the last persist to the
118        #   cache db, so repeat encounters skip the sqlite write.
119        # The lock is needed because compute_image_id() runs from the executor
120        # thread during outbound websocket serialization.
121        self._image_id_forward: dict[tuple[str, str], str] = {}
122        self._image_id_lru: OrderedDict[str, tuple[str, str]] = OrderedDict()
123        self._image_id_persisted: dict[str, float] = {}
124        self._image_id_lock = threading.Lock()
125        # corrupt metadata rows found by the last scan pass, per table, for diagnostics
126        self._corrupt_metadata_rows: dict[str, list[dict[str, str | int]]] = {}
127
128    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
129        """Return all Config Entries for this core module (if any)."""
130        return (
131            # deliberately without a default_value: only values differing from the entry default
132            # are persisted, so declaring one would make a chosen DEFAULT_LANGUAGE
133            # indistinguishable from "never chosen". The locale property applies it on read.
134            ConfigEntry(
135                key=CONF_LANGUAGE,
136                type=ConfigEntryType.STRING,
137                required=False,
138                options=[ConfigValueOption(key, title=value) for key, value in LOCALES.items()],
139            ),
140            ConfigEntry(
141                key=CONF_ENABLE_ONLINE_METADATA,
142                type=ConfigEntryType.BOOLEAN,
143                required=False,
144                default_value=True,
145            ),
146            ConfigEntry(
147                key=CONF_PREFER_LOCAL_GENRES,
148                type=ConfigEntryType.BOOLEAN,
149                required=False,
150                default_value=False,
151            ),
152            ConfigEntry(
153                key=CONF_ENABLE_RADIO_METADATA_LOOKUP,
154                type=ConfigEntryType.BOOLEAN,
155                required=False,
156                default_value=True,
157            ),
158            ConfigEntry(
159                key=CONF_THUMB_CACHE_MAX_SIZE,
160                type=ConfigEntryType.INTEGER,
161                required=False,
162                default_value=DEFAULT_THUMB_CACHE_MAX_SIZE_MB,
163                range=(50, 5000),
164            ),
165        )
166
167    async def setup(self, config: CoreConfig) -> None:
168        """Async initialize of module."""
169        self.config = config
170        if not self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
171            # silence PIL logger
172            logging.getLogger("PIL").setLevel(logging.WARNING)
173        # make sure that our directory with collage images exists
174        self._collage_images_dir = os.path.join(self.mass.cache_path, "collage_images")
175        if not await asyncio.to_thread(os.path.exists, self._collage_images_dir):
176            await asyncio.to_thread(os.mkdir, self._collage_images_dir)
177
178    async def post_setup(self) -> None:
179        """Handle logic after all core controllers have been set up."""
180        # canonical opaque-id endpoint, served by both the public webserver
181        # and the streams server (the latter is what player metadata URLs hit)
182        self.mass.streams.register_dynamic_route("/imageproxy/*", self.handle_imageproxy)
183        self.mass.webserver.register_dynamic_route("/imageproxy/*", self.handle_imageproxy)
184        self._register_maintenance_tasks()
185
186    async def close(self) -> None:
187        """Handle logic on server stop."""
188        self.mass.streams.unregister_dynamic_route("/imageproxy/*")
189        self.mass.webserver.unregister_dynamic_route("/imageproxy/*")
190
191    @property
192    def providers(self) -> list[MetadataProvider]:
193        """Return all loaded/running MetadataProviders."""
194        return sorted(
195            cast("list[MetadataProvider]", self.mass.get_providers(ProviderType.METADATA)),
196            key=lambda p: p.priority,
197        )
198
199    @property
200    def preferred_language(self) -> str:
201        """Return preferred language for metadata (as 2 letter language code 'en')."""
202        return self.locale.split("_")[0]
203
204    @property
205    def locale(self) -> str:
206        """Return preferred language for metadata (as full locale code 'en_EN')."""
207        value = self.mass.config.get_raw_core_config_value(
208            self.domain, CONF_LANGUAGE, DEFAULT_LANGUAGE
209        )
210        return str(value)
211
212    @api_command("metadata/set_default_preferred_language", required_scope=Scope.CONFIG_CORE_WRITE)
213    def set_default_preferred_language(self, lang: str) -> None:
214        """
215        Set the default preferred language.
216
217        Reasoning behind this is that the backend can not make a wise choice for the default,
218        so relies on some external source that knows better to set this info, like the frontend
219        or a streaming provider.
220        Can only be set once (by this call or the user).
221        """
222        if self.mass.config.get_raw_core_config_value(self.domain, CONF_LANGUAGE):
223            return  # already set
224        self.set_preferred_language(lang)
225
226    @api_command("metadata/set_preferred_language", required_scope=Scope.LIBRARY_MANAGE)
227    def set_preferred_language(self, lang: str) -> None:
228        """
229        Set the preferred language.
230
231        Note that this will not modify any existing metadata,
232        but will be used for future lookups.
233        """
234        # prefer exact match
235        if lang in LOCALES:
236            self.mass.config.set_raw_core_config_value(self.domain, CONF_LANGUAGE, lang)
237            return
238        # try strict matching on either locale code or region
239        lang = lang.lower().replace("-", "_")
240        for locale_code, lang_name in LOCALES.items():
241            if lang in (locale_code.lower(), lang_name.lower()):
242                self.mass.config.set_raw_core_config_value(self.domain, CONF_LANGUAGE, locale_code)
243                return
244        # attempt loose match on language code or region code
245        for lang_part in (lang[:2], lang[:-2]):
246            for locale_code in tuple(LOCALES):
247                language_code, region_code = locale_code.lower().split("_", 1)
248                if lang_part in (language_code, region_code):
249                    self.mass.config.set_raw_core_config_value(
250                        self.domain, CONF_LANGUAGE, locale_code
251                    )
252                    return
253        # if we reach this point, we couldn't match the language
254        self.logger.warning("%s is not a valid language", lang)
255
256    @api_command("metadata/update_metadata", required_scope=Scope.LIBRARY_MANAGE)
257    async def update_metadata(
258        self, item: str | MediaItemType, force_refresh: bool = False
259    ) -> MediaItemType:
260        """Get/update extra/enhanced metadata for/on given MediaItem."""
261        async with self.cache.handle_refresh(force_refresh):
262            if isinstance(item, str):
263                retrieved_item = await self.mass.music.get_item_by_uri(item)
264                if isinstance(retrieved_item, BrowseFolder):
265                    raise TypeError("Cannot update metadata on a BrowseFolder item.")
266                item = retrieved_item
267
268        if item.provider != "library":
269            # this shouldn't happen but just in case.
270            raise RuntimeError("Metadata can only be updated for library items")
271
272        async with self._throttler:
273            if item.media_type == MediaType.ARTIST:
274                await self._update_artist_metadata(
275                    cast("Artist", item), force_refresh=force_refresh
276                )
277            if item.media_type == MediaType.ALBUM:
278                await self._update_album_metadata(cast("Album", item), force_refresh=force_refresh)
279            if item.media_type == MediaType.TRACK:
280                await self._update_track_metadata(cast("Track", item), force_refresh=force_refresh)
281            if item.media_type == MediaType.PLAYLIST:
282                await self._update_playlist_metadata(
283                    cast("Playlist", item), force_refresh=force_refresh
284                )
285            if item.media_type == MediaType.AUDIOBOOK:
286                await self._update_audiobook_metadata(
287                    cast("Audiobook", item), force_refresh=force_refresh
288                )
289            if item.media_type == MediaType.PODCAST:
290                await self._update_podcast_metadata(
291                    cast("Podcast", item), force_refresh=force_refresh
292                )
293        return item
294
295    def schedule_update_metadata(self, item: MediaItemType) -> None:
296        """Schedule metadata update for given MediaItem."""
297        if item.provider != "library":
298            # this shouldn't happen but just in case.
299            return
300        last_refresh = item.metadata.last_refresh or 0
301        needs_update = (time() - last_refresh) > REFRESH_INTERVAL
302        if not needs_update:
303            return
304        assert item.uri is not None
305        task_id = self._get_metadata_lookup_task_id(item.uri)
306        _item = item
307
308        self.mass.tasks.run_background_task(
309            task_id=task_id,
310            name=f"Update metadata for {item.name}",
311            handler=lambda: self.update_metadata(_item),
312            translation_key="update_metadata",
313            translation_args=[item.name],
314            translation_owner=self.translation_owner,
315            metadata={
316                "task_domain": "metadata_lookup",
317                "item_uri": item.uri,
318            },
319        )
320
321    @api_command("metadata/get_track_lyrics", required_scope=Scope.LIBRARY_READ)
322    async def get_track_lyrics(
323        self,
324        track: Track,
325    ) -> tuple[str | None, str | None]:
326        """
327        Get lyrics for given track from metadata providers.
328
329        Returns a tuple of (lyrics, lrc_lyrics) if found.
330        """
331        lyrics, lrc_lyrics = await self._get_track_lyrics(track)
332        # on-demand lookups are not stored in the library db, so normalize on the way out
333        # promoting LRC formatted text stored in the plain lyrics tag
334        return lyrics, normalize_lrc_lyrics(lrc_lyrics or extract_lrc_lyrics(lyrics))
335
336    async def get_diagnostics(self) -> dict[str, SerializableType] | None:
337        """Return diagnostics info for this controller to include in diagnostics reports."""
338        if not self._corrupt_metadata_rows:
339            return None
340        return {"corrupt_metadata_rows": cast("SerializableType", self._corrupt_metadata_rows)}
341
342    async def _get_track_lyrics(
343        self,
344        track: Track,
345    ) -> tuple[str | None, str | None]:
346        """Look up (lyrics, lrc_lyrics) for the given track."""
347        if track.metadata and track.metadata.lyrics:
348            return track.metadata.lyrics, track.metadata.lrc_lyrics
349
350        if track.provider == "library":
351            # try to update metadata first
352            await self._update_track_metadata(track, force_refresh=False)
353            return track.metadata.lyrics, track.metadata.lrc_lyrics
354
355        # prefer lyrics from the track's own provider
356        track_provider = self.mass.get_provider(track.provider, provider_type=MusicProvider)
357        if track_provider and ProviderFeature.LYRICS in track_provider.supported_features:
358            full_track = await self.mass.music.tracks.get_provider_item(
359                track.item_id, track.provider
360            )
361            if full_track.metadata and full_track.metadata.lyrics:
362                return full_track.metadata.lyrics, full_track.metadata.lrc_lyrics
363
364        # fallback to other metadata providers
365        for provider in self.providers:
366            if ProviderFeature.LYRICS not in provider.supported_features:
367                continue
368            try:
369                metadata = await provider.get_track_metadata(track)
370            except Exception as err:
371                # a provider failure must not abort the lookup — skip to the next provider
372                self.logger.warning(
373                    "Error fetching lyrics for %s from provider %s: %s",
374                    track.name,
375                    provider.name,
376                    err,
377                    exc_info=err if self.logger.isEnabledFor(10) else None,
378                )
379                continue
380            if metadata and (metadata.lyrics or metadata.lrc_lyrics):
381                return metadata.lyrics, metadata.lrc_lyrics
382        return None, None
383
384    def _register_maintenance_tasks(self) -> None:
385        """Register the recurring metadata maintenance background tasks."""
386        # Spread across the full day so instances don't all hit the shared MusicBrainz mirror at once
387        utc_hour, utc_minute = divmod(random.randint(0, 24 * 60 - 1), 60)
388        desired_schedule = TaskSchedule.daily(hour=utc_hour, minute=utc_minute)
389        self.mass.tasks.register_scheduled_task(
390            task_id=MISSING_ARTIST_METADATA_SCAN_TASK_ID,
391            name="Scan missing artist metadata",
392            handler=self._scan_missing_artist_metadata,
393            schedule=desired_schedule,
394            translation_key="scan_missing_artist_metadata",
395            translation_owner=self.translation_owner,
396            metadata={"task_domain": "metadata_missing_artist_metadata_scan"},
397            allow_retry=True,
398        )
399        self.mass.tasks.register_scheduled_task(
400            task_id=PLAYLIST_METADATA_SCAN_TASK_ID,
401            name="Refresh playlist metadata",
402            handler=self._refresh_playlist_metadata_batch,
403            schedule=desired_schedule,
404            translation_key="refresh_playlist_metadata",
405            translation_owner=self.translation_owner,
406            metadata={"task_domain": "metadata_playlist_metadata_scan"},
407            allow_retry=True,
408        )
409        self.mass.tasks.register_scheduled_task(
410            task_id=THUMB_CACHE_CLEANUP_TASK_ID,
411            name="Cleanup thumbnail cache",
412            handler=self._cleanup_thumb_cache,
413            schedule=desired_schedule,
414            translation_key="cleanup_thumbnail_cache",
415            translation_owner=self.translation_owner,
416            metadata={"task_domain": "metadata_thumb_cache_cleanup"},
417            allow_retry=True,
418        )
419        # runs every hour rather than spread across the day: it is bounded to a small
420        # batch of albums per run, so there is no shared-mirror stampede to avoid
421        self.mass.tasks.register_scheduled_task(
422            task_id=ALBUM_RECONCILIATION_TASK_ID,
423            name="Reconcile duplicate albums",
424            handler=self._reconcile_duplicate_albums,
425            schedule=TaskSchedule.hourly(),
426            translation_key="reconcile_duplicate_albums",
427            translation_owner=self.translation_owner,
428            metadata={"task_domain": "metadata_album_reconciliation"},
429            allow_retry=True,
430        )
431
432    @staticmethod
433    def _get_metadata_lookup_task_id(uri: str) -> str:
434        """Return deterministic task id for a metadata lookup."""
435        return f"{METADATA_LOOKUP_TASK_ID_PREFIX}_{uuid5(NAMESPACE_URL, uri).hex}"
436
437    async def _scan_missing_artist_metadata(self) -> None:
438        """Scan for artists with missing metadata."""
439        update_current_task_progress_text("Searching for artists with missing metadata")
440        missing_images = (
441            f"(json_extract({DB_TABLE_ARTISTS}.metadata,'$.images') ISNULL "
442            f"OR json_extract({DB_TABLE_ARTISTS}.metadata,'$.images') = '[]')"
443        )
444        missing_description = f"json_extract({DB_TABLE_ARTISTS}.metadata,'$.description') ISNULL"
445        never_refreshed = f"json_extract({DB_TABLE_ARTISTS}.metadata,'$.last_refresh') ISNULL"
446        query = f"({missing_images} OR {missing_description}) AND {never_refreshed}"
447        artists = await self._get_scan_batch(self.mass.music.artists, DB_TABLE_ARTISTS, query)
448        if not artists:
449            update_current_task_progress_text("No artists with missing metadata found")
450            return
451        for index, artist in enumerate(artists, 1):
452            try:
453                update_current_task_progress_from_index(
454                    index,
455                    len(artists),
456                    f"Refreshing metadata for artist {index}/{len(artists)}: {artist.name}",
457                )
458                await self._update_artist_metadata(artist, force_refresh=False)
459            except Exception as err:
460                report_current_task_failure(f"{artist.name}: {err}")
461                self.logger.warning(
462                    "Error while updating artist metadata for %s: %s",
463                    artist.name,
464                    str(err),
465                    exc_info=err if self.logger.isEnabledFor(10) else None,
466                )
467        update_current_task_progress(100, f"Processed {len(artists)} artist(s)")
468
469    async def _refresh_playlist_metadata_batch(self) -> None:
470        """Refresh metadata for a small batch of library playlists."""
471        update_current_task_progress_text("Searching for playlists needing metadata refresh")
472        refresh_before = int(time() - REFRESH_INTERVAL)
473        query = (
474            f"{DB_TABLE_PLAYLISTS}.is_dynamic = 0 AND ("
475            f"json_extract({DB_TABLE_PLAYLISTS}.metadata,'$.last_refresh') ISNULL "
476            f"OR json_extract({DB_TABLE_PLAYLISTS}.metadata,'$.last_refresh') < {refresh_before})"
477        )
478        playlists = await self._get_scan_batch(self.mass.music.playlists, DB_TABLE_PLAYLISTS, query)
479        if not playlists:
480            update_current_task_progress_text("No playlists require metadata refresh")
481            return
482        for index, playlist in enumerate(playlists, 1):
483            try:
484                update_current_task_progress_from_index(
485                    index,
486                    len(playlists),
487                    f"Refreshing playlist metadata {index}/{len(playlists)}: {playlist.name}",
488                )
489                await self._update_playlist_metadata(playlist, force_refresh=False)
490            except Exception as err:
491                report_current_task_failure(f"{playlist.name}: {err}")
492                self.logger.warning(
493                    "Error while refreshing playlist metadata for %s: %s",
494                    playlist.name,
495                    str(err),
496                    exc_info=err if self.logger.isEnabledFor(10) else None,
497                )
498        update_current_task_progress(100, f"Processed {len(playlists)} playlist(s)")
499
500    async def _reconcile_duplicate_albums(self) -> None:
501        """Enrich and re-match a small batch of sparse or possibly duplicated albums."""
502        update_current_task_progress_text("Searching for albums needing reconciliation")
503        # candidates keep retrying at the normal REFRESH_INTERVAL cadence (e.g. after a
504        # transient provider outage), rather than only ever once
505        refresh_before = int(time() - REFRESH_INTERVAL)
506        query = (
507            f"({DB_TABLE_ALBUMS}.album_type = '{AlbumType.UNKNOWN.value}' "
508            f"OR {_duplicate_album_sibling_guard()}) AND ("
509            f"json_extract({DB_TABLE_ALBUMS}.metadata,'$.last_refresh') ISNULL "
510            f"OR json_extract({DB_TABLE_ALBUMS}.metadata,'$.last_refresh') < {refresh_before})"
511        )
512        albums = await self._get_scan_batch(self.mass.music.albums, DB_TABLE_ALBUMS, query)
513        if not albums:
514            update_current_task_progress_text("No albums require reconciliation")
515            return
516        for index, album in enumerate(albums, 1):
517            try:
518                update_current_task_progress_from_index(
519                    index,
520                    len(albums),
521                    f"Reconciling album {index}/{len(albums)}: {album.name}",
522                )
523                # enrich sparse provider data (type/year/metadata) first so the follow-up
524                # match has full album details to work with, then re-fetch the now-enriched
525                # library row before re-matching: match_providers merges a confirmed mapping
526                # into an existing duplicate through the safe add_provider_mappings path
527                try:
528                    await self._update_album_metadata(album, force_refresh=False)
529                    reconciled_album = await self.mass.music.albums.get_library_item(album.item_id)
530                except MediaNotFoundError:
531                    # both rows of a duplicate pair can share a batch, so this row may
532                    # already have been merged into its duplicate earlier in the run
533                    continue
534                await self.mass.music.albums.match_providers(reconciled_album)
535            except (MusicAssistantError, aiohttp.ClientError, TimeoutError) as err:
536                report_current_task_failure(f"{album.name}: {err}")
537                self.logger.warning(
538                    "Error while reconciling album %s: %s",
539                    album.name,
540                    str(err),
541                    exc_info=err if self.logger.isEnabledFor(10) else None,
542                )
543        update_current_task_progress(100, f"Processed {len(albums)} album(s)")
544
545    async def _cleanup_thumb_cache(self) -> None:
546        """Remove oldest thumbnails when the cache folder exceeds the configured limit."""
547        max_size_mb = (
548            try_parse_int(
549                self.config.get_value(CONF_THUMB_CACHE_MAX_SIZE), DEFAULT_THUMB_CACHE_MAX_SIZE_MB
550            )
551            or DEFAULT_THUMB_CACHE_MAX_SIZE_MB
552        )
553        removed = await cleanup_thumb_cache(self.mass.cache_path, max_size_mb * 1024 * 1024)
554        if removed:
555            self.logger.debug("Thumbnail cache cleanup: removed %s file(s)", removed)
556
557    async def _get_scan_batch[ItemCls: MediaItemType](
558        self,
559        media_controller: MediaControllerBase[ItemCls],
560        table: str,
561        query: str,
562    ) -> list[ItemCls]:
563        """Fetch a metadata-scan batch, tolerating rows with corrupt metadata JSON."""
564        try:
565            items = await media_controller.get_library_items_by_query(
566                limit=METADATA_SCAN_BATCH_SIZE,
567                order_by="random",
568                extra_query_parts=[query],
569            )
570        except sqlite3.OperationalError as err:
571            if "malformed JSON" not in str(err):
572                raise
573            await self._report_corrupt_metadata_rows(table)
574            return await media_controller.get_library_items_by_query(
575                limit=METADATA_SCAN_BATCH_SIZE,
576                order_by="random",
577                extra_query_parts=[f"{_valid_metadata_guard(table)} AND {query}"],
578            )
579        # a clean scan proves the table currently holds no corrupt rows
580        self._corrupt_metadata_rows.pop(table, None)
581        return items
582
583    async def _report_corrupt_metadata_rows(self, table: str) -> None:
584        """Report library rows whose metadata column holds invalid JSON."""
585        rows = await self.mass.music.database.get_rows_from_query(
586            f"SELECT item_id, name FROM {table} "
587            f"WHERE {table}.metadata IS NOT NULL AND NOT json_valid({table}.metadata)",
588            limit=25,
589        )
590        # keep the findings for the diagnostics report, replacing the previous
591        # pass so repaired rows drop out again
592        if rows:
593            self._corrupt_metadata_rows[table] = [
594                {"item_id": row["item_id"], "name": row["name"]} for row in rows
595            ]
596        else:
597            self._corrupt_metadata_rows.pop(table, None)
598        for row in rows:
599            message = (
600                f"'{row['name']}' has corrupt metadata and was skipped. To repair, remove "
601                f"'{row['name']}' from the library; it will be re-added with fresh metadata "
602                f"on the next library sync ({table} id {row['item_id']})."
603            )
604            report_current_task_failure(message)
605            self.logger.warning(message)
606
607
608def _duplicate_album_sibling_guard() -> str:
609    """Return a query part that selects albums which may be a duplicate of another library row."""
610    shares_artist = (
611        f"EXISTS (SELECT 1 FROM {DB_TABLE_ALBUM_ARTISTS} own "
612        f"JOIN {DB_TABLE_ALBUM_ARTISTS} other ON other.artist_id = own.artist_id "
613        f"WHERE own.album_id = {DB_TABLE_ALBUMS}.item_id AND other.album_id = dup.item_id)"
614    )
615    # a title that normalizes to nothing (e.g. Ed Sheeran's '+', '=' and '÷') matches every
616    # other such title, so those fall back to their raw spelling like the album comparison does
617    same_title = (
618        f"({DB_TABLE_ALBUMS}.search_name != '' OR "
619        f"REPLACE({DB_TABLE_ALBUMS}.name,' ','') = REPLACE(dup.name,' ',''))"
620    )
621    # deliberately an identity-only pre-filter: which editions may be merged is decided by
622    # the album comparison, which escalates an ambiguous edition to tracklists and
623    # MusicBrainz and rejects a recording-changing one (live, remix, ...) outright
624    return (
625        f"EXISTS (SELECT 1 FROM {DB_TABLE_ALBUMS} dup "
626        f"WHERE dup.item_id != {DB_TABLE_ALBUMS}.item_id "
627        f"AND dup.search_name = {DB_TABLE_ALBUMS}.search_name "
628        f"AND {same_title} AND {shares_artist})"
629    )
630
631
632def _valid_metadata_guard(table: str) -> str:
633    """Return a query part that excludes rows with invalid JSON in the metadata column."""
634    # sqlite's json functions raise a fatal 'malformed JSON' error on invalid input,
635    # which would fail the entire scan query because of a single corrupt row
636    return f"({table}.metadata IS NULL OR json_valid({table}.metadata))"
637