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