Backport to Python 3.7

devscripts/install_deps.py

@@ -48,7 +48,8 @@ def main():
 
     def yield_deps(group):
         for dep in group:
-            if mobj := recursive_pattern.fullmatch(dep):
+            mobj = recursive_pattern.fullmatch(dep)
+            if mobj:
                 yield from optional_groups.get(mobj.group('group_name'), [])
             else:
                 yield dep

devscripts/make_changelog.py

@@ -35,7 +35,7 @@ class CommitGroup(enum.Enum):
     MISC = 'Misc.'
 
     @classmethod
-    @lru_cache
+    @lru_cache()
     def subgroup_lookup(cls):
         return {
             name: group
@@ -56,7 +56,7 @@ def subgroup_lookup(cls):
         }
 
     @classmethod
-    @lru_cache
+    @lru_cache()
     def group_lookup(cls):
         result = {
             'fd': cls.DOWNLOADER,

devscripts/tomlparse.py

@@ -64,19 +64,22 @@ def get_target(root: dict, paths: list[str], is_list=False):
 def parse_enclosed(data: str, index: int, end: str, ws_re: re.Pattern):
     index += 1
 
-    if match := ws_re.match(data, index):
+    match = ws_re.match(data, index)
+    if match:
         index = match.end()
 
     while data[index] != end:
         index = yield True, index
 
-        if match := ws_re.match(data, index):
+        match = ws_re.match(data, index)
+        if match:
             index = match.end()
 
         if data[index] == ',':
             index += 1
 
-        if match := ws_re.match(data, index):
+        match = ws_re.match(data, index)
+        if match:
             index = match.end()
 
     assert data[index] == end
@@ -106,7 +109,8 @@ def parse_value(data: str, index: int):
 
         return index, result
 
-    if match := STRING_RE.match(data, index):
+    match = STRING_RE.match(data, index)
+    if match:
         return match.end(), json.loads(match[0]) if match[0][0] == '"' else match[0][1:-1]
 
     match = LEFTOVER_VALUE_RE.match(data, index)

pyproject.toml

@@ -12,7 +12,7 @@ maintainers = [
 ]
 description = "A feature-rich command-line audio/video downloader"
 readme = "README.md"
-requires-python = ">=3.8"
+requires-python = ">=3.7"
 keywords = [
     "youtube-dl",
     "video-downloader",
@@ -28,6 +28,7 @@ classifiers = [
     "Environment :: Console",
     "Programming Language :: Python",
     "Programming Language :: Python :: 3 :: Only",
+    "Programming Language :: Python :: 3.7",
     "Programming Language :: Python :: 3.8",
     "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
@@ -48,7 +49,7 @@ dependencies = [
     "pycryptodomex",
     "requests>=2.31.0,<3",
     "urllib3>=1.26.17,<3",
-    "websockets>=12.0",
+    "websockets>=11.0.0",
 ]
 
 [project.optional-dependencies]

setup.cfg

@@ -16,20 +16,23 @@ remove-unused-variables = true
 
 [tox:tox]
 skipsdist = true
-envlist = py{38,39,310,311,312},pypy{38,39,310}
+envlist = py{37,38,39,310,311,312},pypy{37,38,39,310}
 skip_missing_interpreters = true
 
 [testenv]  # tox
 deps =
     pytest
+    requests
+    curl-cffi
+    websockets
 commands = pytest {posargs:"-m not download"}
 passenv = HOME  # For test_compat_expanduser
 setenv =
     # PYTHONWARNINGS = error  # Catches PIP's warnings too
 
 
 [isort]
-py_version = 38
+py_version = 37
 multi_line_output = VERTICAL_HANGING_INDENT
 line_length = 80
 reverse_relative = true

test/test_traversal.py

@@ -402,7 +402,7 @@ def test_traversal_morsel(self):
             'secure': 'f',
             'httponly': 'g',
             'version': 'h',
-            'samesite': 'i',
+            # 'samesite': 'i',
         }
         morsel = http.cookies.Morsel()
         morsel.set('item_key', 'item_value', 'coded_value')

yt_dlp/YoutubeDL.py

@@ -718,7 +718,8 @@ def check_deprecated(param, option, suggestion):
         for msg in self.params.get('_deprecation_warnings', []):
             self.deprecated_feature(msg)
 
-        if impersonate_target := self.params.get('impersonate'):
+        impersonate_target = self.params.get('impersonate')
+        if impersonate_target:
             if not self._impersonate_target_available(impersonate_target):
                 raise YoutubeDLError(
                     f'Impersonate target "{impersonate_target}" is not available. '
@@ -2680,10 +2681,14 @@ def _fill_common_fields(self, info_dict, final=True):
             if new_key in info_dict and old_key in info_dict:
                 if '_version' not in info_dict:  # HACK: Do not warn when using --load-info-json
                     self.deprecation_warning(f'Do not return {old_key!r} when {new_key!r} is present')
-            elif old_value := info_dict.get(old_key):
-                info_dict[new_key] = old_value.split(', ')
-            elif new_value := info_dict.get(new_key):
-                info_dict[old_key] = ', '.join(v.replace(',', '\N{FULLWIDTH COMMA}') for v in new_value)
+            else:
+                old_value = info_dict.get(old_key)
+                if old_value:
+                    info_dict[new_key] = old_value.split(', ')
+                else:
+                    new_value = info_dict.get(new_key)
+                    if new_value:
+                        info_dict[old_key] = ', '.join(v.replace(',', '\N{FULLWIDTH COMMA}') for v in new_value)
 
     def _raise_pending_errors(self, info):
         err = info.pop('__pending_error', None)

yt_dlp/__init__.py

@@ -1,8 +1,8 @@
 import sys
 
-if sys.version_info < (3, 8):
+if sys.version_info < (3, 7):
     raise ImportError(
-        f'You are using an unsupported version of Python. Only Python versions 3.8 and above are supported by yt-dlp')  # noqa: F541
+        f'You are using an unsupported version of Python. Only Python versions 3.7 and above are supported by yt-dlp')  # noqa: F541
 
 __license__ = 'The Unlicense'
 

yt_dlp/compat/functools.py

@@ -10,3 +10,17 @@
     cache  # >= 3.9
 except NameError:
     cache = lru_cache(maxsize=None)
+
+try:
+    cached_property  # >= 3.8
+except NameError:
+    class cached_property:
+        def __init__(self, func):
+            update_wrapper(self, func)
+            self.func = func
+
+        def __get__(self, instance, _):
+            if instance is None:
+                return self
+            setattr(instance, self.func.__name__, self.func(instance))
+            return getattr(instance, self.func.__name__)

yt_dlp/compat/shlex.py

@@ -0,0 +1,15 @@
+# flake8: noqa: F405
+from shlex import *  # noqa: F403
+
+from .compat_utils import passthrough_module
+
+passthrough_module(__name__, 'shlex')
+del passthrough_module
+
+
+try:
+    join
+except NameError:
+    def join(split_command):
+        """Return a shell-escaped string from *split_command*."""
+        return ' '.join(quote(arg) for arg in split_command)

yt_dlp/downloader/common.py

@@ -371,7 +371,9 @@ def with_fields(*tups, default=''):
         if s['status'] != 'downloading':
             return
 
-        if update_delta := self.params.get('progress_delta'):
+        update_delta = self.params.get('progress_delta')
+
+        if update_delta:
             with self._progress_delta_lock:
                 if time.monotonic() < self._progress_delta_time:
                     return

yt_dlp/extractor/asobistage.py

@@ -1,4 +1,4 @@
-import functools
+from ..compat import functools
 
 from .common import InfoExtractor
 from ..utils import str_or_none, url_or_none

yt_dlp/extractor/bbc.py

@@ -1424,14 +1424,16 @@ def extract_all(pattern):
             'model', 'blocks', is_type('media'),
             'model', 'blocks', is_type('mediaMetadata'),
             'model', {dict}, any))
