Repository: streamlink/streamlink

Issue:
plugins.youtube.py: 404 Intermittent and Repeatable (and related, but not duplicate of 3724 of other 404 issues)

## Plugin Issue

<!-- Replace the space character between the square brackets with an x in order to check the boxes -->
- [X] This is a plugin issue and I have read the contribution guidelines.
- [ ] I am using the latest development version from the master branch. [I apologize, I'm on latest stable, but plugin hasn't been updated in 28 days and we have that installed]


### Description

There have been issues over the past couple of months with YouTube reporting 404s. It's been tricky because it's not always reproducable. We had no problems last two days, and today, half our captures have this problem.

The symptom is no different than 3724 and other previously reported items. The error is the same. Using the simplest execution.
```
ubuntu@ip-XXX-XXX-XXX-XXX:~$ streamlink https://www.youtube.com/watch?v=YertdWavZdk
[cli][info] Found matching plugin youtube for URL https://www.youtube.com/watch?v=YertdWavZdk
[cli][error] Unable to open URL: https://youtube.com/get_video_info (404 Client Error: Not Found for url: https://www.youtube.com/get_video_info?video_id=YertdWavZdk&html5=1&el=detailpage)
```
However, what we are seeing is similar to what we couldn't narrow down when this occurred a month ago. If you wait 3-5 minutes, or do from different machines, the same call can work, or it can report an error.

From the same machine:
```
ubuntu@ip-XXX-XXX-XXX-XXX:~$ streamlink https://www.youtube.com/watch?v=YertdWavZdk
[cli][info] Found matching plugin youtube for URL https://www.youtube.com/watch?v=YertdWavZdk
[cli][error] Unable to open URL: https://youtube.com/get_video_info (404 Client Error: Not Found for url: https://www.youtube.com/get_video_info?video_id=YertdWavZdk&html5=1&el=detailpage)
[Note: Paused 10 minutes manually and tried again]
ubuntu@ip-XXX-XXX-XXX-XXX:~$ streamlink https://www.youtube.com/watch?v=YertdWavZdk
[cli][info] Found matching plugin youtube for URL https://www.youtube.com/watch?v=YertdWavZdk
Available streams: audio_mp4a, audio_opus, 144p (worst), 240p, 360p, 480p (best)
```
If you can test from machines with different IP addresses, you can run the same plain command a few times at the same time, and you'll see about half work, half return the 404.

I'm not sure if I'm missing something on the retry or other recommended fixes (e.g. cookies) but regardless of how complex or simple the query is, we see this intermittent vs. consistent behavior.

Thinking out loud / speculation: a partial rollout of something new in the youtube cluster, a potential change being tested. Just plain don't know. Clearly it's not the plug-in since it behaves some of the time on the same call, but some new behavior (or bug) on YouTube?


### Reproduction steps / Explicit stream URLs to test

<!-- How can we reproduce this? Please note the exact steps below using the list format supplied. If you need more steps please add them. -->

