[script.module.youtube.dl@matrix] 25.05.18+matrix.1

based on GIT dated of May 18 2025

Author: joaopa00

script.module.youtube.dl/lib/youtube_dl/YoutubeDL.py

@@ -5,7 +5,6 @@
 
 __license__ = 'Public Domain'
 
-import codecs
 import io
 import os
 import random
@@ -17,10 +16,12 @@
 )
 from .compat import (
     compat_getpass,
+    compat_register_utf8,
     compat_shlex_split,
-    workaround_optparse_bug9161,
+    _workaround_optparse_bug9161,
 )
 from .utils import (
+    _UnsafeExtensionError,
     DateRange,
     decodeOption,
     DEFAULT_OUTTMPL,
@@ -46,12 +47,10 @@
 
 
 def _real_main(argv=None):
-    # Compatibility fixes for Windows
-    if sys.platform == 'win32':
-        # https://github.com/ytdl-org/youtube-dl/issues/820
-        codecs.register(lambda name: codecs.lookup('utf-8') if name == 'cp65001' else None)
+    # Compatibility fix for Windows
+    compat_register_utf8()
 
-    workaround_optparse_bug9161()
+    _workaround_optparse_bug9161()
 
     setproctitle('youtube-dl')
 
@@ -175,6 +174,9 @@ def _real_main(argv=None):
     if opts.ap_mso and opts.ap_mso not in MSO_INFO:
         parser.error('Unsupported TV Provider, use --ap-list-mso to get a list of supported TV Providers')
 
+    if opts.no_check_extensions:
+        _UnsafeExtensionError.lenient = True
+
     def parse_retries(retries):
         if retries in ('inf', 'infinite'):
             parsed_retries = float('inf')

script.module.youtube.dl/lib/youtube_dl/__init__.py

@@ -11,6 +11,7 @@
     # direct call of __main__.py
     import os.path
     path = os.path.realpath(os.path.abspath(__file__))
+    sys.path.insert(0, os.path.dirname(os.path.dirname(path)))
 
 import youtube_dl
 

script.module.youtube.dl/lib/youtube_dl/__main__.py

@@ -1,19 +1,24 @@
+# coding: utf-8
 from __future__ import unicode_literals
 
 import errno
-import io
 import json
 import os
 import re
 import shutil
 import traceback
 
-from .compat import compat_getenv
+from .compat import (
+    compat_getenv,
+    compat_open as open,
+    compat_os_makedirs,
+)
 from .utils import (
     error_to_compat_str,
+    escape_rfc3986,
     expand_path,
     is_outdated_version,
-    try_get,
+    traverse_obj,
     write_json_file,
 )
 from .version import __version__
@@ -28,23 +33,35 @@ class Cache(object):
     def __init__(self, ydl):
         self._ydl = ydl
 
+    def _write_debug(self, *args, **kwargs):
+        self._ydl.write_debug(*args, **kwargs)
+
+    def _report_warning(self, *args, **kwargs):
+        self._ydl.report_warning(*args, **kwargs)
+
+    def _to_screen(self, *args, **kwargs):
+        self._ydl.to_screen(*args, **kwargs)
+
+    def _get_param(self, k, default=None):
+        return self._ydl.params.get(k, default)
+
     def _get_root_dir(self):
-        res = self._ydl.params.get('cachedir')
+        res = self._get_param('cachedir')
         if res is None:
             cache_root = compat_getenv('XDG_CACHE_HOME', '~/.cache')
             res = os.path.join(cache_root, self._YTDL_DIR)
         return expand_path(res)
 
     def _get_cache_fn(self, section, key, dtype):
-        assert re.match(r'^[a-zA-Z0-9_.-]+$', section), \
+        assert re.match(r'^[\w.-]+$', section), \
             'invalid section %r' % section
-        assert re.match(r'^[a-zA-Z0-9_.-]+$', key), 'invalid key %r' % key
+        key = escape_rfc3986(key, safe='').replace('%', ',')  # encode non-ascii characters
         return os.path.join(
             self._get_root_dir(), section, '%s.%s' % (key, dtype))
 
     @property
     def enabled(self):
-        return self._ydl.params.get('cachedir') is not False
+        return self._get_param('cachedir') is not False
 
     def store(self, section, key, data, dtype='json'):
         assert dtype in ('json',)
@@ -54,61 +71,75 @@ def store(self, section, key, data, dtype='json'):
 
         fn = self._get_cache_fn(section, key, dtype)
         try:
-            try:
-                os.makedirs(os.path.dirname(fn))
-            except OSError as ose:
-                if ose.errno != errno.EEXIST:
-                    raise
+            compat_os_makedirs(os.path.dirname(fn), exist_ok=True)
+            self._write_debug('Saving {section}.{key} to cache'.format(section=section, key=key))
             write_json_file({self._VERSION_KEY: __version__, 'data': data}, fn)
         except Exception:
             tb = traceback.format_exc()
-            self._ydl.report_warning(
-                'Writing cache to %r failed: %s' % (fn, tb))
+            self._report_warning('Writing cache to {fn!r} failed: {tb}'.format(fn=fn, tb=tb))
+
+    def clear(self, section, key, dtype='json'):
+
+        if not self.enabled:
+            return
+
+        fn = self._get_cache_fn(section, key, dtype)
+        self._write_debug('Clearing {section}.{key} from cache'.format(section=section, key=key))
+        try:
+            os.remove(fn)
+        except Exception as e:
+            if getattr(e, 'errno') == errno.ENOENT:
+                # file not found
+                return
+            tb = traceback.format_exc()
+            self._report_warning('Clearing cache from {fn!r} failed: {tb}'.format(fn=fn, tb=tb))
 
     def _validate(self, data, min_ver):
-        version = try_get(data, lambda x: x[self._VERSION_KEY])
+        version = traverse_obj(data, self._VERSION_KEY)
         if not version:  # Backward compatibility
             data, version = {'data': data}, self._DEFAULT_VERSION
         if not is_outdated_version(version, min_ver or '0', assume_new=False):
             return data['data']
-        self._ydl.to_screen(
-            'Discarding old cache from version {version} (needs {min_ver})'.format(**locals()))
+        self._write_debug('Discarding old cache from version {version} (needs {min_ver})'.format(version=version, min_ver=min_ver))
 
-    def load(self, section, key, dtype='json', default=None, min_ver=None):
+    def load(self, section, key, dtype='json', default=None, **kw_min_ver):
         assert dtype in ('json',)
+        min_ver = kw_min_ver.get('min_ver')
 
         if not self.enabled:
             return default
 
         cache_fn = self._get_cache_fn(section, key, dtype)
         try:
+            with open(cache_fn, encoding='utf-8') as cachef:
+                self._write_debug('Loading {section}.{key} from cache'.format(section=section, key=key), only_once=True)
+                return self._validate(json.load(cachef), min_ver)
+        except (ValueError, KeyError):
             try:
-                with io.open(cache_fn, 'r', encoding='utf-8') as cachef:
-                    return self._validate(json.load(cachef), min_ver)
-            except ValueError:
-                try:
-                    file_size = os.path.getsize(cache_fn)
-                except (OSError, IOError) as oe:
-                    file_size = error_to_compat_str(oe)
-                self._ydl.report_warning(
-                    'Cache retrieval from %s failed (%s)' % (cache_fn, file_size))
-        except IOError:
-            pass  # No cache available
+                file_size = 'size: %d' % os.path.getsize(cache_fn)
+            except (OSError, IOError) as oe:
+                file_size = error_to_compat_str(oe)
+            self._report_warning('Cache retrieval from %s failed (%s)' % (cache_fn, file_size))
+        except Exception as e:
+            if getattr(e, 'errno') == errno.ENOENT:
+                # no cache available
+                return
+            self._report_warning('Cache retrieval from %s failed' % (cache_fn,))
 
         return default
 
     def remove(self):
         if not self.enabled:
-            self._ydl.to_screen('Cache is disabled (Did you combine --no-cache-dir and --rm-cache-dir?)')
+            self._to_screen('Cache is disabled (Did you combine --no-cache-dir and --rm-cache-dir?)')
             return
 
         cachedir = self._get_root_dir()
         if not any((term in cachedir) for term in ('cache', 'tmp')):
-            raise Exception('Not removing directory %s - this does not look like a cache dir' % cachedir)
+            raise Exception('Not removing directory %s - this does not look like a cache dir' % (cachedir,))
 
-        self._ydl.to_screen(
-            'Removing cache dir %s .' % cachedir, skip_eol=True)
+        self._to_screen(
+            'Removing cache dir %s .' % (cachedir,), skip_eol=True, ),
         if os.path.exists(cachedir):
-            self._ydl.to_screen('.', skip_eol=True)
+            self._to_screen('.', skip_eol=True)
             shutil.rmtree(cachedir)
-        self._ydl.to_screen('.')
+        self._to_screen('.')

script.module.youtube.dl/lib/youtube_dl/cache.py

@@ -10,9 +10,10 @@
 # https://github.com/unicode-org/icu/blob/main/icu4c/source/data/unidata/CaseFolding.txt
 # In case newly foldable Unicode characters are defined, paste the new version
 # of the text inside the ''' marks.
-# The text is expected to have only blank lines andlines with 1st character #,
+# The text is expected to have only blank lines and lines with 1st character #,
 # all ignored, and fold definitions like this:
-# `from_hex_code; space_separated_to_hex_code_list; comment`
+# `from_hex_code; status; space_separated_to_hex_code_list; comment`
+# Only `status` C/F are used.
 
 _map_str = '''
 # CaseFolding-15.0.0.txt
@@ -1657,11 +1658,6 @@ def _parse_unichr(s):
 del _map_str
 
 
-def casefold(s):
+def _casefold(s):
     assert isinstance(s, compat_str)
     return ''.join((_map.get(c, c) for c in s))
-
-
-__all__ = [
-    casefold
-]