/
/
/
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, WEB_BASE_URL
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 all_tidal_configs = await self.mass.config.get_provider_configs(ProviderType.MUSIC)
54 tidal_configs = [
55 config for config in all_tidal_configs if config.domain == self.provider.domain
56 ]
57 sorted_instances = sorted(tidal_configs, key=lambda x: x.instance_id)
58 show_user_identifier = len(sorted_instances) > 1
59
60 for page_path in pages:
61 parser = await self.get_page_content(page_path)
62 page_name = page_path.split("/")[-1].replace("_", " ").title()
63
64 for module_info in parser.modules:
65 title = module_info.get("title", "Unknown")
66 if not title or title == "Unknown" or "Videos" in title:
67 continue
68
69 items, content_type = parser.get_module_items(module_info)
70 if not items:
71 continue
72
73 key = f"{self.auth.user_id}_{title}"
74 if key not in combined_modules:
75 combined_modules[key] = []
76 module_content_types[key] = content_type
77 module_page_names[key] = page_name
78
79 combined_modules[key].extend(items)
80
81 for key, items in combined_modules.items():
82 user_id_prefix = f"{self.auth.user_id}_"
83 title = key.removeprefix(user_id_prefix)
84
85 unique_items = UniqueList(items)
86 item_id = "".join(c for c in key.lower().replace(" ", "_") if c.isalnum() or c == "_")
87 content_type = module_content_types.get(key, MediaType.PLAYLIST)
88 page_name = module_page_names.get(key, "Tidal")
89
90 folder_name = title
91 if show_user_identifier:
92 raw_user_name = (
93 self.auth.user.profile_name
94 or self.auth.user.user_name
95 or str(self.auth.user_id)
96 )
97 user_name = raw_user_name.split("@", 1)[0]
98 folder_name = f"{title} ({user_name})"
99
100 results.append(
101 RecommendationFolder(
102 item_id=item_id,
103 name=folder_name,
104 provider=self.provider.instance_id,
105 items=UniqueList[MediaItemType | ItemMapping | BrowseFolder](unique_items),
106 subtitle=f"From {page_name} ⢠{len(unique_items)} items",
107 translation_key=item_id,
108 icon="mdi-playlist-music"
109 if content_type == MediaType.PLAYLIST
110 else "mdi-album",
111 )
112 )
113
114 return results
115
116 async def get_page_content(self, page_path: str = "pages/home") -> TidalPageParser:
117 """Get a lazy page parser for a Tidal page."""
118 if cached := await TidalPageParser.from_cache(self.provider, page_path):
119 return cached
120
121 # Let fetch/parse errors propagate: swallowing them here would make
122 # recommendations() succeed with an empty result, which the caching
123 # wrapper would then store for the full TTL.
124 locale = self.mass.metadata.locale.replace("_", "-")
125 data = await self.api.get(
126 page_path,
127 base_url=WEB_BASE_URL,
128 params={
129 "locale": locale,
130 "deviceType": "BROWSER",
131 "countryCode": self.auth.country_code or "US",
132 },
133 )
134
135 parser = TidalPageParser(self.provider)
136 parser.parse_page_structure(data or {}, page_path)
137
138 await self.mass.cache.set(
139 key=page_path,
140 data=parser.to_cache(),
141 provider=self.provider.instance_id,
142 category=CACHE_CATEGORY_RECOMMENDATIONS,
143 expiration=self.page_cache_ttl,
144 )
145 return parser
146