music-assistant-server

1"""User adapter for nicovideo."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6
7from music_assistant.providers.nicovideo.constants import SENSITIVE_CONTENTS
8from music_assistant.providers.nicovideo.services.base import NicovideoBaseService
9
10if TYPE_CHECKING:
11    from typing import Literal
12
13    from music_assistant_models.media_items import Artist, Track
14
15    from music_assistant.providers.nicovideo.services.manager import NicovideoServiceManager
16
17# Import at runtime for isinstance checks
18from niconico.objects.video import EssentialVideo
19
20
21class NicovideoUserService(NicovideoBaseService):
22    """Get user details from nicovideo."""
23
24    def __init__(self, service_manager: NicovideoServiceManager) -> None:
25        """Initialize NicovideoUserService with reference to parent service manager."""
26        super().__init__(service_manager)
27
28    async def get_user(self, user_id: str) -> Artist | None:
29        """Get user details as Artist."""
30        user = await self.service_manager._call_with_throttler(
31            self.niconico_py_client.user.get_user, user_id
32        )
33        return self.converter_manager.artist.convert_by_owner_or_user(user) if user else None
34
35    async def get_recommendations(
36        self,
37        recipe_id: Literal[
38            "video_watch_recommendation", "video_recommendation_recommend", "video_top_recommend"
39        ] = "video_watch_recommendation",
40        limit: int = 25,
41    ) -> list[Track]:
42        """Get recommendations from nicovideo."""
43        recommendations = await self.service_manager._call_with_throttler(
44            self.niconico_py_client.user.get_recommendations,
45            recipe_id,
46            limit=limit,
47            sensitive_contents=SENSITIVE_CONTENTS,
48        )
49        if not recommendations or not recommendations.items:
50            return []
51
52        tracks = []
53        for item in recommendations.items:
54            # Only process video content, skip user recommendations
55            if item.content_type != "video":
56                continue
57
58            # Type check to ensure content is EssentialVideo
59            if isinstance(item.content, EssentialVideo):
60                track = self.converter_manager.track.convert_by_essential_video(item.content)
61                if track:
62                    tracks.append(track)
63        return tracks
64
65    async def get_similar_tracks(self, track_id: str, limit: int = 25) -> list[Track]:
66        """Get tracks similar to the given track."""
67        recommendation_api_item = await self.service_manager._call_with_throttler(
68            self.niconico_py_client.user.get_recommendations,
69            "video_watch_recommendation",
70            video_id=track_id,
71            limit=limit,
72            sensitive_contents=SENSITIVE_CONTENTS,
73        )
74        if not recommendation_api_item or not recommendation_api_item.items:
75            return []
76
77        tracks = []
78        for item in recommendation_api_item.items:
79            # Only process video content
80            if item.content_type != "video":
81                continue
82
83            # Type check to ensure content is EssentialVideo
84            if isinstance(item.content, EssentialVideo):
85                track = self.converter_manager.track.convert_by_essential_video(item.content)
86                if track:
87                    tracks.append(track)
88        return tracks
89
90    async def get_like_history(self, limit: int = 25) -> list[Track]:
91        """Get user's like history from nicovideo."""
92        # Calculate page_size based on limit
93        page_size = min(limit, 25)  # API max is 25 for like history
94        like_history = await self.service_manager._call_with_throttler(
95            self.niconico_py_client.video.get_like_history,
96            page_size=page_size,
97            page=1,
98        )
99        if not like_history or not like_history.items:
100            return []
101
102        tracks = []
103        for item in like_history.items:
104            track = self.converter_manager.track.convert_by_essential_video(item.video)
105            if track:
106                tracks.append(track)
107        return tracks
108
109    async def get_user_history(self, limit: int = 30) -> list[Track]:
110        """Get user's history from nicovideo."""
111        # Calculate page_size based on limit
112        page_size = min(limit, 100)  # API max is 100
113        history = await self.service_manager._call_with_throttler(
114            self.niconico_py_client.video.get_history,
115            page_size=page_size,
116            page=1,
117        )
118        if not history or not history.items:
119            return []
120
121        tracks = []
122        for item in history.items:
123            track = self.converter_manager.track.convert_by_essential_video(item.video)
124            if track:
125                tracks.append(track)
126        return tracks
127
128    async def get_following_activities(self, limit: int = 50) -> list[Track]:
129        """Get latest activities from followed users."""
130        feed_data = await self.service_manager._call_with_throttler(
131            self.niconico_py_client.user.get_following_activities,
132            endpoint="video",
133            context="header_timeline",
134            cursor=None,
135        )
136
137        if not feed_data:
138            return []
139
140        # Convert activities directly to tracks using lightweight conversion
141        tracks = []
142        for activity in feed_data.activities:
143            if activity.content and activity.content.video and "video" in activity.kind.lower():
144                track = self.converter_manager.track.convert_by_activity(activity)
145                if track:
146                    tracks.append(track)
147                if len(tracks) >= limit:
148                    break
149
150        return tracks
151
152    async def get_own_followings(self) -> list[Artist]:
153        """Get users the current user is following and convert them to Artists."""
154        followings_data = await self.service_manager._call_with_throttler(
155            self.niconico_py_client.user.get_own_followings,
156            page_size=25,
157            page=1,
158        )
159
160        if not followings_data or not followings_data.items:
161            return []
162
163        artists = []
164        for user in followings_data.items:
165            artist = self.converter_manager.artist.convert_by_owner_or_user(user)
166            artists.append(artist)
167        return artists
168