Repository: streamlink/streamlink
Pull request #3804 (2021-06-19 20:12:03)

plugins.youtube: translate embed_live URLs
- translate embedded live-channel URLs to avoid unnecessary HTTP
  redirections and having to find the canonical URL in the metadata
- add URL translation tests
- fix plugins can_handle_url pytest collection function

----

In regards to https://github.com/streamlink/streamlink/pull/3797#issuecomment-864449988, channel URLs like `youtube.com/CHANNELNAME` or `youtube.com/channel/CHANNELID` **could** be translated to `youtube.com/CHANNELNAME/live` or `youtube.com/channel/CHANNELID/live` respectively, but that would also require a change of the logic when the embedded data is missing, eg. when a channel is not live, and it'd have to always raise a `NoStreamsError` instead of a `PluginError` for a more graceful error message, which would be misleading in some cases.


Patch:
diff --git a/src/streamlink/plugins/youtube.py b/src/streamlink/plugins/youtube.py
index 4b2852e4..fe5aa716 100644
--- a/src/streamlink/plugins/youtube.py
+++ b/src/streamlink/plugins/youtube.py
@@ -3,7 +3,6 @@ import re
 from html import unescape
 from urllib.parse import urlparse, urlunparse
 
-from streamlink.exceptions import NoStreamsError
 from streamlink.plugin import Plugin, PluginError
 from streamlink.plugin.api import useragents, validate
 from streamlink.plugin.api.utils import itertags
@@ -28,11 +27,9 @@ class YouTube(Plugin):
                 )(?P<video_id>[0-9A-z_-]{11})
             )
             |
-            (?:
-                (?P<embed_live>embed)/live_stream\?channel=[^/?&]+
-                |
-                (?:c(?:hannel)?/|user/)?[^/?]+/live/?$
-            )
+            embed/live_stream\?channel=(?P<embed_live>[^/?&]+)
+            |
+            (?:c(?:hannel)?/|user/)?[^/?]+/live/?$
         )
         |
         https?://youtu\.be/(?P<video_id_short>[0-9A-z_-]{11})
@@ -42,6 +39,7 @@ class YouTube(Plugin):
     _re_mime_type = re.compile(r"""^(?P<type>\w+)/(?P<container>\w+); codecs="(?P<codecs>.+)"$""")
 
     _url_canonical = "https://www.youtube.com/watch?v={video_id}"
+    _url_channelid_live = "https://www.youtube.com/channel/{channel_id}/live"
 
     # There are missing itags
     adp_video = {
@@ -72,17 +70,19 @@ class YouTube(Plugin):
         match = self._re_url.match(url)
         parsed = urlparse(url)
 
+        # translate input URLs to be able to find embedded data and to avoid unnecessary HTTP redirects
         if parsed.netloc == "gaming.youtube.com":
             url = urlunparse(parsed._replace(scheme="https", netloc="www.youtube.com"))
         elif match.group("video_id_short") is not None:
             url = self._url_canonical.format(video_id=match.group("video_id_short"))
         elif match.group("embed") is not None:
             url = self._url_canonical.format(video_id=match.group("video_id"))
+        elif match.group("embed_live") is not None:
+            url = self._url_channelid_live.format(channel_id=match.group("embed_live"))
         else:
             url = urlunparse(parsed._replace(scheme="https"))
 
         super().__init__(url)
-        self._find_canonical_url = match.group("embed_live") is not None
         self.author = None
         self.title = None
         self.session.http.headers.update({'User-Agent': useragents.CHROME})
@@ -114,12 +114,6 @@ class YouTube(Plugin):
 
         return weight, group
 
-    @classmethod
-    def _get_canonical_url(cls, html):
-        for link in itertags(html, "link"):
-            if link.attributes.get("rel") == "canonical":
-                return link.attributes.get("href")
-
     @classmethod
     def _schema_playabilitystatus(cls, data):
         schema = validate.Schema(
@@ -224,13 +218,6 @@ class YouTube(Plugin):
             log.debug(f"c_data_keys: {', '.join(c_data.keys())}")
             res = self.session.http.post("https://consent.youtube.com/s", data=c_data)
 
-        if self._find_canonical_url:
-            self._find_canonical_url = False
-            canonical_url = self._get_canonical_url(res.text)
-            if canonical_url is None:
-                raise NoStreamsError(url)
-            return self._get_data(canonical_url)
-
         match = re.search(self._re_ytInitialPlayerResponse, res.text)
         if not match:
             raise PluginError("Missing initial player response data")


Test patch:
diff --git a/tests/plugins/conftest.py b/tests/plugins/conftest.py
index 3802f0f9..2dc40205 100644
--- a/tests/plugins/conftest.py
+++ b/tests/plugins/conftest.py
@@ -2,7 +2,7 @@ from tests.plugins import PluginCanHandleUrl, generic_negative_matches
 
 
 def pytest_generate_tests(metafunc):
-    if issubclass(metafunc.cls, PluginCanHandleUrl):  # pragma: no branch
+    if metafunc.cls is not None and issubclass(metafunc.cls, PluginCanHandleUrl):  # pragma: no branch
         if metafunc.function.__name__ == "test_can_handle_url_positive":
             metafunc.parametrize("url", metafunc.cls.should_match)
         elif metafunc.function.__name__ == "test_can_handle_url_negative":
diff --git a/tests/plugins/test_youtube.py b/tests/plugins/test_youtube.py
index ef3dc8d9..3ae6da56 100644
--- a/tests/plugins/test_youtube.py
+++ b/tests/plugins/test_youtube.py
@@ -1,5 +1,9 @@
 import unittest
+from unittest.mock import Mock
 
+import pytest
+
+from streamlink.plugin import Plugin
 from streamlink.plugins.youtube import YouTube
 from tests.plugins import PluginCanHandleUrl
 
@@ -57,3 +61,15 @@ class TestPluginYouTube(unittest.TestCase):
     def test_regex_video_id_watch(self):
         self._test_regex("https://www.youtube.com/watch?v=aqz-KE-bpKQ",
                          "aqz-KE-bpKQ", "video_id")
+
+
+@pytest.mark.parametrize("url,expected", [
+    ("http://gaming.youtube.com/watch?v=0123456789A", "https://www.youtube.com/watch?v=0123456789A"),
+    ("http://youtu.be/0123456789A", "https://www.youtube.com/watch?v=0123456789A"),
+    ("http://youtube.com/embed/0123456789A", "https://www.youtube.com/watch?v=0123456789A"),
+    ("http://youtube.com/embed/live_stream?channel=CHANNELID", "https://www.youtube.com/channel/CHANNELID/live"),
+    ("http://www.youtube.com/watch?v=0123456789A", "https://www.youtube.com/watch?v=0123456789A"),
+])
+def test_translate_url(url, expected):
+    Plugin.bind(Mock(), "tests.plugins.test_youtube")
+    assert YouTube(url).url == expected