/
/
/
1"""
2HLS seek optimizer for nicovideo provider.
3
4This module implements a workaround for FFmpeg's seeking limitations with fragmented MP4
5HLS playlists (see https://trac.ffmpeg.org/ticket/7359).
6
7NOTE: This entire module can be removed once Music Assistant requires FFmpeg 8.0+,
8 which fixes the input-side -ss seeking issue (commit 380a518c, 2024-11-10).
9"""
10
11from __future__ import annotations
12
13import logging
14from dataclasses import dataclass
15from typing import TYPE_CHECKING
16
17from music_assistant.helpers.ffmpeg import get_ffmpeg_hls_cmaf_input_args
18from music_assistant.helpers.hls import HLSMediaPlaylist
19from music_assistant.providers.nicovideo.constants import (
20 DOMAND_BID_COOKIE_NAME,
21 NICOVIDEO_USER_AGENT,
22)
23from music_assistant.providers.nicovideo.helpers.utils import log_verbose
24
25if TYPE_CHECKING:
26 from music_assistant.providers.nicovideo.converters.stream import NicovideoStreamData
27
28LOGGER = logging.getLogger(__name__)
29
30
31@dataclass
32class SeekOptimizedStreamContext:
33 """
34 Context for seek-optimized HLS streaming.
35
36 Contains all information needed to set up streaming with fast seeking:
37 - Dynamic playlist content to serve
38 - FFmpeg extra input arguments (headers, seeking)
39 """
40
41 dynamic_playlist_text: str
42 extra_input_args: list[str]
43
44
45class HLSSeekOptimizer:
46 """
47 Optimizes HLS streaming with fast seeking support.
48
49 Generates dynamic HLS playlists and FFmpeg arguments for efficient
50 seeking by calculating optimal segment start positions.
51
52 This eliminates the need to decode all segments before the target position,
53 enabling instant seeking in long nicovideo streams.
54 """
55
56 def __init__(
57 self,
58 hls_data: NicovideoStreamData,
59 ) -> None:
60 """
61 Initialize seek optimizer with HLS data.
62
63 Args:
64 hls_data: HLS streaming data containing parsed playlist and authentication info
65 """
66 self.parsed_playlist: HLSMediaPlaylist = hls_data.parsed_hls_playlist
67 self.domand_bid = hls_data.domand_bid
68
69 def _calculate_start_segment(self, seek_position: int) -> tuple[int, float]:
70 """
71 Calculate which segment to start from based on seek position.
72
73 Args:
74 seek_position: Desired seek position in seconds
75
76 Returns:
77 Tuple of (segment_index, offset_within_segment)
78 - segment_index: Index of the segment to start from
79 - offset_within_segment: Seconds to skip within that segment
80 """
81 if seek_position <= 0:
82 return (0, 0.0)
83
84 accumulated_time = 0.0
85 for idx, segment in enumerate(self.parsed_playlist.segments):
86 segment_duration = segment.duration
87 if segment_duration > 0:
88 if accumulated_time + segment_duration > seek_position:
89 # Found the segment containing seek_position
90 offset = seek_position - accumulated_time
91 return (idx, offset)
92 accumulated_time += segment_duration
93
94 # If seek position is beyond total duration, start from last segment
95 return (max(0, len(self.parsed_playlist.segments) - 1), 0.0)
96
97 def _generate_dynamic_playlist(self, start_segment_idx: int) -> str:
98 """
99 Generate dynamic HLS playlist with segments from start_segment_idx onward.
100
101 Args:
102 start_segment_idx: Index to start from
103
104 Returns:
105 Dynamic HLS playlist text
106 """
107 lines = []
108
109 # Add header lines
110 lines.extend(self.parsed_playlist.header_lines)
111
112 # Add segments from start_segment_idx onward
113 # Track previous segment's key_line and map_line to emit only when changed
114 prev_key_line: str | None = None
115 prev_map_line: str | None = None
116
117 for segment in self.parsed_playlist.segments[start_segment_idx:]:
118 # Add discontinuity marker if present
119 if segment.discontinuity:
120 lines.append("#EXT-X-DISCONTINUITY")
121
122 # Add program date/time if present
123 if segment.program_date_time:
124 lines.append(segment.program_date_time)
125
126 # Add map line only if it changed from previous segment
127 # Note: MAP must come before KEY according to RFC 8216
128 if segment.map_line and segment.map_line != prev_map_line:
129 lines.append(segment.map_line)
130 prev_map_line = segment.map_line
131
132 # Add key line only if it changed from previous segment
133 if segment.key_line and segment.key_line != prev_key_line:
134 lines.append(segment.key_line)
135 prev_key_line = segment.key_line
136
137 # Add segment info and URL
138 lines.append(segment.extinf_line)
139
140 # Add byte range if present
141 if segment.byterange_line:
142 lines.append(segment.byterange_line)
143
144 lines.append(segment.segment_url)
145
146 # Add end tag
147 lines.extend(self.parsed_playlist.footer_lines)
148
149 return "\n".join(lines)
150
151 def create_stream_context(self, seek_position: int) -> SeekOptimizedStreamContext:
152 """
153 Create seek-optimized streaming context.
154
155 This method combines segment calculation, playlist generation,
156 and FFmpeg arguments preparation for fast seeking.
157
158 Args:
159 seek_position: Position to seek to in seconds
160
161 Returns:
162 SeekOptimizedStreamContext with all streaming setup information
163 """
164 # Stage 1: Calculate which segment contains the seek position (coarse seek)
165 # This avoids processing unnecessary segments before the target position
166 start_segment_idx, offset_within_segment = self._calculate_start_segment(seek_position)
167 if seek_position > 0:
168 log_verbose(
169 LOGGER,
170 "HLS seek: position=%ds â segment %d/%d (offset %.2fs)",
171 seek_position,
172 start_segment_idx,
173 len(self.parsed_playlist.segments),
174 offset_within_segment,
175 )
176
177 # Generate HLS playlist starting from the calculated segment
178 dynamic_playlist_text = self._generate_dynamic_playlist(start_segment_idx)
179
180 # Build FFmpeg extra input arguments
181 headers = (
182 f"User-Agent: {NICOVIDEO_USER_AGENT}\r\n"
183 f"Cookie: {DOMAND_BID_COOKIE_NAME}={self.domand_bid}\r\n"
184 )
185 # nicovideo serves CMAF (.cmfa) audio segments, which some FFmpeg builds reject
186 extra_input_args = ["-headers", headers, *get_ffmpeg_hls_cmaf_input_args()]
187
188 # Stage 2: Apply input-side -ss for fine-tuning within the segment
189 if offset_within_segment > 0:
190 extra_input_args.extend(["-ss", str(offset_within_segment)])
191
192 return SeekOptimizedStreamContext(
193 dynamic_playlist_text=dynamic_playlist_text,
194 extra_input_args=extra_input_args,
195 )
196