Merge branch 'master' into regex-refactor

.deepsource.toml

@@ -0,0 +1,16 @@
+version = 1
+
+test_patterns = [
+  "*/tests/**"
+]
+
+exclude_patterns = [
+  "setup.py"
+]
+
+[[analyzers]]
+name = "python"
+enabled = true
+
+  [analyzers.meta]
+  runtime_version = "3.x.x"

.flake8

@@ -1,3 +1,3 @@
 [flake8]
-ignore = E231,E203,W605
+ignore = E231,E203,W503
 max-line-length = 88

.readthedocs.yml

@@ -0,0 +1,19 @@
+# .readthedocs.yml
+# Read the Docs configuration file
+# See https://docs.readthedocs.io/en/stable/config-file/v2.html for details
+
+# Required
+version: 2
+
+# Build documentation in the docs/ directory with Sphinx
+sphinx:
+  configuration: docs/conf.py
+
+# Optionally build your docs in additional formats such as PDF and ePub
+formats: all
+
+# Optionally set the version of Python and requirements required to build your docs
+python:
+  version: 3.7
+  install:
+    - requirements: docs/requirements.txt

README.md

@@ -2,10 +2,11 @@
 <div align="center">
   <p align="center">
 	  <a href="https://pypi.org/project/pytube3/"><img src="https://img.shields.io/pypi/v/pytube3.svg" alt="pypi"></a>
+	  <a href="https://pypi.python.org/pypi/pytube3/"><img src="https://img.shields.io/pypi/pyversions/pytube3.svg" /></a>
 	  <a href="https://travis-ci.com/hbmartin/pytube3/"><img src="https://travis-ci.org/hbmartin/pytube3.svg?branch=master" /></a>
 	  <a href='https://pytube3.readthedocs.io/en/latest/?badge=latest'><img src='https://readthedocs.org/projects/pytube3/badge/?version=latest' alt='Documentation Status' /></a>
 	  <a href="https://coveralls.io/github/hbmartin/pytube3?branch=master"><img src="https://coveralls.io/repos/github/hbmartin/pytube3/badge.svg?branch=master" /></a>
-	  <a href="https://pypi.python.org/pypi/pytube3/"><img src="https://img.shields.io/pypi/pyversions/pytube3.svg" /></a>
+	  <a href="https://www.codefactor.io/repository/github/hbmartin/pytube3/overview/master"><img src="https://www.codefactor.io/repository/github/hbmartin/pytube3/badge/master" alt="CodeFactor" /></a>
 	  <a href="https://github.com/ambv/black"><img src="https://img.shields.io/badge/code%20style-black-000000.svg" /></a>
 	  <a href="https://twitter.com/hmartin"><img src="https://img.shields.io/badge/[email protected]?style=flat" /></a>
   </p>
@@ -243,6 +244,8 @@ Finally, if you're filing a bug report, the cli contains a switch called ``--bui
 
 ## Development
 
+<a href="https://deepsource.io/gh/hbmartin/pytube3/?ref=repository-badge" target="_blank"><img alt="DeepSource" title="DeepSource" src="https://static.deepsource.io/deepsource-badge-light-mini.svg"></a>
+
 Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.
 
 To run code checking before a PR use ``make test``

docs/requirements.txt

@@ -0,0 +1 @@
+typing_extensions

pytube/__main__.py

@@ -23,13 +23,12 @@ from pytube import Stream
 from pytube import StreamQuery
 from pytube.mixins import install_proxy
 from pytube.exceptions import VideoUnavailable
-from pytube.helpers import apply_mixin
 from pytube.monostate import OnProgress, OnComplete, Monostate
 
 logger = logging.getLogger(__name__)
 
 
