Merge pull request #48 from hbmartin/error-404

Pipfile

@@ -21,11 +21,14 @@ flake8-executable = "*"
 flake8-if-expr = "*"
 flake8-isort = "*"
 flake8-logging-format = "*"
+flake8-mock = "*"
+flake8-mutable = "*"
 flake8-print = "*"
 flake8-pytest = "*"
 flake8-pytest-style = "*"
 flake8-quotes = "*"
 flake8-return = "*"
+flake8-strict = "*"
 flake8-string-format = "*"
 mypy = "*"
 pep8-naming = "*"

pytube/__main__.py

@@ -20,7 +20,7 @@ from pytube import extract
 from pytube import request
 from pytube import Stream
 from pytube import StreamQuery
-from pytube.extract import apply_descrambler, apply_signature
+from pytube.extract import apply_descrambler, apply_signature, get_ytplayer_config
 from pytube.helpers import install_proxy
 from pytube.exceptions import VideoUnavailable
 from pytube.monostate import OnProgress, OnComplete, Monostate
@@ -76,12 +76,10 @@ class YouTube:
         # video_id part of /watch?v=<video_id>
         self.video_id = extract.video_id(url)
 
-        # https://www.youtube.com/watch?v=<video_id>
-        self.watch_url = extract.watch_url(self.video_id)
+        self.watch_url = f"https://youtube.com/watch?v={self.video_id}"
+        self.embed_url = f"https://www.youtube.com/embed/{self.video_id}"
 
-        self.embed_url = extract.embed_url(self.video_id)
-        # A dictionary shared between all instances of :class:`Stream <Stream>`
-        # (Borg pattern). Boooooo.
+        # Shared between all instances of `Stream` (Borg pattern).
         self.stream_monostate = Monostate(
             on_progress=on_progress_callback, on_complete=on_complete_callback
         )
@@ -111,9 +109,7 @@ class YouTube:
             self.player_config_args = self.vid_info
         else:
             assert self.watch_html is not None
-            self.player_config_args = extract.get_ytplayer_config(self.watch_html,)[
-                "args"
-            ]
+            self.player_config_args = get_ytplayer_config(self.watch_html)["args"]
 
             # Fix for KeyError: 'title' issue #434
             if "title" not in self.player_config_args:  # type: ignore
@@ -140,8 +136,9 @@ class YouTube:
                     self.player_config_args, fmt, self.js  # type: ignore
                 )
             except TypeError:
-                assert self.embed_html is not None
-                self.js_url = extract.js_url(self.embed_html, self.age_restricted)
+                if not self.embed_html:
+                    self.embed_html = request.get(url=self.embed_url)
+                self.js_url = extract.js_url(self.embed_html)
                 self.js = request.get(self.js_url)
                 assert self.js is not None
                 apply_signature(self.player_config_args, fmt, self.js)
@@ -152,6 +149,8 @@ class YouTube:
         # load the player_response object (contains subtitle information)
         self.player_response = json.loads(self.player_config_args["player_response"])
         del self.player_config_args["player_response"]
+        self.stream_monostate.title = self.title
+        self.stream_monostate.duration = self.length
 
         logger.info("init finished successfully")
 
