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