[dependencies] Create module with all dependency imports

test/test_aes.py

@@ -23,7 +23,7 @@
     aes_gcm_decrypt_and_verify,
     aes_gcm_decrypt_and_verify_bytes,
 )
-from yt_dlp.compat import compat_pycrypto_AES
+from yt_dlp.dependencies import Cryptodome_AES
 from yt_dlp.utils import bytes_to_intlist, intlist_to_bytes
 
 # the encrypted data can be generate with 'devscripts/generate_aes_testdata.py'
@@ -45,7 +45,7 @@ def test_cbc_decrypt(self):
         data = b'\x97\x92+\xe5\x0b\xc3\x18\x91ky9m&\xb3\xb5@\xe6\x27\xc2\x96.\xc8u\x88\xab9-[\x9e|\xf1\xcd'
         decrypted = intlist_to_bytes(aes_cbc_decrypt(bytes_to_intlist(data), self.key, self.iv))
         self.assertEqual(decrypted.rstrip(b'\x08'), self.secret_msg)
-        if compat_pycrypto_AES:
+        if Cryptodome_AES:
             decrypted = aes_cbc_decrypt_bytes(data, intlist_to_bytes(self.key), intlist_to_bytes(self.iv))
             self.assertEqual(decrypted.rstrip(b'\x08'), self.secret_msg)
 
@@ -75,7 +75,7 @@ def test_gcm_decrypt(self):
         decrypted = intlist_to_bytes(aes_gcm_decrypt_and_verify(
             bytes_to_intlist(data), self.key, bytes_to_intlist(authentication_tag), self.iv[:12]))
         self.assertEqual(decrypted.rstrip(b'\x08'), self.secret_msg)
-        if compat_pycrypto_AES:
+        if Cryptodome_AES:
             decrypted = aes_gcm_decrypt_and_verify_bytes(
                 data, intlist_to_bytes(self.key), authentication_tag, intlist_to_bytes(self.iv[:12]))
             self.assertEqual(decrypted.rstrip(b'\x08'), self.secret_msg)

yt_dlp/YoutubeDL.py

@@ -27,10 +27,8 @@
 
 from .cache import Cache
 from .compat import (
-    compat_brotli,
     compat_get_terminal_size,
     compat_os_name,
-    compat_pycrypto_AES,
     compat_shlex_quote,
     compat_str,
     compat_urllib_error,
@@ -109,7 +107,6 @@
     format_field,
     formatSeconds,
     get_domain,
-    has_certifi,
     int_or_none,
     iri_to_uri,
     join_nonempty,
@@ -3656,20 +3653,11 @@ def python_implementation():
         ) or 'none'
         write_debug('exe versions: %s' % exe_str)
 
-        from .cookies import SECRETSTORAGE_AVAILABLE, SQLITE_AVAILABLE
+        from .dependencies import available_dependencies
-        from .downloader.websocket import has_websockets
-        from .postprocessor.embedthumbnail import has_mutagen
 
-        lib_str = join_nonempty(
+        write_debug('Optional libraries: %s' % (', '.join(sorted({
-            compat_brotli and compat_brotli.__name__,
+            module.__name__.split('.')[0] for module in available_dependencies.values()
-            has_certifi and 'certifi',
+        })) or 'none'))
-            compat_pycrypto_AES and compat_pycrypto_AES.__name__.split('.')[0],
-            SECRETSTORAGE_AVAILABLE and 'secretstorage',
-            has_mutagen and 'mutagen',
-            SQLITE_AVAILABLE and 'sqlite',
-            has_websockets and 'websockets',
-            delim=', ') or 'none'
-        write_debug('Optional libraries: %s' % lib_str)
 
         self._setup_opener()
         proxy_map = {}

yt_dlp/aes.py

@@ -1,16 +1,17 @@
 from math import ceil
 
-from .compat import compat_b64decode, compat_ord, compat_pycrypto_AES
+from .compat import compat_b64decode, compat_ord
+from .dependencies import Cryptodome_AES
 from .utils import bytes_to_intlist, intlist_to_bytes
 
