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