music-assistant-server

1"""Recommendations subcontroller: aggregates library + provider recommendation rows."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from itertools import zip_longest
8from typing import TYPE_CHECKING, cast
9
10from music_assistant_models.auth import Scope
11from music_assistant_models.enums import ProviderFeature
12from music_assistant_models.media_items import UniqueList
13
14from music_assistant.constants import MASS_LOGGER_NAME
15from music_assistant.controllers.music.constants import (
16    RECOMMENDATIONS_ITEMS_TIMEOUT,
17    RECOMMENDATIONS_ROWS_TIMEOUT,
18)
19
20if TYPE_CHECKING:
21    from music_assistant_models.media_items import (
22        BrowseFolder,
23        ItemMapping,
24        MediaItemType,
25        RecommendationFolder,
26    )
27
28    from music_assistant.mass import MusicAssistant
29    from music_assistant.models.metadata_provider import MetadataProvider
30    from music_assistant.models.music_provider import MusicProvider
31    from music_assistant.models.plugin import PluginProvider
32
33
34class RecommendationsController:
35    """Serves the recommendations API: default library rows plus provider rows."""
36
37    def __init__(self, mass: MusicAssistant) -> None:
38        """Initialize the controller and register its api commands."""
39        self.mass = mass
40        self.logger = logging.getLogger(f"{MASS_LOGGER_NAME}.music.recommendations")
41        self.mass.register_api_command(
42            "music/recommendations",
43            self.get_recommendations,
44            required_scope=Scope.LIBRARY_READ,
45        )
46        self.mass.register_api_command(
47            "music/recommendations/items",
48            self.get_recommendation_items,
49            required_scope=Scope.LIBRARY_READ,
50        )
51
52    async def get_recommendations(self) -> list[RecommendationFolder]:
53        """Get all available recommendation rows (library + providers, interleaved), without items."""
54        providers = self.mass.music._apply_user_provider_filter(
55            self.mass.get_providers_supporting_feature(ProviderFeature.RECOMMENDATIONS)
56        )
57        rows_per_source: list[list[RecommendationFolder]] = [
58            *await asyncio.gather(
59                *[
60                    self._provider_rows(
61                        cast("MusicProvider | MetadataProvider | PluginProvider", provider)
62                    )
63                    for provider in providers
64                ]
65            ),
66        ]
67        # interleave: one folder per source per pass, preserving each source's ordering
68        return [item for sublist in zip_longest(*rows_per_source) for item in sublist if item]
69
70    async def get_recommendation_items(
71        self, provider: str, item_id: str
72    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
73        """
74        Get the items for a single recommendation row.
75
76        :param provider: The provider instance id owning the row.
77        :param item_id: The item_id of the row, as returned by the recommendations listing.
78        """
79        try:
80            prov = self.mass.get_provider(provider)
81            # re-apply the user provider filter the rows listing applies, so a user
82            # can not fetch items from a music provider an admin has restricted them from
83            if prov is None or not self.mass.music._apply_user_provider_filter([prov]):
84                return UniqueList()
85            if ProviderFeature.RECOMMENDATIONS not in prov.supported_features:
86                # keep the base-model guarantee that this method is only called for
87                # providers declaring the feature, matching the rows listing
88                return UniqueList()
89            async with asyncio.timeout(RECOMMENDATIONS_ITEMS_TIMEOUT):
90                return await cast(
91                    "MusicProvider | MetadataProvider | PluginProvider", prov
92                ).get_recommendation_items(item_id)
93        except TimeoutError:
94            self.logger.warning(
95                "Timeout while fetching recommendation items for %s/%s; skipping",
96                provider,
97                item_id,
98            )
99            return UniqueList()
100        except Exception as err:
101            self.logger.warning(
102                "Error while fetching recommendation items for %s/%s: %s",
103                provider,
104                item_id,
105                str(err),
106                exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
107            )
108            return UniqueList()
109
110    async def _provider_rows(
111        self, provider: MusicProvider | MetadataProvider | PluginProvider
112    ) -> list[RecommendationFolder]:
113        """Return a provider's recommendation rows, or an empty list if it times out or raises."""
114        try:
115            async with asyncio.timeout(RECOMMENDATIONS_ROWS_TIMEOUT):
116                return await provider.get_recommendations()
117        except TimeoutError:
118            self.logger.warning(
119                "Timeout while fetching recommendation rows from %s; skipping for this request",
120                provider.name,
121            )
122            return []
123        except Exception as err:
124            self.logger.warning(
125                "Error while fetching recommendation rows from %s: %s",
126                provider.name,
127                str(err),
128                exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
129            )
130            return []
131