Merge pull request #155 from alexmercerind/ytdlp

Migrate to yt-dlp

Author: mytja

.github/FUNDING.yml

@@ -16,7 +16,7 @@ jobs:
       with:
         python-version: '3.8'
     - name: Install dependencies
-      run: python -m pip install httpx git+https://github.com/pytube/pytube
+      run: python -m pip install httpx yt-dlp
     - name: Sync
       run: python syncExample.py
     - name: Async

.github/workflows/automatic-testing.yml

@@ -52,10 +52,9 @@ async def get(self, videoFormats: dict, itag: int) -> Union[str, None]:
             >>> print(url)
             "https://r6---sn-gwpa-5bgk.googlevideo.com/videoplayback?expire=1610798125&ei=zX8CYITXEIGKz7sP9MWL0AE&ip=2409%3A4053%3A803%3A2b22%3Adc68%3Adfb9%3Aa676%3A26a3&id=o-APBakKSE2_eMDMegtCmeWXfuhhUfAzJTmOCWj4lkEjAM&itag=251&source=youtube&requiressl=yes&mh=aP&mm=31%2C29&mn=sn-gwpa-5bgk%2Csn-gwpa-qxad&ms=au%2Crdu&mv=m&mvi=6&pl=36&initcwndbps=146250&vprv=1&mime=audio%2Fwebm&ns=ULL4mkMO31KDtEhOjkOrmpkF&gir=yes&clen=10210834&dur=634.601&lmt=1544629945422176&mt=1610776131&fvip=6&keepalive=yes&c=WEB&txp=5511222&n=uEjSqtzBZaJyVn&sparams=expire%2Cei%2Cip%2Cid%2Citag%2Csource%2Crequiressl%2Cvprv%2Cmime%2Cns%2Cgir%2Cclen%2Cdur%2Clmt&sig=AOq0QJ8wRAIgKKIEiwQTgXsdKPEyOckgVPs_LMH6KJoeaYmZic_lelECIHXHs1ZnSP5mgtpffNlIMJM3DhxcvDbA-4udFFE6AmVP&lsparams=mh%2Cmm%2Cmn%2Cms%2Cmv%2Cmvi%2Cpl%2Cinitcwndbps&lsig=AG3C_xAwRQIhAPmhL745RYeL_ffgUJk_xJLC-8riXKMylLTLA_pITYWWAiB2qUIXur8ThW7cLfQ73mIVK61mMZc2ncK6FZWjUHGcUw%3D%3D"
         '''
-        self._getDecipheredURLs(videoFormats)
-        for stream in self._streams:
-            if stream["itag"] == itag:
-                return stream["url"]
+        self._getDecipheredURLs(videoFormats, itag)
+        if len(self._streams) == 1:
+            return self._streams[0]["url"]
         return None
 
     async def getAll(self, videoFormats: dict) -> dict:

youtubesearchpython/__future__/streamurlfetcher.py

@@ -1,96 +1,147 @@
-import json
+import copy
 import urllib.request
+import urllib.parse
 
+import re
+
+from youtubesearchpython.core.constants import ResultMode
+from youtubesearchpython.core.video import VideoCore
+from youtubesearchpython.core.componenthandler import getValue
 from youtubesearchpython.core.requests import RequestCore
 
-isPyTubeInstalled = False
+isYtDLPinstalled = False
 
-import httpx
 try:
-    from pytube.extract import apply_descrambler, apply_signature
-    from pytube import YouTube, extract
-    import pytube
-    isPyTubeInstalled = True
+    from yt_dlp.extractor.youtube import YoutubeBaseInfoExtractor, YoutubeIE
+    from yt_dlp import YoutubeDL
+    from yt_dlp.utils import url_or_none, try_get, update_url_query, ExtractorError
+
+    isYtDLPinstalled = True
 except:
-    class YouTube:
-        def __init__(self):
-            pass
+    pass
 
 
-class StreamURLFetcherCore(YouTube):
+class StreamURLFetcherCore(RequestCore):
     '''
     Overrided parent's constructor.
     '''
     def __init__(self):
-        if isPyTubeInstalled:
+        if isYtDLPinstalled:
+            super().__init__()
             self._js_url = None
             self._js = None
+            #self.ytdlp = YoutubeBaseInfoExtractor()
+            self.ytie = YoutubeIE()
+            self.ytie.set_downloader(YoutubeDL())
+            self._streams = []
         else:
-            raise Exception('ERROR: PyTube is not installed. To use this functionality of youtube-search-python, PyTube must be installed.')
+            raise Exception('ERROR: yt-dlp is not installed. To use this functionality of youtube-search-python, yt-dlp must be installed.')
 
     '''
     Saving videoFormats inside a dictionary with key "player_response" for apply_descrambler & apply_signature methods.
     '''
-    def _getDecipheredURLs(self, videoFormats: dict) -> None:
-        # For some reason, we cannot fetch JavaScript the old way, as PyTube's RegEx doesn't like it.
-        self.video_id = videoFormats["id"]
-        self._player_response = videoFormats
-        if not videoFormats['streamingData']:
-            try:
-                self.use_oauth = False
-                self.allow_oauth_cache = False
-                self.bypass_age_gate()
-                self._player_response = self._vid_info
-            except:
-                raise Exception('ERROR: Could not make request.')
-
-        # We use this to retrieve JavaScript
-        url = f"https://www.youtube.com/watch?v={self.video_id}"
-        self.youtube = pytube.YouTube(url)
+    def _getDecipheredURLs(self, videoFormats: dict, formatId: int = None) -> None:
+        # We reset our stream list
+        # See https://github.com/alexmercerind/youtube-search-python/pull/155#discussion_r790165920
+        # If we don't reset it, then it's going to cache older URLs and as we are using length comparison in upper class
+        # it would return None, because length is not 1
+        self._streams = []
 
+        self.video_id = videoFormats["id"]
+        if not videoFormats["streamingData"]:
+            # Video is age-restricted. Try to retrieve it using ANDROID_EMBED client and override old response.
+            # This works most time.
+            vc = VideoCore(self.video_id, None, ResultMode.dict, None, False, overridedClient="ANDROID_EMBED")
+            vc.sync_create()
+            videoFormats = vc.result
+            if not videoFormats["streamingData"]:
+                # Video is:
+                # 1. Either age-restricted on so called level 3
+                # 2. Needs payment (is only for users that use so called "Join feature")
+                raise Exception("streamingData is not present in Video.get. This is most likely a age-restricted video")
+        # We deepcopy a list, otherwise it would duplicate
+        # See https://github.com/alexmercerind/youtube-search-python/pull/155#discussion_r790165920
+        self._player_response = copy.deepcopy(videoFormats["streamingData"]["formats"])
+        self._player_response.extend(videoFormats["streamingData"]["adaptiveFormats"])
+        self.format_id = formatId
         self._decipher()
 
-    '''
-    This method is derived from YouTube.prefetch.
-    This method fetches player JavaScript & its URL from /watch endpoint on YouTube.
-    Removed unnecessary methods & web requests as we already have metadata.
-    Uses httpx.AsyncClient in place of requests.
-    Removed v parameter from the query. (No idea about why PyTube bothered with that)
-    '''
+    def extract_js_url(self, res: str):
+        if res:
+            # My modified RegEx derived from yt-dlp, that retrieves JavaScript version
+            # Source: https://github.com/yt-dlp/yt-dlp/blob/e600a5c90817f4caac221679f6639211bba1f3a2/yt_dlp/extractor/youtube.py#L2258
+            player_version = re.search(
+                r'([0-9a-fA-F]{8})\\?', res)
+            player_version = player_version.group().replace("\\", "")
+            self._js_url = f'https://www.youtube.com/s/player/{player_version}/player_ias.vflset/en_US/base.js'
+        else:
+            raise Exception("Failed to retrieve JavaScript for this video")
+
     def _getJS(self) -> None:
-        self._js = self.youtube.js
+        # Here we get a JavaScript that links to specific Player JavaScript
+        self.url = 'https://www.youtube.com/iframe_api'
+        res = self.syncGetRequest()
+        self.extract_js_url(res.text)
 
     async def getJavaScript(self):
-        # we don't wanna break compatibility, so we just pass.
-        # We retrieve Player JavaScript using _getDecipheredURLs()
-        pass
+        # Same as in _getJS(), except it's asynchronous
+        self.url = 'https://www.youtube.com/iframe_api'
+        res = await self.asyncGetRequest()
+        self.extract_js_url(res.text)
 
-    '''
-    Not fetching for new player JavaScript if pytube.__js__ is not None or exception is not caused.
-    '''
     def _decipher(self, retry: bool = False):
-        if not pytube.__js__ or retry:
-            self.youtube._js = None
-            self.youtube._js_url = None
-            pytube.__js__ = None
-            pytube.__js_url__ = None
+        if not self._js_url or retry:
+            self._js_url = None
+            self._js = None
             self._getJS()
         try:
-            '''
-            These two are the main methods being used from PyTube.
-            Used to decipher the stream URLs using player JavaScript & the player_response passed from the getStream method of this derieved class.
-            These methods operate on the value of "player_response" key in dictionary of self._player_response & save _deciphered information in the "url_encoded_fmt_stream_map" key.
-            '''
+            # We need to decipher one URL at time.
+            for yt_format in self._player_response:
+                # If format_id is specified, then it means that we requested only for one URL (ITAG), thus we can skip
+                # all other ITAGs, which would take up our precious system resources and our valuable time
+                if self.format_id == yt_format["itag"] or self.format_id is None:
+                    # If "url" is specified in JSON, it is definitely an unciphered URL.
+                    # Thus we can skip deciphering completely.
+                    if getValue(yt_format, ["url"]):
+                        # This is a non-ciphered URL
+                        yt_format["throttled"] = False
+                        self._streams.append(yt_format)
+                        continue
+                    else:
+                        cipher = yt_format["signatureCipher"]
+                    # Some deciphering magic from yt-dlp
+                    # Source: https://github.com/yt-dlp/yt-dlp/blob/master/yt_dlp/extractor/youtube.py#L2972-L2981
+                    sc = urllib.parse.parse_qs(cipher)
+                    fmt_url = url_or_none(try_get(sc, lambda x: x['url'][0]))
+                    encrypted_sig = try_get(sc, lambda x: x['s'][0])
+                    if not (sc and fmt_url and encrypted_sig):
+                        # It's not ciphered
+                        yt_format["throttled"] = False
+                        self._streams.append(yt_format)
+                        continue
+                    if not cipher:
+                        continue
+                    signature = self.ytie._decrypt_signature(sc['s'][0], self.video_id, self._js_url)
+                    sp = try_get(sc, lambda x: x['sp'][0]) or 'signature'
+                    fmt_url += '&' + sp + '=' + signature
 
-            stream = apply_descrambler(self._player_response["streamingData"])
-            apply_signature(
-                stream, self._player_response, pytube.__js__
-            )
-            self._streams = stream
+                    # Some magic to unthrottle streams
+                    # Source: https://github.com/yt-dlp/yt-dlp/blob/master/yt_dlp/extractor/youtube.py#L2983-L2993
+                    query = urllib.parse.parse_qs(fmt_url)
+                    throttled = False
+                    if query.get('n'):
+                        try:
+                            fmt_url = update_url_query(fmt_url, {
+                                'n': self.ytie._decrypt_nsig(query['n'][0], self.video_id, self._js_url)})
+                        except ExtractorError as e:
+                            throttled = True
+                    yt_format["url"] = fmt_url
+                    yt_format["throttled"] = throttled
+                    self._streams.append(yt_format)
         except Exception as e:
             if retry:
                 raise e
             '''
             Fetch updated player JavaScript to get new cipher algorithm.
             '''
-            self._decipher(retry = True)
+            self._decipher(retry=True)

youtubesearchpython/core/streamurlfetcher.py

@@ -8,14 +8,47 @@
 from youtubesearchpython.core.componenthandler import getValue, getVideoId
 
 
+CLIENTS = {
+    "MWEB": {
+        'context': {
+            'client': {
+                'clientName': 'MWEB',
+                'clientVersion': '2.20211109.01.00'
+            }
+        },
+        'api_key': 'AIzaSyAO_FJ2SlqU8Q4STEHLGCilw_Y9_11qcW8'
+    },
+    "ANDROID": {
+        'context': {
+            'client': {
+                'clientName': 'ANDROID',
+                'clientVersion': '16.20'
+            }
+        },
+        'api_key': 'AIzaSyAO_FJ2SlqU8Q4STEHLGCilw_Y9_11qcW8'
+    },
+    "ANDROID_EMBED": {
+        'context': {
+            'client': {
+                'clientName': 'ANDROID',
+                'clientVersion': '16.20',
+                'clientScreen': 'EMBED'
+            }
+        },
+        'api_key': 'AIzaSyAO_FJ2SlqU8Q4STEHLGCilw_Y9_11qcW8'
+    }
+}
+
+
 class VideoCore(RequestCore):
-    def __init__(self, videoLink: str, componentMode: str, resultMode: int, timeout: int, enableHTML: bool):
+    def __init__(self, videoLink: str, componentMode: str, resultMode: int, timeout: int, enableHTML: bool, overridedClient: str = "ANDROID"):
         super().__init__()
         self.timeout = timeout
         self.resultMode = resultMode
         self.componentMode = componentMode
         self.videoLink = videoLink
         self.enableHTML = enableHTML
+        self.overridedClient = overridedClient
 
     # We call this when we use only HTML
     def post_request_only_html_processing(self):
@@ -34,15 +67,7 @@ def prepare_innertube_request(self):
             'racyCheckOk': True,
             "videoId": getVideoId(self.videoLink)
         })
-        self.data = {
-            'context': {
-                'client': {
-                    'clientName': 'ANDROID',
-                    'clientVersion': '16.20'
-                }
-            },
-            'api_key': 'AIzaSyAO_FJ2SlqU8Q4STEHLGCilw_Y9_11qcW8'
-        }
+        self.data = CLIENTS[self.overridedClient]
 
     async def async_create(self):
         self.prepare_innertube_request()
@@ -69,15 +94,7 @@ def prepare_html_request(self):
             'racyCheckOk': True,
             "videoId": getVideoId(self.videoLink)
         })
-        self.data = {
-            'context': {
-                'client': {
-                    'clientName': 'MWEB',
-                    'clientVersion': '2.20211109.01.00'
-                }
-            },
-            'api_key': 'AIzaSyAO_FJ2SlqU8Q4STEHLGCilw_Y9_11qcW8'
-        }
+        self.data = CLIENTS["MWEB"]
 
     def sync_html_create(self):
         self.prepare_html_request()

youtubesearchpython/core/video.py

@@ -49,10 +49,9 @@ def get(self, videoFormats: dict, itag: int) -> Union[str, None]:
             >>> print(url)
             "https://r6---sn-gwpa-5bgk.googlevideo.com/videoplayback?expire=1610798125&ei=zX8CYITXEIGKz7sP9MWL0AE&ip=2409%3A4053%3A803%3A2b22%3Adc68%3Adfb9%3Aa676%3A26a3&id=o-APBakKSE2_eMDMegtCmeWXfuhhUfAzJTmOCWj4lkEjAM&itag=251&source=youtube&requiressl=yes&mh=aP&mm=31%2C29&mn=sn-gwpa-5bgk%2Csn-gwpa-qxad&ms=au%2Crdu&mv=m&mvi=6&pl=36&initcwndbps=146250&vprv=1&mime=audio%2Fwebm&ns=ULL4mkMO31KDtEhOjkOrmpkF&gir=yes&clen=10210834&dur=634.601&lmt=1544629945422176&mt=1610776131&fvip=6&keepalive=yes&c=WEB&txp=5511222&n=uEjSqtzBZaJyVn&sparams=expire%2Cei%2Cip%2Cid%2Citag%2Csource%2Crequiressl%2Cvprv%2Cmime%2Cns%2Cgir%2Cclen%2Cdur%2Clmt&sig=AOq0QJ8wRAIgKKIEiwQTgXsdKPEyOckgVPs_LMH6KJoeaYmZic_lelECIHXHs1ZnSP5mgtpffNlIMJM3DhxcvDbA-4udFFE6AmVP&lsparams=mh%2Cmm%2Cmn%2Cms%2Cmv%2Cmvi%2Cpl%2Cinitcwndbps&lsig=AG3C_xAwRQIhAPmhL745RYeL_ffgUJk_xJLC-8riXKMylLTLA_pITYWWAiB2qUIXur8ThW7cLfQ73mIVK61mMZc2ncK6FZWjUHGcUw%3D%3D"
         '''
-        self._getDecipheredURLs(videoFormats)
-        for stream in self._streams:
-            if stream["itag"] == itag:
-                return stream["url"]
+        self._getDecipheredURLs(videoFormats, itag)
+        if len(self._streams) == 1:
+            return self._streams[0]["url"]
         return None
 
     def getAll(self, videoFormats: dict) -> Union[dict, None]:
@@ -134,4 +133,4 @@ def getAll(self, videoFormats: dict) -> Union[dict, None]:
             }
         '''
         self._getDecipheredURLs(videoFormats)
-        return {"streams": self._player_response}
+        return {"streams": self._streams}