return os.path.expandvars(compat_expanduser(s))
-def orderedSet(iterable):
- """ Remove all duplicates from the input iterable """
- res = []
- for el in iterable:
- if el not in res:
- res.append(el)
- return res
+def orderedSet(iterable, *, lazy=False):
+ """Remove all duplicates from the input iterable"""
+ def _iter():
+ seen = [] # Do not use set since the items can be unhashable
+ for x in iterable:
+ if x not in seen:
+ seen.append(x)
+ yield x
+
+ return _iter() if lazy else list(_iter())
def _htmlentity_transform(entity_with_semicolon):
self.exc_info = sys.exc_info() # preserve original exception
super().__init__(''.join((
- format_field(ie, template='[%s] '),
- format_field(video_id, template='%s: '),
+ format_field(ie, None, '[%s] '),
+ format_field(video_id, None, '%s: '),
msg,
- format_field(cause, template=' (caused by %r)'),
+ format_field(cause, None, ' (caused by %r)'),
'' if expected else bug_reports_message())))
def format_traceback(self):
is_exhausted = False
def __init__(self, ydl, info_dict):
- self.ydl, self.info_dict = ydl, info_dict
+ self.ydl = ydl
+
+ # _entries must be assigned now since infodict can change during iteration
+ entries = info_dict.get('entries')
+ if entries is None:
+ raise EntryNotInPlaylist('There are no entries')
+ elif isinstance(entries, list):
+ self.is_exhausted = True
+
+ requested_entries = info_dict.get('requested_entries')
+ self.is_incomplete = bool(requested_entries)
+ if self.is_incomplete:
+ assert self.is_exhausted
+ self._entries = [self.MissingEntry] * max(requested_entries)
+ for i, entry in zip(requested_entries, entries):
+ self._entries[i - 1] = entry
+ elif isinstance(entries, (list, PagedList, LazyList)):
+ self._entries = entries
+ else:
+ self._entries = LazyList(entries)
PLAYLIST_ITEMS_RE = re.compile(r'''(?x)
(?P<start>[+-]?\d+)?
except (ExistingVideoReached, RejectedVideoReached):
return
- @property
- def full_count(self):
- if self.info_dict.get('playlist_count'):
- return self.info_dict['playlist_count']
- elif self.is_exhausted and not self.is_incomplete:
+ def get_full_count(self):
+ if self.is_exhausted and not self.is_incomplete:
return len(self)
elif isinstance(self._entries, InAdvancePagedList):
if self._entries._pagesize == 1:
return self._entries._pagecount
- @functools.cached_property
- def _entries(self):
- entries = self.info_dict.get('entries')
- if entries is None:
- raise EntryNotInPlaylist('There are no entries')
- elif isinstance(entries, list):
- self.is_exhausted = True
-
- indices = self.info_dict.get('requested_entries')
- self.is_incomplete = bool(indices)
- if self.is_incomplete:
- assert self.is_exhausted
- ret = [self.MissingEntry] * max(indices)
- for i, entry in zip(indices, entries):
- ret[i - 1] = entry
- return ret
-
- if isinstance(entries, (list, PagedList, LazyList)):
- return entries
- return LazyList(entries)
-
@functools.cached_property
def _getter(self):
if isinstance(self._entries, list):
if i < 0:
continue
try:
- try:
- entry = self._getter(i)
- except self.IndexError:
- self.is_exhausted = True
- if step > 0:
- break
- continue
- except IndexError:
- if self.is_exhausted:
+ entry = self._getter(i)
+ except self.IndexError:
+ self.is_exhausted = True
+ if step > 0:
break
- raise
+ continue
yield i + 1, entry
def __len__(self):
return '"%s"' % v
+ def create_map(mobj):
+ return json.dumps(dict(json.loads(js_to_json(mobj.group(1) or '[]', vars=vars))))
+
code = re.sub(r'new Date\((".+")\)', r'\g<1>', code)
+ code = re.sub(r'new Map\((\[.*?\])?\)', create_map, code)
return re.sub(r'''(?sx)
"(?:[^"\\]*(?:\\\\|\\['"nurtbfx/\n]))*[^"\\]*"|
return q
-POSTPROCESS_WHEN = ('pre_process', 'after_filter', 'before_dl', 'after_move', 'post_process', 'after_video', 'playlist')
+POSTPROCESS_WHEN = ('pre_process', 'after_filter', 'before_dl', 'post_process', 'after_move', 'after_video', 'playlist')
DEFAULT_OUTTMPL = {
try:
_, stderr, returncode = Popen.run(
[exe, '-w', key, value, path] if exe == 'xattr' else [exe, '-n', key, '-v', value, path],
- stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE)
+ text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE)
except OSError as e:
raise XAttrMetadataError(e.errno, e.strerror)
if returncode:
# FIXME: https://github.com/ytdl-org/youtube-dl/commit/dfe5fa49aed02cf36ba9f743b11b0903554b5e56
contents = optionf.read()
res = shlex.split(contents, comments=True)
+ except Exception as err:
+ raise ValueError(f'Unable to parse "{filename}": {err}')
finally:
optionf.close()
return res