@@ -166,23 +165,29 @@ class YouTube:
 
         """
         self.watch_html = request.get(url=self.watch_url)
-        if (
-            self.watch_html is None
-            or '<img class="icon meh" src="/yts/img' not in self.watch_html
+        if self.watch_html is None:
+            raise VideoUnavailable(video_id=self.video_id)
+        self.age_restricted = extract.is_age_restricted(self.watch_html)
+        if not self.age_restricted and (
+            "yt-badge-live" in self.watch_html
+            or "This video is private" in self.watch_html
         ):
             raise VideoUnavailable(video_id=self.video_id)
 
-        self.embed_html = request.get(url=self.embed_url)
-        self.age_restricted = extract.is_age_restricted(self.watch_html)
-        self.vid_info_url = extract.video_info_url(
-            video_id=self.video_id,
-            watch_url=self.watch_url,
-            embed_html=self.embed_html,
-            age_restricted=self.age_restricted,
-        )
+        if self.age_restricted:
+            if not self.embed_html:
+                self.embed_html = request.get(url=self.embed_url)
+            self.vid_info_url = extract.video_info_url_age_restricted(
+                self.video_id, self.watch_url
+            )
+        else:
+            self.vid_info_url = extract.video_info_url(
+                video_id=self.video_id, watch_url=self.watch_url
+            )
+
         self.vid_info_raw = request.get(self.vid_info_url)
         if not self.age_restricted:
-            self.js_url = extract.js_url(self.watch_html, self.age_restricted)
+            self.js_url = extract.js_url(self.watch_html)
             self.js = request.get(self.js_url)
 
     def initialize_stream_objects(self, fmt: str) -> None:
@@ -275,7 +280,7 @@ class YouTube:
         """
         return self.player_response.get("videoDetails", {}).get(
             "shortDescription"
-        ) or extract.get_vid_descr(self.watch_html)
+        ) or extract._get_vid_descr(self.watch_html)
 
     @property
     def rating(self) -> float:

pytube/extract.py

@@ -77,66 +77,58 @@ def video_id(url: str) -> str:
     return regex_search(r"(?:v=|\/)([0-9A-Za-z_-]{11}).*", url, group=1)
 
 
-def watch_url(video_id: str) -> str:
-    """Construct a sanitized YouTube watch url, given a video id.
+def video_info_url(video_id: str, watch_url: str) -> str:
+    """Construct the video_info url.
 
     :param str video_id:
         A YouTube video identifier.
+    :param str watch_url:
+        A YouTube watch url.
     :rtype: str
     :returns:
-        Sanitized YouTube watch url.
+        :samp:`https://youtube.com/get_video_info` with necessary GET
+        parameters.
     """
-    return "https://youtube.com/watch?v=" + video_id
-
-
-def embed_url(video_id: str) -> str:
-    return f"https://www.youtube.com/embed/{video_id}"
-
-
-def eurl(video_id: str) -> str:
-    return f"https://youtube.googleapis.com/v/{video_id}"
+    params = OrderedDict(
+        [
+            ("video_id", video_id),
+            ("el", "$el"),
+            ("ps", "default"),
+            ("eurl", quote(watch_url)),
+            ("hl", "en_US"),
+        ]
+    )
+    return _video_info_url(params)
 
 
-def video_info_url(
-    video_id: str, watch_url: str, embed_html: Optional[str], age_restricted: bool,
-) -> str:
+def video_info_url_age_restricted(video_id: str, embed_html: str) -> str:
     """Construct the video_info url.
 
     :param str video_id:
         A YouTube video identifier.
-    :param str watch_url:
-        A YouTube watch url.
     :param str embed_html:
         The html contents of the embed page (for age restricted videos).
-    :param bool age_restricted:
-        Is video age restricted.
     :rtype: str
     :returns:
         :samp:`https://youtube.com/get_video_info` with necessary GET
         parameters.
     """
-    if age_restricted:
-        assert embed_html is not None
+    try:
         sts = regex_search(r'"sts"\s*:\s*(\d+)', embed_html, group=1)
-        # Here we use ``OrderedDict`` so that the output is consistent between
-        # Python 2.7+.
-        params = OrderedDict(
-            [("video_id", video_id), ("eurl", eurl(video_id)), ("sts", sts),]
-        )
-    else:
-        params = OrderedDict(
-            [
-                ("video_id", video_id),
-                ("el", "$el"),
-                ("ps", "default"),
-                ("eurl", quote(watch_url)),
-                ("hl", "en_US"),
-            ]
-        )
+    except RegexMatchError:
+        sts = ""
+    # Here we use ``OrderedDict`` so that the output is consistent between
+    # Python 2.7+.
+    eurl = f"https://youtube.googleapis.com/v/{video_id}"
+    params = OrderedDict([("video_id", video_id), ("eurl", eurl), ("sts", sts),])
+    return _video_info_url(params)
+
+
+def _video_info_url(params: OrderedDict) -> str:
     return "https://youtube.com/get_video_info?" + urlencode(params)
 
 
-def js_url(html: str, age_restricted: Optional[bool] = False) -> str:
+def js_url(html: str) -> str:
     """Get the base JavaScript url.
 
     Construct the base JavaScript url, which contains the decipher
