music-assistant-server
5.8 KB•PY
recommendations.py
5.8 KB • 157 lines • python
1"""Recommendation logic for Tidal."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6
7from music_assistant_models.enums import MediaType, ProviderType
8from music_assistant_models.media_items import (
9 Album,
10 Artist,
11 BrowseFolder,
12 ItemMapping,
13 MediaItemType,
14 Playlist,
15 RecommendationFolder,
16 Track,
17 UniqueList,
18)
19
20from .constants import CACHE_CATEGORY_RECOMMENDATIONS
21from .tidal_page_parser import TidalPageParser
22
23if TYPE_CHECKING:
24 from .provider import TidalProvider
25
26
27class TidalRecommendationManager:
28 """Manages Tidal recommendations."""
29
30 def __init__(self, provider: TidalProvider):
31 """Initialize recommendation manager."""
32 self.provider = provider
33 self.api = provider.api
34 self.auth = provider.auth
35 self.logger = provider.logger
36 self.mass = provider.mass
37 self.page_cache_ttl = 3 * 3600
38
39 async def get_recommendations(self) -> list[RecommendationFolder]:
40 """Get this provider's recommendations organized into folders."""
41 results: list[RecommendationFolder] = []
42 pages = [
43 "pages/home",
44 "pages/for_you",
45 "pages/hi_res",
46 "pages/explore_new_music",
47 "pages/explore_top_music",
48 ]
49 combined_modules: dict[str, list[Playlist | Album | Track | Artist]] = {}
50 module_content_types: dict[str, MediaType] = {}
51 module_page_names: dict[str, str] = {}
52
53 try:
54 all_tidal_configs = await self.mass.config.get_provider_configs(ProviderType.MUSIC)
55 tidal_configs = [
56 config for config in all_tidal_configs if config.domain == self.provider.domain
57 ]
58 sorted_instances = sorted(tidal_configs, key=lambda x: x.instance_id)
59 show_user_identifier = len(sorted_instances) > 1
60
61 for page_path in pages:
62 parser = await self.get_page_content(page_path)
63 page_name = page_path.split("/")[-1].replace("_", " ").title()
64
65 for module_info in parser._module_map:
66 title = module_info.get("title", "Unknown")
67 if not title or title == "Unknown" or "Videos" in title:
68 continue
69
70 items, content_type = parser.get_module_items(module_info)
71 if not items:
72 continue
73
74 key = f"{self.auth.user_id}_{title}"
75 if key not in combined_modules:
76 combined_modules[key] = []
77 module_content_types[key] = content_type
78 module_page_names[key] = page_name
79
80 combined_modules[key].extend(items)
81
82 for key, items in combined_modules.items():
83 user_id_prefix = f"{self.auth.user_id}_"
84 title = key.removeprefix(user_id_prefix)
85
86 unique_items = UniqueList(items)
87 item_id = "".join(
88 c for c in key.lower().replace(" ", "_") if c.isalnum() or c == "_"
89 )
90 content_type = module_content_types.get(key, MediaType.PLAYLIST)
91 page_name = module_page_names.get(key, "Tidal")
92
93 folder_name = title
94 if show_user_identifier:
95 raw_user_name = (
96 self.auth.user.profile_name
97 or self.auth.user.user_name
98 or str(self.auth.user_id)
99 )
100 user_name = raw_user_name.split("@", 1)[0]
101 folder_name = f"{title} ({user_name})"
102
103 results.append(
104 RecommendationFolder(
105 item_id=item_id,
106 name=folder_name,
107 provider=self.provider.instance_id,
108 items=UniqueList[MediaItemType | ItemMapping | BrowseFolder](unique_items),
109 subtitle=f"From {page_name} ⢠{len(unique_items)} items",
110 translation_key=item_id,
111 icon="mdi-playlist-music"
112 if content_type == MediaType.PLAYLIST
113 else "mdi-album",
114 )
115 )
116
117 except Exception as err:
118 self.logger.warning("Error fetching recommendations: %s", err)
119
120 return results
121
122 async def get_page_content(self, page_path: str = "pages/home") -> TidalPageParser:
123 """Get a lazy page parser for a Tidal page."""
124 if cached := await TidalPageParser.from_cache(self.provider, page_path):
125 return cached
126
127 try:
128 locale = self.mass.metadata.locale.replace("_", "-")
129 api_result = await self.api.get(
130 page_path,
131 base_url="https://listen.tidal.com/v1",
132 params={
133 "locale": locale,
134 "deviceType": "BROWSER",
135 "countryCode": self.auth.country_code or "US",
136 },
137 )
138
139 data = api_result[0] if isinstance(api_result, tuple) else api_result
140 parser = TidalPageParser(self.provider)
141 parser.parse_page_structure(data or {}, page_path)
142
143 await self.mass.cache.set(
144 key=page_path,
145 data={
146 "module_map": parser._module_map,
147 "content_map": parser._content_map,
148 "parsed_at": parser._parsed_at,
149 },
150 provider=self.provider.instance_id,
151 category=CACHE_CATEGORY_RECOMMENDATIONS,
152 expiration=self.page_cache_ttl,
153 )
154 return parser
155 except Exception:
156 return TidalPageParser(self.provider)
157