music-assistant-server

1"""Stream converter for nicovideo objects."""
2
3from __future__ import annotations
4
5from dataclasses import dataclass
6
7from music_assistant_models.enums import MediaType, StreamType
8from music_assistant_models.errors import UnplayableMediaError
9from music_assistant_models.streamdetails import StreamDetails, StreamMetadata
10from niconico.objects.video.watch import (  # noqa: TC002 - Using by StreamConversionData(BaseModel Serialization)
11    WatchData,
12    WatchMediaDomandAudio,
13)
14from pydantic import BaseModel
15
16from music_assistant.helpers.hls import HLSMediaPlaylist, HLSMediaPlaylistParser
17from music_assistant.providers.nicovideo.converters.base import NicovideoConverterBase
18from music_assistant.providers.nicovideo.helpers import create_audio_format
19
20
21@dataclass
22class NicovideoStreamData:
23    """Type-safe container for nicovideo HLS streaming data.
24
25    This dataclass is stored in StreamDetails.data to pass
26    HLS-specific information to get_audio_stream().
27
28    Attributes:
29        domand_bid: Authentication cookie value
30        parsed_hls_playlist: Pre-parsed HLS playlist data (fetched once during conversion)
31    """
32
33    domand_bid: str
34    parsed_hls_playlist: HLSMediaPlaylist
35
36
37class StreamConversionData(BaseModel):
38    """Data needed for StreamDetails conversion."""
39
40    watch_data: WatchData
41    selected_audio: WatchMediaDomandAudio
42    hls_url: str
43    domand_bid: str
44    hls_playlist_text: str
45
46
47class NicovideoStreamConverter(NicovideoConverterBase):
48    """Handles StreamDetails conversion for nicovideo.
49
50    This converter transforms nicovideo video data into MusicAssistant StreamDetails
51    using StreamType.CUSTOM for optimized HLS streaming with fast seeking support.
52    """
53
54    def convert_from_conversion_data(self, conversion_data: StreamConversionData) -> StreamDetails:
55        """Convert StreamConversionData into StreamDetails.
56
57        Args:
58            conversion_data: Data containing video info, audio selection, and HLS details
59
60        Returns:
61            StreamDetails configured for custom HLS streaming with seek optimization
62
63        Raises:
64            UnplayableMediaError: If track data cannot be converted
65        """
66        watch_data = conversion_data.watch_data
67        selected_audio = conversion_data.selected_audio
68        video_id = watch_data.video.id_
69
70        # Get track information for stream metadata
71        track = self.converter_manager.track.convert_by_watch_data(watch_data)
72        if not track:
73            raise UnplayableMediaError(f"Cannot convert track data for video {video_id}")
74
75        # Get album and image information
76        album = track.album
77        # Do not use album image intentionally
78        image = track.image if track else None
79
80        parsed_playlist = HLSMediaPlaylistParser(conversion_data.hls_playlist_text).parse()
81
82        return StreamDetails(
83            provider=self.provider.instance_id,
84            item_id=video_id,
85            audio_format=create_audio_format(
86                sample_rate=selected_audio.sampling_rate,
87                bit_rate=selected_audio.bit_rate,
88            ),
89            media_type=MediaType.TRACK,
90            # CUSTOM stream type enables optimized seeking for nicovideo's fMP4-based HLS:
91            # 1. Generate dynamic playlist starting near target position (coarse seek)
92            # 2. Use input-side -ss for precise positioning (fine-tune)
93            # Without playlist reconstruction, input-side -ss on HLS results in empty output
94            # because FFmpeg cannot identify target segments before parsing the playlist.
95            stream_type=StreamType.CUSTOM,
96            duration=watch_data.video.duration,
97            stream_metadata=StreamMetadata(
98                title=track.name,
99                artist=track.artist_str,
100                album=album.name if album else None,
101                image_url=image.path if image else None,
102            ),
103            loudness=selected_audio.integrated_loudness,
104            data=NicovideoStreamData(
105                domand_bid=conversion_data.domand_bid,
106                parsed_hls_playlist=parsed_playlist,
107            ),
108            allow_seek=True,
109            can_seek=True,
110        )
111