-import asyncio
-import atexit
import base64
import binascii
import calendar
compat_os_name,
compat_shlex_quote,
)
-from ..dependencies import websockets, xattr
+from ..dependencies import xattr
__name__ = __name__.rsplit('.', 1)[0] # Pretend to be the parent module
s = self._close_object(e)
if s is not None:
continue
- raise type(e)(f'{e.msg} in {s[e.pos-10:e.pos+10]!r}', s, e.pos)
+ raise type(e)(f'{e.msg} in {s[e.pos - 10:e.pos + 10]!r}', s, e.pos)
assert False, 'Too many attempts to decode JSON'
elif char in '\\/|*<>':
return '\0_'
if restricted and (char in '!&\'()[]{}$;`^,#' or char.isspace() or ord(char) > 127):
- return '\0_'
+ return '' if unicodedata.category(char)[0] in 'CM' else '\0_'
return char
# Replace look-alike Unicode glyphs
def sanitize_path(s, force=False):
"""Sanitizes and normalizes path on Windows"""
+ # XXX: this handles drive relative paths (c:sth) incorrectly
if sys.platform == 'win32':
force = False
drive_or_unc, _ = os.path.splitdrive(s)
sanitized_path.insert(0, drive_or_unc + os.path.sep)
elif force and s and s[0] == os.path.sep:
sanitized_path.insert(0, os.path.sep)
- return os.path.join(*sanitized_path)
+ # TODO: Fix behavioral differences <3.12
+ # The workaround using `normpath` only superficially passes tests
+ # Ref: https://github.com/python/cpython/pull/100351
+ return os.path.normpath(os.path.join(*sanitized_path))
def sanitize_url(url, *, scheme='http'):
_fix('LD_LIBRARY_PATH') # Linux
_fix('DYLD_LIBRARY_PATH') # macOS
- def __init__(self, *args, env=None, text=False, **kwargs):
+ def __init__(self, args, *remaining, env=None, text=False, shell=False, **kwargs):
if env is None:
env = os.environ.copy()
self._fix_pyinstaller_ld_path(env)
kwargs['universal_newlines'] = True # For 3.6 compatibility
kwargs.setdefault('encoding', 'utf-8')
kwargs.setdefault('errors', 'replace')
- super().__init__(*args, env=env, **kwargs, startupinfo=self._startupinfo)
+
+ if shell and compat_os_name == 'nt' and kwargs.get('executable') is None:
+ if not isinstance(args, str):
+ args = ' '.join(compat_shlex_quote(a) for a in args)
+ shell = False
+ args = f'{self.__comspec()} /Q /S /D /V:OFF /C "{args}"'
+
+ super().__init__(args, *remaining, env=env, shell=shell, **kwargs, startupinfo=self._startupinfo)
+
+ def __comspec(self):
+ comspec = os.environ.get('ComSpec') or os.path.join(
+ os.environ.get('SystemRoot', ''), 'System32', 'cmd.exe')
+ if os.path.isabs(comspec):
+ return comspec
+ raise FileNotFoundError('shell not found: neither %ComSpec% nor %SystemRoot% is set')
def communicate_or_kill(self, *args, **kwargs):
try:
if precision == 'auto':
auto_precision = True
precision = 'microsecond'
- today = datetime_round(datetime.datetime.utcnow(), precision)
+ today = datetime_round(datetime.datetime.now(datetime.timezone.utc), precision)
if date_str in ('now', 'today'):
return today
if date_str == 'yesterday':
'second': 1,
}
roundto = lambda x, n: ((x + n / 2) // n) * n
- timestamp = calendar.timegm(dt.timetuple())
- return datetime.datetime.utcfromtimestamp(roundto(timestamp, unit_seconds[precision]))
+ timestamp = roundto(calendar.timegm(dt.timetuple()), unit_seconds[precision])
+ return datetime.datetime.fromtimestamp(timestamp, datetime.timezone.utc)
def hyphenate_date(date_str):
def __repr__(self):
return f'{__name__}.{type(self).__name__}({self.start.isoformat()!r}, {self.end.isoformat()!r})'
+ def __str__(self):
+ return f'{self.start} to {self.end}'
+
def __eq__(self, other):
return (isinstance(other, DateRange)
and self.start == other.start and self.end == other.end)
s = re.sub(r'([\r\n]+)', r' \1', s)
enc, buffer = None, out
- if 'b' in getattr(out, 'mode', ''):
+ # `mode` might be `None` (Ref: https://github.com/yt-dlp/yt-dlp/issues/8816)
+ if 'b' in (getattr(out, 'mode', None) or ''):
enc = encoding or preferredencoding()
elif hasattr(out, 'buffer'):
buffer = out.buffer
buf = ctypes.create_string_buffer(len(title_bytes))
buf.value = title_bytes
try:
+ # PR_SET_NAME = 15 Ref: /usr/include/linux/prctl.h
libc.prctl(15, buf, 0, 0, 0)
except AttributeError:
return # Strange libc, just skip this
)?
T)?
(?:
- (?P<hours>[0-9]+)\s*h(?:ours?)?,?\s*
+ (?P<hours>[0-9]+)\s*h(?:(?:ou)?rs?)?,?\s*
)?
(?:
(?P<mins>[0-9]+)\s*m(?:in(?:ute)?s?)?,?\s*
raise self.IndexError()
return entries[0]
+ def __bool__(self):
+ return bool(self.getslice(0, 1))
+
class OnDemandPagedList(PagedList):
"""Download pages until a page with less than maximum results"""
s)
-def escape_rfc3986(s):
- """Escape non-ASCII characters as suggested by RFC 3986"""
- return urllib.parse.quote(s, b"%/;:@&=+$,!~*'()?#[]")
-
-
-def escape_url(url):
- """Escape URL as suggested by RFC 3986"""
- url_parsed = urllib.parse.urlparse(url)
- return url_parsed._replace(
- netloc=url_parsed.netloc.encode('idna').decode('ascii'),
- path=escape_rfc3986(url_parsed.path),
- params=escape_rfc3986(url_parsed.params),
- query=escape_rfc3986(url_parsed.query),
- fragment=escape_rfc3986(url_parsed.fragment)
- ).geturl()
-
-
def parse_qs(url, **kwargs):
return urllib.parse.parse_qs(urllib.parse.urlparse(url).query, **kwargs)
def create_map(mobj):
return json.dumps(dict(json.loads(js_to_json(mobj.group(1) or '[]', vars=vars))))
+ code = re.sub(r'(?:new\s+)?Array\((.*?)\)', r'[\g<1>]', code)
code = re.sub(r'new Map\((\[.*?\])?\)', create_map, code)
if not strict:
- code = re.sub(r'new Date\((".+")\)', r'\g<1>', code)
+ code = re.sub(rf'new Date\(({STRING_RE})\)', r'\g<1>', code)
code = re.sub(r'new \w+\((.*?)\)', lambda m: json.dumps(m.group(0)), code)
code = re.sub(r'parseInt\([^\d]+(\d+)[^\d]+\)', r'\1', code)
code = re.sub(r'\(function\([^)]*\)\s*\{[^}]*\}\s*\)\s*\(\s*(["\'][^)]*["\'])\s*\)', r'\1', code)
'quicktime': 'mov',
'webm': 'webm',
'vp9': 'vp9',
+ 'video/ogg': 'ogv',
'x-flv': 'flv',
'x-m4v': 'm4v',
'x-matroska': 'mkv',
def match_filter_func(filters, breaking_filters=None):
if not filters and not breaking_filters:
return None
+ repr_ = f'{match_filter_func.__module__}.{match_filter_func.__qualname__}({filters}, {breaking_filters})'
+
breaking_filters = match_filter_func(breaking_filters) or (lambda _, __: None)
filters = set(variadic(filters or []))
if interactive:
filters.remove('-')
+ @function_with_repr.set_repr(repr_)
def _match_func(info_dict, incomplete=False):
ret = breaking_filters(info_dict, incomplete)
if ret is not None:
raise XAttrMetadataError(e.errno, e.strerror)
return
- # UNIX Method 1. Use xattrs/pyxattrs modules
+ # UNIX Method 1. Use os.setxattr/xattrs/pyxattrs modules
setxattr = None
- if getattr(xattr, '_yt_dlp__identifier', None) == 'pyxattr':
+ if callable(getattr(os, 'setxattr', None)):
+ setxattr = os.setxattr
+ elif getattr(xattr, '_yt_dlp__identifier', None) == 'pyxattr':
# Unicode arguments are not supported in pyxattr until version 0.5.0
# See https://github.com/ytdl-org/youtube-dl/issues/5498
if version_tuple(xattr.__version__) >= (0, 5, 0):
else 'xattr' if check_executable('xattr', ['-h']) else None)
if not exe:
raise XAttrUnavailableError(
- 'Couldn\'t find a tool to set the xattrs. Install either the python "xattr" or "pyxattr" modules or the '
+ 'Couldn\'t find a tool to set the xattrs. Install either the "xattr" or "pyxattr" Python modules or the '
+ ('"xattr" binary' if sys.platform != 'linux' else 'GNU "attr" package (which contains the "setfattr" tool)'))
value = value.decode()
def read_stdin(what):
- eof = 'Ctrl+Z' if compat_os_name == 'nt' else 'Ctrl+D'
- write_string(f'Reading {what} from STDIN - EOF ({eof}) to end:\n')
+ if what:
+ eof = 'Ctrl+Z' if compat_os_name == 'nt' else 'Ctrl+D'
+ write_string(f'Reading {what} from STDIN - EOF ({eof}) to end:\n')
return sys.stdin
return self.parser.parse_args(self.all_args)
-class WebSocketsWrapper:
- """Wraps websockets module to use in non-async scopes"""
- pool = None
-
- def __init__(self, url, headers=None, connect=True):
- self.loop = asyncio.new_event_loop()
- # XXX: "loop" is deprecated
- self.conn = websockets.connect(
- url, extra_headers=headers, ping_interval=None,
- close_timeout=float('inf'), loop=self.loop, ping_timeout=float('inf'))
- if connect:
- self.__enter__()
- atexit.register(self.__exit__, None, None, None)
-
- def __enter__(self):
- if not self.pool:
- self.pool = self.run_with_loop(self.conn.__aenter__(), self.loop)
- return self
-
- def send(self, *args):
- self.run_with_loop(self.pool.send(*args), self.loop)
-
- def recv(self, *args):
- return self.run_with_loop(self.pool.recv(*args), self.loop)
-
- def __exit__(self, type, value, traceback):
- try:
- return self.run_with_loop(self.conn.__aexit__(type, value, traceback), self.loop)
- finally:
- self.loop.close()
- self._cancel_all_tasks(self.loop)
-
- # taken from https://github.com/python/cpython/blob/3.9/Lib/asyncio/runners.py with modifications
- # for contributors: If there's any new library using asyncio needs to be run in non-async, move these function out of this class
- @staticmethod
- def run_with_loop(main, loop):
- if not asyncio.iscoroutine(main):
- raise ValueError(f'a coroutine was expected, got {main!r}')
-
- try:
- return loop.run_until_complete(main)
- finally:
- loop.run_until_complete(loop.shutdown_asyncgens())
- if hasattr(loop, 'shutdown_default_executor'):
- loop.run_until_complete(loop.shutdown_default_executor())
-
- @staticmethod
- def _cancel_all_tasks(loop):
- to_cancel = asyncio.all_tasks(loop)
-
- if not to_cancel:
- return
-
- for task in to_cancel:
- task.cancel()
-
- # XXX: "loop" is removed in python 3.10+
- loop.run_until_complete(
- asyncio.gather(*to_cancel, loop=loop, return_exceptions=True))
-
- for task in to_cancel:
- if task.cancelled():
- continue
- if task.exception() is not None:
- loop.call_exception_handler({
- 'message': 'unhandled exception during asyncio.run() shutdown',
- 'exception': task.exception(),
- 'task': task,
- })
-
-
def merge_headers(*dicts):
"""Merge dicts of http headers case insensitively, prioritizing the latter ones"""
return {k.title(): v for k, v in itertools.chain.from_iterable(map(dict.items, dicts))}
def __call__(self, *args, **kwargs):
return self.func(*args, **kwargs)
+ @classmethod
+ def set_repr(cls, repr_):
+ return functools.partial(cls, repr_=repr_)
+
def __repr__(self):
if self.__repr:
return self.__repr
assert left > 3 and right >= 0
if s is None or len(s) <= left + right:
return s
- return f'{s[:left-3]}...{s[-right:] if right else ""}'
+ return f'{s[:left - 3]}...{s[-right:] if right else ""}'
def orderedSet_from_options(options, alias_dict, *, use_regex=False, start=None):
return tuple(self._calculate_field_preference(format, field) for field in self._order)
+def filesize_from_tbr(tbr, duration):
+ """
+ @param tbr: Total bitrate in kbps (1000 bits/sec)
+ @param duration: Duration in seconds
+ @returns Filesize in bytes
+ """
+ if tbr is None or duration is None:
+ return None
+ return int(duration * tbr * (1000 / 8))
+
+
# XXX: Temporary
class _YDLLogger:
def __init__(self, ydl=None):