@@ -144,12 +136,8 @@ def js_url(html: str, age_restricted: Optional[bool] = False) -> str:
 
     :param str html:
         The html contents of the watch page.
-    :param bool age_restricted:
-        Is video age restricted.
-
     """
-    ytplayer_config = get_ytplayer_config(html, age_restricted or False)
-    base_js = ytplayer_config["assets"]["js"]
+    base_js = get_ytplayer_config(html)["assets"]["js"]
     return "https://youtube.com" + base_js
 
 
@@ -180,7 +168,7 @@ def mime_type_codec(mime_type_codec: str) -> Tuple[str, List[str]]:
     return mime_type, [c.strip() for c in codecs.split(",")]
 
 
-def get_ytplayer_config(html: str, age_restricted: bool = False) -> Any:
+def get_ytplayer_config(html: str) -> Any:
     """Get the YouTube player configuration data from the watch html.
 
     Extract the ``ytplayer_config``, which is json data embedded within the
@@ -189,21 +177,29 @@ def get_ytplayer_config(html: str, age_restricted: bool = False) -> Any:
 
     :param str html:
         The html contents of the watch page.
-    :param bool age_restricted:
-        Is video age restricted.
     :rtype: str
     :returns:
         Substring of the html containing the encoded manifest data.
     """
-    if age_restricted:
-        pattern = r";yt\.setConfig\(\{'PLAYER_CONFIG':\s*({.*})(,'EXPERIMENT_FLAGS'|;)"  # noqa: E501
-    else:
-        pattern = r";ytplayer\.config\s*=\s*({.*?});"
-    yt_player_config = regex_search(pattern, html, group=1)
-    return json.loads(yt_player_config)
-
-
-def get_vid_descr(html: Optional[str]) -> str:
+    config_patterns = [
+        r";ytplayer\.config\s*=\s*({.*?});",
+        r";ytplayer\.config\s*=\s*({.+?});ytplayer",
+        r";yt\.setConfig\(\{'PLAYER_CONFIG':\s*({.*})}\);",
+        r";yt\.setConfig\(\{'PLAYER_CONFIG':\s*({.*})(,'EXPERIMENT_FLAGS'|;)",  # noqa: E501
+    ]
+    logger.debug("finding initial function name")
+    for pattern in config_patterns:
+        regex = re.compile(pattern)
+        function_match = regex.search(html)
+        if function_match:
+            logger.debug("finished regex search, matched: %s", pattern)
+            yt_player_config = function_match.group(1)
+            return json.loads(yt_player_config)
+
+    raise RegexMatchError(caller="get_ytplayer_config", pattern="config_patterns")
+
+
+def _get_vid_descr(html: Optional[str]) -> str:
     html_parser = PytubeHTMLParser()
     if html:
         html_parser.feed(html)
@@ -278,6 +274,8 @@ def apply_descrambler(stream_data: Dict, key: str) -> None:
     {'foo': [{'bar': '1', 'var': 'test'}, {'em': '5', 't': 'url encoded'}]}
 
     """
+    otf_type = "FORMAT_STREAM_TYPE_OTF"
+
     if key == "url_encoded_fmt_stream_map" and not stream_data.get(
         "url_encoded_fmt_stream_map"
     ):
@@ -294,6 +292,8 @@ def apply_descrambler(stream_data: Dict, key: str) -> None:
                     "type": format_item["mimeType"],
                     "quality": format_item["quality"],
                     "itag": format_item["itag"],
+                    "bitrate": format_item.get("bitrate"),
+                    "is_otf": (format_item.get("type") == otf_type),
                 }
                 for format_item in formats
             ]
@@ -308,6 +308,8 @@ def apply_descrambler(stream_data: Dict, key: str) -> None:
                     "type": format_item["mimeType"],
                     "quality": format_item["quality"],
                     "itag": format_item["itag"],
+                    "bitrate": format_item.get("bitrate"),
+                    "is_otf": (format_item.get("type") == otf_type),
                 }
                 for i, format_item in enumerate(formats)
             ]