-        if model and (entry := parse_model(model)):
-            if not entry.get('timestamp'):
-                entry['timestamp'] = traverse_obj(next_data, (
-                    ..., 'contents', is_type('timestamp'), 'model',
-                    'timestamp', {functools.partial(int_or_none, scale=1000)}, any))
-            entries.append(entry)
-            return self.playlist_result(
-                entries, playlist_id, playlist_title, playlist_description)
+        if model:
+            entry = parse_model(model)
+            if entry:
+                if not entry.get('timestamp'):
+                    entry['timestamp'] = traverse_obj(next_data, (
+                        ..., 'contents', is_type('timestamp'), 'model',
+                        'timestamp', {functools.partial(int_or_none, scale=1000)}, any))
+                entries.append(entry)
+                return self.playlist_result(
+                    entries, playlist_id, playlist_title, playlist_description)
 
         # Multiple video article (e.g.
         # http://www.bbc.co.uk/blogs/adamcurtis/entries/3662a707-0af9-3149-963f-47bea720b460)

yt_dlp/extractor/common.py

@@ -1,6 +1,5 @@
 import base64
 import collections
-import functools
 import getpass
 import hashlib
 import http.client
@@ -22,6 +21,7 @@
 import urllib.request
 import xml.etree.ElementTree
 
