/
/
/
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