[utils] Add `try_call`

pull/3266/head
pukkandan 2 years ago
parent f189faf1ce
commit c4f60dd7cd
No known key found for this signature in database
GPG Key ID: 7EEE9E1E817D0A39

@ -643,7 +643,7 @@ Wrap all extracted numeric data into safe functions from [`yt_dlp/utils.py`](yt_
Use `url_or_none` for safe URL processing. Use `url_or_none` for safe URL processing.
Use `try_get`, `dict_get` and `traverse_obj` for safe metadata extraction from parsed JSON. Use `traverse_obj` and `try_call` (superseeds `dict_get` and `try_get`) for safe metadata extraction from parsed JSON.
Use `unified_strdate` for uniform `upload_date` or any `YYYYMMDD` meta field extraction, `unified_timestamp` for uniform `timestamp` extraction, `parse_filesize` for `filesize` extraction, `parse_count` for count meta fields extraction, `parse_resolution`, `parse_duration` for `duration` extraction, `parse_age_limit` for `age_limit` extraction. Use `unified_strdate` for uniform `upload_date` or any `YYYYMMDD` meta field extraction, `unified_timestamp` for uniform `timestamp` extraction, `parse_filesize` for `filesize` extraction, `parse_count` for count meta fields extraction, `parse_resolution`, `parse_duration` for `duration` extraction, `parse_age_limit` for `age_limit` extraction.

@ -18,7 +18,7 @@ from ..utils import (
parse_http_range, parse_http_range,
sanitized_Request, sanitized_Request,
ThrottledDownload, ThrottledDownload,
try_get, try_call,
write_xattr, write_xattr,
XAttrMetadataError, XAttrMetadataError,
XAttrUnavailableError, XAttrUnavailableError,
@ -120,12 +120,12 @@ class HttpFD(FileDownloader):
else: else:
range_end = None range_end = None
if try_get(None, lambda _: range_start > range_end): if try_call(lambda: range_start > range_end):
ctx.resume_len = 0 ctx.resume_len = 0
ctx.open_mode = 'wb' ctx.open_mode = 'wb'
raise RetryDownload(Exception(f'Conflicting range. (start={range_start} > end={range_end})')) raise RetryDownload(Exception(f'Conflicting range. (start={range_start} > end={range_end})'))
if try_get(None, lambda _: range_end >= ctx.content_len): if try_call(lambda: range_end >= ctx.content_len):
range_end = ctx.content_len - 1 range_end = ctx.content_len - 1
request = sanitized_Request(url, request_data, headers) request = sanitized_Request(url, request_data, headers)

@ -14,6 +14,7 @@ from ..utils import (
float_or_none, float_or_none,
mimetype2ext, mimetype2ext,
str_or_none, str_or_none,
try_call,
try_get, try_get,
unescapeHTML, unescapeHTML,
unsmuggle_url, unsmuggle_url,
@ -145,11 +146,11 @@ class MediasiteIE(InfoExtractor):
'duration': slide['Time'] / 1000, 'duration': slide['Time'] / 1000,
}) })
next_time = try_get(None, [ next_time = try_call(
lambda _: Stream['Slides'][i + 1]['Time'], lambda: Stream['Slides'][i + 1]['Time'],
lambda _: duration, lambda: duration,
lambda _: slide['Time'], lambda: slide['Time'],
], expected_type=(int, float)) expected_type=(int, float))
fragments.append({ fragments.append({
'path': fname_template.format(slide.get('Number', i + 1)), 'path': fname_template.format(slide.get('Number', i + 1)),

@ -5,6 +5,7 @@ from .common import InfoExtractor
from ..utils import ( from ..utils import (
int_or_none, int_or_none,
qualities, qualities,
try_call,
try_get, try_get,
ExtractorError, ExtractorError,
) )
@ -26,10 +27,10 @@ class WhoWatchIE(InfoExtractor):
metadata = self._download_json('https://api.whowatch.tv/lives/%s' % video_id, video_id) metadata = self._download_json('https://api.whowatch.tv/lives/%s' % video_id, video_id)
live_data = self._download_json('https://api.whowatch.tv/lives/%s/play' % video_id, video_id) live_data = self._download_json('https://api.whowatch.tv/lives/%s/play' % video_id, video_id)
title = try_get(None, ( title = try_call(
lambda x: live_data['share_info']['live_title'][1:-1], lambda: live_data['share_info']['live_title'][1:-1],
lambda x: metadata['live']['title'], lambda: metadata['live']['title'],
), compat_str) expected_type=str)
hls_url = live_data.get('hls_url') hls_url = live_data.get('hls_url')
if not hls_url: if not hls_url:

@ -3096,15 +3096,19 @@ def dict_get(d, key_or_keys, default=None, skip_false_values=True):
return d.get(key_or_keys, default) return d.get(key_or_keys, default)
def try_get(src, getter, expected_type=None): def try_call(*funcs, expected_type=None, args=[], kwargs={}):
for get in variadic(getter): for f in funcs:
try: try:
v = get(src) val = f(*args, **kwargs)
except (AttributeError, KeyError, TypeError, IndexError): except (AttributeError, KeyError, TypeError, IndexError, ZeroDivisionError):
pass pass
else: else:
if expected_type is None or isinstance(v, expected_type): if expected_type is None or isinstance(val, expected_type):
return v return val
def try_get(src, getter, expected_type=None):
return try_call(*variadic(getter), args=(src,), expected_type=expected_type)
def filter_dict(dct, cndn=lambda _, v: v is not None): def filter_dict(dct, cndn=lambda _, v: v is not None):

Loading…
Cancel
Save