Merge pull request #36 from kutu-dev/fix/add-missing-parameters-in-st…

…ream-and-hls

Fix/add missing parameters in stream and hls

src/knuckles/media_retrieval.py

@@ -52,25 +52,61 @@ def _download_file(
 
         return download_path
 
-    def stream(self, id: str) -> str:
-        """Returns a valid url for streaming the requested song
-
-        :param id: The id of the song to stream
+    def stream(
+        self,
+        id: str,
+        max_bitrate_rate: int | None = None,
+        format: str | None = None,
+        time_offset: int | None = None,
+        size: str | None = None,
+        estimate_content_length: bool | None = None,
+        converted: bool | None = None,
+    ) -> str:
+        """Returns a valid url for streaming the requested song or bideo
+
+        :param id: The id of the song or video to stream
         :type id: str
+        :param max_bitrate_rate: A limit for the stream bitrate
+        :type max_bitrate_rate: int | None
+        :param format: The file format of preference to be used in the stream.
+        :type format: str | None
+        :param time_offset: Only applicable to video streaming.
+            An offset in seconds from where the video should start.
+        :type time_offset: int | None
+        :param size: Only applicable to video streaming.
+            The resolution for the streamed video, in the format of "WIDTHxHEIGHT".
+        :type size: str | None
+        :param estimate_content_length: If the response should set a
+            Content-Length HTTP header with an estimation of the duration of the media.
+        :type estimate_content_length: bool | None
+        :param converted: Only applicable to video streaming.
+            Try to retrieve from the server an optimize video in MP4 if it's available.
+        :type converted: bool | None
         :return A url that points to the given song in the stream endpoint
         :rtype str
         """
 
-        return self._generate_url("stream", {"id": id})
+        return self._generate_url(
+            "stream",
+            {
+                "id": id,
+                "maxBitRate": max_bitrate_rate,
+                "format": format,
+                "timeOffset": time_offset,
+                "size": size,
+                "estimateContentLength": estimate_content_length,
+                "converted": converted,
+            },
+        )
 
     def download(self, id: str, file_or_directory_path: Path) -> Path:
         """Calls the "download" endpoint of the API.
 
         :param id: The id of the song or video to download.
         :type id: str
         :param file_or_directory_path: If a directory path is passed the file will be
-        inside of it with the default filename given by the API,
-        if not the file will be saved directly in the given path.
+            inside of it with the default filename given by the API,
+            if not the file will be saved directly in the given path.
         :type file_or_directory_path: Path
         :return The path of the downloaded file
         :rtype Path
@@ -90,16 +126,31 @@ def download(self, id: str, file_or_directory_path: Path) -> Path:
 
         return self._download_file(response, file_or_directory_path, filename)
 
-    def hls(self, id: str) -> str:
+    def hls(
+        self,
+        id: str,
+        custom_bitrates: list[str] | None = None,
+        audio_track_id: str | None = None,
+    ) -> str:
         """Returns a valid url for streaming the requested song with hls.m3u8
 
         :param id: The id of the song to stream.
         :type id: str
+        :param custom_bitrates: A list of bitrates to be added to the hls playlist
+            for video streaming, the resolution can also be specified with
+            this format: "BITRATE@WIDTHxHEIGHT".
+        :type custom_bitrates: list[str] | None
+        :param audio_track_id: The id of the audio track to be used
+            if the playlist is for a video.
+        :type audio_track_id: str | None
         :return A url that points to the given song in the hls.m3u8 endpoint
         :rtype str
         """
 
-        return self._generate_url("hls.m3u8", {"id": id})
+        return self._generate_url(
+            "hls.m3u8",
+            {"id": id, "bitRate": custom_bitrates, "audioTrack": audio_track_id},
+        )
 
     def get_captions(
         self,
@@ -112,8 +163,8 @@ def get_captions(
         :param id: The ID of the video to get the captions
         :type id: str
         :param file_or_directory_path: If a directory path is passed the file will be
-        inside of it with the default filename given by the API,
-        if not the file will be saved directly in the given path.
+            inside of it with the default filename given by the API,
+            if not the file will be saved directly in the given path.
         :type file_or_directory_path: Path
         :param subtitles_file_format: The preferred captions file format.
         :type subtitles_file_format: SubtitlesFileFormat

tests/api/test_media_retrieval.py

@@ -12,11 +12,35 @@
 from tests.mocks.media_retrieval import FileMetadata
 
 
-def test_stream(subsonic: Subsonic, song: dict[str, Any]) -> None:
-    stream_url = parse.urlparse(subsonic.media_retrieval.stream(song["id"]))
+def test_stream_song(subsonic: Subsonic, song: dict[str, Any]) -> None:
+    stream_url = parse.urlparse(
+        subsonic.media_retrieval.stream(
+            song["id"], 0, song["suffix"], estimate_content_length=True
+        )
+    )
 
     assert stream_url.path == "/rest/stream"
     assert parse.parse_qs(stream_url.query)["id"][0] == song["id"]
+    assert parse.parse_qs(stream_url.query)["maxBitRate"][0] == "0"
+    assert parse.parse_qs(stream_url.query)["format"][0] == song["suffix"]
+    assert parse.parse_qs(stream_url.query)["estimateContentLength"][0] == "True"
+
+
+def test_stream_video(subsonic: Subsonic, video: dict[str, Any]) -> None:
+    stream_url = parse.urlparse(
+        subsonic.media_retrieval.stream(
+            video["id"], 0, video["suffix"], 809, "640x480", True, True
+        )
+    )
+
+    assert stream_url.path == "/rest/stream"
+    assert parse.parse_qs(stream_url.query)["id"][0] == video["id"]
+    assert parse.parse_qs(stream_url.query)["maxBitRate"][0] == "0"
+    assert parse.parse_qs(stream_url.query)["format"][0] == video["suffix"]
+    assert parse.parse_qs(stream_url.query)["timeOffset"][0] == "809"
+    assert parse.parse_qs(stream_url.query)["size"][0] == "640x480"
+    assert parse.parse_qs(stream_url.query)["estimateContentLength"][0] == "True"
+    assert parse.parse_qs(stream_url.query)["converted"][0] == "True"
 
 
 @responses.activate
@@ -61,13 +85,34 @@ def test_download_without_a_given_filename(
     assert download_path == tmp_path / download_metadata.default_filename
 
 
-def test_hls(subsonic: Subsonic, song: dict[str, Any]) -> None:
+def test_hls_song(subsonic: Subsonic, song: dict[str, Any]) -> None:
     stream_url = parse.urlparse(subsonic.media_retrieval.hls(song["id"]))
 
     assert stream_url.path == "/rest/hls.m3u8"
     assert parse.parse_qs(stream_url.query)["id"][0] == song["id"]
 
 
+def test_hls_video(
+    subsonic: Subsonic, video: dict[str, Any], video_details: dict[str, Any]
+) -> None:
+    custom_bitrates = ["1000@480x360", "820@1920x1080"]
+
+    stream_url = parse.urlparse(
+        subsonic.media_retrieval.hls(
+            video["id"], custom_bitrates, video_details["audioTrack"][0]["id"]
+        )
+    )
+
+    assert stream_url.path == "/rest/hls.m3u8"
+    assert parse.parse_qs(stream_url.query)["id"][0] == video["id"]
+    assert custom_bitrates[0] in parse.parse_qs(stream_url.query)["bitRate"]
+    assert custom_bitrates[1] in parse.parse_qs(stream_url.query)["bitRate"]
+    assert (
+        parse.parse_qs(stream_url.query)["audioTrack"][0]
+        == video_details["audioTrack"][0]["id"]
+    )
+
+
 @responses.activate
 def test_get_captions_with_a_given_filename(
     subsonic: Subsonic,

tests/mocks/browsing.py

@@ -141,6 +141,23 @@ def mock_get_song(mock_generator: MockGenerator, song: dict[str, Any]) -> Respon
     return mock_generator("getSong", {"id": song["id"]}, {"song": song})
 
 
+@pytest.fixture
+def video() -> dict[str, Any]:
+    return {"id": "videoId", "suffix": "mpv"}
+
+
+@pytest.fixture()
+def video_details() -> dict[str, Any]:
+    return {
+        "captions": {"id": "0", "name": "Planes 2.srt"},
+        "audioTrack": [
+            {"id": "1", "name": "English", "languageCode": "eng"},
+        ],
+        "conversion": {"id": "37", "bitRate": "1000"},
+        "id": "7058",
+    }
+
+
 @pytest.fixture
 def album_info(base_url: str) -> dict[str, Any]:
     return {