-class YouTube(object):
+class YouTube:
     """Core developer interface for pytube."""
 
     def __init__(
@@ -91,16 +90,8 @@ class YouTube(object):
             install_proxy(proxies)
 
         if not defer_prefetch_init:
-            self.prefetch_descramble()
-
-    def prefetch_descramble(self) -> None:
-        """Download data, descramble it, and build Stream instances.
-
-        :rtype: None
-
-        """
-        self.prefetch()
-        self.descramble()
+            self.prefetch()
+            self.descramble()
 
     def descramble(self) -> None:
         """Descramble the stream data and build Stream instances.
@@ -161,7 +152,9 @@ class YouTube(object):
             self.initialize_stream_objects(fmt)
 
         # load the player_response object (contains subtitle information)
-        apply_mixin(self.player_config_args, "player_response", json.loads)
+        self.player_config_args["player_response"] = json.loads(
+            self.player_config_args["player_response"]
+        )
 
         self.initialize_caption_objects()
         logger.info("init finished successfully")
@@ -283,7 +276,11 @@ class YouTube(object):
         :rtype: str
 
         """
-        return self.player_config_args["title"]
+        return self.player_config_args.get("title") or (
+            self.player_config_args.get("player_response", {})
+            .get("videoDetails", {})
+            .get("title")
+        )
 
     @property
     def description(self) -> str:
@@ -292,7 +289,11 @@ class YouTube(object):
         :rtype: str
 
         """
-        return self.vid_descr
+        return self.vid_descr or (
+            self.player_config_args.get("player_response", {})
+            .get("videoDetails", {})
+            .get("shortDescription")
+        )
 
     @property
     def rating(self) -> float:
@@ -314,7 +315,7 @@ class YouTube(object):
         :rtype: str
 
         """
-        return (
+        return self.player_config_args.get("length_seconds") or (
             self.player_config_args.get("player_response", {})
             .get("videoDetails", {})
             .get("lengthSeconds")

pytube/captions.py

@@ -34,7 +34,8 @@ class Caption:
         """
         return self.xml_caption_to_srt(self.xml_captions)
 
-    def float_to_srt_time_format(self, d: float) -> str:
+    @staticmethod
+    def float_to_srt_time_format(d: float) -> str:
         """Convert decimal durations into proper srt format.
 
         :rtype: str
@@ -56,7 +57,7 @@ class Caption:
         """
         segments = []
         root = ElementTree.fromstring(xml_captions)
-        for i, child in enumerate(root.getchildren()):
+        for i, child in enumerate(list(root)):
             text = child.text or ""
             caption = unescape(text.replace("\n", " ").replace("  ", " "),)
             duration = float(child.attrib["dur"])

pytube/cipher.py

@@ -205,14 +205,14 @@ def map_functions(js_func: str) -> Callable:
     """
     mapper = (
         # function(a){a.reverse()}
-        ("{\w\.reverse\(\)}", reverse),
+        (r"{\w\.reverse\(\)}", reverse),
         # function(a,b){a.splice(0,b)}
-        ("{\w\.splice\(0,\w\)}", splice),
+        (r"{\w\.splice\(0,\w\)}", splice),
         # function(a,b){var c=a[0];a[0]=a[b%a.length];a[b]=c}
-        ("{var\s\w=\w\[0\];\w\[0\]=\w\[\w\%\w.length\];\w\[\w\]=\w}", swap),
+        (r"{var\s\w=\w\[0\];\w\[0\]=\w\[\w\%\w.length\];\w\[\w\]=\w}", swap),
         # function(a,b){var c=a[0];a[0]=a[b%a.length];a[b%a.length]=c}
         (
-            "{var\s\w=\w\[0\];\w\[0\]=\w\[\w\%\w.length\];" "\w\[\w\%\w.length\]=\w}",
+            r"{var\s\w=\w\[0\];\w\[0\]=\w\[\w\%\w.length\];\w\[\w\%\w.length\]=\w}",
             swap,
         ),
     )

pytube/cli.py

@@ -8,7 +8,8 @@ import json
 import logging
 import os
 import sys
-from typing import Tuple
+from io import BufferedWriter
+from typing import Tuple, Any
 
 from pytube import __version__
 from pytube import YouTube
@@ -19,6 +20,7 @@ logger = logging.getLogger(__name__)
 
 def main():
     """Command line application to download youtube videos."""
+    # noinspection PyTypeChecker
     parser = argparse.ArgumentParser(description=main.__doc__)
     parser.add_argument("url", help="The YouTube /watch url", nargs="?")
     parser.add_argument(
@@ -128,14 +130,18 @@ def display_progress_bar(
 
     filled = int(round(max_width * bytes_received / float(filesize)))
     remaining = max_width - filled
-    bar = ch * filled + " " * remaining
+    progress_bar = ch * filled + " " * remaining
     percent = round(100.0 * bytes_received / float(filesize), 1)
-    text = " ↳ |{bar}| {percent}%\r".format(bar=bar, percent=percent)
+    text = " ↳ |{progress_bar}| {percent}%\r".format(
+        progress_bar=progress_bar, percent=percent
+    )
     sys.stdout.write(text)
     sys.stdout.flush()
 
 
-def on_progress(stream, chunk, file_handle, bytes_remaining):
+def on_progress(
+    stream: Any, chunk: Any, file_handler: BufferedWriter, bytes_remaining: int
+) -> None:
     filesize = stream.filesize
     bytes_received = filesize - bytes_remaining
     display_progress_bar(bytes_received, filesize)

pytube/contrib/playlist.py

@@ -1,7 +1,6 @@
 # -*- coding: utf-8 -*-
-"""
-Module to download a complete playlist from a youtube channel
-"""
+"""Module to download a complete playlist from a youtube channel"""
+
 import json
 import logging
 import re
@@ -14,7 +13,7 @@ from pytube.__main__ import YouTube
 logger = logging.getLogger(__name__)
 
 
-class Playlist(object):
+class Playlist:
     """Handles all the task of manipulating and downloading a whole YouTube
     playlist
     """
@@ -40,7 +39,8 @@ class Playlist(object):
         # url is already in the desired format, so just return it
         return self.playlist_url
 
-    def _load_more_url(self, req):
+    @staticmethod
+    def _load_more_url(req):
         """Given an html page or a fragment thereof, looks for
         and returns the "load more" url if found.
         """
@@ -70,8 +70,8 @@ class Playlist(object):
         # The above only returns 100 or fewer links
         # Simulating a browser request for the load more link
         load_more_url = self._load_more_url(req)
-        while len(load_more_url):  # there is an url found
-            logger.debug("load more url: %s" % load_more_url)
+        while len(load_more_url) > 0:  # there is an url found
+            logger.debug("load more url: %s", load_more_url)
             req = request.get(load_more_url)
             load_more = json.loads(req)
             videos = re.findall(
@@ -155,8 +155,6 @@ class Playlist(object):
                 logger.debug(e)
                 if not self.suppress_exception:
                     raise e
-                else:
-                    logger.debug("Exception suppressed")
             else:
                 # TODO: this should not be hardcoded to a single user's
                 # preference
@@ -177,26 +175,21 @@ class Playlist(object):
                 logger.debug("download complete")
 
     def title(self) -> Optional[str]:
-        """return playlist title (name)
-        """
-        try:
-            url = self.construct_playlist_url()
-            req = request.get(url)
-            open_tag = "<title>"
-            end_tag = "</title>"
-            pattern = re.compile(open_tag + "(.+?)" + end_tag)
-            match = pattern.search(req)
-
-            if match is None:
-                return None
-
-            return (
-                match.group()
-                .replace(open_tag, "")
-                .replace(end_tag, "")
-                .replace("- YouTube", "")
-                .strip()
-            )
-        except Exception as e:
-            logger.debug(e)
+        """return playlist title (name)"""
+        url = self.construct_playlist_url()
+        req = request.get(url)
+        open_tag = "<title>"
+        end_tag = "</title>"
+        pattern = re.compile(open_tag + "(.+?)" + end_tag)
+        match = pattern.search(req)
+
+        if match is None:
             return None
+
+        return (
+            match.group()
+            .replace(open_tag, "")
+            .replace(end_tag, "")
+            .replace("- YouTube", "")
+            .strip()
+        )

pytube/helpers.py

@@ -37,25 +37,6 @@ def regex_search(pattern: str, string: str, group: int) -> str:
     return results.group(group)
 
 
-def apply_mixin(dct, key, func, *args, **kwargs):
-    r"""Apply in-place data mutation to a dictionary.
-
-    :param dict dct:
-        Dictionary to apply mixin function to.
-    :param str key:
-        Key within dictionary to apply mixin function to.
-    :param callable func:
-        Transform function to apply to ``dct[key]``.
-    :param \*args:
-        (optional) positional arguments that ``func`` takes.
-    :param \*\*kwargs:
-        (optional) keyword arguments that ``func`` takes.
-    :rtype:
-        None
-    """
-    dct[key] = func(dct[key], *args, **kwargs)
-
-
 def safe_filename(s: str, max_length: int = 255) -> str:
     """Sanitize a string making it safe to use as a filename.
 
@@ -73,26 +54,26 @@ def safe_filename(s: str, max_length: int = 255) -> str:
     # Characters in range 0-31 (0x00-0x1F) are not allowed in ntfs filenames.
     ntfs_characters = [chr(i) for i in range(0, 31)]
     characters = [
-        '"',
-        "\#",
-        "\$",
-        "\%",
-        "'",
-        "\*",
-        "\,",
-        "\.",
-        "\/",
-        "\:",
-        '"',
-        "\;",
-        "\<",
-        "\>",
-        "\?",
-        "\\",
-        "\^",
-        "\|",
-        "\~",
-        "\\\\",
+        r'"',
+        r"\#",
+        r"\$",
+        r"\%",
+        r"'",
+        r"\*",
+        r"\,",
+        r"\.",
+        r"\/",
+        r"\:",
+        r'"',
+        r"\;",
+        r"\<",
+        r"\>",
+        r"\?",
+        r"\\",
+        r"\^",
+        r"\|",
+        r"\~",
+        r"\\\\",
     ]
     pattern = "|".join(ntfs_characters + characters)
     regex = re.compile(pattern, re.UNICODE)

pytube/query.py

@@ -167,7 +167,7 @@ class StreamQuery:
         return StreamQuery(fmt_streams)
 
     def order_by(self, attribute_name: str) -> "StreamQuery":
-        """Apply a sort order to a resultset. Filters out stream the do not have the attribute.
+        """Apply a sort order. Filters out stream the do not have the attribute.
 
         :param str attribute_name:
             The name of the attribute to sort by.
@@ -191,7 +191,6 @@ class StreamQuery:
                 }
             except ValueError:
                 integer_attr_repr = None
-                pass
 
         # lookup integer values if we have them
         if integer_attr_repr is not None:
@@ -201,10 +200,10 @@ class StreamQuery:
                     key=lambda s: integer_attr_repr[getattr(s, attribute_name)],  # type: ignore  # noqa: E501
                 )
             )
-        else:
-            return StreamQuery(
-                sorted(has_attribute, key=lambda s: getattr(s, attribute_name))
-            )
+
+        return StreamQuery(
+            sorted(has_attribute, key=lambda s: getattr(s, attribute_name))
+        )
 
     def desc(self) -> "StreamQuery":
         """Sort streams in descending order.

pytube/request.py

@@ -17,15 +17,16 @@ def get(url, headers=False, streaming=False, chunk_size=8192):
         The size in bytes of each chunk. Defaults to 8*1024
     """
 
-    # https://github.com/nficano/pytube/pull/465
     req = Request(url, headers={"User-Agent": "Mozilla/5.0"})
     response = urlopen(req)
 
     if streaming:
         return stream_response(response, chunk_size)
-    elif headers:
+
+    if headers:
         # https://github.com/nficano/pytube/issues/160
         return {k.lower(): v for k, v in response.info().items()}
+
     return response.read().decode("utf-8")
 
 

pytube/streams.py

@@ -23,7 +23,7 @@ from pytube.monostate import Monostate
 logger = logging.getLogger(__name__)
 
 
-class Stream(object):
+class Stream:
     """Container for stream manifest data."""
 
     def __init__(self, stream: Dict, player_config_args: Dict, monostate: Monostate):
@@ -176,20 +176,16 @@ class Stream(object):
         :returns:
             Youtube video title
         """
-        player_config_args = self.player_config_args or {}
-
-        if "title" in player_config_args:
-            return player_config_args["title"]
-
-        details = self.player_config_args.get("player_response", {},).get(
-            "videoDetails", {}
+        return (
+            self.player_config_args.get("title")
+            or (
+                self.player_config_args.get("player_response", {})
+                .get("videoDetails", {})
+                .get("title")
+            )
+            or "Unknown YouTube Video Title"
         )
 
-        if "title" in details:
-            return details["title"]
-
-        return "Unknown YouTube Video Title"
-
     @property
     def default_filename(self) -> str:
         """Generate filename based on the video title.

tests/test_request.py

@@ -18,9 +18,8 @@ def test_get_streaming(mock_urlopen):
     response.read.side_effect = fake_stream_binary
     mock_urlopen.return_value = response
     response = request.get("http://fakeassurl.gov", streaming=True)
-    call_count = 0
-    for i in response:
-        call_count += 1
+    call_count = len(list(response))
+
     assert call_count == 3