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