Merge branch 'master' into request-head

.flake8

@@ -1,5 +1,5 @@
 [flake8]
-ignore = E231,E203,W503,Q000,WPS111,WPS305,WPS348,WPS602,D400,DAR201,S101,DAR101,C812,D104,I001,WPS306,WPS214,D401,WPS229,WPS420,WPS230,WPS414,WPS114,WPS226,WPS442,C819,WPS601,T001,RST304,WPS410,WPS428,A003,A002,I003,WPS221,WPS326,WPS201,S405,DAR301,WPS210,WPS202,WPS213,WPS301,P103
+ignore = E231,E203,W503,Q000,WPS111,WPS305,WPS348,WPS602,D400,DAR201,S101,DAR101,C812,D104,I001,WPS306,WPS214,D401,WPS229,WPS420,WPS230,WPS414,WPS114,WPS226,WPS442,C819,WPS601,T001,RST304,WPS410,WPS428,A003,A002,I003,WPS221,WPS326,WPS201,S405,DAR301,WPS210,WPS202,WPS213,WPS301,P103,WPS407,WPS432,WPS211
 max-line-length = 89
 
 [isort]

.gitignore

@@ -54,6 +54,11 @@ coverage.xml
 *.cover
 .hypothesis/
 .pytest_cache/
+*.mp4
+
+# Performance profiling
+prof/
+*.cprof
 
 # Debian Files
 debian/files

Pipfile

@@ -23,4 +23,5 @@ flake8-eradicate = "*"
 flake8-broken-line = "*"
 pep8-naming = "*"
 flake8-string-format = "*"
-flake8-quotes = "*"
+flake8-quotes = "*"
+pytest-profiling = "*"

pytube/__main__.py

@@ -70,7 +70,6 @@ class YouTube:
         self.player_response: Dict = {}
         # streams
         self.age_restricted: Optional[bool] = None
-        self.vid_descr: Optional[str] = None
 
         self.fmt_streams: List[Stream] = []
 
@@ -125,8 +124,6 @@ class YouTube:
                 title = title[:index] if index > 0 else title
                 self.player_config_args["title"] = unescape(title)
 
-        if self.watch_html:
-            self.vid_descr = extract.get_vid_descr(self.watch_html)
         # https://github.com/nficano/pytube/issues/165
         stream_maps = ["url_encoded_fmt_stream_map"]
         if "adaptive_fmts" in self.player_config_args:
@@ -276,9 +273,9 @@ class YouTube:
         :rtype: str
 
         """
-        return self.vid_descr or (
-            self.player_response.get("videoDetails", {}).get("shortDescription")
-        )
+        return self.player_response.get("videoDetails", {}).get(
+            "shortDescription"
+        ) or extract.get_vid_descr(self.watch_html)
 
     @property
     def rating(self) -> float:

pytube/cipher.py

@@ -20,11 +20,75 @@ from itertools import chain
 from typing import List, Tuple, Dict, Callable, Any, Optional
 
 from pytube.exceptions import RegexMatchError
-from pytube.helpers import regex_search, create_logger
+from pytube.helpers import regex_search, create_logger, cache
 
 logger = create_logger()
 
 
+class Cipher:
+    def __init__(self, js: str):
+        self.transform_plan: List[str] = get_transform_plan(js)
+        var, _ = self.transform_plan[0].split(".")
+        self.transform_map = get_transform_map(js, var)
+        self.js_func_regex = re.compile(r"\w+\.(\w+)\(\w,(\d+)\)")
+
+    def get_signature(self, ciphered_signature: str) -> str:
+        """Decipher the signature.
+
+        Taking the ciphered signature, applies the transform functions.
+
+        :param str ciphered_signature:
+            The ciphered signature sent in the ``player_config``.
+        :rtype: str
+        :returns:
+           Decrypted signature required to download the media content.
+        """
+        signature = list(ciphered_signature)
+
+        for js_func in self.transform_plan:
+            name, argument = self.parse_function(js_func)  # type: ignore
+            signature = self.transform_map[name](signature, argument)
+            logger.debug(
+                "applied transform function\n"
+                "output: %s\n"
+                "js_function: %s\n"
+                "argument: %d\n"
+                "function: %s",
+                "".join(signature),
+                name,
+                argument,
+                self.transform_map[name],
+            )
+
+        return "".join(signature)
+
+    @cache
+    def parse_function(self, js_func: str) -> Tuple[str, int]:
+        """Parse the Javascript transform function.
+
+        Break a JavaScript transform function down into a two element ``tuple``
+        containing the function name and some integer-based argument.
+
+        :param str js_func:
+            The JavaScript version of the transform function.
+        :rtype: tuple
+        :returns:
+            two element tuple containing the function name and an argument.
+
+        **Example**:
+
+        >>> parse_function('DE.AJ(a,15)')
+        ('AJ', 15)
+
+        """
+        logger.debug("parsing transform function")
+        parse_match = self.js_func_regex.search(js_func)
+        if not parse_match:
+            raise RegexMatchError(caller="parse_function", pattern="js_func_regex")
+        fn_name, fn_arg = parse_match.groups()
+        return fn_name, int(fn_arg)
+
+
 def get_initial_function_name(js: str) -> str:
     """Extract the name of the function responsible for computing the signature.
     :param str js:
