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