-if compat_pycrypto_AES:
+if Cryptodome_AES:
     def aes_cbc_decrypt_bytes(data, key, iv):
         """ Decrypt bytes with AES-CBC using pycryptodome """
-        return compat_pycrypto_AES.new(key, compat_pycrypto_AES.MODE_CBC, iv).decrypt(data)
+        return Cryptodome_AES.new(key, Cryptodome_AES.MODE_CBC, iv).decrypt(data)
 
     def aes_gcm_decrypt_and_verify_bytes(data, key, tag, nonce):
         """ Decrypt bytes with AES-GCM using pycryptodome """
-        return compat_pycrypto_AES.new(key, compat_pycrypto_AES.MODE_GCM, nonce).decrypt_and_verify(data, tag)
+        return Cryptodome_AES.new(key, Cryptodome_AES.MODE_GCM, nonce).decrypt_and_verify(data, tag)
 
 else:
     def aes_cbc_decrypt_bytes(data, key, iv):

yt_dlp/compat/__init__.py

@@ -54,11 +54,6 @@ def compat_realpath(path):
     compat_realpath = os.path.realpath
 
 
-try:
-    import websockets as compat_websockets
-except ImportError:
-    compat_websockets = None
-
 # Python 3.8+ does not honor %HOME% on windows, but this breaks compatibility with youtube-dl
 # See https://github.com/yt-dlp/yt-dlp/issues/792
 # https://docs.python.org/3/library/os.path.html#os.path.expanduser
@@ -78,22 +73,6 @@ def compat_expanduser(path):
     compat_expanduser = os.path.expanduser
 
 
-try:
-    from Cryptodome.Cipher import AES as compat_pycrypto_AES
-except ImportError:
-    try:
-        from Crypto.Cipher import AES as compat_pycrypto_AES
-    except ImportError:
-        compat_pycrypto_AES = None
-
-try:
-    import brotlicffi as compat_brotli
-except ImportError:
-    try:
-        import brotli as compat_brotli
-    except ImportError:
-        compat_brotli = None
-
 WINDOWS_VT_MODE = False if compat_os_name == 'nt' else None
 
 

yt_dlp/compat/_legacy.py

@@ -17,6 +17,9 @@
 from .asyncio import run as compat_asyncio_run  # noqa: F401
 from .re import Pattern as compat_Pattern  # noqa: F401
 from .re import match as compat_Match  # noqa: F401
+from ..dependencies import Cryptodome_AES as compat_pycrypto_AES  # noqa: F401
+from ..dependencies import brotli as compat_brotli  # noqa: F401
+from ..dependencies import websockets as compat_websockets  # noqa: F401
 
 
 # compat_ctypes_WINFUNCTYPE = ctypes.WINFUNCTYPE

yt_dlp/cookies.py

@@ -17,31 +17,14 @@
     unpad_pkcs7,
 )
 from .compat import compat_b64decode, compat_cookiejar_Cookie
+from .dependencies import (
+    _SECRETSTORAGE_UNAVAILABLE_REASON,
+    secretstorage,
+    sqlite3,
+)
 from .minicurses import MultilinePrinter, QuietMultilinePrinter
 from .utils import Popen, YoutubeDLCookieJar, error_to_str, expand_path
 
-try:
-    import sqlite3
-    SQLITE_AVAILABLE = True
-except ImportError:
-    # although sqlite3 is part of the standard library, it is possible to compile python without
-    # sqlite support. See: https://github.com/yt-dlp/yt-dlp/issues/544
-    SQLITE_AVAILABLE = False
-
-
-try:
-    import secretstorage
-    SECRETSTORAGE_AVAILABLE = True
-except ImportError:
-    SECRETSTORAGE_AVAILABLE = False
-    SECRETSTORAGE_UNAVAILABLE_REASON = (
-        'as the `secretstorage` module is not installed. '
-        'Please install by running `python3 -m pip install secretstorage`.')
-except Exception as _err:
-    SECRETSTORAGE_AVAILABLE = False
-    SECRETSTORAGE_UNAVAILABLE_REASON = f'as the `secretstorage` module could not be initialized. {_err}'
-
-
 CHROMIUM_BASED_BROWSERS = {'brave', 'chrome', 'chromium', 'edge', 'opera', 'vivaldi'}
 SUPPORTED_BROWSERS = CHROMIUM_BASED_BROWSERS | {'firefox', 'safari'}
 
