/
/
/
1"""
2nicovideo playlist mixin for Music Assistant.
3
4In this section, "Mylist" on niconico is treated as a playlist.
5"""
6
7from __future__ import annotations
8
9from collections.abc import AsyncGenerator
10from typing import TYPE_CHECKING, override
11
12from music_assistant_models.errors import MediaNotFoundError
13from music_assistant_models.media_items import Playlist, Track # noqa: TC002 - used in @use_cache
14
15from music_assistant.controllers.cache import use_cache
16from music_assistant.providers.nicovideo.provider_mixins.base import (
17 NicovideoMusicProviderMixinBase,
18)
19
20if TYPE_CHECKING:
21 from music_assistant_models.enums import MediaType
22
23
24class NicovideoMusicProviderPlaylistMixin(NicovideoMusicProviderMixinBase):
25 """Mixin class for handling playlist-related operations in NicovideoMusicProvider."""
26
27 @override
28 @use_cache(3600 * 24 * 14) # Cache for 14 days
29 async def get_playlist(self, prov_playlist_id: str) -> Playlist:
30 """Get full playlist details by id."""
31 playlist_with_tracks = await self.service_manager.mylist.get_mylist_or_own_mylist(
32 prov_playlist_id, page_size=500
33 )
34 if not playlist_with_tracks:
35 raise MediaNotFoundError(f"Playlist with id {prov_playlist_id} not found on nicovideo.")
36 return playlist_with_tracks.playlist
37
38 @override
39 @use_cache(3600 * 3, allow_expired_cache=True) # Cache for 3 hours
40 async def get_playlist_tracks(
41 self,
42 prov_playlist_id: str,
43 page: int = 0,
44 ) -> list[Track]:
45 """Get all playlist tracks for given playlist id."""
46 playlist_with_tracks = await self.service_manager.mylist.get_mylist_or_own_mylist(
47 prov_playlist_id, page_size=500, page=page + 1
48 )
49
50 return playlist_with_tracks.tracks if playlist_with_tracks else []
51
52 @override
53 async def get_library_playlists(
54 self,
55 ) -> AsyncGenerator[Playlist]:
56 """Retrieve library playlists from the provider."""
57 # Get own mylists (editable playlists)
58 own_mylists = await self.service_manager.mylist.get_own_mylists()
59 for mylist in own_mylists:
60 yield mylist
61 # Following mylists are not included in simplified config
62 return
63
64 @override
65 async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
66 """Add track(s) to playlist."""
67 for track_id in prov_track_ids:
68 success = await self.service_manager.mylist.add_mylist_item(prov_playlist_id, track_id)
69 if success:
70 self.logger.debug(
71 "Successfully added track %s to playlist %s",
72 track_id,
73 prov_playlist_id,
74 )
75 else:
76 self.logger.warning(
77 "Failed to add track %s to playlist %s", track_id, prov_playlist_id
78 )
79
80 @override
81 async def remove_playlist_tracks(
82 self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
83 ) -> None:
84 """Remove track(s) from playlist."""
85 # Get current playlist tracks to find track IDs at the specified positions
86 # Note: NicoNico's mylist does not allow duplicate entries of the same video_id
87 # within a single playlist. Therefore, mapping from 1-based positions to
88 # video_id is safe and uniquely identifies the target items.
89 playlist_tracks = await self.get_playlist_tracks(prov_playlist_id)
90
91 # Extract track IDs to remove based on positions
92 # Note: positions_to_remove uses 1-based indexing, so convert to 0-based
93 track_ids_to_remove = []
94 for position in positions_to_remove:
95 index = position - 1 # Convert from 1-based to 0-based indexing
96 if 0 <= index < len(playlist_tracks):
97 track_ids_to_remove.append(playlist_tracks[index].item_id)
98
99 if not track_ids_to_remove:
100 self.logger.warning(
101 "No valid tracks found to remove from playlist %s", prov_playlist_id
102 )
103 return
104
105 success = await self.service_manager.mylist.remove_mylist_items(
106 prov_playlist_id, track_ids_to_remove
107 )
108 if success:
109 self.logger.debug(
110 "Successfully removed %d tracks from playlist %s",
111 len(track_ids_to_remove),
112 prov_playlist_id,
113 )
114 else:
115 self.logger.warning("Failed to remove tracks from playlist %s", prov_playlist_id)
116
117 @override
118 async def create_playlist(self, name: str, media_types: set[MediaType]) -> Playlist:
119 """Create a new playlist on provider with given name."""
120 # Create a new mylist using niconico.py
121 create_result = await self.service_manager.mylist.create_mylist(
122 name, description="Created by Music Assistant", is_public=False
123 )
124
125 if not create_result:
126 raise MediaNotFoundError(
127 f"Failed to create playlist '{name}' on nicovideo.",
128 translation_key="create_playlist_failed",
129 translation_owner=self.translation_owner,
130 translation_args=[name],
131 )
132
133 # Get the created mylist details
134 mylist_id = str(create_result.mylist.id_)
135 playlist_with_tracks = await self.service_manager.mylist.get_own_mylist(
136 mylist_id, page_size=1
137 )
138
139 if not playlist_with_tracks:
140 raise MediaNotFoundError(
141 f"Failed to retrieve created playlist '{name}' from nicovideo.",
142 translation_key="retrieve_created_playlist_failed",
143 translation_owner=self.translation_owner,
144 translation_args=[name],
145 )
146
147 self.logger.info("Successfully created playlist '%s' with ID %s", name, mylist_id)
148 return playlist_with_tracks.playlist
149