@@ -48,7 +112,6 @@ def get_initial_function_name(js: str) -> str:
         r"\bc\s*&&\s*[a-zA-Z0-9]+\.set\([^,]+\s*,\s*\([^)]*\)\s*\(\s*(?P<sig>[a-zA-Z0-9$]+)\(",  # noqa: E501
         r"\bc\s*&&\s*[a-zA-Z0-9]+\.set\([^,]+\s*,\s*\([^)]*\)\s*\(\s*(?P<sig>[a-zA-Z0-9$]+)\(",  # noqa: E501
     ]
-
     logger.debug("finding initial function name")
     for pattern in function_patterns:
         regex = re.compile(pattern)
@@ -71,7 +134,6 @@ def get_transform_plan(js: str) -> List[str]:
 
     **Example**:
 
-    >>> get_transform_plan(js)
     ['DE.AJ(a,15)',
     'DE.VR(a,3)',
     'DE.AJ(a,51)',
@@ -222,68 +284,3 @@ def map_functions(js_func: str) -> Callable:
         if re.search(pattern, js_func):
             return fn
     raise RegexMatchError(caller="map_functions", pattern="multiple")
-
-
-def parse_function(js_func: str) -> Tuple[str, int]:
-    """Parse the Javascript transform function.
-
-    Break a JavaScript transform function down into a two element ``tuple``
-    containing the function name and some integer-based argument.
-
-    :param str js_func:
-        The JavaScript version of the transform function.
-    :rtype: tuple
-    :returns:
-        two element tuple containing the function name and an argument.
-
-    **Example**:
-
-    >>> parse_function('DE.AJ(a,15)')
-    ('AJ', 15)
-
-    """
-    logger.debug("parsing transform function")
-    pattern = r"\w+\.(\w+)\(\w,(\d+)\)"
-    regex = re.compile(pattern)
-    parse_match = regex.search(js_func)
-    if not parse_match:
-        raise RegexMatchError(caller="parse_function", pattern=pattern)
-    fn_name, fn_arg = parse_match.groups()
-    return fn_name, int(fn_arg)
-
-
-def get_signature(js: str, ciphered_signature: str) -> str:
-    """Decipher the signature.
-
-    Taking the ciphered signature, applies the transform functions.
-
-    :param str js:
-        The contents of the base.js asset file.
-    :param str ciphered_signature:
-        The ciphered signature sent in the ``player_config``.
-    :rtype: str
-    :returns:
-       Decrypted signature required to download the media content.
-
-    """
-    transform_plan = get_transform_plan(js)
-    var, _ = transform_plan[0].split(".")
-    transform_map = get_transform_map(js, var)
-    signature = list(ciphered_signature)
-
-    for js_func in transform_plan:
-        name, argument = parse_function(js_func)
-        signature = transform_map[name](signature, argument)
-        logger.debug(
-            "applied transform function\n"
-            "output: %s\n"
-            "js_function: %s\n"
-            "argument: %d\n"
-            "function: %s",
-            "".join(signature),
-            name,
-            argument,
-            transform_map[name],
-        )
-
-    return "".join(signature)

pytube/extract.py

@@ -1,7 +1,6 @@
 # -*- coding: utf-8 -*-
 """This module contains all non-cipher related data extraction logic."""
 import json
-import pprint
 import re
 from collections import OrderedDict
 
@@ -10,7 +9,7 @@ from typing import Any, Optional, Tuple, List, Dict
 from urllib.parse import quote, parse_qs, unquote, parse_qsl
 from urllib.parse import urlencode
 
-from pytube import cipher
+from pytube.cipher import Cipher
 from pytube.exceptions import RegexMatchError, HTMLParseError, LiveStreamError
 from pytube.helpers import regex_search, logger
 
@@ -205,9 +204,10 @@ def get_ytplayer_config(html: str, age_restricted: bool = False) -> Any:
     return json.loads(yt_player_config)
 
 
-def get_vid_descr(html: str) -> str:
+def get_vid_descr(html: Optional[str]) -> str:
     html_parser = PytubeHTMLParser()
-    html_parser.feed(html)
+    if html:
+        html_parser.feed(html)
     return html_parser.vid_descr
 
 
@@ -224,6 +224,7 @@ def apply_signature(config_args: Dict, fmt: str, js: str) -> None:
         The contents of the base.js asset file.
 
     """
+    cipher = Cipher(js=js)
     stream_manifest = config_args[fmt]
     live_stream = (
         json.loads(config_args["player_response"])
@@ -247,17 +248,13 @@ def apply_signature(config_args: Dict, fmt: str, js: str) -> None:
             continue
 
         if js is not None:
-            signature = cipher.get_signature(js, stream["s"])
+            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")
 
-        logger.debug(
-            "finished descrambling signature for itag=%s\n%s",
-            stream["itag"],
-            pprint.pformat({"s": stream["s"], "signature": signature,}, indent=2,),
-        )
+        logger.debug("finished descrambling signature for itag=%s", stream["itag"])
         # 403 forbidden fix
         stream_manifest[i]["url"] = url + "&sig=" + signature
 
@@ -320,6 +317,5 @@ def apply_descrambler(stream_data: Dict, key: str) -> None:
             {k: unquote(v) for k, v in parse_qsl(i)}
             for i in stream_data[key].split(",")
         ]
-    logger.debug(
-        "applying descrambler\n%s", pprint.pformat(stream_data[key], indent=2),
-    )
+
+    logger.debug("applying descrambler")

pytube/request.py

@@ -6,9 +6,12 @@ from urllib.request import Request
 from urllib.request import urlopen
 
 
-def _execute_request(url: str, method: Optional[str] = None) -> Any:
+def _execute_request(url: str, method: Optional[str] = None, headers: Optional[Dict[str, str]] = None) -> Any:
+    base_headers = {"User-Agent": "Mozilla/5.0"}
+    if headers:
+        base_headers.update(headers)
     if url.lower().startswith("http"):
-        request = Request(url, headers={"User-Agent": "Mozilla/5.0"}, method=method)
+        request = Request(url, headers=base_headers, method=method)
     else:
         raise ValueError
     return urlopen(request)  # nosec
@@ -32,7 +35,7 @@ def stream(url: str, chunk_size: int = 8192) -> Iterable[bytes]:
     :param int chunk_size: The size in bytes of each chunk. Defaults to 8*1024
     :rtype: Iterable[bytes]
     """
-    response = _execute_request(url)
+    response = _execute_request(url, headers={"Range": "bytes=0-"})
     while True:
         buf = response.read(chunk_size)
         if not buf:
@@ -40,7 +43,7 @@ def stream(url: str, chunk_size: int = 8192) -> Iterable[bytes]:
         yield buf
 
 
-def headers(url: str) -> Dict:
+def head(url: str) -> Dict:
     """Fetch headers returned http GET request.
 
     :param str url:

pytube/streams.py

@@ -140,7 +140,7 @@ class Stream:
             Filesize (in bytes) of the stream.
         """
         if self._filesize is None:
-            headers = request.headers(self.url)
+            headers = request.head(self.url)
             self._filesize = int(headers["content-length"])
         return self._filesize
 

tests/test_cipher.py

@@ -20,17 +20,6 @@ def test_get_transform_object_with_no_match_should_error():
         cipher.get_transform_object("asdf", var="lt")
 
 
-def test_parse_function_with_match():
-    fn_name, fn_arg = cipher.parse_function("DE.AJ(a,15)")
-    assert fn_name == "AJ"
-    assert fn_arg == 15
-
-
-def test_parse_function_with_no_match_should_error():
-    with pytest.raises(RegexMatchError):
-        cipher.parse_function("asdf")
-
-
 def test_reverse():
     reversed_array = cipher.reverse([1, 2, 3, 4], None)
     assert reversed_array == [4, 3, 2, 1]

tests/test_request.py

@@ -30,7 +30,7 @@ def test_headers(mock_urlopen):
     response = mock.Mock()
     response.info.return_value = {"content-length": "16384"}
     mock_urlopen.return_value = response
-    response = request.headers("http://fakeassurl.gov")
+    response = request.head("http://fakeassurl.gov")
     assert response == {"content-length": "16384"}
 
 

tests/test_streams.py

@@ -11,7 +11,7 @@ from pytube import Stream, streams
 
 def test_filesize(cipher_signature, mocker):
     mocker.patch.object(request, "headers")
-    request.headers.return_value = {"content-length": "6796391"}
+    request.head.return_value = {"content-length": "6796391"}
     assert cipher_signature.streams.first().filesize == 6796391
 
 
@@ -40,11 +40,11 @@ def test_title(cipher_signature):
 def test_description(cipher_signature):
     expected = (
         "PSY - ‘I LUV IT’ M/V @ https://youtu.be/Xvjnoagk6GU\n"
-        "PSY - ‘New Face’ M/V @https://youtu.be/OwJPPaEyqhI\n"
+        "PSY - ‘New Face’ M/V @https://youtu.be/OwJPPaEyqhI\n\n"
         "PSY - 8TH ALBUM '4X2=8' on iTunes @\n"
-        "https://smarturl.it/PSY_8thAlbum\n"
-        "PSY - GANGNAM STYLE(강남스타일) on iTunes @ http://smarturl.it/PsyGangnam\n"
-        "#PSY #싸이 #GANGNAMSTYLE #강남스타일\n"
+        "https://smarturl.it/PSY_8thAlbum\n\n"
+        "PSY - GANGNAM STYLE(강남스타일) on iTunes @ http://smarturl.it/PsyGangnam\n\n"
+        "#PSY #싸이 #GANGNAMSTYLE #강남스타일\n\n"
         "More about PSY@\nhttp://www.youtube.com/officialpsy\n"
         "http://www.facebook.com/officialpsy\n"
         "http://twitter.com/psy_oppa\n"
@@ -55,14 +55,14 @@ def test_description(cipher_signature):
     )
     assert cipher_signature.description == expected
 
-    cipher_signature.vid_descr = None
+    cipher_signature.player_response = {}
     expected = (
         "PSY - ‘I LUV IT’ M/V @ https://youtu.be/Xvjnoagk6GU\n"
-        "PSY - ‘New Face’ M/V @https://youtu.be/OwJPPaEyqhI\n\n"
+        "PSY - ‘New Face’ M/V @https://youtu.be/OwJPPaEyqhI\n"
         "PSY - 8TH ALBUM '4X2=8' on iTunes @\n"
-        "https://smarturl.it/PSY_8thAlbum\n\n"
-        "PSY - GANGNAM STYLE(강남스타일) on iTunes @ http://smarturl.it/PsyGangnam\n\n"
-        "#PSY #싸이 #GANGNAMSTYLE #강남스타일\n\n"
+        "https://smarturl.it/PSY_8thAlbum\n"
+        "PSY - GANGNAM STYLE(강남스타일) on iTunes @ http://smarturl.it/PsyGangnam\n"
+        "#PSY #싸이 #GANGNAMSTYLE #강남스타일\n"
         "More about PSY@\nhttp://www.youtube.com/officialpsy\n"
         "http://www.facebook.com/officialpsy\n"
         "http://twitter.com/psy_oppa\n"
@@ -88,7 +88,7 @@ def test_views(cipher_signature):
 
 def test_download(cipher_signature, mocker):
     mocker.patch.object(request, "headers")
-    request.headers.return_value = {"content-length": "16384"}
+    request.head.return_value = {"content-length": "16384"}
     mocker.patch.object(request, "stream")
     request.stream.return_value = iter([str(random.getrandbits(8 * 1024))])
     with mock.patch("pytube.streams.open", mock.mock_open(), create=True):
@@ -98,7 +98,7 @@ def test_download(cipher_signature, mocker):
 
 def test_download_with_prefix(cipher_signature, mocker):
     mocker.patch.object(request, "headers")
-    request.headers.return_value = {"content-length": "16384"}
+    request.head.return_value = {"content-length": "16384"}
     mocker.patch.object(request, "stream")
     request.stream.return_value = iter([str(random.getrandbits(8 * 1024))])
     streams.target_directory = MagicMock(return_value="/target")
@@ -110,7 +110,7 @@ def test_download_with_prefix(cipher_signature, mocker):
 
 def test_download_with_filename(cipher_signature, mocker):
     mocker.patch.object(request, "headers")
-    request.headers.return_value = {"content-length": "16384"}
+    request.head.return_value = {"content-length": "16384"}
     mocker.patch.object(request, "stream")
     request.stream.return_value = iter([str(random.getrandbits(8 * 1024))])
     streams.target_directory = MagicMock(return_value="/target")
@@ -122,7 +122,7 @@ def test_download_with_filename(cipher_signature, mocker):
 
 def test_download_with_existing(cipher_signature, mocker):
     mocker.patch.object(request, "headers")
-    request.headers.return_value = {"content-length": "16384"}
+    request.head.return_value = {"content-length": "16384"}
     mocker.patch.object(request, "stream")
     streams.target_directory = MagicMock(return_value="/target")
     mocker.patch.object(os.path, "isfile")
@@ -138,7 +138,7 @@ def test_download_with_existing(cipher_signature, mocker):
 
 def test_download_with_existing_no_skip(cipher_signature, mocker):
     mocker.patch.object(request, "headers")
-    request.headers.return_value = {"content-length": "16384"}
+    request.head.return_value = {"content-length": "16384"}
     mocker.patch.object(request, "stream")
     request.stream.return_value = iter([str(random.getrandbits(8 * 1024))])
     streams.target_directory = MagicMock(return_value="/target")
@@ -168,7 +168,7 @@ def test_on_progress_hook(cipher_signature, mocker):
     cipher_signature.register_on_progress_callback(callback_fn)
 
     mocker.patch.object(request, "headers")
-    request.headers.return_value = {"content-length": "16384"}
+    request.head.return_value = {"content-length": "16384"}
     mocker.patch.object(request, "stream")
     request.stream.return_value = iter([str(random.getrandbits(8 * 1024))])
 
@@ -187,7 +187,7 @@ def test_on_complete_hook(cipher_signature, mocker):
     cipher_signature.register_on_complete_callback(callback_fn)
 
     mocker.patch.object(request, "headers")
-    request.headers.return_value = {"content-length": "16384"}
+    request.head.return_value = {"content-length": "16384"}
     mocker.patch.object(request, "stream")
     request.stream.return_value = iter([str(random.getrandbits(8 * 1024))])