pytube/monostate.py

@@ -52,7 +52,13 @@ class OnComplete(Protocol):
 
 class Monostate:
     def __init__(
-        self, on_progress: Optional[OnProgress], on_complete: Optional[OnComplete]
+        self,
+        on_progress: Optional[OnProgress],
+        on_complete: Optional[OnComplete],
+        title: Optional[str] = None,
+        duration: Optional[int] = None,
     ):
         self.on_progress = on_progress
         self.on_complete = on_complete
+        self.title = title
+        self.duration = duration

pytube/query.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 
 """This module provides a query interface for media streams and captions."""
-from typing import List, Optional
+from typing import List, Optional, Callable
 
 from pytube import Stream, Caption
 
@@ -168,9 +168,12 @@ class StreamQuery:
         if is_dash is not None:
             filters.append(lambda s: s.is_dash == is_dash)
 
+        return self._filter(filters)
+
+    def _filter(self, filters: List[Callable]) -> "StreamQuery":
         fmt_streams = self.fmt_streams
-        for fn in filters:
-            fmt_streams = filter(fn, fmt_streams)
+        for filter_lambda in filters:
+            fmt_streams = filter(filter_lambda, fmt_streams)
         return StreamQuery(list(fmt_streams))
 
     def order_by(self, attribute_name: str) -> "StreamQuery":
@@ -281,10 +284,18 @@ class StreamQuery:
         :returns:
             The :class:`Stream <Stream>` matching the given itag or None if
             not found.
-
         """
         return self.filter(only_audio=True, subtype=subtype).order_by("abr").last()
 
+    def otf(self, is_otf: bool = False) -> "StreamQuery":
+        """Filter stream by OTF, useful if some streams have 404 URLs
+
+        :param bool is_otf: Set to False to retrieve only non-OTF streams
+        :rtype: :class:`StreamQuery <StreamQuery>`
+        :returns: A StreamQuery object with otf filtered streams
+        """
+        return self._filter([lambda s: s.is_otf == is_otf])
+
     def first(self) -> Optional[Stream]:
         """Get the first :class:`Stream <Stream>` in the results.
 

pytube/streams.py

@@ -58,6 +58,9 @@ class Stream:
         # streams return NoneType for audio/video depending.
         self.video_codec, self.audio_codec = self.parse_codecs()
 
+        self.is_otf: bool = stream["is_otf"]
+        self.bitrate: Optional[int] = stream["bitrate"]
+
         self._filesize: Optional[int] = None  # filesize in bytes
 
         # Additional information about the stream format, such as resolution,
