/
/
/
1"""MixIn for NicovideoMusicProvider: search and recommendations methods."""
2
3from __future__ import annotations
4
5from typing import override
6
7from music_assistant_models.enums import MediaType
8from music_assistant_models.media_items import (
9 BrowseFolder,
10 ItemMapping,
11 MediaItemType,
12 RecommendationFolder,
13 SearchResults,
14 Track,
15)
16from music_assistant_models.unique_list import UniqueList
17
18from music_assistant.controllers.cache import use_cache
19from music_assistant.providers.nicovideo.provider_mixins.base import (
20 NicovideoMusicProviderMixinBase,
21)
22
23
24class NicovideoMusicProviderExplorerMixin(NicovideoMusicProviderMixinBase):
25 """Search and recommendations methods for NicovideoMusicProvider."""
26
27 @override
28 @use_cache(3600 * 3) # Cache for 3 hours
29 async def search(
30 self,
31 search_query: str,
32 media_types: list[MediaType],
33 limit: int = 5,
34 ) -> SearchResults:
35 """
36 Perform search on musicprovider.
37
38 :param search_query: Search query.
39 :param media_types: A list of media_types to include.
40 :param limit: Number of items to return in the search (per type).
41 """
42 search_result = SearchResults()
43
44 if MediaType.TRACK in media_types:
45 tracks = await self.service_manager.search.search_videos_by_keyword(search_query, limit)
46 search_result.tracks = tracks
47
48 # Search for both playlists and albums in a single API call for efficiency
49 list_media_types: list[MediaType] = [
50 mt for mt in media_types if mt in (MediaType.PLAYLIST, MediaType.ALBUM)
51 ]
52
53 if list_media_types:
54 await self.service_manager.search.search_playlists_and_albums_by_keyword(
55 search_query, limit, search_result, list_media_types
56 )
57
58 return search_result
59
60 @override
61 async def get_recommendations(self) -> list[RecommendationFolder]:
62 """
63 Get this provider's available recommendation rows, without items.
64
65 Must be fast: return static or cached row descriptors only, without
66 live backend calls. The items for a row are fetched separately
67 through get_recommendation_items.
68 """
69 return [
70 RecommendationFolder(
71 item_id="nicovideo_recommendations",
72 name="Recommended tracks",
73 translation_key="recommended_tracks",
74 provider=self.instance_id,
75 icon="mdi-star-circle-outline",
76 ),
77 RecommendationFolder(
78 item_id="nicovideo_history",
79 name="Recently played",
80 translation_key="recently_played",
81 provider=self.instance_id,
82 icon="mdi-history",
83 ),
84 RecommendationFolder(
85 item_id="nicovideo_following_activities",
86 name="New Tracks from Followed Users",
87 translation_key="nicovideo_following_activities",
88 provider=self.instance_id,
89 icon="mdi-account-plus-outline",
90 ),
91 RecommendationFolder(
92 item_id="nicovideo_like_history",
93 name="Recently favorited tracks",
94 translation_key="recent_favorite_tracks",
95 provider=self.instance_id,
96 icon="mdi-heart-outline",
97 ),
98 ]
99
100 @override
101 async def get_recommendation_items(
102 self, item_id: str
103 ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
104 """
105 Get the items for a single recommendation row.
106
107 :param item_id: The item_id of the row, as returned by get_recommendations.
108 """
109 folder = await self._get_recommendation_folder(item_id)
110 if folder is None:
111 return UniqueList()
112 return folder.items
113
114 @override
115 @use_cache(3600 * 6, allow_expired_cache=True) # Cache for 6 hours
116 async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
117 """Retrieve a dynamic list of similar tracks based on the provided track."""
118 return await self.service_manager.user.get_similar_tracks(prov_track_id, limit)
119
120 @use_cache(1800, base_class=RecommendationFolder) # Cache for 30 minutes
121 async def _get_recommendation_folder(self, item_id: str) -> RecommendationFolder | None:
122 """
123 Build a single recommendation row with its items fetched (cached per row).
124
125 :param item_id: The item_id of the row, as returned by get_recommendations.
126 :return: The populated folder, or None for an unknown item_id.
127 """
128 items: list[Track]
129 if item_id == "nicovideo_recommendations":
130 items = await self.service_manager.user.get_recommendations(
131 "video_recommendation_recommend", limit=25
132 )
133 elif item_id == "nicovideo_history":
134 items = await self.service_manager.user.get_user_history(limit=50)
135 elif item_id == "nicovideo_following_activities":
136 items = await self.service_manager.user.get_following_activities(limit=30)
137 elif item_id == "nicovideo_like_history":
138 items = await self.service_manager.user.get_like_history(limit=50)
139 else:
140 return None
141 folder = next(
142 (row for row in await self.get_recommendations() if row.item_id == item_id),
143 None,
144 )
145 if folder is None:
146 return None
147 folder.items = UniqueList(items)
148 return folder
149