+from ..compat import functools  # isort: split
 from ..compat import (
     compat_etree_fromstring,
     compat_expanduser,

yt_dlp/extractor/crtvg.py

@@ -39,7 +39,10 @@ def _real_extract(self, url):
         formats.extend(self._extract_mpd_formats(video_url + '/manifest.mpd', video_id, fatal=False))
 
         old_video_id = None
-        if mobj := re.fullmatch(r'[^/#?]+-(?P<old_id>\d{7})', video_id):
+
+        mobj = re.fullmatch(r'[^/#?]+-(?P<old_id>\d{7})', video_id)
+
+        if mobj:
             old_video_id = [make_archive_id(self, mobj.group('old_id'))]
 
         return {

yt_dlp/extractor/crunchyroll.py

@@ -62,7 +62,8 @@ def _request_token(self, headers, data, note='Requesting token', errnote='Failed
         except ExtractorError as error:
             if not isinstance(error.cause, HTTPError) or error.cause.status != 403:
                 raise
-            if target := error.cause.response.extensions.get('impersonate'):
+            target = error.cause.response.extensions.get('impersonate')
+            if target:
                 raise ExtractorError(f'Got HTTP Error 403 when using impersonate target "{target}"')
             raise ExtractorError(
                 'Request blocked by Cloudflare. '
@@ -234,7 +235,8 @@ def _extract_stream(self, identifier, display_id=None):
 
         # Invalidate stream token to avoid rate-limit
         error_msg = 'Unable to invalidate stream token; you may experience rate-limiting'
-        if stream_token := stream_response.get('token'):
+        stream_token = stream_response.get('token')
+        if stream_token:
             self._request_webpage(Request(
                 f'https://cr-play-service.prd.crunchyrollsvc.com/v1/token/{identifier}/{stream_token}/inactive',
                 headers=headers, method='PATCH'), display_id, 'Invalidating stream token', error_msg, fatal=False)

yt_dlp/extractor/dangalplay.py

@@ -132,8 +132,10 @@ def _real_extract(self, url):
                 if error_info.get('code') == '1016':
                     self.raise_login_required(
                         f'Your token has expired or is invalid. {self._LOGIN_HINT}', method=None)
-                elif msg := error_info.get('message'):
-                    raise ExtractorError(msg)
+                else:
+                    msg = error_info.get('message')
+                    if msg:
+                        raise ExtractorError(msg)
             raise
 
         m3u8_url = traverse_obj(details, (

yt_dlp/extractor/elementorembed.py

@@ -56,13 +56,16 @@ class ElementorEmbedIE(InfoExtractor):
     def _extract_from_webpage(self, url, webpage):
         for data_settings in re.findall(self._WIDGET_REGEX, webpage):
             data = self._parse_json(data_settings, None, fatal=False, transform_source=unescapeHTML)
-            if youtube_url := traverse_obj(data, ('youtube_url', {url_or_none})):
+            youtube_url = traverse_obj(data, ('youtube_url', {url_or_none}))
+            if youtube_url:
                 yield self.url_result(youtube_url, ie=YoutubeIE)
 
             for video in traverse_obj(data, ('tabs', lambda _, v: v['_id'], {dict})):
-                if youtube_url := traverse_obj(video, ('youtube_url', {url_or_none})):
+                youtube_url = traverse_obj(video, ('youtube_url', {url_or_none}))
+                if youtube_url:
                     yield self.url_result(youtube_url, ie=YoutubeIE)
-                if vimeo_url := traverse_obj(video, ('vimeo_url', {url_or_none})):
+                vimeo_url = traverse_obj(video, ('vimeo_url', {url_or_none}))
+                if vimeo_url:
                     yield self.url_result(vimeo_url, ie=VimeoIE)
                 for direct_url in traverse_obj(video, (('hosted_url', 'external_url'), 'url', {url_or_none})):
                     yield {

yt_dlp/extractor/err.py

@@ -194,7 +194,9 @@ def _real_extract(self, url):
                 format_url, video_id, mpd_id='dash', fatal=False)
             formats.extend(fmts)
             self._merge_subtitles(subs, target=subtitles)
-        if format_url := traverse_obj(media_data, ('src', 'file', {url_or_none})):
+
+        format_url = traverse_obj(media_data, ('src', 'file', {url_or_none}))
+        if format_url:
             formats.append({
                 'url': format_url,
                 'format_id': 'http',

yt_dlp/extractor/francetv.py

@@ -119,7 +119,8 @@ def _extract_video(self, video_id, hostname=None):
             video_url = video['url']
             format_id = video.get('format')
 
-            if token_url := url_or_none(video.get('token')):
+            token_url = url_or_none(video.get('token'))
+            if token_url:
                 tokenized_url = traverse_obj(self._download_json(
                     token_url, video_id, f'Downloading signed {format_id} manifest URL',
                     fatal=False, query={
@@ -138,7 +139,8 @@ def _extract_video(self, video_id, hostname=None):
                 fmts, subs = self._extract_m3u8_formats_and_subtitles(
                     video_url, video_id, 'mp4', m3u8_id=format_id, fatal=False)
                 for f in traverse_obj(fmts, lambda _, v: v['vcodec'] == 'none' and v.get('tbr') is None):
-                    if mobj := re.match(rf'{format_id}-[Aa]udio-\w+-(?P<bitrate>\d+)', f['format_id']):
+                    mobj = re.match(rf'{format_id}-[Aa]udio-\w+-(?P<bitrate>\d+)', f['format_id'])
+                    if mobj:
                         f.update({
                             'tbr': int_or_none(mobj.group('bitrate')),
                             'acodec': 'mp4a',

yt_dlp/extractor/gbnews.py

@@ -49,7 +49,7 @@ class GBNewsIE(InfoExtractor):
         'params': {'skip_download': 'm3u8'},
     }]
 
-    @functools.lru_cache
+    @functools.lru_cache()
     def _get_ss_endpoint(self, data_id, data_env):
         if not data_id:
             data_id = 'GB003'

yt_dlp/extractor/godresource.py

@@ -50,7 +50,8 @@ def _real_extract(self, url):
 
         video_url = api_data['streamUrl']
         is_live = api_data.get('isLive') or False
-        if (ext := determine_ext(video_url)) == 'm3u8':
+        ext = determine_ext(video_url)
+        if ext == 'm3u8':
             formats, subtitles = self._extract_m3u8_formats_and_subtitles(
                 video_url, display_id, live=is_live)
         elif ext == 'mp4':