@@ -152,15 +155,22 @@ class Stream:
         :returns:
             Youtube video title
         """
-        return (
-            self.player_config_args.get("title")
-            or (
-                self.player_config_args.get("player_response", {})
-                .get("videoDetails", {})
-                .get("title")
-            )
-            or "Unknown YouTube Video Title"
-        )
+        return self._monostate.title or "Unknown YouTube Video Title"
+
+    @property
+    def filesize_approx(self) -> int:
+        """Get approximate filesize of the video
+
+        Falls back to HTTP call if there is not sufficient information to approximate
+
+        :rtype: int
+        :returns: size of video in bytes
+        """
+        if self._monostate.duration and self.bitrate:
+            bits_in_byte = 8
+            return int((self._monostate.duration * self.bitrate) / bits_in_byte)
+
+        return self.filesize
 
     @property
     def default_filename(self) -> str:

tests/conftest.py

@@ -48,7 +48,7 @@ def presigned_video():
 @pytest.fixture
 def age_restricted():
     """Youtube instance initialized with video id zRbsm3e2ltw."""
-    filename = "yt-video-zRbsm3e2ltw-1507777044.json.gz"
+    filename = "yt-video-irauhITDrsE.json.gz"
     return load_playback_file(filename)
 
 

tests/generate_fixture.py

@@ -19,6 +19,7 @@ output = {
     "watch_html": yt.watch_html,
     "video_info": yt.vid_info,
     "js": yt.js,
+    "embed_html": yt.embed_html,
 }
 
 outpath = path.join(currentdir, "mocks", "yt-video-" + yt.video_id + ".json")

tests/test_extract.py

@@ -12,18 +12,20 @@ def test_extract_video_id():
     assert video_id == "9bZkp7q19f0"
 
 
-def test_extract_watch_url():
-    video_id = "9bZkp7q19f0"
-    watch_url = extract.watch_url(video_id)
-    assert watch_url == "https://youtube.com/watch?v=9bZkp7q19f0"
+def test_info_url(age_restricted):
+    video_info_url = extract.video_info_url_age_restricted(
+        video_id="QRS8MkLhQmM", embed_html=age_restricted["embed_html"],
+    )
+    expected = (
+        "https://youtube.com/get_video_info?video_id=QRS8MkLhQmM&eurl"
+        "=https%3A%2F%2Fyoutube.googleapis.com%2Fv%2FQRS8MkLhQmM&sts="
+    )
+    assert video_info_url == expected
 
 
-def test_info_url(cipher_signature):
+def test_info_url_age_restricted(cipher_signature):
     video_info_url = extract.video_info_url(
-        video_id=cipher_signature.video_id,
-        watch_url=cipher_signature.watch_url,
-        embed_html="",
-        age_restricted=False,
+        video_id=cipher_signature.video_id, watch_url=cipher_signature.watch_url
     )
     expected = (
         "https://youtube.com/get_video_info?video_id=9bZkp7q19f0&el=%24el"
@@ -63,12 +65,7 @@ def test_get_vid_desc(cipher_signature):
         "http://sptfy.com/PSY\n"
         "http://weibo.com/psyoppa"
     )
-    assert extract.get_vid_descr(cipher_signature.watch_html) == expected
-
-
-def test_eurl():
-    url = extract.eurl("videoid")
-    assert url == "https://youtube.googleapis.com/v/videoid"
+    assert extract._get_vid_descr(cipher_signature.watch_html) == expected
 
 
 def test_mime_type_codec():

tests/test_query.py

@@ -162,3 +162,11 @@ def test_get_audio_only(cipher_signature):
 
 def test_get_audio_only_with_subtype(cipher_signature):
     assert cipher_signature.streams.get_audio_only(subtype="webm").itag == 251
+
+
+def test_otf(cipher_signature):
+    non_otf = cipher_signature.streams.otf().all()
+    assert len(non_otf) == 22
+
+    otf = cipher_signature.streams.otf(True).all()
+    assert len(otf) == 0

tests/test_streams.py

@@ -14,6 +14,15 @@ def test_filesize(cipher_signature, mocker):
     assert cipher_signature.streams.first().filesize == 6796391
 
 
+def test_filesize_approx(cipher_signature, mocker):
+    mocker.patch.object(request, "head")
+    request.head.return_value = {"content-length": "123"}
+    stream = cipher_signature.streams.first()
+    assert stream.filesize_approx == 22350604
+    stream.bitrate = None
+    assert stream.filesize_approx == 123
+
+
 def test_default_filename(cipher_signature):
     expected = "PSY - GANGNAM STYLE(강남스타일) MV.mp4"
     stream = cipher_signature.streams.first()
@@ -21,19 +30,14 @@ def test_default_filename(cipher_signature):
 
 
 def test_title(cipher_signature):
-    expected = "PSY - GANGNAM STYLE(강남스타일) M/V"
-    stream = cipher_signature.streams.first()
-    assert stream.title == expected
-
-    expected = "PSY - GANGNAM STYLE(강남스타일)"
-    stream.player_config_args = {
-        "player_response": {"videoDetails": {"title": expected}},
-    }
-    assert stream.title == expected
-
-    expected = "Unknown YouTube Video Title"
-    stream.player_config_args = {}
-    assert stream.title == expected
+    expected = "title"
+    cipher_signature.player_config_args["title"] = expected
+    assert cipher_signature.title == expected
+
+    expected = "title2"
+    del cipher_signature.player_config_args["title"]
+    cipher_signature.player_response = {"videoDetails": {"title": expected}}
+    assert cipher_signature.title == expected
 
 
 def test_caption_tracks(presigned_video):