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