@@ -122,7 +105,7 @@ def extract_cookies_from_browser(browser_name, profile=None, logger=YDLLogger(),
 
 def _extract_firefox_cookies(profile, logger):
     logger.info('Extracting cookies from firefox')
-    if not SQLITE_AVAILABLE:
+    if not sqlite3:
         logger.warning('Cannot extract cookies from firefox without sqlite3 support. '
                        'Please use a python interpreter compiled with sqlite3 support')
         return YoutubeDLCookieJar()
@@ -236,7 +219,7 @@ def _get_chromium_based_browser_settings(browser_name):
 def _extract_chrome_cookies(browser_name, profile, keyring, logger):
     logger.info(f'Extracting cookies from {browser_name}')
 
-    if not SQLITE_AVAILABLE:
+    if not sqlite3:
         logger.warning(f'Cannot extract cookies from {browser_name} without sqlite3 support. '
                        'Please use a python interpreter compiled with sqlite3 support')
         return YoutubeDLCookieJar()
@@ -806,8 +789,8 @@ def _get_kwallet_password(browser_keyring_name, logger):
 
 
 def _get_gnome_keyring_password(browser_keyring_name, logger):
-    if not SECRETSTORAGE_AVAILABLE:
+    if not secretstorage:
-        logger.error(f'secretstorage not available {SECRETSTORAGE_UNAVAILABLE_REASON}')
+        logger.error(f'secretstorage not available {_SECRETSTORAGE_UNAVAILABLE_REASON}')
         return b''
     # the Gnome keyring does not seem to organise keys in the same way as KWallet,
     # using `dbus-monitor` during startup, it can be observed that chromium lists all keys

yt_dlp/dependencies.py

@@ -0,0 +1,77 @@
+# flake8: noqa: F401
+
+try:
+    import brotlicffi as brotli
+except ImportError:
+    try:
+        import brotli
+    except ImportError:
+        brotli = None
+
+
+try:
+    import certifi
+except ImportError:
+    certifi = None
+else:
+    from os.path import exists as _path_exists
+
+    # The certificate may not be bundled in executable
+    if not _path_exists(certifi.where()):
+        certifi = None
+
+
+try:
+    from Cryptodome.Cipher import AES as Cryptodome_AES
+except ImportError:
+    try:
+        from Crypto.Cipher import AES as Cryptodome_AES
+    except ImportError:
+        Cryptodome_AES = None
+
+
+try:
+    import mutagen
+except ImportError:
+    mutagen = None
+
+
+secretstorage = None
+try:
+    import secretstorage
+    _SECRETSTORAGE_UNAVAILABLE_REASON = None
+except ImportError:
+    _SECRETSTORAGE_UNAVAILABLE_REASON = (
+        'as the `secretstorage` module is not installed. '
+        'Please install by running `python3 -m pip install secretstorage`')
+except Exception as _err:
+    _SECRETSTORAGE_UNAVAILABLE_REASON = f'as the `secretstorage` module could not be initialized. {_err}'
+
+
+try:
+    import sqlite3
+except ImportError:
+    # although sqlite3 is part of the standard library, it is possible to compile python without
+    # sqlite support. See: https://github.com/yt-dlp/yt-dlp/issues/544
+    sqlite3 = None
+
+
+try:
+    import websockets
+except (ImportError, SyntaxError):
+    # websockets 3.10 on python 3.6 causes SyntaxError
+    # See https://github.com/yt-dlp/yt-dlp/issues/2633
+    websockets = None
+
+
+all_dependencies = {k: v for k, v in globals().items() if not k.startswith('_')}
+
+
+available_dependencies = {k: v for k, v in all_dependencies.items() if v}
+
+
+__all__ = [
+    'all_dependencies',
+    'available_dependencies',
+    *all_dependencies.keys(),
+]

yt_dlp/downloader/hls.py

@@ -5,7 +5,8 @@
 from .external import FFmpegFD
 from .fragment import FragmentFD
 from .. import webvtt
-from ..compat import compat_pycrypto_AES, compat_urlparse
+from ..compat import compat_urlparse
+from ..dependencies import Cryptodome_AES
 from ..downloader import get_suitable_downloader
 from ..utils import bug_reports_message, parse_m3u8_attributes, update_url_query
 
@@ -60,7 +61,7 @@ def real_download(self, filename, info_dict):
         s = urlh.read().decode('utf-8', 'ignore')
 
         can_download, message = self.can_download(s, info_dict, self.params.get('allow_unplayable_formats')), None
-        if can_download and not compat_pycrypto_AES and '#EXT-X-KEY:METHOD=AES-128' in s:
+        if can_download and not Cryptodome_AES and '#EXT-X-KEY:METHOD=AES-128' in s:
             if FFmpegFD.available():
                 can_download, message = False, 'The stream has AES-128 encryption and pycryptodomex is not available'
             else:

yt_dlp/downloader/websocket.py

@@ -3,18 +3,10 @@
 import signal
 import threading
 
-try:
-    import websockets
-except (ImportError, SyntaxError):
-    # websockets 3.10 on python 3.6 causes SyntaxError
-    # See https://github.com/yt-dlp/yt-dlp/issues/2633
-    has_websockets = False
-else:
-    has_websockets = True
-
 from .common import FileDownloader
 from .external import FFmpegFD
 from ..compat import asyncio
+from ..dependencies import websockets
 
 
 class FFmpegSinkFD(FileDownloader):

yt_dlp/extractor/fc2.py

@@ -4,10 +4,10 @@
 from ..compat import (
     compat_parse_qs,
 )
+from ..dependencies import websockets
 from ..utils import (
     ExtractorError,
     WebSocketsWrapper,
-    has_websockets,
     js_to_json,
     sanitized_Request,
     std_headers,
@@ -170,7 +170,7 @@ class FC2LiveIE(InfoExtractor):
     }]
 
     def _real_extract(self, url):
