Merge branch 'master' into added-tests-cli

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 = "*"

README.md

@@ -36,14 +36,12 @@ $ pip install pytube3 --upgrade
 ## Quick start
 ```python
  >>> from pytube import YouTube
- >>> YouTube('https://youtu.be/9bZkp7q19f0').streams.first().download()
+ >>> YouTube('https://youtu.be/9bZkp7q19f0').streams[0].download()
  >>>
  >>> yt = YouTube('http://youtube.com/watch?v=9bZkp7q19f0')
  >>> yt.streams
   ... .filter(progressive=True, file_extension='mp4')
-  ... .order_by('resolution')
-  ... .desc()
-  ... .first()
+  ... .order_by('resolution')[-1]
   ... .download()
 ```
 
@@ -64,7 +62,7 @@ Let's begin with showing how easy it is to download a video with pytube:
 
 ```python
 >>> from pytube import YouTube
->>> YouTube('http://youtube.com/watch?v=9bZkp7q19f0').streams.first().download()
+>>> YouTube('http://youtube.com/watch?v=9bZkp7q19f0').streams[0].download()
 ```
 This example will download the highest quality progressive download stream available.
 
@@ -72,7 +70,7 @@ Next, let's explore how we would view what video streams are available:
 
 ```python
 >>> yt = YouTube('http://youtube.com/watch?v=9bZkp7q19f0')
->>> yt.streams.all()
+>>> print(yt.streams)
  [<Stream: itag="22" mime_type="video/mp4" res="720p" fps="30fps" vcodec="avc1.64001F" acodec="mp4a.40.2">,
  <Stream: itag="43" mime_type="video/webm" res="360p" fps="30fps" vcodec="vp8.0" acodec="vorbis">,
  <Stream: itag="18" mime_type="video/mp4" res="360p" fps="30fps" vcodec="avc1.42001E" acodec="mp4a.40.2">,
@@ -108,7 +106,7 @@ The legacy streams that contain the audio and video in a single file (referred t
 To only view these progressive download streams:
 
 ```python
