plugins.twitch: add --twitch-supported-codecs

src/streamlink/plugins/twitch.py

@@ -22,7 +22,7 @@
 from datetime import datetime, timedelta
 from json import dumps as json_dumps
 from random import random
-from typing import ClassVar, Mapping, Optional, Tuple, Type
+from typing import ClassVar, List, Mapping, Optional, Tuple, Type
 from urllib.parse import urlparse
 
 from requests.exceptions import HTTPError
@@ -203,18 +203,22 @@ def __init__(self, *args, disable_ads: bool = False, low_latency: bool = False,
 
 
 class UsherService:
-    def __init__(self, session):
+    def __init__(self, session: Streamlink, supported_codecs: Optional[List[str]] = None):
         self.session = session
+        self.supported_codecs = supported_codecs or ["h264"]
 
     def _create_url(self, endpoint, **extra_params):
         url = f"https://usher.ttvnw.net{endpoint}"
         params = {
+            "platform": "web",
             "player": "twitchweb",
             "p": int(random() * 999999),
             "type": "any",
             "allow_source": "true",
             "allow_audio_only": "true",
             "allow_spectre": "false",
+            "playlist_include_framerate": "true",
+            "supported_codecs": ",".join(self.supported_codecs),
         }
         params.update(extra_params)
 
@@ -236,7 +240,7 @@ def channel(self, channel, **extra_params):
                 },
             ).validate(extra_params)
             log.debug(f"{extra_params_debug!r}")
-        except PluginError:
+        except PluginError:  # pragma: no cover
             pass
         return self._create_url(f"/api/channel/hls/{channel}.m3u8", **extra_params)
 
@@ -666,6 +670,23 @@ def decode_client_integrity_token(data: str, schema: Optional[validate.Schema] =
         Regular streams can cause buffering issues with this option enabled due to the reduced --hls-live-edge value.
     """,
 )
+@pluginargument(
+    "supported-codecs",
+    type="comma_list_filter",
+    type_kwargs={
+        "acceptable": ["h264", "h265", "av1"],
+        "unique": True,
+    },
+    default=["h264"],
+    help="""
+        A comma-separated list of codec names which signals Twitch the client's stream codec preference.
+        Which streams and which codecs are available depends on the specific channel and broadcast.
+
+        Default is "h264".
+
+        Supported codecs are "h264", "h265", "av1". Set to "h264,h265,av1" to enable all codecs.
+    """,
+)
 @pluginargument(
     "api-header",
     metavar="KEY=VALUE",
@@ -727,7 +748,10 @@ def __init__(self, *args, **kwargs):
             api_header=self.get_option("api-header"),
             access_token_param=self.get_option("access-token-param"),
         )
-        self.usher = UsherService(session=self.session)
+        self.usher = UsherService(
+            session=self.session,
+            supported_codecs=self.get_option("supported-codecs"),
+        )
 
         self._checked_metadata = False
 

tests/plugins/test_twitch.py

@@ -1,8 +1,10 @@
+import json
 import unittest
 from contextlib import nullcontext
 from datetime import datetime, timedelta, timezone
 from typing import Optional
 from unittest.mock import MagicMock, Mock, call, patch
+from urllib.parse import parse_qsl, urlparse
 
 import pytest
 import requests_mock as rm
@@ -416,6 +418,84 @@ def test_hls_low_latency_no_ads_reload_time(self):
         assert self.thread.reader.worker.playlist_reload_time == pytest.approx(23 / 3)
 
 
+class TestUsherService:
+    @pytest.fixture(autouse=True)
+    def caplog(self, caplog: pytest.LogCaptureFixture):
+        caplog.set_level(1, "streamlink.plugins.twitch")
+        return caplog
+
+    @pytest.fixture()
+    def plugin(self, request: pytest.FixtureRequest, session: Streamlink):
+        return Twitch(
+            session,
+            "https://twitch.tv/twitch",
+            options=Options(getattr(request, "param", {})),
+        )
+
+    @pytest.fixture()
+    def endpoint(self, request: pytest.FixtureRequest, caplog: pytest.LogCaptureFixture, plugin: Twitch):
+        param = getattr(request, "param", {})
+        service = param.get("service", "channel")
+
+        token = {
+            "expires": 9876543210,
+            "channel": "twitch",
+            "channel_id": 123,
+            "user_id": 456,
+            "user_ip": "127.0.0.1",
+            "adblock": False,
+            "geoblock_reason": "",
+            "hide_ads": False,
+            "server_ads": True,
+            "show_ads": True,
+        }
+
+        return getattr(plugin.usher, service)("twitch", token=json.dumps(token), sig="tokensignature")
+
+    @pytest.mark.parametrize(("endpoint", "expected_path", "logs"), [
+        pytest.param(
+            {"service": "channel"},
+            "/api/channel/hls/twitch.m3u8",
+            [(
+                "streamlink.plugins.twitch",
+                "debug",
+                "{'adblock': False, 'geoblock_reason': '', 'hide_ads': False, 'server_ads': True, 'show_ads': True}",
+            )],
+            id="channel",
+        ),
+        pytest.param(
+            {"service": "video"},
+            "/vod/twitch",
+            [],
+            id="video",
+        ),
+    ], indirect=["endpoint"])
+    def test_service(self, caplog: pytest.LogCaptureFixture, endpoint: str, expected_path: str, logs: list):
+        url = urlparse(endpoint)
+        assert url.path == expected_path
+
+        qs = dict(parse_qsl(url.query))
+        assert qs.get("token")
+        assert qs.get("sig")
+
+        assert [(r.name, r.levelname, r.message) for r in caplog.get_records(when="setup")] == logs
+
+    @pytest.mark.parametrize("endpoint", [
+        pytest.param({"service": "channel"}, id="channel"),
+        pytest.param({"service": "video"}, id="video"),
+    ], indirect=True)
+    @pytest.mark.parametrize(("plugin", "expected"), [
+        pytest.param({}, "h264", id="unset"),
+        pytest.param({"supported_codecs": []}, "h264", id="empty"),
+        pytest.param({"supported_codecs": ["h264"]}, "h264", id="h264"),
+        pytest.param({"supported_codecs": ["av1", "h264"]}, "av1,h264", id="av1,h264"),
+        pytest.param({"supported_codecs": ["av1", "h264", "h265"]}, "av1,h264,h265", id="av1,h264,h265"),
+    ], indirect=["plugin"])
+    def test_supported_codecs(self, plugin: Twitch, endpoint: str, expected: str):
+        qs = dict(parse_qsl(urlparse(endpoint).query))
+        assert qs.get("supported_codecs") == expected
+
+
 class TestTwitchAPIAccessToken:
     @pytest.fixture(autouse=True)
     def _client_integrity_token(self, monkeypatch: pytest.MonkeyPatch):