return '_'
return char
+ if s == '':
+ return ''
# Handle timestamps
s = re.sub(r'[0-9]+(?::[0-9]+)+', lambda m: m.group(0).replace(':', '_'), s)
result = ''.join(map(replace_insane, s))
r'&([^&;]+;)', lambda m: _htmlentity_transform(m.group(1)), s)
+def escapeHTML(text):
+ return (
+ text
+ .replace('&', '&')
+ .replace('<', '<')
+ .replace('>', '>')
+ .replace('"', '"')
+ .replace("'", ''')
+ )
+
+
def process_communicate_or_kill(p, *args, **kwargs):
try:
return p.communicate(*args, **kwargs)
return optval
-def formatSeconds(secs, delim=':'):
+def formatSeconds(secs, delim=':', msec=False):
if secs > 3600:
- return '%d%s%02d%s%02d' % (secs // 3600, delim, (secs % 3600) // 60, delim, secs % 60)
+ ret = '%d%s%02d%s%02d' % (secs // 3600, delim, (secs % 3600) // 60, delim, secs % 60)
elif secs > 60:
- return '%d%s%02d' % (secs // 60, delim, secs % 60)
+ ret = '%d%s%02d' % (secs // 60, delim, secs % 60)
else:
- return '%d' % secs
+ ret = '%d' % secs
+ return '%s.%03d' % (ret, secs % 1) if msec else ret
def make_HTTPS_handler(params, **kwargs):
pass
+class ThrottledDownload(YoutubeDLError):
+ """ Download speed below --throttled-rate. """
+ pass
+
+
class MaxDownloadsReached(YoutubeDLError):
""" --max-downloads limit has been reached. """
pass
def __init__(self, iterable):
self.__iterable = iter(iterable)
self.__cache = []
+ self.__reversed = False
def __iter__(self):
- for item in self.__cache:
- yield item
+ if self.__reversed:
+ # We need to consume the entire iterable to iterate in reverse
+ yield from self.exhaust()
+ return
+ yield from self.__cache
for item in self.__iterable:
self.__cache.append(item)
yield item
+ def __exhaust(self):
+ self.__cache.extend(self.__iterable)
+ return self.__cache
+
def exhaust(self):
''' Evaluate the entire iterable '''
- self.__cache.extend(self.__iterable)
+ return self.__exhaust()[::-1 if self.__reversed else 1]
+
+ @staticmethod
+ def __reverse_index(x):
+ return -(x + 1)
def __getitem__(self, idx):
if isinstance(idx, slice):
step = idx.step or 1
- start = idx.start if idx.start is not None else 1 if step > 0 else -1
+ start = idx.start if idx.start is not None else 0 if step > 0 else -1
stop = idx.stop if idx.stop is not None else -1 if step > 0 else 0
+ if self.__reversed:
+ (start, stop), step = map(self.__reverse_index, (start, stop)), -step
+ idx = slice(start, stop, step)
elif isinstance(idx, int):
+ if self.__reversed:
+ idx = self.__reverse_index(idx)
start = stop = idx
else:
raise TypeError('indices must be integers or slices')
if start < 0 or stop < 0:
# We need to consume the entire iterable to be able to slice from the end
# Obviously, never use this with infinite iterables
- self.exhaust()
- else:
- n = max(start, stop) - len(self.__cache) + 1
- if n > 0:
- self.__cache.extend(itertools.islice(self.__iterable, n))
+ return self.__exhaust()[idx]
+
+ n = max(start, stop) - len(self.__cache) + 1
+ if n > 0:
+ self.__cache.extend(itertools.islice(self.__iterable, n))
return self.__cache[idx]
def __bool__(self):
try:
- self[0]
+ self[-1] if self.__reversed else self[0]
except IndexError:
return False
return True
self.exhaust()
return len(self.__cache)
+ def reverse(self):
+ self.__reversed = not self.__reversed
+ return self
+
+ def __repr__(self):
+ # repr and str should mimic a list. So we exhaust the iterable
+ return repr(self.exhaust())
+
+ def __str__(self):
+ return repr(self.exhaust())
+
class PagedList(object):
def __len__(self):
def try_get(src, getter, expected_type=None):
- if not isinstance(getter, (list, tuple)):
- getter = [getter]
- for get in getter:
+ for get in variadic(getter):
try:
v = get(src)
except (AttributeError, KeyError, TypeError, IndexError):
# As of [1] format syntax is:
# %[mapping_key][conversion_flags][minimum_width][.precision][length_modifier]type
# 1. https://docs.python.org/2/library/stdtypes.html#string-formatting
-FORMAT_RE = r'''(?x)
+STR_FORMAT_RE = r'''(?x)
(?<!%)
%
- \({0}\) # mapping key
- (?:[#0\-+ ]+)? # conversion flags (optional)
- (?:\d+)? # minimum field width (optional)
- (?:\.\d+)? # precision (optional)
- [hlL]? # length modifier (optional)
- (?P<type>[diouxXeEfFgGcrs%]) # conversion type
+ (?P<has_key>\((?P<key>{0})\))? # mapping key
+ (?P<format>
+ (?:[#0\-+ ]+)? # conversion flags (optional)
+ (?:\d+)? # minimum field width (optional)
+ (?:\.\d+)? # precision (optional)
+ [hlL]? # length modifier (optional)
+ [diouxXeEfFgGcrs] # conversion type
+ )
'''
assert isinstance(keys, (list, tuple))
for key_list in keys:
- if isinstance(key_list, compat_str):
- key_list = (key_list,)
arg_list = list(filter(
lambda x: x is not None,
- [argdict.get(key.lower()) for key in key_list]))
+ [argdict.get(key.lower()) for key in variadic(key_list)]))
if arg_list:
return [arg for args in arg_list for arg in args]
return default
return classes
-def traverse_dict(dictn, keys, casesense=True):
- keys = list(keys)[::-1]
- while keys:
- key = keys.pop()
- if isinstance(dictn, dict):
- if not casesense:
- dictn = {k.lower(): v for k, v in dictn.items()}
- key = key.lower()
- dictn = dictn.get(key)
- elif isinstance(dictn, (list, tuple, compat_str)):
- if ':' in key:
- key = slice(*map(int_or_none, key.split(':')))
+def traverse_obj(
+ obj, *key_list, default=None, expected_type=None,
+ casesense=True, is_user_input=False, traverse_string=False):
+ ''' Traverse nested list/dict/tuple
+ @param default Default value to return
+ @param expected_type Only accept final value of this type
+ @param casesense Whether to consider dictionary keys as case sensitive
+ @param is_user_input Whether the keys are generated from user input. If True,
+ strings are converted to int/slice if necessary
+ @param traverse_string Whether to traverse inside strings. If True, any
+ non-compatible object will also be converted into a string
+ '''
+ if not casesense:
+ _lower = lambda k: k.lower() if isinstance(k, str) else k
+ key_list = ((_lower(k) for k in keys) for keys in key_list)
+
+ def _traverse_obj(obj, keys):
+ for key in list(keys):
+ if isinstance(obj, dict):
+ obj = (obj.get(key) if casesense or (key in obj)
+ else next((v for k, v in obj.items() if _lower(k) == key), None))
else:
- key = int_or_none(key)
- dictn = try_get(dictn, lambda x: x[key])
- else:
- return None
- return dictn
+ if is_user_input:
+ key = (int_or_none(key) if ':' not in key
+ else slice(*map(int_or_none, key.split(':'))))
+ if not isinstance(key, (int, slice)):
+ return None
+ if not isinstance(obj, (list, tuple)):
+ if not traverse_string:
+ return None
+ obj = str(obj)
+ try:
+ obj = obj[key]
+ except IndexError:
+ return None
+ return obj
+
+ for keys in key_list:
+ val = _traverse_obj(obj, keys)
+ if val is not None:
+ if expected_type is None or isinstance(val, expected_type):
+ return val
+ return default
+
+
+def traverse_dict(dictn, keys, casesense=True):
+ ''' For backward compatibility. Do not use '''
+ return traverse_obj(dictn, keys, casesense=casesense,
+ is_user_input=True, traverse_string=True)
+
+
+def variadic(x, allowed_types=str):
+ return x if isinstance(x, collections.Iterable) and not isinstance(x, allowed_types) else (x,)
projects / yt-dlp.git / blobdiff