/
/
/
1"""Search adapter for nicovideo."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6
7from music_assistant_models.enums import MediaType
8
9if TYPE_CHECKING:
10 from music_assistant_models.media_items import Album, Playlist, Track
11from niconico.objects.video.search import (
12 EssentialMylist,
13 EssentialSeries,
14)
15
16from music_assistant.providers.nicovideo.services.base import NicovideoBaseService
17
18if TYPE_CHECKING:
19 from music_assistant_models.media_items import SearchResults
20
21 from music_assistant.providers.nicovideo.services.manager import NicovideoServiceManager
22
23
24class NicovideoSearchService(NicovideoBaseService):
25 """Handles search related operations for nicovideo."""
26
27 def __init__(self, adapter: NicovideoServiceManager) -> None:
28 """Initialize NicovideoSearchService with reference to parent adapter."""
29 super().__init__(adapter)
30
31 async def search_playlists_and_albums_by_keyword(
32 self,
33 search_query: str,
34 limit: int,
35 search_result: SearchResults,
36 media_types: list[MediaType],
37 ) -> None:
38 """Search for playlists (mylists) and albums (series) by keyword."""
39 if not media_types:
40 return
41
42 search_playlists = MediaType.PLAYLIST in media_types
43 search_albums = MediaType.ALBUM in media_types
44
45 playlists_to_add = []
46 albums_to_add = []
47
48 # Search for mylists and series separately to work around API bug
49 # where specifying both types returns only series
50 if search_playlists:
51 mylists = await self._search_mylists_by_keyword(search_query, limit)
52 playlists_to_add.extend(mylists)
53
54 if search_albums:
55 albums = await self._search_series_by_keyword(search_query, limit)
56 albums_to_add.extend(albums)
57
58 # Add items to search result
59 if playlists_to_add:
60 current_playlists = list(search_result.playlists)
61 current_playlists.extend(playlists_to_add)
62 search_result.playlists = current_playlists
63 if albums_to_add:
64 current_albums = list(search_result.albums)
65 current_albums.extend(albums_to_add)
66 search_result.albums = current_albums
67
68 async def _search_mylists_by_keyword(self, search_query: str, limit: int) -> list[Playlist]:
69 """Search for mylists by keyword."""
70 list_search_data = await self.service_manager._call_with_throttler(
71 self.niconico_py_client.video.search.search_lists,
72 search_query,
73 page_size=limit,
74 types=["mylist"],
75 )
76
77 if not list_search_data:
78 return []
79
80 playlists = []
81 for item in list_search_data.items:
82 if isinstance(item, EssentialMylist):
83 playlists.append(self.converter_manager.playlist.convert_by_mylist(item))
84
85 return playlists
86
87 async def _search_series_by_keyword(self, search_query: str, limit: int) -> list[Album]:
88 """Search for series by keyword."""
89 list_search_data = await self.service_manager._call_with_throttler(
90 self.niconico_py_client.video.search.search_lists,
91 search_query,
92 page_size=limit,
93 types=["series"],
94 )
95
96 if not list_search_data:
97 return []
98
99 albums = []
100 for item in list_search_data.items:
101 if isinstance(item, EssentialSeries):
102 albums.append(self.converter_manager.album.convert_by_series(item))
103
104 return albums
105
106 async def search_videos_by_keyword(self, search_query: str, limit: int) -> list[Track]:
107 """Search for videos by keyword."""
108 video_search_data = await self.service_manager._call_with_throttler(
109 self.niconico_py_client.video.search.search_videos_by_keyword,
110 search_query,
111 page_size=limit,
112 search_by_user=True,
113 )
114 if not video_search_data:
115 return []
116
117 tracks = []
118 for item in video_search_data.items:
119 if item.id_:
120 track = self.converter_manager.track.convert_by_essential_video(item)
121 if track:
122 tracks.append(track)
123 return tracks
124