music-assistant-server

1"""Stream handling for the Internet Archive provider."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any
6
7from music_assistant_models.enums import ContentType, MediaType, StreamType
8from music_assistant_models.errors import MediaNotFoundError
9from music_assistant_models.media_items import AudioFormat
10from music_assistant_models.streamdetails import StreamDetails
11
12if TYPE_CHECKING:
13    from .provider import InternetArchiveProvider
14
15
16class InternetArchiveStreaming:
17    """Handles stream details and multi-file streaming for Internet Archive."""
18
19    def __init__(self, provider: InternetArchiveProvider) -> None:
20        """
21        Initialize the streaming handler.
22
23        Args:
24            provider: The Internet Archive provider instance
25        """
26        self.provider = provider
27
28    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
29        """Get streamdetails for a track or audiobook."""
30        if "#" in item_id:
31            return self._get_single_file_stream(item_id, {}, media_type)
32        audio_files = await self.provider.client.get_audio_files(item_id)
33        if not audio_files:
34            raise MediaNotFoundError(f"No audio files found for {item_id}")
35
36        if media_type == MediaType.AUDIOBOOK and len(audio_files) > 1:
37            return await self._get_multi_file_audiobook_stream(item_id, audio_files)
38        return self._get_single_file_stream(item_id, audio_files[0], media_type)
39
40    async def _get_multi_file_audiobook_stream(
41        self, item_id: str, audio_files: list[dict[str, Any]]
42    ) -> StreamDetails:
43        """Get stream details for a multi-file audiobook."""
44        # Create list of download URLs for all chapters
45        chapter_urls = []
46
47        # Use provider's helper method for consistent duration calculation
48        total_duration, _ = await self.provider._calculate_audiobook_duration_and_chapters(item_id)
49
50        for file_info in audio_files:
51            filename = file_info["name"]
52            download_url = self.provider.client.get_download_url(item_id, filename)
53            chapter_urls.append(download_url)
54
55        duration_to_set = total_duration if total_duration > 0 else None
56
57        return StreamDetails(
58            provider=self.provider.instance_id,
59            item_id=item_id,
60            audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
61            media_type=MediaType.AUDIOBOOK,
62            stream_type=StreamType.CUSTOM,
63            duration=duration_to_set,
64            data={"chapters": chapter_urls, "chapters_data": audio_files},
65            allow_seek=True,
66            can_seek=True,
67        )
68
69    def _get_single_file_stream(
70        self, item_id: str, file_info: dict[str, Any], media_type: MediaType
71    ) -> StreamDetails:
72        """Get stream details for a single file."""
73        if "#" in item_id:
74            # This is a track from an album - extract parent_id and filename
75            parent_id, filename = item_id.split("#", 1)
76            download_url = self.provider.client.get_download_url(parent_id, filename)
77        else:
78            # This is a single item
79            filename = file_info["name"]
80            download_url = self.provider.client.get_download_url(item_id, filename)
81
82        return StreamDetails(
83            provider=self.provider.instance_id,
84            item_id=item_id,
85            audio_format=AudioFormat(
86                content_type=ContentType.UNKNOWN,  # Let ffmpeg detect format
87            ),
88            media_type=media_type,
89            stream_type=StreamType.HTTP,
90            path=download_url,
91            allow_seek=True,
92            can_seek=True,
93        )
94