1. Run this command on multiple machines (or wait a few minutes in between each call, either works: streamlink `https://www.youtube.com/watch?v=YertdWavZdk` - note you can use any youtube link that was a live stream originally, whether it's not yet live, currently live, or after the stream finishes. URL, have not tested with previously recorded only



### Log output

<!--
DEBUG LOG OUTPUT IS REQUIRED for a plugin issue!
INCLUDE THE ENTIRE COMMAND LINE and make sure to **remove usernames and passwords**

Use the `--loglevel debug` parameter and avoid using parameters which suppress log output.
Debug log includes important details about your platform. Don't remove it.
https://streamlink.github.io/latest/cli.html#cmdoption-loglevel

You can copy the output to https://gist.github.com/ or paste it below.

Don't post screenshots of the log output and instead copy the text from your terminal application.
-->
When the command works / doesn't 404:
```
ubuntu@ip-XXX-XXX-XXX-XXX:~$ streamlink --loglevel debug https://www.youtube.com/watch?v=YertdWavZdk
[cli][debug] OS:         Linux-5.4.0-1049-aws-x86_64-with-Ubuntu-18.04-bionic
[cli][debug] Python:     3.6.9
[cli][debug] Streamlink: 2.1.2
[cli][debug] Requests(2.22.0), Socks(1.6.5), Websocket(0.58.0)
[cli][debug] Arguments:
[cli][debug]  url=https://www.youtube.com/watch?v=YertdWavZdk
[cli][debug]  --loglevel=debug
[cli][info] Found matching plugin youtube for URL https://www.youtube.com/watch?v=YertdWavZdk
[plugins.youtube][debug] Video ID from URL
[plugins.youtube][debug] Using video ID: YertdWavZdk
[plugins.youtube][debug] get_video_info - 1: Found data
[plugins.youtube][debug] MuxedStream: v 135 a 251 = 480p
[plugins.youtube][debug] MuxedStream: v 133 a 251 = 240p
[plugins.youtube][debug] MuxedStream: v 160 a 251 = 144p
Available streams: audio_mp4a, audio_opus, 144p (worst), 240p, 360p, 480p (best)
```
When the command 404s:
```
ubuntu@ip-XXX-XXX-XXX-XXX:~$ streamlink --loglevel debug https://www.youtube.com/watch?v=YertdWavZdk
[cli][debug] OS:         Linux-5.4.0-1049-aws-x86_64-with-Ubuntu-18.04-bionic
[cli][debug] Python:     3.6.9
[cli][debug] Streamlink: 2.1.2
[cli][debug] Requests(2.22.0), Socks(1.6.5), Websocket(0.58.0)
[cli][debug] Arguments:
[cli][debug]  url=https://www.youtube.com/watch?v=YertdWavZdk
[cli][debug]  --loglevel=debug
[cli][info] Found matching plugin youtube for URL https://www.youtube.com/watch?v=YertdWavZdk
[plugins.youtube][debug] Video ID from URL
[plugins.youtube][debug] Using video ID: YertdWavZdk
error: Unable to open URL: https://youtube.com/get_video_info (404 Client Error: Not Found for url: https://www.youtube.com/get_video_info?video_id=YertdWavZdk&html5=1&el=detailpage)
```
Note the two calls above were run on the same machine ~3 minutes apart.

### Additional comments, etc.
I've seen and tried the other suggestions: manually adding a cookie, playing with retries. As best I can tell, this is something on YouTube's end that's now been appearing intermittently for at least 2 months and, in our daily use, has not once been affecting 100% of our calls, which are all structured identically other than the youtube link. Either everything works fine, or half the YouTube calls fail when this appears. Never 100%. In going through our internal Jira, it's always been an intermittent thing.



[Love Streamlink? Please consider supporting our collective. Thanks!](https://opencollective.com/streamlink/donate)


Discussion:
bastimeyer: @back-to 
https://github.com/streamlink/streamlink/commit/3d2848ae5b2d9dd70fd66fd25b8523db38cf741a#r52311523

> won't work for videos with }; in the title
> https://www.youtube.com/watch?v=CqSfYBGfYEc
> `error: Unable to parse JSON: Unterminated string starting at:`

This can't be fixed entirely unless we parse the JS code and get the JSON data from the AST.

I could change the regex and extract the entire JS from the `<script>` tag and then give the regex a stronger anchor at the end, but that doesn't solve the problem entirely, as titles with `};\s*var` would also break the regex.
bastimeyer: @back-to 
Making the regex super strict has the potential to break the plugin as soon as YT adjusts a tiny thing
```diff
diff --git a/src/streamlink/plugins/youtube.py b/src/streamlink/plugins/youtube.py
index 3b5b2531..eaf3f26a 100644
--- a/src/streamlink/plugins/youtube.py
+++ b/src/streamlink/plugins/youtube.py
@@ -38,7 +38,13 @@ class YouTube(Plugin):
         https?://youtu\.be/(?P<video_id_short>[0-9A-z_-]{11})
     """, re.VERBOSE)
 
-    _re_ytInitialPlayerResponse = re.compile(r"""var ytInitialPlayerResponse\s*=\s*({.*?});""", re.DOTALL)
+    _re_ytInitialPlayerResponse = re.compile(r"""
+        var\s+ytInitialPlayerResponse\s*=\s*({.*?});\s*
+        var\s+meta\s*=\s*document\.createElement\('meta'\);\s*
+        meta\.name\s*=\s*'referrer';\s*
+        meta\.content\s*=\s*'origin-when-cross-origin';\s*
+        document.getElementsByTagName\('head'\)\[0]\.appendChild\(meta\);
+    """, re.VERBOSE | re.DOTALL)
     _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}"
```
bastimeyer: I've added `\s*var\s+meta\s*=` to the regex, as a compromise, since it's way more unlikely that someone has set a title that matches the regex's anchor `};\s*var\s+meta\s*=` than just `};` and there's more room for YT to make changes without breaking the plugin.

Diff:
https://github.com/streamlink/streamlink/compare/cfd7329455781f92e8b45cc8e81ad810072905b4..eabf22875d72b223ccd47c08ede25c14a9a13144

I've also squashed the first two commits into one, which is better if you want to keep the individual commits when merging.
tfdahlin: > 
> 
> @back-to
> [3d2848a#r52311523](https://github.com/streamlink/streamlink/commit/3d2848ae5b2d9dd70fd66fd25b8523db38cf741a#r52311523)
> 
> > won't work for videos with }; in the title
> > https://www.youtube.com/watch?v=CqSfYBGfYEc
> > `error: Unable to parse JSON: Unterminated string starting at:`
> 
> This can't be fixed entirely unless we parse the JS code and get the JSON data from the AST.
> 
> I could change the regex and extract the entire JS from the `<script>` tag and then give the regex a stronger anchor at the end, but that doesn't solve the problem entirely, as titles with `};\s*var` would also break the regex.
> 
> This is the entire script tag content:
> 
> ```
> <script nonce="...">var ytInitialPlayerResponse = {};var meta = document.createElement('meta'); meta.name = 'referrer'; meta.content = 'origin-when-cross-origin'; document.getElementsByTagName('head')[0].appendChild(meta);</script>
> ```

I'm here because somebody linked to this in a pytube issue, wanted to look at your solution to the get_video_info problem. If you're interested in doing parsing like this, I actually wrote a parser that I haven't had problems with for doing exactly this over [here](https://github.com/pytube/pytube/blob/master/pytube/parser.py).

Out of curiosity, did you test your changes against age-restricted videos? That's the stumbling block that I'm running into right now using the innertube API endpoints, and I'm wondering if you happen to know if the stream data for age-restricted videos is available on the page or not
bastimeyer: Custom parsing solutions are always bad, same with cheap regexes like this, and the only true solution when trying to extract data from JS is using a parser which supports the latest ECMA versions and their syntaxes and then reading and traversing the generated AST. We've had these discussions several times now (eg. #2534), but the proposed python dependencies were either outdated or unmaintained and thus not ideal. I hope we can spark up a discussion again now that the youtube plugin is affected by this problem. But let's not discuss this here.

No, I have not checked all types of videos with these changes. My OP post has a list of types I've checked. YT-Authentication was not supported before and testing age-restricted videos therefore doesn't make much sense.
tfdahlin: I wasn't sure if your OP was extensive or not for what you had tested. Pytube doesn't support yt auth either, but by passing certain parameters to the `get_video_info` endpoint, we were able to extract stream data for age-restricted videos without auth which is why I was asking. 

Patch:
diff --git a/src/streamlink/plugins/youtube.py b/src/streamlink/plugins/youtube.py
index 9d8e455e..4b2852e4 100644
--- a/src/streamlink/plugins/youtube.py
+++ b/src/streamlink/plugins/youtube.py
@@ -1,104 +1,47 @@
 import logging
 import re
-from urllib.parse import parse_qsl, urlparse, urlunparse
+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, parse_query
+from streamlink.plugin.api.utils import itertags
 from streamlink.stream import HLSStream, HTTPStream
 from streamlink.stream.ffmpegmux import MuxedStream
-from streamlink.utils import parse_json, search_dict
+from streamlink.utils import parse_json
 
 log = logging.getLogger(__name__)
 
 
-_config_schema = validate.Schema(
-    {
-        validate.optional("player_response"): validate.all(
-            validate.text,
-            validate.transform(parse_json),
-            {
-                validate.optional("streamingData"): {
-                    validate.optional("hlsManifestUrl"): validate.text,
-                    validate.optional("formats"): [{
-                        "itag": int,
-                        validate.optional("url"): validate.text,
-                        validate.optional("cipher"): validate.text,
-                        "qualityLabel": validate.text
-                    }],
-                    validate.optional("adaptiveFormats"): [{
-                        "itag": int,
-                        "mimeType": validate.all(
-                            validate.text,
-                            validate.transform(
-                                lambda t:
-                                    [t.split(';')[0].split('/')[0], t.split(';')[1].split('=')[1].strip('"')]
-                            ),
-                            [validate.text, validate.text],
-                        ),
-                        validate.optional("url"): validate.url(scheme="http"),
-                        validate.optional("cipher"): validate.text,
-                        validate.optional("signatureCipher"): validate.text,
-                        validate.optional("qualityLabel"): validate.text,
-                        validate.optional("bitrate"): int
-                    }]
-                },
-                validate.optional("videoDetails"): {
-                    validate.optional("isLive"): validate.transform(bool),
-                    validate.optional("author"): validate.text,
-                    validate.optional("title"): validate.text
-                },
-                validate.optional("playabilityStatus"): {
-                    validate.optional("status"): validate.text,
-                    validate.optional("reason"): validate.text
-                },
-            },
-        ),
-        "status": validate.text
-    }
-)
-
-_ytdata_re = re.compile(r'ytInitialData\s*=\s*({.*?});', re.DOTALL)
-_url_re = re.compile(r"""
-    https?://(?:\w+\.)?youtube\.com
-    (?:
-        (?:
-            /(?:
-                watch.+v=
-                |
-                embed/(?!live_stream)
-                |
-                v/
-            )(?P<video_id>[0-9A-z_-]{11})
-        )
-        |
+class YouTube(Plugin):
+    _re_url = re.compile(r"""
+        https?://(?:\w+\.)?youtube\.com/
         (?:
-            /(?:
-                (?:user|c(?:hannel)?)/
+            (?:
+                (?:
+                    watch\?(?:.*&)*v=
+                    |
+                    (?P<embed>embed)/(?!live_stream)
+                    |
+                    v/
+                )(?P<video_id>[0-9A-z_-]{11})
+            )
+            |
+            (?:
+                (?P<embed_live>embed)/live_stream\?channel=[^/?&]+
                 |
-                embed/live_stream\?channel=
-            )[^/?&]+
+                (?:c(?:hannel)?/|user/)?[^/?]+/live/?$
+            )
         )
         |
-        (?:
-            /(?:c/)?[^/?]+/live/?$
-        )
-    )
-    |
-    https?://youtu\.be/(?P<video_id_short>[0-9A-z_-]{11})
-""", re.VERBOSE)
-
+        https?://youtu\.be/(?P<video_id_short>[0-9A-z_-]{11})
+    """, re.VERBOSE)
 
-class YouTube(Plugin):
-    _oembed_url = "https://www.youtube.com/oembed"
-    _video_info_url = "https://youtube.com/get_video_info"
+    _re_ytInitialPlayerResponse = re.compile(r"""var\s+ytInitialPlayerResponse\s*=\s*({.*?});\s*var\s+meta\s*=""", re.DOTALL)
+    _re_mime_type = re.compile(r"""^(?P<type>\w+)/(?P<container>\w+); codecs="(?P<codecs>.+)"$""")
 
-    _oembed_schema = validate.Schema(
-        {
-            "author_name": validate.text,
-            "title": validate.text
-        }
-    )
+    _url_canonical = "https://www.youtube.com/watch?v={video_id}"
 
     # There are missing itags
     adp_video = {
@@ -126,29 +69,33 @@ class YouTube(Plugin):
     }
 
     def __init__(self, url):
-        super().__init__(url)
-        parsed = urlparse(self.url)
-        if parsed.netloc == 'gaming.youtube.com':
-            self.url = urlunparse(parsed._replace(netloc='www.youtube.com'))
+        match = self._re_url.match(url)
+        parsed = urlparse(url)
+
+        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"))
+        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.video_id = None
         self.session.http.headers.update({'User-Agent': useragents.CHROME})
 
     def get_author(self):
-        if self.author is None:
-            self.get_oembed
         return self.author
 
     def get_title(self):
-        if self.title is None:
-            self.get_oembed
         return self.title
 
     @classmethod
     def can_handle_url(cls, url):
-        return _url_re.match(url)
+        return cls._re_url.match(url)
 
     @classmethod
     def stream_weight(cls, stream):
@@ -167,42 +114,86 @@ class YouTube(Plugin):
 
         return weight, group
 
-    @property
-    def get_oembed(self):
-        if self.video_id is None:
-            self.video_id = self._find_video_id(self.url)
-
-        params = {
-            "url": "https://www.youtube.com/watch?v={0}".format(self.video_id),
-            "format": "json"
-        }
-        res = self.session.http.get(self._oembed_url, params=params)
-        data = self.session.http.json(res, schema=self._oembed_schema)
-        self.author = data["author_name"]
-        self.title = data["title"]
-
-    def _create_adaptive_streams(self, info, streams):
+    @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(
+            {"playabilityStatus": {
+                "status": str,
+                validate.optional("reason"): str
+            }},
+            validate.get("playabilityStatus"),
+            validate.union_get("status", "reason")
+        )
+        return validate.validate(schema, data)
+
+    @classmethod
+    def _schema_videodetails(cls, data):
+        schema = validate.Schema(
+            {"videoDetails": {
+                "videoId": str,
+                "author": str,
+                "title": str,
+                validate.optional("isLiveContent"): validate.transform(bool)
+            }},
+            validate.get("videoDetails"),
+            validate.union_get("videoId", "author", "title", "isLiveContent")
+        )
+        return validate.validate(schema, data)
+
+    @classmethod
+    def _schema_streamingdata(cls, data):
+        schema = validate.Schema(
+            {"streamingData": {
+                validate.optional("hlsManifestUrl"): str,
+                validate.optional("formats"): [validate.all(
+                    {
+                        "itag": int,
+                        "qualityLabel": str,
+                        validate.optional("url"): validate.url(scheme="http")
+                    },
+                    validate.union_get("url", "qualityLabel")
+                )],
+                validate.optional("adaptiveFormats"): [validate.all(
+                    {
+                        "itag": int,
+                        "mimeType": validate.all(
+                            str,
+                            validate.transform(cls._re_mime_type.search),
+                            validate.union_get("type", "codecs"),
+                        ),
+                        validate.optional("url"): validate.url(scheme="http"),
+                        validate.optional("qualityLabel"): str
+                    },
+                    validate.union_get("url", "qualityLabel", "itag", "mimeType")
+                )]
+            }},
+            validate.get("streamingData"),
+            validate.union_get("hlsManifestUrl", "formats", "adaptiveFormats")
+        )
+        hls_manifest, formats, adaptive_formats = validate.validate(schema, data)
+        return hls_manifest, formats or [], adaptive_formats or []
+
+    def _create_adaptive_streams(self, adaptive_formats):
+        streams = {}
         adaptive_streams = {}
         best_audio_itag = None
 
         # Extract audio streams from the adaptive format list
-        streaming_data = info.get("player_response", {}).get("streamingData", {})
-        for stream_info in streaming_data.get("adaptiveFormats", []):
-            if "url" not in stream_info:
-                continue
-            stream_params = dict(parse_qsl(stream_info["url"]))
-            if "itag" not in stream_params:
+        for url, label, itag, mimeType in adaptive_formats:
+            if url is None:
                 continue
-            itag = int(stream_params["itag"])
             # extract any high quality streams only available in adaptive formats
-            adaptive_streams[itag] = stream_info["url"]
-
-            stream_type, stream_format = stream_info["mimeType"]
+            adaptive_streams[itag] = url
+            stream_type, stream_codecs = mimeType
 
             if stream_type == "audio":
-                stream = HTTPStream(self.session, stream_info["url"])
-                name = "audio_{0}".format(stream_format)
-                streams[name] = stream
+                streams[f"audio_{stream_codecs}"] = HTTPStream(self.session, url)
 
                 # find the best quality audio stream m4a, opus or vorbis
                 if best_audio_itag is None or self.adp_audio[itag] > self.adp_audio[best_audio_itag]:
@@ -211,151 +202,75 @@ class YouTube(Plugin):
         if best_audio_itag and adaptive_streams and MuxedStream.is_usable(self.session):
             aurl = adaptive_streams[best_audio_itag]
             for itag, name in self.adp_video.items():
-                if itag in adaptive_streams:
-                    vurl = adaptive_streams[itag]
-                    log.debug("MuxedStream: v {video} a {audio} = {name}".format(
-                        audio=best_audio_itag,
-                        name=name,
-                        video=itag,
-                    ))
-                    streams[name] = MuxedStream(self.session,
-                                                HTTPStream(self.session, vurl),
-                                                HTTPStream(self.session, aurl))
+                if itag not in adaptive_streams:
+                    continue
+                vurl = adaptive_streams[itag]
+                log.debug(f"MuxedStream: v {itag} a {best_audio_itag} = {name}")
+                streams[name] = MuxedStream(
+                    self.session,
+                    HTTPStream(self.session, vurl),
+                    HTTPStream(self.session, aurl)
+                )
 
         return streams
 
-    def _find_video_id(self, url):
-        m = _url_re.match(url)
-        video_id = m.group("video_id") or m.group("video_id_short")
-        if video_id:
-            log.debug("Video ID from URL")
-            return video_id
-
+    def _get_data(self, url):
         res = self.session.http.get(url)
         if urlparse(res.url).netloc == "consent.youtube.com":
             c_data = {}
             for _i in itertags(res.text, "input"):
                 if _i.attributes.get("type") == "hidden":
-                    c_data[_i.attributes.get("name")] = _i.attributes.get("value")
+                    c_data[_i.attributes.get("name")] = unescape(_i.attributes.get("value"))
             log.debug(f"c_data_keys: {', '.join(c_data.keys())}")
             res = self.session.http.post("https://consent.youtube.com/s", data=c_data)
 
-        datam = _ytdata_re.search(res.text)
-        if datam:
-            data = parse_json(datam.group(1))
-            # find the videoRenderer object, where there is a LVE NOW badge
-            for vid_ep in search_dict(data, 'currentVideoEndpoint'):
-                video_id = vid_ep.get("watchEndpoint", {}).get("videoId")
-                if video_id:
-                    log.debug("Video ID from currentVideoEndpoint")
-                    return video_id
-            for x in search_dict(data, 'videoRenderer'):
-                if x.get("viewCountText", {}).get("runs"):
-                    if x.get("videoId"):
-                        log.debug("Video ID from videoRenderer (live)")
-                        return x["videoId"]
-                for bstyle in search_dict(x.get("badges", {}), "style"):
-                    if bstyle == "BADGE_STYLE_TYPE_LIVE_NOW":
-                        if x.get("videoId"):
-                            log.debug("Video ID from videoRenderer (live)")
-                            return x["videoId"]
-
-        if urlparse(url).path.endswith(("/embed/live_stream", "/live")):
-            for link in itertags(res.text, "link"):
-                if link.attributes.get("rel") == "canonical":
-                    canon_link = link.attributes.get("href")
-                    if canon_link != url:
-                        if canon_link.endswith("v=live_stream"):
-                            log.debug("The video is not available")
-                            break
-                        else:
-                            log.debug("Re-directing to canonical URL: {0}".format(canon_link))
-                            return self._find_video_id(canon_link)
-
-        raise PluginError("Could not find a video on this page")
-
-    def _get_stream_info(self, video_id):
-        # normal
-        _params_1 = {"el": "detailpage"}
-        # age restricted
-        _params_2 = {"el": "embedded"}
-        # embedded restricted
-        _params_3 = {"eurl": "https://youtube.googleapis.com/v/{0}".format(video_id)}
-
-        count = 0
-        info_parsed = None
-        for _params in (_params_1, _params_2, _params_3):
-            count += 1
-            params = {"video_id": video_id, "html5": "1"}
-            params.update(_params)
-
-            res = self.session.http.get(self._video_info_url, params=params)
-            info_parsed = parse_query(res.text, name="config", schema=_config_schema)
-            player_response = info_parsed.get("player_response", {})
-            playability_status = player_response.get("playabilityStatus", {})
-            if (playability_status.get("status") != "OK"):
-                reason = playability_status.get("reason")
-                log.debug("get_video_info - {0}: {1}".format(
-                    count, reason)
-                )
-                continue
-            self.author = player_response.get("videoDetails", {}).get("author")
-            self.title = player_response.get("videoDetails", {}).get("title")
-            log.debug("get_video_info - {0}: Found data".format(count))
-            break
+        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)
 
-        return info_parsed
+        match = re.search(self._re_ytInitialPlayerResponse, res.text)
+        if not match:
+            raise PluginError("Missing initial player response data")
 
-    def _get_streams(self):
-        is_live = False
+        return parse_json(match.group(1))
 
-        self.video_id = self._find_video_id(self.url)
-        log.debug(f"Using video ID: {self.video_id}")
+    def _get_streams(self):
+        data = self._get_data(self.url)
 
-        info = self._get_stream_info(self.video_id)
-        if info and info.get("status") == "fail":
-            log.error("Could not get video info: {0}".format(info.get("reason")))
-            return
-        elif not info:
-            log.error("Could not get video info")
+        status, reason = self._schema_playabilitystatus(data)
+        if status != "OK":
+            log.error(f"Could not get video info: {reason}")
             return
 
-        if info.get("player_response", {}).get("videoDetails", {}).get("isLive"):
+        video_id, self.author, self.title, is_live = self._schema_videodetails(data)
+        log.debug(f"Using video ID: {video_id}")
+
+        if is_live:
             log.debug("This video is live.")
-            is_live = True
 
         streams = {}
-        protected = False
-        if (info.get("player_response", {}).get("streamingData", {}).get("adaptiveFormats", [{}])[0].get("cipher")
-           or info.get("player_response", {}).get("streamingData", {}).get("adaptiveFormats", [{}])[0].get("signatureCipher")
-           or info.get("player_response", {}).get("streamingData", {}).get("formats", [{}])[0].get("cipher")):
-            protected = True
+        hls_manifest, formats, adaptive_formats = self._schema_streamingdata(data)
+
+        protected = next((True for url, *_ in formats + adaptive_formats if url is None), False)
+        if protected:
             log.debug("This video may be protected.")
 
-        for stream_info in info.get("player_response", {}).get("streamingData", {}).get("formats", []):
-            if "url" not in stream_info:
+        for url, label in formats:
+            if url is None:
                 continue
-            stream = HTTPStream(self.session, stream_info["url"])
-            name = stream_info["qualityLabel"]
-
-            streams[name] = stream
+            streams[label] = HTTPStream(self.session, url)
 
         if not is_live:
-            streams = self._create_adaptive_streams(info, streams)
+            streams.update(self._create_adaptive_streams(adaptive_formats))
 
-        hls_manifest = info.get("player_response", {}).get("streamingData", {}).get("hlsManifestUrl")
         if hls_manifest:
-            try:
-                hls_streams = HLSStream.parse_variant_playlist(
-                    self.session, hls_manifest, name_key="pixels"
-                )
-                streams.update(hls_streams)
-            except OSError as err:
-                log.warning(f"Failed to extract HLS streams: {err}")
+            streams.update(HLSStream.parse_variant_playlist(self.session, hls_manifest, name_key="pixels"))
 
         if not streams and protected:
-            raise PluginError("This plugin does not support protected videos, "
-                              "try youtube-dl instead")
+            raise PluginError("This plugin does not support protected videos, try youtube-dl instead")
 
         return streams
 


Test patch:
diff --git a/tests/plugins/test_youtube.py b/tests/plugins/test_youtube.py
index 7bcbce62..ef3dc8d9 100644
--- a/tests/plugins/test_youtube.py
+++ b/tests/plugins/test_youtube.py
@@ -1,6 +1,6 @@
 import unittest
 
-from streamlink.plugins.youtube import YouTube, _url_re
+from streamlink.plugins.youtube import YouTube
 from tests.plugins import PluginCanHandleUrl
 
 
@@ -9,19 +9,18 @@ class TestPluginCanHandleUrlYouTube(PluginCanHandleUrl):
 
     should_match = [
         "https://www.youtube.com/EXAMPLE/live",
-        "https://www.youtube.com/c/EXAMPLE",
-        "https://www.youtube.com/c/EXAMPLE/",
+        "https://www.youtube.com/EXAMPLE/live/",
         "https://www.youtube.com/c/EXAMPLE/live",
         "https://www.youtube.com/c/EXAMPLE/live/",
-        "https://www.youtube.com/channel/EXAMPLE",
-        "https://www.youtube.com/channel/EXAMPLE/",
+        "https://www.youtube.com/channel/EXAMPLE/live",
+        "https://www.youtube.com/channel/EXAMPLE/live/",
+        "https://www.youtube.com/user/EXAMPLE/live",
+        "https://www.youtube.com/user/EXAMPLE/live/",
         "https://www.youtube.com/embed/aqz-KE-bpKQ",
         "https://www.youtube.com/embed/live_stream?channel=UCNye-wNBqNL5ZzHSJj3l8Bg",
-        "https://www.youtube.com/user/EXAMPLE",
-        "https://www.youtube.com/user/EXAMPLE/",
-        "https://www.youtube.com/user/EXAMPLE/live",
         "https://www.youtube.com/v/aqz-KE-bpKQ",
         "https://www.youtube.com/watch?v=aqz-KE-bpKQ",
+        "https://www.youtube.com/watch?foo=bar&baz=qux&v=aqz-KE-bpKQ",
         "https://youtu.be/0123456789A",
     ]
 
@@ -31,6 +30,9 @@ class TestPluginCanHandleUrlYouTube(PluginCanHandleUrl):
         "https://www.youtube.com/account",
         "https://www.youtube.com/feed/guide_builder",
         "https://www.youtube.com/t/terms",
+        "https://www.youtube.com/c/EXAMPLE",
+        "https://www.youtube.com/channel/EXAMPLE",
+        "https://www.youtube.com/user/EXAMPLE",
         "https://youtu.be",
         "https://youtu.be/",
         "https://youtu.be/c/CHANNEL",
@@ -40,7 +42,7 @@ class TestPluginCanHandleUrlYouTube(PluginCanHandleUrl):
 
 class TestPluginYouTube(unittest.TestCase):
     def _test_regex(self, url, expected_string, expected_group):
-        m = _url_re.match(url)
+        m = YouTube._re_url.match(url)
         self.assertIsNotNone(m)
         self.assertEqual(expected_string, m.group(expected_group))