Observed string: "import enumimport functoolsimport jsonimport osimport reimport subprocessimport sysimport tempfileimport timeimport uuidfrom .fragment import FragmentFDfrom ..networking import Requestfrom ..postprocessor.ffmpeg import EXT_TO_OUT_FORMATS, FFmpegPostProcessorfrom ..utils import ( Popen, RetryManager, _configuration_args, check_executable, classproperty, cli_bool_option, cli_option, cli_valueless_option, determine_ext, encodeArgument, find_available_port, remove_end, traverse_obj,class Features(enum.Enum): TO_STDOUT = enum.auto() MULTIPLE_FORMATS = enum.auto()class ExternalFD(FragmentFD): SUPPORTED_PROTOCOLS = ('http', 'https', 'ftp', 'ftps') SUPPORTED_FEATURES = () _CAPTURE_STDERR = True def real_download(self, filename, info_dict): self.report_destination(filename) tmpfilename = self.temp_name(filename) self._cookies_tempfile = None try: started = time.time() retval = self._call_downloader(tmpfilename, info_dict) except KeyboardInterrupt: if not info_dict.get('is_live'): raise # Live stream downloading cancellation should be considered as # correct and expected termination thus all postprocessing # should take place retval = 0 self.to_screen(f'[{self.get_basename()}] Interrupted by user') finally: if self._cookies_tempfile: self.try_remove(self._cookies_tempfile) if retval == 0: status = { 'filename': filename, 'status': 'finished', 'elapsed': time.time() - started, } if filename != '-': fsize = os.path.getsize(tmpfilename) self.try_rename(tmpfilename, filename) status.update({ 'downloaded_bytes': fsize, 'total_bytes': fsize, }) self._hook_progress(status, info_dict) return True else: self.to_stderr('\n') self.report_error('%s exited with code %d' % ( self.get_basename(), retval)) return False @classmethod def get_basename(cls): return cls.__name__[:-2].lower() @classproperty def EXE_NAME(cls): return cls.get_basename() @functools.cached_property def exe(self): return self.EXE_NAME @classmethod def available(cls, path=None): path = check_executable( cls.EXE_NAME if path in (None, cls.get_basename()) else path, [cls.AVAILABLE_OPT]) if not path: return False cls.exe = path return path @classmethod def supports(cls, info_dict): return all(( not info_dict.get('to_stdout') or Features.TO_STDOUT in cls.SUPPORTED_FEATURES, '+' not in info_dict['protocol'] or Features.MULTIPLE_FORMATS in cls.SUPPORTED_FEATURES, not traverse_obj(info_dict, ('hls_aes', ...), 'extra_param_to_segment_url', 'extra_param_to_key_url'), all(proto in cls.SUPPORTED_PROTOCOLS for proto in info_dict['protocol'].split('+')), )) @classmethod def can_download(cls, info_dict, path=None): return cls.available(path) and cls.supports(info_dict) def _option(self, command_option, param): return cli_option(self.params, command_option, param) def _bool_option(self, command_option, param, true_value='true', false_value='false', separator=None): return cli_bool_option(self.params, command_option, param, true_value, false_value, separator) def _valueless_option(self, command_option, param, expected_value=True): return cli_valueless_option(self.params, command_option, param, expected_value) def _configuration_args(self, keys=None, *args, **kwargs): return _configuration_args( self.get_basename(), self.params.get('external_downloader_args'), self.EXE_NAME, keys, *args, **kwargs) def _write_cookies(self): if not self.ydl.cookiejar.filename: tmp_cookies = tempfile.NamedTemporaryFile(suffix='.cookies', delete=False) tmp_cookies.close() self._cookies_tempfile = tmp_cookies.name self.to_screen(f'[download] Writing temporary cookies file to "{self._cookies_tempfile}"') # real_download resets _cookies_tempfile; if it's None then save() will write to cookiejar.filename self.ydl.cookiejar.save(self._cookies_tempfile) return self.ydl.cookiejar.filename or self._cookies_tempfile def _call_downloader(self, tmpfilename, info_dict): """ Either overwrite this or implement _make_cmd """ cmd = [encodeArgument(a) for a in self._make_cmd(tmpfilename, info_dict)] self._debug_cmd(cmd) if 'fragments' not in info_dict: _, stderr, returncode = self._call_process(cmd, info_dict) if returncode and stderr: self.to_stderr(stderr) return returncode skip_unavailable_fragments = self.params.get('skip_unavailable_fragments', True) retry_manager = RetryManager(self.params.get('fragment_retries'), self.report_retry, frag_index=None, fatal=not skip_unavailable_fragments) for retry in retry_manager: _, stderr, returncode = self._call_process(cmd, info_dict) if not returncode: break # TODO: Decide whether to retry based on error code # https://aria2.github.io/manual/en/html/aria2c.html#exit-status if stderr: self.to_stderr(stderr) retry.error = Exception() continue if not skip_unavailable_fragments and retry_manager.error: return -1 decrypt_fragment = self.decrypter(info_dict) dest, _ = self.sanitize_open(tmpfilename, 'wb') for frag_index, fragment in enumerate(info_dict['fragments']): fragment_filename = f'{tmpfilename}-Frag{frag_index}' try: src, _ = self.sanitize_open(fragment_filename, 'rb') except OSError as err: if skip_unavailable_fragments and frag_index > 1: self.report_skip_fragment(frag_index, err) continue self.report_error(f'Unable to open fragment {frag_index}; {err}') return -1 dest.write(decrypt_fragment(fragment, src.read())) src.close() if not self.params.get('keep_fragments', False): self.try_remove(fragment_filename) dest.close() self.try_remove(f'{tmpfilename}.frag.urls') return 0 def _call_process(self, cmd, info_dict): return Popen.run(cmd, text=True, stderr=subprocess.PIPE if self._CAPTURE_STDERR else None)class CurlFD(ExternalFD): AVAILABLE_OPT = '-V' _CAPTURE_STDERR = False # curl writes the progress to stderr def _make_cmd(self, tmpfilename, info_dict): cmd = [self.exe, '--location', '-o', tmpfilename, '--compressed'] cookie_header = self.ydl.cookiejar.get_cookie_header(info_dict['url']) if cookie_header: cmd += ['--cookie', cookie_header] if info_dict.get('http_headers') is not None: for key, val in info_dict['http_headers'].items(): cmd += ['--header', f'{key}: {val}'] cmd += self._bool_option('--continue-at', 'continuedl', '-', '0') cmd += self._valueless_option('--silent', 'noprogress') cmd += self._valueless_option('--verbose', 'verbose') cmd += self._option('--limit-rate', 'ratelimit') retry = self._option('--retry', 'retries') if len(retry) == 2: if retry[1] in ('inf', 'infinite'): retry[1] = '2147483647' cmd += retry cmd += self._option('--max-filesize', 'max_filesize') cmd += self._option('--interface', 'source_address') cmd += self._option('--proxy', 'proxy') cmd += self._valueless_option('--insecure', 'nocheckcertificate') cmd += self._configuration_args() cmd += ['--', info_dict['url']] return cmdclass AxelFD(ExternalFD): AVAILABLE_OPT = '-V' def _make_cmd(self, tmpfilename, info_dict): cmd = [self.exe, '-o', tmpfilename] if info_dict.get('http_headers') is not None: for key, val in info_dict['http_headers'].items(): cmd += ['-H', f'{key}: {val}'] cookie_header = self.ydl.cookiejar.get_cookie_header(info_dict['url']) if cookie_header: cmd += ['-H', f'Cookie: {cookie_header}', '--max-redirect=0'] cmd += self._configuration_args() cmd += ['--', info_dict['url']] return cmdclass WgetFD(ExternalFD): AVAILABLE_OPT = '--version' def _make_cmd(self, tmpfilename, info_dict): cmd = [self.exe, '-O', tmpfilename, '-nv', '--compression=auto'] if self.ydl.cookiejar.get_cookie_header(info_dict['url']): cmd += ['--load-cookies', self._write_cookies()] if info_dict.get('http_headers') is not None: for key, val in info_dict['http_headers'].items(): cmd += ['--header', f'{key}: {val}'] cmd += self._option('--limit-rate', 'ratelimit') retry = self._option('--tries', 'retries') if len(retry) == 2: if retry[1] in ('inf', 'infinite'): retry[1] = '0' cmd += retry cmd += self._option('--bind-address', 'source_address') proxy = self.params.get('proxy') if proxy: for var in ('http_proxy', 'https_proxy'): cmd += ['--execute', f'{var}={proxy}'] cmd += self._valueless_option('--no-check-certificate', 'nocheckcertificate') cmd += self._configuration_args() cmd += ['--', info_dict['url']] return cmdclass Aria2cFD(ExternalFD): AVAILABLE_OPT = '-v' SUPPORTED_PROTOCOLS = ('http', 'https', 'ftp', 'ftps', 'dash_frag_urls', 'm3u8_frag_urls') @staticmethod def supports_manifest(manifest): UNSUPPORTED_FEATURES = [ r'#EXT-X-BYTERANGE', # playlists composed of byte ranges of media files [1] # 1. https://tools.ietf.org/html/draft-pantos-http-live-streaming-17#section-4.3.2.2 ] check_results = (not re.search(feature, manifest) for feature in UNSUPPORTED_FEATURES) return all(check_results) @staticmethod def _aria2c_filename(fn): return fn if os.path.isabs(fn) else f'.{os.path.sep}{fn}' def _call_downloader(self, tmpfilename, info_dict): # FIXME: Disabled due to https://github.com/yt-dlp/yt-dlp/issues/5931 if False and 'no-external-downloader-progress' not in self.params.get('compat_opts', []): info_dict['__rpc'] = { 'port': find_available_port() or 19190, 'secret': str(uuid.uuid4()), } return super()._call_downloader(tmpfilename, info_dict) def _make_cmd(self, tmpfilename, info_dict): cmd = [self.exe, '-c', '--no-conf', '--console-log-level=warn', '--summary-interval=0', '--download-result=hide', '--http-accept-gzip=true', '--file-allocation=none', '-x16', '-j16', '-s16'] if 'fragments' in info_dict: cmd += ['--allow-overwrite=true', '--allow-piece-length-change=true'] else: cmd += ['--min-split-size', '1M'] if self.ydl.cookiejar.get_cookie_header(info_dict['url']): cmd += [f'--load-cookies={self._write_cookies()}'] if info_dict.get('http_headers') is not None: for key, val in info_dict['http_headers'].items(): cmd += ['--header', f'{key}: {val}'] cmd += self._option('--max-overall-download-limit', 'ratelimit') cmd += self._option('--interface', 'source_address') cmd += self._option('--all-proxy', 'proxy') cmd += self._bool_option('--check-certificate', 'nocheckcertificate', 'false', 'true', '=') cmd += self._bool_option('--remote-time', 'updatetime', 'true', 'false', '=') cmd += self._bool_option('--show-console-readout', 'noprogress', 'false', 'true', '=') cmd += self._configuration_args() if '__rpc' in info_dict: cmd += [ '--enable-rpc', f'--rpc-listen-port={info_dict["__rpc"]["port"]}', f'--rpc-secret={info_dict["__rpc"]["secret"]}'] # aria2c strips out spaces from the beginning/end of filenames and paths. # We work around this issue by adding a "./" to the beginning of the # filename and relative path, and adding a "/" at the end of the path. # See: https://github.com/yt-dlp/yt-dlp/issues/276 # https://github.com/ytdl-org/youtube-dl/issues/20312 # https://github.com/aria2/aria2/issues/1373 dn = os.path.dirname(tmpfilename) if dn: cmd += ['--dir', self._aria2c_filename(dn) + os.path.sep] if 'fragments' not in info_dict: cmd += ['--out', self._aria2c_filename(os.path.basename(tmpfilename))] cmd += ['--auto-file-renaming=false'] if 'fragments' in info_dict: cmd += ['--uri-selector=inorder'] url_list_file = f'{tmpfilename}.frag.urls' url_list = [] for frag_index, fragment in enumerate(info_dict['fragments']): fragment_filename = f'{os.path.basename(tmpfilename)}-Frag{frag_index}' url_list.append('{}\n\tout={}'.format(fragment['url'], self._aria2c_filename(fragment_filename))) stream, _ = self.sanitize_open(url_list_file, 'wb') stream.write('\n'.join(url_list).encode()) stream.close() cmd += ['-i', self._aria2c_filename(url_list_file)] else: cmd += ['--', info_dict['url']] return cmd def aria2c_rpc(self, rpc_port, rpc_secret, method, params=()): # Does not actually need to be UUID, just unique sanitycheck = str(uuid.uuid4()) d = json.dumps({ 'jsonrpc': '2.0', 'id': sanitycheck, 'method': method, 'params': [f'token:{rpc_secret}', *params], }).encode() request = Request( f'http://localhost:{rpc_port}/jsonrpc', data=d, headers={ 'Content-Type': 'application/json', 'Content-Length': f'{len(d)}', }, proxies={'all': None}) with self.ydl.urlopen(request) as r: resp = json.load(r) assert resp.get('id') == sanitycheck, 'Something went wrong with RPC server' return resp['result'] def _call_process(self, cmd, info_dict): if '__rpc' not in info_dict: return super()._call_process(cmd, info_dict) send_rpc = functools.partial(self.aria2c_rpc, info_dict['__rpc']['port'], info_dict['__rpc']['secret']) started = time.time() fragmented = 'fragments' in info_dict frag_count = len(info_dict['fragments']) if fragmented else 1 status = { 'filename': info_dict.get('_filename'), 'status': 'downloading', 'elapsed': 0, 'downloaded_bytes': 0, 'fragment_count': frag_count if fragmented else None, 'fragment_index': 0 if fragmented else None, } self._hook_progress(status, info_dict) def get_stat(key, *obj, average=False): val = tuple(filter(None, map(float, traverse_obj(obj, (..., ..., key))))) or [0] return sum(val) / (len(val) if average else 1) with Popen(cmd, text=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) as p: # Add a small sleep so that RPC client can receive response, # or the connection stalls infinitely time.sleep(0.2) retval = p.poll() while retval is None: # We don't use tellStatus as we won't know the GID without reading stdout # Ref: https://aria2.github.io/manual/en/html/aria2c.html#aria2.tellActive activ" (Indicator: " curl ") in Source: external.py Observed string: "from __future__ import annotationsimport ioimport itertoolsimport mathimport reimport urllib.parsefrom ._helper import InstanceStoreMixinfrom ..utils.networking import select_proxyfrom .common import ( Features, Request, Response, register_preference, register_rh,from .exceptions import ( CertificateVerifyError, HTTPError, IncompleteRead, ProxyError, SSLError, TransportError,from .impersonate import ImpersonateRequestHandler, ImpersonateTargetfrom ..dependencies import curl_cffi, certififrom ..utils import int_or_noneif curl_cffi is None: raise ImportError('curl_cffi is not installed')curl_cffi_version = tuple(map(int, re.split(r'[^\d]+', curl_cffi.__version__)[:3]))if curl_cffi_version != (0, 5, 10) and not (0, 10) <= curl_cffi_version < (0, 14): curl_cffi._yt_dlp__version = f'{curl_cffi.__version__} (unsupported)' raise ImportError('Only curl_cffi versions 0.5.10, 0.10.x, 0.11.x, 0.12.x, 0.13.x are supported')import curl_cffi.requestsfrom curl_cffi.const import CurlECode, CurlOptclass CurlCFFIResponseReader(io.IOBase): def __init__(self, response: curl_cffi.requests.Response): self._response = response self._iterator = response.iter_content() self._buffer = b'' self.bytes_read = 0 def readable(self): return True def read(self, size=None): exception_raised = True try: while self._iterator and (size is None or len(self._buffer) < size): chunk = next(self._iterator, None) if chunk is None: self._iterator = None break self._buffer += chunk self.bytes_read += len(chunk) if size is None: size = len(self._buffer) data = self._buffer[:size] self._buffer = self._buffer[size:] # "free" the curl instance if the response is fully read. # curl_cffi doesn't do this automatically and only allows one open response per thread if not self._iterator and not self._buffer: self.close() exception_raised = False return data finally: if exception_raised: self.close() def close(self): if not self.closed: self._response.close() self._buffer = b'' super().close()class CurlCFFIResponseAdapter(Response): fp: CurlCFFIResponseReader def __init__(self, response: curl_cffi.requests.Response): super().__init__( fp=CurlCFFIResponseReader(response), headers=response.headers, url=response.url, status=response.status_code) def read(self, amt=None): try: return self.fp.read(amt) except curl_cffi.requests.errors.RequestsError as e: if e.code == CurlECode.PARTIAL_FILE: content_length = e.response and int_or_none(e.response.headers.get('Content-Length')) raise IncompleteRead( partial=self.fp.bytes_read, expected=content_length - self.fp.bytes_read if content_length is not None else None, cause=e) from e raise TransportError(cause=e) from e# See: https://github.com/lexiforest/curl_cffi?tab=readme-ov-file#supported-impersonate-browsers# https://github.com/lexiforest/curl-impersonate?tab=readme-ov-file#supported-browsersBROWSER_TARGETS: dict[tuple[int, ...], dict[str, ImpersonateTarget]] = { (0, 5): { 'chrome99': ImpersonateTarget('chrome', '99', 'windows', '10'), 'chrome99_android': ImpersonateTarget('chrome', '99', 'android', '12'), 'chrome100': ImpersonateTarget('chrome', '100', 'windows', '10'), 'chrome101': ImpersonateTarget('chrome', '101', 'windows', '10'), 'chrome104': ImpersonateTarget('chrome', '104', 'windows', '10'), 'chrome107': ImpersonateTarget('chrome', '107', 'windows', '10'), 'chrome110': ImpersonateTarget('chrome', '110', 'windows', '10'), 'edge99': ImpersonateTarget('edge', '99', 'windows', '10'), 'edge101': ImpersonateTarget('edge', '101', 'windows', '10'), 'safari153': ImpersonateTarget('safari', '15.3', 'macos', '11'), 'safari155': ImpersonateTarget('safari', '15.5', 'macos', '12'), }, (0, 7): { 'chrome116': ImpersonateTarget('chrome', '116', 'windows', '10'), 'chrome119': ImpersonateTarget('chrome', '119', 'macos', '14'), 'chrome120': ImpersonateTarget('chrome', '120', 'macos', '14'), 'chrome123': ImpersonateTarget('chrome', '123', 'macos', '14'), 'chrome124': ImpersonateTarget('chrome', '124', 'macos', '14'), 'safari170': ImpersonateTarget('safari', '17.0', 'macos', '14'), 'safari172_ios': ImpersonateTarget('safari', '17.2', 'ios', '17.2'), }, (0, 9): { 'safari153': ImpersonateTarget('safari', '15.3', 'macos', '14'), 'safari155': ImpersonateTarget('safari', '15.5', 'macos', '14'), 'chrome119': ImpersonateTarget('chrome', '119', 'macos', '14'), 'chrome120': ImpersonateTarget('chrome', '120', 'macos', '14'), 'chrome123': ImpersonateTarget('chrome', '123', 'macos', '14'), 'chrome124': ImpersonateTarget('chrome', '124', 'macos', '14'), 'chrome131': ImpersonateTarget('chrome', '131', 'macos', '14'), 'chrome131_android': ImpersonateTarget('chrome', '131', 'android', '14'), 'chrome133a': ImpersonateTarget('chrome', '133', 'macos', '15'), 'firefox133': ImpersonateTarget('firefox', '133', 'macos', '14'), 'safari180': ImpersonateTarget('safari', '18.0', 'macos', '15'), 'safari180_ios': ImpersonateTarget('safari', '18.0', 'ios', '18.0'), }, (0, 10): { 'firefox135': ImpersonateTarget('firefox', '135', 'macos', '14'), }, (0, 11): { 'tor145': ImpersonateTarget('tor', '14.5', 'macos', '14'), 'safari184': ImpersonateTarget('safari', '18.4', 'macos', '15'), 'safari184_ios': ImpersonateTarget('safari', '18.4', 'ios', '18.4'), 'chrome136': ImpersonateTarget('chrome', '136', 'macos', '15'), }, (0, 12): { 'safari260': ImpersonateTarget('safari', '26.0', 'macos', '26'), 'safari260_ios': ImpersonateTarget('safari', '26.0', 'ios', '26.0'), },# Needed for curl_cffi < 0.11# See: https://github.com/lexiforest/curl_cffi/commit/d2f15c7a31506a08d217fcc04ae7570c39f5f5bb_TARGETS_COMPAT_LOOKUP = { 'safari153': 'safari15_3', 'safari155': 'safari15_5', 'safari170': 'safari17_0', 'safari172_ios': 'safari17_2_ios', 'safari180': 'safari18_0', 'safari180_ios': 'safari18_0_ios',@register_rhclass CurlCFFIRH(ImpersonateRequestHandler, InstanceStoreMixin): RH_NAME = 'curl_cffi' _SUPPORTED_URL_SCHEMES = ('http', 'https') _SUPPORTED_FEATURES = (Features.NO_PROXY, Features.ALL_PROXY) _SUPPORTED_PROXY_SCHEMES = ('http', 'https', 'socks4', 'socks4a', 'socks5', 'socks5h') _SUPPORTED_IMPERSONATE_TARGET_MAP = { target: ( name if curl_cffi_version >= (0, 11) else _TARGETS_COMPAT_LOOKUP.get(name, name) if curl_cffi_version >= (0, 9) else curl_cffi.requests.BrowserType[_TARGETS_COMPAT_LOOKUP.get(name, name)] ) for name, target in dict(sorted(itertools.chain.from_iterable( targets.items() for version, targets in BROWSER_TARGETS.items() if curl_cffi_version >= version ), key=lambda x: ( # deprioritize mobile targets since they give very different behavior x[1].os not in ('ios', 'android'), # prioritize tor < edge < firefox < safari < chrome ('tor', 'edge', 'firefox', 'safari', 'chrome').index(x[1].client), # prioritize newest version float(x[1].version) if x[1].version else 0, # group by os name x[1].os, ), reverse=True)).items() } def _create_instance(self, cookiejar=None): return curl_cffi.requests.Session(cookies=cookiejar) def _check_extensions(self, extensions): super()._check_extensions(extensions) extensions.pop('impersonate', None) extensions.pop('cookiejar', None) extensions.pop('timeout', None) # CurlCFFIRH ignores legacy ssl options currently. # Impersonation generally uses a looser SSL configuration than urllib/requests. extensions.pop('legacy_ssl', None) def send(self, request: Request) -> Response: target = self._get_request_target(request) try: response = super().send(request) except HTTPError as e: e.response.extensions['impersonate'] = target raise response.extensions['impersonate'] = target return response def _send(self, request: Request): max_redirects_exceeded = False session: curl_cffi.requests.Session = self._get_instance( cookiejar=self._get_cookiejar(request) if 'cookie' not in request.headers else None) if self.verbose: session.curl.setopt(CurlOpt.VERBOSE, 1) proxies = self._get_proxies(request) if 'no' in proxies: session.curl.setopt(CurlOpt.NOPROXY, proxies['no']) proxies.pop('no', None) # curl doesn't support per protocol proxies, so we select the one that matches the request protocol proxy = select_proxy(request.url, proxies=proxies) if proxy: session.curl.setopt(CurlOpt.PROXY, proxy) scheme = urllib.parse.urlparse(request.url).scheme.lower() if scheme != 'http': # Enable HTTP CONNECT for HTTPS urls. # Don't use CONNECT for http for compatibility with urllib behaviour. # See: https://curl.se/libcurl/c/CURLOPT_HTTPPROXYTUNNEL.html session.curl.setopt(CurlOpt.HTTPPROXYTUNNEL, 1) # curl_cffi does not currently set these for proxies session.curl.setopt(CurlOpt.PROXY_CAINFO, certifi.where()) if not self.verify: session.curl.setopt(CurlOpt.PROXY_SSL_VERIFYPEER, 0) session.curl.setopt(CurlOpt.PROXY_SSL_VERIFYHOST, 0) headers = self._get_impersonate_headers(request) if self._client_cert: session.curl.setopt(CurlOpt.SSLCERT, self._client_cert['client_certificate']) client_certificate_key = self._client_cert.get('client_certificate_key') client_certificate_password = self._client_cert.get('client_certificate_password') if client_certificate_key: session.curl.setopt(CurlOpt.SSLKEY, client_certificate_key) if client_certificate_password: session.curl.setopt(CurlOpt.KEYPASSWD, client_certificate_password) timeout = self._calculate_timeout(request) # set CURLOPT_LOW_SPEED_LIMIT and CURLOPT_LOW_SPEED_TIME to act as a read timeout. [1] # This is required only for 0.5.10 [2] # Note: CURLOPT_LOW_SPEED_TIME is in seconds, so we need to round up to the nearest second. [3] # [1] https://unix.stackexchange.com/a/305311 # [2] https://github.com/yifeikong/curl_cffi/issues/156 # [3] https://curl.se/libcurl/c/CURLOPT_LOW_SPEED_TIME.html session.curl.setopt(CurlOpt.LOW_SPEED_LIMIT, 1) # 1 byte per second session.curl.setopt(CurlOpt.LOW_SPEED_TIME, math.ceil(timeout)) try: curl_response = session.request( method=request.method, url=request.url, headers=headers, data=request.data, verify=self.verify, max_redirects=5, timeout=(timeout, timeout), impersonate=self._SUPPORTED_IMPERSONATE_TARGET_MAP.get( self._get_request_target(request)), interface=self.source_address, stream=True, ) except curl_cffi.requests.errors.RequestsError as e: if e.code == CurlECode.PEER_FAILED_VERIFICATION: raise CertificateVerifyError(cause=e) from e elif e.code == CurlECode.SSL_CONNECT_ERROR: raise SSLError(cause=e) from e elif e.code == CurlECode.TOO_MANY_REDIRECTS: max_redirects_exceeded = True curl_response = e.response elif ( e.code == CurlECode.PROXY or (e.code == CurlECode.RECV_ERROR and 'CONNECT' in str(e)) ): raise ProxyError(cause=e) from e else: raise TransportError(cause=e) from e response = CurlCFFIResponseAdapter(curl_response) if not 200 <= response.status < 300: raise HTTPError(response, redirect_loop=max_redirects_exceeded) return response@register_preference(CurlCFFIRH)def curl_cffi_preference(rh, request): return -100" (Indicator: " curl ") in Source: _curlcffi.py Observed string: "# flake8: noqa: F401import warningsfrom .common import ( HEADRequest, PATCHRequest, PUTRequest, Request, RequestDirector, RequestHandler, Response,# isort: split# TODO: all request handlers should be safely importedfrom . import _urllibfrom ..utils import bug_reports_messagetry: from . import _requestsexcept ImportError: passexcept Exception as e: warnings.warn(f'Failed to import "requests" request handler: {e}' + bug_reports_message())try: from . import _websocketsexcept ImportError: passexcept Exception as e: warnings.warn(f'Failed to import "websockets" request handler: {e}' + bug_reports_message())try: from . import _curlcffiexcept ImportError: passexcept Exception as e: warnings.warn(f'Failed to import "curl_cffi" request handler: {e}' + bug_reports_message())" (Indicator: ""curl") in Source: __init__.py