- >>> yt.streams.filter(progressive=True).all()
+ >>> yt.streams.filter(progressive=True)
   [<Stream: itag="22" mime_type="video/mp4" res="720p" fps="30fps" vcodec="avc1.64001F" acodec="mp4a.40.2">,
   <Stream: itag="43" mime_type="video/webm" res="360p" fps="30fps" vcodec="vp8.0" acodec="vorbis">,
   <Stream: itag="18" mime_type="video/mp4" res="360p" fps="30fps" vcodec="avc1.42001E" acodec="mp4a.40.2">,
@@ -119,7 +117,7 @@ To only view these progressive download streams:
 Conversely, if you only want to see the DASH streams (also referred to as "adaptive") you can do:
 
 ```python
->>> yt.streams.filter(adaptive=True).all()
+>>> yt.streams.filter(adaptive=True)
  [<Stream: itag="137" mime_type="video/mp4" res="1080p" fps="30fps" vcodec="avc1.640028">,
   <Stream: itag="248" mime_type="video/webm" res="1080p" fps="30fps" vcodec="vp9">,
   <Stream: itag="136" mime_type="video/mp4" res="720p" fps="30fps" vcodec="avc1.4d401f">,
@@ -146,7 +144,7 @@ You can also download a complete Youtube playlist:
 ```python
 >>> from pytube import Playlist
 >>> playlist = Playlist("https://www.youtube.com/playlist?list=PLynhp4cZEpTbRs_PYISQ8v_uwO0_mDg_X")
->>> for video in playlist.videos:
+>>> for video in playlist:
 >>> 	video.streams.get_highest_resolution().download()
 ```
 This will download the highest progressive stream available (generally 720p) from the given playlist.
@@ -158,7 +156,7 @@ Pytube allows you to filter on every property available (see the documentation f
 To list the audio only streams:
 
 ```python
->>> yt.streams.filter(only_audio=True).all()
+>>> yt.streams.filter(only_audio=True)
   [<Stream: itag="140" mime_type="audio/mp4" abr="128kbps" acodec="mp4a.40.2">,
   <Stream: itag="171" mime_type="audio/webm" abr="128kbps" acodec="vorbis">,
   <Stream: itag="249" mime_type="audio/webm" abr="50kbps" acodec="opus">,
@@ -169,7 +167,7 @@ To list the audio only streams:
 To list only ``mp4`` streams:
 
 ```python
->>> yt.streams.filter(subtype='mp4').all()
+>>> yt.streams.filter(subtype='mp4')
  [<Stream: itag="22" mime_type="video/mp4" res="720p" fps="30fps" vcodec="avc1.64001F" acodec="mp4a.40.2">,
   <Stream: itag="18" mime_type="video/mp4" res="360p" fps="30fps" vcodec="avc1.42001E" acodec="mp4a.40.2">,
   <Stream: itag="137" mime_type="video/mp4" res="1080p" fps="30fps" vcodec="avc1.640028">,
@@ -184,9 +182,9 @@ To list only ``mp4`` streams:
 Multiple filters can also be specified:
 
 ```python
->>> yt.streams.filter(subtype='mp4', progressive=True).all()
+>>> yt.streams.filter(subtype='mp4', progressive=True)
 >>> # this can also be expressed as:
->>> yt.streams.filter(subtype='mp4').filter(progressive=True).all()
+>>> yt.streams.filter(subtype='mp4').filter(progressive=True)
   [<Stream: itag="22" mime_type="video/mp4" res="720p" fps="30fps" vcodec="avc1.64001F" acodec="mp4a.40.2">,
   <Stream: itag="18" mime_type="video/mp4" res="360p" fps="30fps" vcodec="avc1.42001E" acodec="mp4a.40.2">]
 ```
@@ -200,7 +198,7 @@ You also have an interface to select streams by their itag, without needing to f
 If you need to optimize for a specific feature, such as the "highest resolution" or "lowest average bitrate":
 
 ```python
->>> yt.streams.filter(progressive=True).order_by('resolution').desc().all()
+>>> yt.streams.filter(progressive=True).order_by('resolution').desc()
 ```
 Note: Using ``order_by`` on a given attribute will filter out all streams missing that attribute.
 

pytube/__main__.py

@@ -20,9 +20,9 @@ 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.exceptions import VideoUnavailable, LiveStreamError
 from pytube.monostate import OnProgress, OnComplete, Monostate
 
 logger = logging.getLogger(__name__)
@@ -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
@@ -135,16 +131,13 @@ class YouTube:
                 apply_descrambler(self.vid_info, fmt)
             apply_descrambler(self.player_config_args, fmt)
 
-            try:
-                apply_signature(
-                    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.js:
+                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)
+
+            apply_signature(self.player_config_args, fmt, self.js)
 
             # build instances of :class:`Stream <Stream>`
             self.initialize_stream_objects(fmt)
@@ -152,6 +145,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")
 
@@ -163,26 +158,32 @@ class YouTube:
         which blocks for long periods of time.
 
         :rtype: None
-
         """
         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.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 not self.age_restricted:
+            if "yt-badge-live" in self.watch_html:
+                raise LiveStreamError(self.video_id)
+            if "This video is private" in self.watch_html:
+                raise VideoUnavailable(video_id=self.video_id)
+
+        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 +276,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/cli.py

@@ -400,12 +400,12 @@ def display_streams(youtube: YouTube) -> None:
         A valid YouTube watch URL.
 
     """
-    for stream in youtube.streams.all():
+    for stream in youtube.streams:
         print(stream)
 
 
 def _print_available_captions(captions: CaptionQuery) -> None:
-    print(f"Available caption codes are: {', '.join(c.code for c in captions.all())}")
+    print(f"Available caption codes are: {', '.join(c.code for c in captions)}")
 
 
 def download_caption(

pytube/contrib/playlist.py

@@ -6,8 +6,9 @@ import json
 import logging
 import re
 from datetime import date, datetime
-from typing import List, Optional, Iterable, Dict
+from typing import List, Optional, Iterable, Dict, Union
 from urllib.parse import parse_qs
+from collections.abc import Sequence
 
 from pytube import request, YouTube
 from pytube.helpers import cache, deprecated, install_proxy, uniqueify
@@ -15,7 +16,7 @@ from pytube.helpers import cache, deprecated, install_proxy, uniqueify
 logger = logging.getLogger(__name__)
 
 
-class Playlist:
+class Playlist(Sequence):
     """Load a YouTube playlist with URL or ID"""
 
     def __init__(self, url: str, proxies: Optional[Dict[str, str]] = None):
@@ -142,10 +143,19 @@ class Playlist:
         """
         yield from (YouTube(url) for url in self.video_urls)
 
+    def __getitem__(self, i: Union[slice, int]) -> Union[str, List[str]]:
+        return self.video_urls[i]
+
+    def __len__(self) -> int:
+        return len(self.video_urls)
+
+    def __repr__(self) -> str:
+        return f"{self.video_urls}"
+
     @deprecated(
         "This call is unnecessary, you can directly access .video_urls or .videos"
     )
-    def populate_video_urls(self) -> List[str]:
+    def populate_video_urls(self) -> List[str]:  # pragma: no cover
         """Complete links of all the videos in playlist
 
         :rtype: List[str]

pytube/exceptions.py

@@ -35,6 +35,15 @@ class RegexMatchError(ExtractError):
 class LiveStreamError(ExtractError):
     """Video is a live stream."""
 
+    def __init__(self, video_id: str):
+        """
+        :param str video_id:
+            A YouTube video identifier.
+        """
+        super().__init__(f"{video_id} is streaming live and cannot be loaded")
+
+        self.video_id = video_id
+
 
 class VideoUnavailable(PytubeError):
     """Video is unavailable."""

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)
@@ -235,7 +231,7 @@ def apply_signature(config_args: Dict, fmt: str, js: str) -> None:
             url: str = stream["url"]
         except KeyError:
             if live_stream:
-                raise LiveStreamError("Video is currently being streamed live")
+                raise LiveStreamError("UNKNOWN")
         # 403 Forbidden fix.
         if "signature" in url or (
             "s" not in stream and ("&sig=" in url or "&lsig=" in url)
@@ -246,12 +242,7 @@ def apply_signature(config_args: Dict, fmt: str, js: str) -> None:
             logger.debug("signature found, skip decipher")
             continue
 
-        if js is not None:
-            signature = cipher.get_signature(ciphered_signature=stream["s"])
-        else:
-            # signature not present in url (line 33), need js to descramble
-            # TypeError caught in __main__
-            raise TypeError("JS is None")
+        signature = cipher.get_signature(ciphered_signature=stream["s"])
 
         logger.debug("finished descrambling signature for itag=%s", stream["itag"])
         # 403 forbidden fix
@@ -278,6 +269,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 +287,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 +303,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,12 +1,14 @@
 # -*- coding: utf-8 -*-
 
 """This module provides a query interface for media streams and captions."""
-from typing import List, Optional
+from typing import Callable, List, Optional, Union
+from collections.abc import Mapping, Sequence
 
 from pytube import Stream, Caption
+from pytube.helpers import deprecated
 
 
-class StreamQuery:
+class StreamQuery(Sequence):
     """Interface for querying the available media streams."""
 
     def __init__(self, fmt_streams):
@@ -168,9 +170,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 +286,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.
 
@@ -313,15 +326,19 @@ class StreamQuery:
         except IndexError:
             pass
 
-    def count(self) -> int:
-        """Get the count the query would return.
+    @deprecated("Get the size of this list directly using len()")
+    def count(self, value: Optional[str] = None) -> int:  # pragma: no cover
+        """Get the count of items in the list.
 
         :rtype: int
-
         """
-        return len(self.fmt_streams)
+        if value:
+            return self.fmt_streams.count(value)
+
+        return len(self)
 
-    def all(self) -> List[Stream]:
+    @deprecated("This object can be treated as a list, all() is useless")
+    def all(self) -> List[Stream]:  # pragma: no cover
         """Get all the results represented by this query as a list.
 
         :rtype: list
@@ -329,8 +346,17 @@ class StreamQuery:
         """
         return self.fmt_streams
 
+    def __getitem__(self, i: Union[slice, int]):
+        return self.fmt_streams[i]
 
-class CaptionQuery:
+    def __len__(self) -> int:
+        return len(self.fmt_streams)
+
+    def __repr__(self) -> str:
+        return f"{self.fmt_streams}"
+
+
+class CaptionQuery(Mapping):
     """Interface for querying the available captions."""
 
     def __init__(self, captions: List[Caption]):
@@ -340,9 +366,9 @@ class CaptionQuery:
             list of :class:`Caption <Caption>` instances.
 
         """
-        self.captions = captions
         self.lang_code_index = {c.code: c for c in captions}
 
+    @deprecated("This object can be treated as a dictionary, i.e. captions['en']")
     def get_by_language_code(self, lang_code: str) -> Optional[Caption]:
         """Get the :class:`Caption <Caption>` for a given ``lang_code``.
 
@@ -355,10 +381,23 @@ class CaptionQuery:
         """
         return self.lang_code_index.get(lang_code)
 
-    def all(self) -> List[Caption]:
+    @deprecated("This object can be treated as a dictionary")
+    def all(self) -> List[Caption]:  # pragma: no cover
         """Get all the results represented by this query as a list.
 
         :rtype: list
 
         """
-        return self.captions
+        return list(self.lang_code_index.values())
+
+    def __getitem__(self, i: str):
+        return self.lang_code_index[i]
+
+    def __len__(self) -> int:
+        return len(self.lang_code_index)
+
+    def __iter__(self):
+        return iter(self.lang_code_index)
+
+    def __repr__(self) -> str:
+        return f"{self.lang_code_index}"

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/contrib/test_playlist.py

@@ -81,6 +81,39 @@ def test_video_urls(request_get, playlist_html):
     ]
 
 
+@mock.patch("pytube.contrib.playlist.request.get")
+def test_repr(request_get, playlist_html):
+    url = "https://www.fakeurl.com/playlist?list=whatever"
+    request_get.return_value = playlist_html
+    playlist = Playlist(url)
+    playlist._find_load_more_url = MagicMock(return_value=None)
+    request_get.assert_called()
+    assert (
+        repr(playlist) == "['https://www.youtube.com/watch?v=ujTCoH21GlA', "
+        "'https://www.youtube.com/watch?v=45ryDIPHdGg', "
+        "'https://www.youtube.com/watch?v=1BYu65vLKdA', "
+        "'https://www.youtube.com/watch?v=3AQ_74xrch8', "
+        "'https://www.youtube.com/watch?v=ddqQUz9mZaM', "
+        "'https://www.youtube.com/watch?v=vwLT6bZrHEE', "
+        "'https://www.youtube.com/watch?v=TQKI0KE-JYY', "
+        "'https://www.youtube.com/watch?v=dNBvQ38MlT8', "
+        "'https://www.youtube.com/watch?v=JHxyrMgOUWI', "
+        "'https://www.youtube.com/watch?v=l2I8NycJMCY', "
+        "'https://www.youtube.com/watch?v=g1Zbuk1gAfk', "
+        "'https://www.youtube.com/watch?v=zixd-si9Q-o']"
+    )
+
+
+@mock.patch("pytube.contrib.playlist.request.get")
+def test_sequence(request_get, playlist_html):
+    url = "https://www.fakeurl.com/playlist?list=whatever"
+    request_get.return_value = playlist_html
+    playlist = Playlist(url)
+    playlist._find_load_more_url = MagicMock(return_value=None)
+    assert playlist[0] == "https://www.youtube.com/watch?v=ujTCoH21GlA"
+    assert len(playlist) == 12
+
+
 @mock.patch("pytube.contrib.playlist.request.get")
 @mock.patch("pytube.cli.YouTube.__init__", return_value=None)
 def test_videos(youtube, request_get, playlist_html):

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_captions.py

@@ -2,6 +2,8 @@
 from unittest import mock
 from unittest.mock import patch, mock_open, MagicMock
 
+import pytest
+
 from pytube import Caption, CaptionQuery, captions
 
 
@@ -12,7 +14,7 @@ def test_float_to_srt_time_format():
     assert caption1.float_to_srt_time_format(3.89) == "00:00:03,890"
 
 
-def test_caption_query_all():
+def test_caption_query_sequence():
     caption1 = Caption(
         {"url": "url1", "name": {"simpleText": "name1"}, "languageCode": "en"}
     )
@@ -20,7 +22,11 @@ def test_caption_query_all():
         {"url": "url2", "name": {"simpleText": "name2"}, "languageCode": "fr"}
     )
     caption_query = CaptionQuery(captions=[caption1, caption2])
-    assert caption_query.captions == [caption1, caption2]
+    assert len(caption_query) == 2
+    assert caption_query["en"] == caption1
+    assert caption_query["fr"] == caption2
+    with pytest.raises(KeyError):
+        caption_query["nada"]
 
 
 def test_caption_query_get_by_language_code_when_exists():
@@ -101,6 +107,9 @@ def test_repr():
     )
     assert str(caption) == '<Caption lang="name1" code="en">'
 
+    caption_query = CaptionQuery(captions=[caption])
+    assert repr(caption_query) == '{\'en\': <Caption lang="name1" code="en">}'
+
 
 @mock.patch("pytube.request.get")
 def test_xml_captions(request_get):

tests/test_cli.py

@@ -19,14 +19,18 @@ def test_main_invalid_url(_parse_args):
         cli.main()
 
 
+@mock.patch("pytube.cli.display_streams")
 @mock.patch("pytube.cli.YouTube")
-def test_download_when_itag_not_found(youtube):
+def test_download_when_itag_not_found(youtube, display_streams):
+    # Given
     youtube.streams = mock.Mock()
-    youtube.streams.all.return_value = []
     youtube.streams.get_by_itag.return_value = None
+    # When
     with pytest.raises(SystemExit):
         cli.download_by_itag(youtube, 123)
+    # Then
     youtube.streams.get_by_itag.assert_called_with(123)
+    display_streams.assert_called_with(youtube)
 
 
 @mock.patch("pytube.cli.YouTube")
@@ -46,26 +50,28 @@ def test_download_when_itag_is_found(youtube, stream):
 @mock.patch("pytube.cli.YouTube")
 @mock.patch("pytube.Stream")
 def test_display_stream(youtube, stream):
+    # Given
     stream.itag = 123
     stream.__repr__ = MagicMock(return_value="")
     youtube.streams = StreamQuery([stream])
-    with patch.object(youtube.streams, "all", wraps=youtube.streams.all) as wrapped_all:
-        cli.display_streams(youtube)
-        wrapped_all.assert_called()
-        stream.__repr__.assert_called()
+    # When
+    cli.display_streams(youtube)
+    # Then
+    stream.__repr__.assert_called()
 
 
+@mock.patch("pytube.cli._print_available_captions")
 @mock.patch("pytube.cli.YouTube")
-def test_download_caption_with_none(youtube):
+def test_download_caption_with_none(youtube, print_available):
+    # Given
     caption = Caption(
         {"url": "url1", "name": {"simpleText": "name1"}, "languageCode": "en"}
     )
     youtube.captions = CaptionQuery([caption])
-    with patch.object(
-        youtube.captions, "all", wraps=youtube.captions.all
-    ) as wrapped_all:
-        cli.download_caption(youtube, None)
-        wrapped_all.assert_called()
+    # When
+    cli.download_caption(youtube, None)
+    # Then
+    print_available.assert_called_with(youtube.captions)
 
 
 @mock.patch("pytube.cli.YouTube")
@@ -80,17 +86,18 @@ def test_download_caption_with_language_found(youtube):
     caption.download.assert_called_with(title="video title", output_path=None)
 
 
+@mock.patch("pytube.cli._print_available_captions")
 @mock.patch("pytube.cli.YouTube")
-def test_download_caption_with_language_not_found(youtube):
+def test_download_caption_with_lang_not_found(youtube, print_available):
+    # Given
     caption = Caption(
         {"url": "url1", "name": {"simpleText": "name1"}, "languageCode": "en"}
     )
     youtube.captions = CaptionQuery([caption])
-    with patch.object(
-        youtube.captions, "all", wraps=youtube.captions.all
-    ) as wrapped_all:
-        cli.download_caption(youtube, "blah")
-        wrapped_all.assert_called()
+    # When
+    cli.download_caption(youtube, "blah")
+    # Then
+    print_available.assert_called_with(youtube.captions)
 
 
 def test_display_progress_bar(capsys):

tests/test_exceptions.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-from pytube.exceptions import VideoUnavailable, RegexMatchError
+from pytube.exceptions import VideoUnavailable, RegexMatchError, LiveStreamError
 
 
 def test_video_unavailable():
@@ -15,3 +15,11 @@ def test_regex_match_error():
         raise RegexMatchError(caller="hello", pattern="*")
     except RegexMatchError as e:
         assert str(e) == "hello: could not find match for *"
+
+
+def test_live_stream_error():
+    try:
+        raise LiveStreamError(video_id="YLnZklYFe7E")
+    except LiveStreamError as e:
+        assert e.video_id == "YLnZklYFe7E"
+        assert str(e) == "YLnZklYFe7E is streaming live and cannot be loaded"

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():
@@ -80,3 +77,8 @@ def test_mime_type_codec():
 def test_mime_type_codec_with_no_match_should_error():
     with pytest.raises(RegexMatchError):
         extract.mime_type_codec("audio/webm")
+
+
+def test_get_ytplayer_config_with_no_match_should_error():
+    with pytest.raises(RegexMatchError):
+        extract.get_ytplayer_config("")

tests/test_mixins.py

@@ -1,3 +0,0 @@
-# -*- coding: utf-8 -*-
-def test_pre_signed_video(presigned_video):
-    assert presigned_video.streams.count() == 12

tests/test_query.py

@@ -3,11 +3,6 @@
 import pytest
 
 
-def test_count(cipher_signature):
-    """Ensure :meth:`~pytube.StreamQuery.count` returns an accurate amount."""
-    assert cipher_signature.streams.count() == 22
-
-
 @pytest.mark.parametrize(
     ("test_input", "expected"),
     [
@@ -30,7 +25,7 @@ def test_count(cipher_signature):
 )
 def test_filters(test_input, expected, cipher_signature):
     """Ensure filters produce the expected results."""
-    result = [s.itag for s in cipher_signature.streams.filter(**test_input).all()]
+    result = [s.itag for s in cipher_signature.streams.filter(**test_input)]
     assert result == expected
 
 
@@ -64,8 +59,7 @@ def test_order_by(cipher_signature):
     :class:`Stream <Stream>` instances in the expected order.
     """
     itags = [
-        s.itag
-        for s in cipher_signature.streams.filter(type="audio").order_by("itag").all()
+        s.itag for s in cipher_signature.streams.filter(type="audio").order_by("itag")
     ]
     assert itags == [140, 249, 250, 251]
 
@@ -77,10 +71,7 @@ def test_order_by_descending(cipher_signature):
     # numerical values
     itags = [
         s.itag
-        for s in cipher_signature.streams.filter(type="audio")
-        .order_by("itag")
-        .desc()
-        .all()
+        for s in cipher_signature.streams.filter(type="audio").order_by("itag").desc()
     ]
     assert itags == [251, 250, 249, 140]
 
@@ -91,7 +82,6 @@ def test_order_by_non_numerical(cipher_signature):
         for s in cipher_signature.streams.filter(res="360p")
         .order_by("mime_type")
         .desc()
-        .all()
     ]
     assert mime_types == ["video/webm", "video/mp4", "video/mp4"]
 
@@ -103,10 +93,7 @@ def test_order_by_ascending(cipher_signature):
     # numerical values
     itags = [
         s.itag
-        for s in cipher_signature.streams.filter(type="audio")
-        .order_by("itag")
-        .asc()
-        .all()
+        for s in cipher_signature.streams.filter(type="audio").order_by("itag").asc()
     ]
     assert itags == [140, 249, 250, 251]
 
@@ -114,16 +101,13 @@ def test_order_by_ascending(cipher_signature):
 def test_order_by_non_numerical_ascending(cipher_signature):
     mime_types = [
         s.mime_type
-        for s in cipher_signature.streams.filter(res="360p")
-        .order_by("mime_type")
-        .asc()
-        .all()
+        for s in cipher_signature.streams.filter(res="360p").order_by("mime_type").asc()
     ]
     assert mime_types == ["video/mp4", "video/mp4", "video/webm"]
 
 
 def test_order_by_with_none_values(cipher_signature):
-    abrs = [s.abr for s in cipher_signature.streams.order_by("abr").asc().all()]
+    abrs = [s.abr for s in cipher_signature.streams.order_by("abr").asc()]
     assert abrs == ["50kbps", "70kbps", "96kbps", "128kbps", "160kbps"]
 
 
@@ -151,7 +135,7 @@ def test_get_highest_resolution(cipher_signature):
 
 
 def test_filter_is_dash(cipher_signature):
-    streams = cipher_signature.streams.filter(is_dash=False).all()
+    streams = cipher_signature.streams.filter(is_dash=False)
     itags = [s.itag for s in streams]
     assert itags == [18, 398, 397, 396, 395, 394]
 
@@ -162,3 +146,28 @@ 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_sequence(cipher_signature):
+    assert len(cipher_signature.streams) == 22
+    assert cipher_signature.streams[0] is not None
+
+
+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
+
+
+def test_repr(cipher_signature):
+    assert repr(
+        cipher_signature.streams.filter(
+            progressive=True, subtype="mp4", resolution="360p"
+        )
+    ) == (
+        '[<Stream: itag="18" mime_type="video/mp4" '
+        'res="360p" fps="30fps" vcodec="avc1.42001E" '
+        'acodec="mp4a.40.2" progressive="True" type="video">]'
+    )

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 = "title"
+    cipher_signature.player_config_args["title"] = expected
+    assert cipher_signature.title == expected
 
-    expected = "Unknown YouTube Video Title"
-    stream.player_config_args = {}
-    assert stream.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):
@@ -41,7 +45,7 @@ def test_caption_tracks(presigned_video):
 
 
 def test_captions(presigned_video):
-    assert len(presigned_video.captions.all()) == 13
+    assert len(presigned_video.captions) == 13
 
 
 def test_description(cipher_signature):