music-assistant-server

1"""Track converter for nicovideo objects."""
2
3from __future__ import annotations
4
5from datetime import datetime
6from typing import TYPE_CHECKING
7
8from music_assistant_models.enums import ImageType, LinkType
9from music_assistant_models.media_items import (
10    Artist,
11    AudioFormat,
12    ItemMapping,
13    MediaItemImage,
14    MediaItemLink,
15    MediaItemMetadata,
16    Track,
17)
18from music_assistant_models.unique_list import UniqueList
19from niconico.objects.video import EssentialVideo, Owner, VideoThumbnail
20
21from music_assistant.providers.nicovideo.converters.base import NicovideoConverterBase
22from music_assistant.providers.nicovideo.helpers import create_audio_format
23
24if TYPE_CHECKING:
25    from niconico.objects.nvapi import Activity
26    from niconico.objects.video.watch import WatchData, WatchVideo, WatchVideoThumbnail
27
28
29class NicovideoTrackConverter(NicovideoConverterBase):
30    """Handles track conversion for nicovideo."""
31
32    def convert_by_activity(self, activity: Activity) -> Track | None:
33        """
34        Convert an Activity object from feed into a Track.
35
36        This is a lightweight conversion optimized for feed display,
37        using only the information available in the activity data.
38        Missing information like view counts and detailed metadata
39        will be absent, but this is acceptable for feed listings.
40        """
41        content = activity.content
42
43        # Only process video content
44        if content.type_ != "video" or not content.video:
45            return None
46
47        # Create audio format with minimal info
48        audio_format = create_audio_format()
49
50        # Build artists from actor information using ItemMapping
51        artists_list: UniqueList[Artist | ItemMapping] = UniqueList()
52        if activity.actor.id_ and activity.actor.name:
53            artist_mapping = ItemMapping(
54                item_id=activity.actor.id_,
55                provider=self.provider.domain,
56                name=activity.actor.name,
57            )
58            artists_list.append(artist_mapping)
59
60        # Create track with available information
61        return Track(
62            item_id=content.id_,
63            provider=self.provider.instance_id,
64            name=content.title,
65            duration=content.video.duration,
66            artists=artists_list,
67            # Assume playable if duration > 0 (we don't have payment info here)
68            is_playable=content.video.duration > 0,
69            metadata=self._create_track_metadata(
70                video_id=content.id_,
71                release_date_str=content.started_at,
72                thumbnail_url=activity.thumbnail_url,
73            ),
74            provider_mappings=self.helper.create_provider_mapping(
75                item_id=content.id_,
76                url_path="watch",
77                # We don't have availability info, so default to True if playable
78                available=content.video.duration > 0,
79                audio_format=audio_format,
80            ),
81        )
82
83    def convert_by_essential_video(self, video: EssentialVideo) -> Track | None:
84        """Convert an EssentialVideo object into a Track."""
85        # Skip muted videos
86        if video.is_muted:
87            return None
88
89        # Calculate popularity using standard formula
90        popularity = self.helper.calculate_popularity(
91            mylist_count=video.count.mylist,
92            like_count=video.count.like,
93        )
94
95        # Since EssentialVideo doesn't have detailed audio format info, we use defaults
96        audio_format = create_audio_format()
97
98        # Build artists using artist converter (prefer full Artist over ItemMapping)
99        artists_list: UniqueList[Artist | ItemMapping] = UniqueList()
100        if video.owner.id_ is not None:
101            artist_obj = self.converter_manager.artist.convert_by_owner_or_user(video.owner)
102            artists_list.append(artist_obj)
103
104        # Create base track with enhanced metadata
105        return Track(
106            item_id=video.id_,
107            provider=self.provider.instance_id,
108            name=video.title,
109            duration=video.duration,
110            artists=artists_list,
111            # Videos that cannot be played will have a duration of 0.
112            is_playable=video.duration > 0 and not video.is_payment_required,
113            metadata=self._create_track_metadata(
114                video_id=video.id_,
115                description=video.short_description,
116                explicit=video.require_sensitive_masking,
117                release_date_str=video.registered_at,
118                popularity=popularity,
119                thumbnail=video.thumbnail,
120            ),
121            provider_mappings=self.helper.create_provider_mapping(
122                item_id=video.id_,
123                url_path="watch",
124                available=self.is_video_available(video),
125                audio_format=audio_format,
126            ),
127        )
128
129    def convert_by_watch_data(self, watch_data: WatchData) -> Track | None:
130        """Convert a WatchData object into a Track."""
131        video = watch_data.video
132
133        # Skip deleted, private, or muted videos
134        if video.is_deleted or video.is_private:
135            return None
136
137        # Calculate popularity using standard formula
138        popularity = self.helper.calculate_popularity(
139            mylist_count=video.count.mylist,
140            like_count=video.count.like,
141        )
142
143        # Create owner object for artist conversion based on channel vs user video
144        if watch_data.channel:
145            # Channel video case
146            owner = Owner(
147                ownerType="channel",
148                type="channel",
149                visibility="visible",
150                id=watch_data.channel.id_,
151                name=watch_data.channel.name,
152                iconUrl=watch_data.channel.thumbnail.url if watch_data.channel.thumbnail else None,
153            )
154        else:
155            # User video case
156            owner = Owner(
157                ownerType="user",
158                type="user",
159                visibility="visible",
160                id=str(watch_data.owner.id_) if watch_data.owner else None,
161                name=watch_data.owner.nickname if watch_data.owner else None,
162                iconUrl=watch_data.owner.icon_url if watch_data.owner else None,
163            )
164
165        # Create audio format from watch data
166        audio_format = self._create_audio_format_from_watch_data(watch_data)
167
168        # Build artists using artist converter (avoid adding if owner id is missing)
169        artists_list: UniqueList[Artist | ItemMapping] = UniqueList()
170        if owner.id_ is not None:
171            artist_obj = self.converter_manager.artist.convert_by_owner_or_user(owner)
172            artists_list.append(artist_obj)
173
174        # Create base track with enhanced metadata
175        track = Track(
176            item_id=video.id_,
177            provider=self.provider.instance_id,
178            name=video.title,
179            duration=video.duration,
180            artists=artists_list,
181            # Videos that cannot be played will have a duration of 0.
182            is_playable=video.duration > 0 and not video.is_authentication_required,
183            metadata=self._create_track_metadata_from_watch_video(
184                video=video,
185                watch_data=watch_data,
186                popularity=popularity,
187            ),
188            provider_mappings=self.helper.create_provider_mapping(
189                item_id=video.id_,
190                url_path="watch",
191                available=self.is_video_available(video),
192                audio_format=audio_format,
193            ),
194        )
195
196        # Add album information if series data is available (prefer full Album over ItemMapping)
197        if watch_data.series is not None:
198            track.album = self.converter_manager.album.convert_by_series(
199                watch_data.series,
200                artists_list=artists_list,
201            )
202
203        return track
204
205    def _create_audio_format_from_watch_data(self, watch_data: WatchData) -> AudioFormat | None:
206        """
207        Create AudioFormat from WatchData audio information.
208
209        Args:
210            watch_data: WatchData object containing media information.
211
212        Returns:
213            AudioFormat object if audio information is available, None otherwise.
214        """
215        if (
216            not watch_data.media
217            or not watch_data.media.domand
218            or not watch_data.media.domand.audios
219        ):
220            return None
221
222        # Use the first available audio stream (typically the highest quality)
223        audio = watch_data.media.domand.audios[0]
224
225        if not audio.is_available:
226            return None
227
228        return create_audio_format(
229            sample_rate=audio.sampling_rate,
230            bit_rate=audio.bit_rate,
231        )
232
233    def _create_track_metadata_from_watch_video(
234        self,
235        video: WatchVideo,
236        watch_data: WatchData,
237        *,
238        popularity: int | None = None,
239    ) -> MediaItemMetadata:
240        """Create track metadata from WatchVideo object."""
241        metadata = MediaItemMetadata()
242
243        if video.description:
244            metadata.description = video.description
245
246        if video.registered_at:
247            try:
248                # Handle both direct ISO format and Z-suffixed format
249                if video.registered_at.endswith("Z"):
250                    clean_date_str = video.registered_at.replace("Z", "+00:00")
251                    metadata.release_date = datetime.fromisoformat(clean_date_str)
252                else:
253                    metadata.release_date = datetime.fromisoformat(video.registered_at)
254            except (ValueError, AttributeError) as err:
255                # Log debug message for date parsing failures to help with troubleshooting
256                self.logger.debug(
257                    "Failed to convert release date '%s': %s", video.registered_at, err
258                )
259
260        if popularity is not None:
261            metadata.popularity = popularity
262
263        # Add tag information as genres
264        if watch_data.tag and watch_data.tag.items:
265            # Extract tag names from tag items and create genres set
266            tag_names: list[str] = []
267            for tag_item in watch_data.tag.items:
268                tag_names.append(tag_item.name)
269
270            if tag_names:
271                metadata.genres = set(tag_names)
272
273        # Add thumbnail images
274        if video.thumbnail:
275            metadata.images = self._convert_watch_video_thumbnails(video.thumbnail)
276
277        # Add video link
278        metadata.links = {
279            MediaItemLink(
280                type=LinkType.WEBSITE,
281                url=f"https://www.nicovideo.jp/watch/{video.id_}",
282            )
283        }
284
285        return metadata
286
287    def _convert_watch_video_thumbnails(
288        self, thumbnail: WatchVideoThumbnail
289    ) -> UniqueList[MediaItemImage]:
290        """Convert WatchVideo thumbnails into multiple image sizes."""
291        images: UniqueList[MediaItemImage] = UniqueList()
292
293        def _add_thumbnail_image(url: str) -> None:
294            images.append(
295                MediaItemImage(
296                    type=ImageType.THUMB,
297                    path=url,
298                    provider=self.provider.instance_id,
299                    remotely_accessible=True,
300                )
301            )
302
303        # Add main thumbnail URLs
304        if thumbnail.url:
305            _add_thumbnail_image(thumbnail.url)
306        if thumbnail.middle_url:
307            _add_thumbnail_image(thumbnail.middle_url)
308        if thumbnail.large_url:
309            _add_thumbnail_image(thumbnail.large_url)
310
311        return images
312
313    def _create_track_metadata(
314        self,
315        video_id: str,
316        *,
317        description: str | None = None,
318        explicit: bool | None = None,
319        release_date_str: str | None = None,
320        popularity: int | None = None,
321        thumbnail: VideoThumbnail | None = None,
322        thumbnail_url: str | None = None,
323    ) -> MediaItemMetadata:
324        """Create track metadata with common fields."""
325        metadata = MediaItemMetadata()
326
327        if description:
328            metadata.description = description
329
330        if explicit is not None:
331            metadata.explicit = explicit
332
333        if release_date_str:
334            try:
335                # Handle both direct ISO format and Z-suffixed format
336                if release_date_str.endswith("Z"):
337                    clean_date_str = release_date_str.replace("Z", "+00:00")
338                    metadata.release_date = datetime.fromisoformat(clean_date_str)
339                else:
340                    metadata.release_date = datetime.fromisoformat(release_date_str)
341            except (ValueError, AttributeError) as err:
342                # Log debug message for date parsing failures to help with troubleshooting
343                self.logger.debug("Failed to convert release date '%s': %s", release_date_str, err)
344
345        if popularity is not None:
346            metadata.popularity = popularity
347
348        # Add thumbnail images with enhanced support
349        if thumbnail:
350            # Use enhanced thumbnail parsing for multiple sizes
351            metadata.images = self._convert_video_thumbnails(thumbnail)
352        elif thumbnail_url:
353            # Fallback to single thumbnail URL
354            metadata.images = UniqueList(
355                [
356                    MediaItemImage(
357                        type=ImageType.THUMB,
358                        path=thumbnail_url,
359                        provider=self.provider.instance_id,
360                        remotely_accessible=True,
361                    )
362                ]
363            )
364
365        # Add video link
366        metadata.links = {
367            MediaItemLink(
368                type=LinkType.WEBSITE,
369                url=f"https://www.nicovideo.jp/watch/{video_id}",
370            )
371        }
372
373        return metadata
374
375    def _convert_video_thumbnails(self, thumbnail: VideoThumbnail) -> UniqueList[MediaItemImage]:
376        """Convert video thumbnails into multiple image sizes."""
377        images: UniqueList[MediaItemImage] = UniqueList()
378
379        # nhd_url is the largest size, use it as primary
380        if thumbnail.nhd_url:
381            images.append(
382                MediaItemImage(
383                    type=ImageType.THUMB,
384                    path=thumbnail.nhd_url,
385                    provider=self.provider.instance_id,
386                    remotely_accessible=True,
387                )
388            )
389
390        # large_url as secondary (if different from nhd_url)
391        if thumbnail.large_url and thumbnail.large_url != thumbnail.nhd_url:
392            images.append(
393                MediaItemImage(
394                    type=ImageType.THUMB,
395                    path=thumbnail.large_url,
396                    provider=self.provider.instance_id,
397                    remotely_accessible=True,
398                )
399            )
400
401        # middle_url and listing_url are same size, skip them if nhd_url exists
402        # Only add if nhd_url is not available
403        if not thumbnail.nhd_url and thumbnail.middle_url:
404            images.append(
405                MediaItemImage(
406                    type=ImageType.THUMB,
407                    path=thumbnail.middle_url,
408                    provider=self.provider.instance_id,
409                    remotely_accessible=True,
410                )
411            )
412
413        return images
414
415    def is_video_available(self, video: EssentialVideo | WatchVideo) -> bool:
416        """
417        Check if a video is available for playback.
418
419        Args:
420            video: Either EssentialVideo or WatchVideo object.
421
422        Returns:
423            True if the video is available for playback, False otherwise.
424        """
425        # Common check: duration must be greater than 0
426        if video.duration <= 0:
427            return False
428
429        # Type-specific availability checks
430        if isinstance(video, EssentialVideo):
431            return not video.is_payment_required and not video.is_muted
432        # WatchVideo
433        return not video.is_deleted
434