music-assistant-server

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 RecommendationFolder, SearchResults, Track
9from music_assistant_models.unique_list import UniqueList
10
11from music_assistant.controllers.cache import use_cache
12from music_assistant.providers.nicovideo.provider_mixins.base import (
13    NicovideoMusicProviderMixinBase,
14)
15
16
17class NicovideoMusicProviderExplorerMixin(NicovideoMusicProviderMixinBase):
18    """Search and recommendations methods for NicovideoMusicProvider."""
19
20    @override
21    @use_cache(3600 * 3)  # Cache for 3 hours
22    async def search(
23        self,
24        search_query: str,
25        media_types: list[MediaType],
26        limit: int = 5,
27    ) -> SearchResults:
28        """Perform search on musicprovider.
29
30        :param search_query: Search query.
31        :param media_types: A list of media_types to include.
32        :param limit: Number of items to return in the search (per type).
33        """
34        search_result = SearchResults()
35
36        if MediaType.TRACK in media_types:
37            tracks = await self.service_manager.search.search_videos_by_keyword(search_query, limit)
38            search_result.tracks = tracks
39
40        # Search for both playlists and albums in a single API call for efficiency
41        list_media_types = [mt for mt in media_types if mt in (MediaType.PLAYLIST, MediaType.ALBUM)]
42
43        if list_media_types:
44            await self.service_manager.search.search_playlists_and_albums_by_keyword(
45                search_query, limit, search_result, list_media_types
46            )
47
48        return search_result
49
50    @override
51    @use_cache(1800)  # Cache for 30 minutes
52    async def recommendations(self) -> list[RecommendationFolder]:
53        """
54        Get this provider's recommendations.
55
56        Returns an actual (and often personalised) list of recommendations
57        from this provider for the user/account.
58        """
59        recommendation_folders = []
60
61        # Main recommendations (default: 25 tracks)
62        main_recommendation_tracks = await self.service_manager.user.get_recommendations(
63            "video_recommendation_recommend", limit=25
64        )
65        if main_recommendation_tracks:
66            recommendation_folders.append(
67                RecommendationFolder(
68                    item_id="nicovideo_recommendations",
69                    name="nicovideo recommendations",
70                    provider=self.instance_id,
71                    icon="mdi-star-circle-outline",
72                    items=UniqueList(main_recommendation_tracks),
73                )
74            )
75
76        # History Tracks (default: 50 tracks)
77        history_tracks = await self.service_manager.user.get_user_history(limit=50)
78        if history_tracks:
79            recommendation_folders.append(
80                RecommendationFolder(
81                    item_id="nicovideo_history",
82                    name="Recently watched (nicovideo history)",
83                    provider=self.instance_id,
84                    icon="mdi-history",
85                    items=UniqueList(history_tracks),
86                )
87            )
88
89        # Following activities recommendations (default: 30 tracks)
90        following_activities_tracks = await self.service_manager.user.get_following_activities(
91            limit=30
92        )
93        if following_activities_tracks:
94            recommendation_folders.append(
95                RecommendationFolder(
96                    item_id="nicovideo_following_activities",
97                    name="New Tracks from Followed Users",
98                    provider=self.instance_id,
99                    icon="mdi-account-plus-outline",
100                    items=UniqueList(following_activities_tracks),
101                )
102            )
103
104        # Like History recommendations (default: 50 tracks)
105        like_history_tracks = await self.service_manager.user.get_like_history(limit=50)
106        if like_history_tracks:
107            recommendation_folders.append(
108                RecommendationFolder(
109                    item_id="nicovideo_like_history",
110                    name="Recently liked (Like history)",
111                    provider=self.instance_id,
112                    icon="mdi-heart-outline",
113                    items=UniqueList(like_history_tracks),
114                )
115            )
116
117        return recommendation_folders
118
119    @override
120    @use_cache(3600 * 6)  # Cache for 6 hours
121    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
122        """Retrieve a dynamic list of similar tracks based on the provided track."""
123        return await self.service_manager.user.get_similar_tracks(prov_track_id, limit)
124