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