-        if not has_websockets:
+        if not websockets:
             raise ExtractorError('websockets library is not available. Please install it.', expected=True)
         video_id = self._match_id(url)
         webpage = self._download_webpage('https://live.fc2.com/%s/' % video_id, video_id)

yt_dlp/extractor/twitcasting.py

@@ -2,7 +2,7 @@
 import re
 
 from .common import InfoExtractor
-from ..downloader.websocket import has_websockets
+from ..dependencies import websockets
 from ..utils import (
     clean_html,
     ExtractorError,
@@ -161,7 +161,7 @@ def find_dmu(x):
                     note='Downloading source quality m3u8',
                     headers=self._M3U8_HEADERS, fatal=False))
 
-            if has_websockets:
+            if websockets:
                 qq = qualities(['base', 'mobilesource', 'main'])
                 streams = traverse_obj(stream_server_data, ('llfmp4', 'streams')) or {}
                 for mode, ws_url in streams.items():

yt_dlp/postprocessor/embedthumbnail.py

@@ -4,17 +4,9 @@
 import re
 import subprocess
 
-try:
-    from mutagen.flac import FLAC, Picture
-    from mutagen.mp4 import MP4, MP4Cover
-    from mutagen.oggopus import OggOpus
-    from mutagen.oggvorbis import OggVorbis
-    has_mutagen = True
-except ImportError:
-    has_mutagen = False
-
 from .common import PostProcessor
 from .ffmpeg import FFmpegPostProcessor, FFmpegThumbnailsConvertorPP
