music-assistant-server

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