music-assistant-server

1"""Parser for Tidal page structures with lazy loading."""
2
3from __future__ import annotations
4
5import json
6import time
7from typing import TYPE_CHECKING, Any
8
9from music_assistant_models.enums import MediaType
10
11from .constants import CACHE_CATEGORY_RECOMMENDATIONS, SKIPPABLE_ITEM_ERRORS
12from .parsers import parse_album, parse_artist, parse_playlist, parse_track
13
14if TYPE_CHECKING:
15    from music_assistant_models.media_items import Album, Artist, Playlist, Track
16
17    from .provider import TidalProvider
18
19
20class TidalPageParser:
21    """Parser for Tidal page structures with lazy loading."""
22
23    def __init__(self, provider: TidalProvider) -> None:
24        """Initialize the parser with the Tidal provider instance."""
25        self.provider = provider
26        self.logger = provider.logger
27        self._content_map: dict[str, dict[str, Any]] = {
28            "MIX": {},
29            "PLAYLIST": {},
30            "ALBUM": {},
31            "TRACK": {},
32            "ARTIST": {},
33        }
34        self._module_map: list[dict[str, Any]] = []
35        self._page_path: str | None = None
36        self._parsed_at: int = 0
37
38    @property
39    def modules(self) -> list[dict[str, Any]]:
40        """Return the parsed page modules."""
41        return self._module_map
42
43    def to_cache(self) -> dict[str, Any]:
44        """Return the parsed page state as a cacheable dict."""
45        return {
46            "module_map": self._module_map,
47            "content_map": self._content_map,
48            "parsed_at": self._parsed_at,
49        }
50
51    def parse_page_structure(self, page_data: dict[str, Any], page_path: str) -> None:
52        """Parse Tidal page structure into indexed modules."""
53        self._page_path = page_path
54        self._parsed_at = int(time.time())
55        self._module_map = []
56
57        # Extract modules from rows
58        module_idx = 0
59        for row_idx, row in enumerate(page_data.get("rows", [])):
60            for module in row.get("modules", []):
61                # Store basic module info for later processing
62                module_info = {
63                    "title": module.get("title", ""),
64                    "type": module.get("type", ""),
65                    "raw_data": module,
66                    "module_idx": module_idx,
67                    "row_idx": row_idx,
68                }
69                self._module_map.append(module_info)
70                module_idx += 1
71
72    def get_module_items(
73        self, module_info: dict[str, Any]
74    ) -> tuple[list[Playlist | Album | Track | Artist], MediaType]:
75        """Extract media items from a module with simplified type handling."""
76        result: list[Playlist | Album | Track | Artist] = []
77        type_counts: dict[MediaType, int] = {
78            MediaType.PLAYLIST: 0,
79            MediaType.ALBUM: 0,
80            MediaType.TRACK: 0,
81            MediaType.ARTIST: 0,
82        }
83
84        module_data = module_info.get("raw_data", {})
85        module_type = module_data.get("type", "")
86
87        self.logger.debug(
88            "Processing module type: %s, title: %s",
89            module_type,
90            module_data.get("title", "Unknown"),
91        )
92
93        # Process module based on type
94        self._process_module_by_type(module_data, module_type, result, type_counts)
95
96        # Determine the primary content type based on counts
97        primary_type = self._determine_primary_type(type_counts)
98
99        self._log_module_results(module_data, result, type_counts)
100
101        return result, primary_type
102
103    def _process_module_by_type(
104        self,
105        module_data: dict[str, Any],
106        module_type: str,
107        result: list[Playlist | Album | Track | Artist],
108        type_counts: dict[MediaType, int],
109    ) -> None:
110        """Process module content based on module type."""
111        # Extract paged list if present (most modules have this)
112        paged_list = module_data.get("pagedList", {})
113        items = paged_list.get("items", [])
114
115        # Different module types have different content structures
116        if module_type == "PLAYLIST_LIST":
117            self._process_playlist_list(items, result, type_counts)
118        elif module_type == "TRACK_LIST":
119            self._process_track_list(items, result, type_counts)
120        elif module_type == "ALBUM_LIST":
121            self._process_album_list(items, result, type_counts)
122        elif module_type == "ARTIST_LIST":
123            self._process_artist_list(items, result, type_counts)
124        elif module_type == "MIX_LIST":
125            self._process_mix_list(items, result, type_counts)
126        elif module_type == "HIGHLIGHT_MODULE":
127            self._process_highlight_module(module_data, result, type_counts)
128        else:
129            # Generic fallback for other module types
130            self._process_generic_items(items, result, type_counts)
131
132    def _process_playlist_list(
133        self,
134        items: list[dict[str, Any]],
135        result: list[Playlist | Album | Track | Artist],
136        type_counts: dict[MediaType, int],
137    ) -> None:
138        """Process items from a PLAYLIST_LIST module."""
139        for item in items:
140            if isinstance(item, dict):
141                # Check if item appears to be a mix
142                is_mix = "mixId" in item or "mixType" in item
143
144                try:
145                    playlist = parse_playlist(self.provider, item, is_mix=is_mix)
146                    result.append(playlist)
147                    type_counts[MediaType.PLAYLIST] += 1
148                except SKIPPABLE_ITEM_ERRORS as err:
149                    self.logger.warning("Error parsing playlist: %s", err)
150
151    def _process_track_list(
152        self,
153        items: list[dict[str, Any]],
154        result: list[Playlist | Album | Track | Artist],
155        type_counts: dict[MediaType, int],
156    ) -> None:
157        """Process items from a TRACK_LIST module."""
158        for item in items:
159            if isinstance(item, dict):
160                try:
161                    track = parse_track(self.provider, item)
162                    result.append(track)
163                    type_counts[MediaType.TRACK] += 1
164                except SKIPPABLE_ITEM_ERRORS as err:
165                    self.logger.warning("Error parsing track: %s", err)
166
167    def _process_album_list(
168        self,
169        items: list[dict[str, Any]],
170        result: list[Playlist | Album | Track | Artist],
171        type_counts: dict[MediaType, int],
172    ) -> None:
173        """Process items from an ALBUM_LIST module."""
174        for item in items:
175            if isinstance(item, dict):
176                try:
177                    album = parse_album(self.provider, item)
178                    result.append(album)
179                    type_counts[MediaType.ALBUM] += 1
180                except SKIPPABLE_ITEM_ERRORS as err:
181                    self.logger.warning("Error parsing album: %s", err)
182
183    def _process_artist_list(
184        self,
185        items: list[dict[str, Any]],
186        result: list[Playlist | Album | Track | Artist],
187        type_counts: dict[MediaType, int],
188    ) -> None:
189        """Process items from an ARTIST_LIST module."""
190        for item in items:
191            if isinstance(item, dict):
192                try:
193                    artist = parse_artist(self.provider, item)
194                    result.append(artist)
195                    type_counts[MediaType.ARTIST] += 1
196                except SKIPPABLE_ITEM_ERRORS as err:
197                    self.logger.warning("Error parsing artist: %s", err)
198
199    def _process_mix_list(
200        self,
201        items: list[dict[str, Any]],
202        result: list[Playlist | Album | Track | Artist],
203        type_counts: dict[MediaType, int],
204    ) -> None:
205        """Process items from a MIX_LIST module."""
206        for item in items:
207            if isinstance(item, dict):
208                try:
209                    mix = parse_playlist(self.provider, item, is_mix=True)
210                    result.append(mix)
211                    type_counts[MediaType.PLAYLIST] += 1
212                except SKIPPABLE_ITEM_ERRORS as err:
213                    self.logger.warning("Error parsing mix: %s", err)
214
215    def _process_generic_items(
216        self,
217        items: list[dict[str, Any]],
218        result: list[Playlist | Album | Track | Artist],
219        type_counts: dict[MediaType, int],
220    ) -> None:
221        """Process items with generic type detection."""
222        for item in items:
223            if isinstance(item, dict):
224                # Try to determine item type from structure
225                try:
226                    parsed_item = self._parse_item(item, type_counts)
227                    if parsed_item:
228                        result.append(parsed_item)
229                except SKIPPABLE_ITEM_ERRORS as err:
230                    self.logger.warning("Error parsing generic item: %s", err)
231
232    def _log_module_results(
233        self,
234        module_data: dict[str, Any],
235        result: list[Playlist | Album | Track | Artist],
236        type_counts: dict[MediaType, int],
237    ) -> None:
238        """Log detailed module processing results."""
239        self.logger.debug(
240            "Module '%s' processed: %d items (%d playlists, %d albums, %d tracks, %d artists)",
241            module_data.get("title", "Unknown"),
242            len(result),
243            type_counts[MediaType.PLAYLIST],
244            type_counts[MediaType.ALBUM],
245            type_counts[MediaType.TRACK],
246            type_counts[MediaType.ARTIST],
247        )
248
249    def _determine_primary_type(self, type_counts: dict[MediaType, int]) -> MediaType:
250        """Determine the primary media type based on item counts."""
251        primary_type = MediaType.PLAYLIST  # Default
252        max_count = 0
253        for media_type, count in type_counts.items():
254            if count > max_count:
255                max_count = count
256                primary_type = media_type
257        return primary_type
258
259    def _process_highlight_module(
260        self,
261        module_data: dict[str, Any],
262        result: list[Playlist | Album | Track | Artist],
263        type_counts: dict[MediaType, int],
264    ) -> None:
265        """Process highlights from a HIGHLIGHT_MODULE."""
266        highlights = module_data.get("highlight", [])
267        for highlight in highlights:
268            if isinstance(highlight, dict):  # Make sure highlight is a dict
269                highlight_item = highlight.get("item", {})
270                highlight_type = highlight.get("type", "")
271                if isinstance(highlight_item, dict):
272                    if parsed_item := self._parse_item(highlight_item, type_counts, highlight_type):
273                        result.append(parsed_item)
274
275    def _process_paged_list(
276        self,
277        module_data: dict[str, Any],
278        module_type: str,
279        result: list[Playlist | Album | Track | Artist],
280        type_counts: dict[MediaType, int],
281    ) -> None:
282        """Process items from a paged list module."""
283        paged_list = module_data.get("pagedList", {})
284        items = paged_list.get("items", [])
285
286        # Handle module-specific type inference
287        inferred_type: str | None = None
288        if module_type in {"ALBUM_LIST", "TRACK_LIST", "PLAYLIST_LIST", "MIX_LIST"}:
289            inferred_type = module_type.replace("_LIST", "")
290
291        # Process each item
292        for item in items:
293            if not item or not isinstance(item, dict):
294                continue
295
296            # Use inferred type if no explicit type
297            item_type = item.get("type", inferred_type) or ""
298            if parsed_item := self._parse_item(item, type_counts, item_type):
299                result.append(parsed_item)
300
301    def _parse_item(
302        self,
303        item: dict[str, Any],
304        type_counts: dict[MediaType, int],
305        item_type: str = "",
306    ) -> Playlist | Album | Track | Artist | None:
307        """
308        Parse a single item from Tidal data into a media item.
309
310        Args:
311            item: Dictionary containing item data
312            type_counts: Dictionary to track counts by media type
313            item_type: Optional item type hint
314
315        Returns:
316            Parsed media item or None if parsing failed
317        """
318        # Handle nested item structure
319        if not item_type and isinstance(item, dict) and "type" in item and "item" in item:
320            item_type = item["type"]
321            item = item["item"]
322
323        # If no explicit type, try to infer from structure
324        if not item_type:
325            if "mixId" in item or "mixType" in item:
326                item_type = "MIX"
327            elif "uuid" in item:
328                item_type = "PLAYLIST"
329            elif "id" in item and "duration" in item and "album" in item:
330                item_type = "TRACK"
331            elif "id" in item and "numberOfTracks" in item and "artists" in item:
332                item_type = "ALBUM"
333            elif "id" in item and "picture" in item and "name" in item and "album" not in item:
334                item_type = "ARTIST"
335
336        # Parse based on detected type
337        try:
338            if item_type == "MIX":
339                media_item: Playlist | Album | Track | Artist = parse_playlist(
340                    self.provider, item, is_mix=True
341                )
342                type_counts[MediaType.PLAYLIST] += 1
343                return media_item
344            if item_type == "PLAYLIST":
345                media_item = parse_playlist(self.provider, item)
346                type_counts[MediaType.PLAYLIST] += 1
347                return media_item
348            if item_type == "ALBUM":
349                media_item = parse_album(self.provider, item)
350                type_counts[MediaType.ALBUM] += 1
351                return media_item
352            if item_type == "TRACK":
353                media_item = parse_track(self.provider, item)
354                type_counts[MediaType.TRACK] += 1
355                return media_item
356            if item_type == "ARTIST":
357                media_item = parse_artist(self.provider, item)
358                type_counts[MediaType.ARTIST] += 1
359                return media_item
360            # Last resort - try to infer from structure for unlabeled items
361            if "uuid" in item:
362                media_item = parse_playlist(self.provider, item)
363                type_counts[MediaType.PLAYLIST] += 1
364                return media_item
365            if "id" in item and "title" in item and "duration" in item:
366                media_item = parse_track(self.provider, item)
367                type_counts[MediaType.TRACK] += 1
368                return media_item
369            if "id" in item and "title" in item and "numberOfTracks" in item:
370                media_item = parse_album(self.provider, item)
371                type_counts[MediaType.ALBUM] += 1
372                return media_item
373
374            self.logger.warning("Unknown item type, could not parse: %s", item)
375            return None
376
377        except SKIPPABLE_ITEM_ERRORS as err:
378            self.logger.debug("Error parsing %s item: %s", item_type, err)
379            return None
380        except (json.JSONDecodeError, UnicodeError) as err:
381            self.logger.debug("JSON/Unicode error parsing %s item: %s", item_type, err)
382            return None
383
384    @classmethod
385    async def from_cache(cls, provider: TidalProvider, page_path: str) -> TidalPageParser | None:
386        """Create a parser instance from cached data if available and valid."""
387        cached_data = await provider.mass.cache.get(
388            page_path,
389            provider=provider.instance_id,
390            category=CACHE_CATEGORY_RECOMMENDATIONS,
391        )
392        if not cached_data:
393            return None
394
395        parser = cls(provider)
396        parser._page_path = page_path
397        parser._module_map = cached_data.get("module_map", [])
398        parser._content_map = cached_data.get("content_map", {})
399        parser._parsed_at = cached_data.get("parsed_at", 0)
400
401        return parser
402
403    @property
404    def content_stats(self) -> dict[str, int | float]:
405        """Get statistics about the parsed content."""
406        stats = {
407            "modules": len(self._module_map),
408            "cache_age_minutes": (time.time() - self._parsed_at) / 60,
409        }
410
411        for media_type, items in self._content_map.items():
412            stats[f"{media_type.lower()}_count"] = len(items)
413
414        return stats
415