+from ..dependencies import mutagen
 from ..utils import (
     Popen,
     PostProcessingError,
@@ -26,6 +18,12 @@
     shell_quote,
 )
 
+if mutagen:
+    from mutagen.flac import FLAC, Picture
+    from mutagen.mp4 import MP4, MP4Cover
+    from mutagen.oggopus import OggOpus
+    from mutagen.oggvorbis import OggVorbis
+
 
 class EmbedThumbnailPPError(PostProcessingError):
     pass
@@ -121,7 +119,7 @@ def run(self, info):
         elif info['ext'] in ['m4a', 'mp4', 'mov']:
             prefer_atomicparsley = 'embed-thumbnail-atomicparsley' in self.get_param('compat_opts', [])
             # Method 1: Use mutagen
-            if not has_mutagen or prefer_atomicparsley:
+            if not mutagen or prefer_atomicparsley:
                 success = False
             else:
                 try:
@@ -194,7 +192,7 @@ def run(self, info):
                     raise EmbedThumbnailPPError(f'Unable to embed using ffprobe & ffmpeg; {err}')
 
         elif info['ext'] in ['ogg', 'opus', 'flac']:
-            if not has_mutagen:
+            if not mutagen:
                 raise EmbedThumbnailPPError('module mutagen was not found. Please install using `python -m pip install mutagen`')
 
             self._report_run('mutagen', filename)

yt_dlp/utils.py

@@ -41,7 +41,6 @@
 
 from .compat import (
     asyncio,
-    compat_brotli,
     compat_chr,
     compat_cookiejar,
     compat_etree_fromstring,
@@ -64,18 +63,10 @@
     compat_urllib_parse_urlparse,
     compat_urllib_request,
     compat_urlparse,
-    compat_websockets,
 )
+from .dependencies import brotli, certifi, websockets
 from .socks import ProxyType, sockssocket
 
-try:
-    import certifi
-
-    # The certificate may not be bundled in executable
-    has_certifi = os.path.exists(certifi.where())
-except ImportError:
-    has_certifi = False
-
 
 def register_socks_protocols():
     # "Register" SOCKS protocols
@@ -138,7 +129,7 @@ def random_user_agent():
 SUPPORTED_ENCODINGS = [
     'gzip', 'deflate'
 ]
-if compat_brotli:
+if brotli:
     SUPPORTED_ENCODINGS.append('br')
 
 std_headers = {
@@ -1267,7 +1258,7 @@ def deflate(data):
     def brotli(data):
         if not data:
             return data
-        return compat_brotli.decompress(data)
+        return brotli.decompress(data)
 
     def http_request(self, req):
         # According to RFC 3986, URLs can not contain non-ASCII characters, however this is not
@@ -5231,7 +5222,7 @@ class WebSocketsWrapper():
 
     def __init__(self, url, headers=None, connect=True):
         self.loop = asyncio.events.new_event_loop()
-        self.conn = compat_websockets.connect(
+        self.conn = websockets.connect(
             url, extra_headers=headers, ping_interval=None,
             close_timeout=float('inf'), loop=self.loop, ping_timeout=float('inf'))
         if connect:
@@ -5294,9 +5285,6 @@ def _cancel_all_tasks(loop):
                 })
 
 
-has_websockets = bool(compat_websockets)
-
-
 def merge_headers(*dicts):
     """Merge dicts of http headers case insensitively, prioritizing the latter ones"""
     return {k.title(): v for k, v in itertools.chain.from_iterable(map(dict.items, dicts))}
@@ -5312,3 +5300,8 @@ def __get__(self, _, cls):
 
 def Namespace(**kwargs):
     return collections.namedtuple('Namespace', kwargs)(**kwargs)
+
+
+# Deprecated
+has_certifi = bool(certifi)
+has_websockets = bool(websockets)