compat_subprocess_get_DEVNULL = lambda: open(os.path.devnull, 'w')
try:
- from urllib.parse import parse_qs as compat_parse_qs
-except ImportError: # Python 2
- # HACK: The following is the correct parse_qs implementation from cpython 3's stdlib.
- # Python 2's version is apparently totally broken
- def _unquote(string, encoding='utf-8', errors='replace'):
+ from urllib.parse import unquote as compat_urllib_parse_unquote
+except ImportError:
+ def compat_urllib_parse_unquote(string, encoding='utf-8', errors='replace'):
if string == '':
return string
res = string.split('%')
string += pct_sequence.decode(encoding, errors)
return string
+
+try:
+ from urllib.parse import parse_qs as compat_parse_qs
+except ImportError: # Python 2
+ # HACK: The following is the correct parse_qs implementation from cpython 3's stdlib.
+ # Python 2's version is apparently totally broken
+
def _parse_qsl(qs, keep_blank_values=False, strict_parsing=False,
encoding='utf-8', errors='replace'):
qs, _coerce_result = qs, unicode
continue
if len(nv[1]) or keep_blank_values:
name = nv[0].replace('+', ' ')
- name = _unquote(name, encoding=encoding, errors=errors)
+ name = compat_urllib_parse_unquote(
+ name, encoding=encoding, errors=errors)
name = _coerce_result(name)
value = nv[1].replace('+', ' ')
- value = _unquote(value, encoding=encoding, errors=errors)
+ value = compat_urllib_parse_unquote(
+ value, encoding=encoding, errors=errors)
value = _coerce_result(value)
r.append((name, value))
return r
if sys.version_info >= (2,7):
def find_xpath_attr(node, xpath, key, val):
""" Find the xpath xpath[@key=val] """
- assert re.match(r'^[a-zA-Z]+$', key)
- assert re.match(r'^[a-zA-Z0-9@\s:._]*$', val)
+ assert re.match(r'^[a-zA-Z-]+$', key)
+ assert re.match(r'^[a-zA-Z0-9@\s:._-]*$', val)
expr = xpath + u"[@%s='%s']" % (key, val)
return node.find(expr)
else:
encoding = 'utf-8'
return s.encode(encoding, 'ignore')
+
+def encodeArgument(s):
+ if not isinstance(s, compat_str):
+ # Legacy code that uses byte strings
+ # Uncomment the following line after fixing all post processors
+ #assert False, 'Internal error: %r should be of type %r, is %r' % (s, compat_str, type(s))
+ s = s.decode('ascii')
+ return encodeFilename(s, True)
+
+
def decodeOption(optval):
if optval is None:
return optval
class ExtractorError(Exception):
"""Error during info extraction."""
- def __init__(self, msg, tb=None, expected=False, cause=None):
+ def __init__(self, msg, tb=None, expected=False, cause=None, video_id=None):
""" tb, if given, is the original traceback (so that it can be printed out).
If expected is set, this is a normal error message and most likely not a bug in youtube-dl.
"""
if sys.exc_info()[0] in (compat_urllib_error.URLError, socket.timeout, UnavailableVideoError):
expected = True
+ if video_id is not None:
+ msg = video_id + ': ' + msg
if not expected:
msg = msg + u'; please report this issue on https://yt-dl.org/bug . Be sure to call youtube-dl with the --verbose flag and include its complete output. Make sure you are using the latest version; type youtube-dl -U to update.'
super(ExtractorError, self).__init__(msg)
self.traceback = tb
self.exc_info = sys.exc_info() # preserve original exception
self.cause = cause
+ self.video_id = video_id
def format_traceback(self):
if self.traceback is None:
https_response = http_response
-def parse_iso8601(date_str):
+def parse_iso8601(date_str, delimiter='T'):
""" Return a UNIX timestamp from the given date """
if date_str is None:
timezone = datetime.timedelta(
hours=sign * int(m.group('hours')),
minutes=sign * int(m.group('minutes')))
-
- dt = datetime.datetime.strptime(date_str, '%Y-%m-%dT%H:%M:%S') - timezone
+ date_format = '%Y-%m-%d{0}%H:%M:%S'.format(delimiter)
+ dt = datetime.datetime.strptime(date_str, date_format) - timezone
return calendar.timegm(dt.timetuple())
'%d %b %Y',
'%B %d %Y',
'%b %d %Y',
+ '%b %dst %Y %I:%M%p',
+ '%b %dnd %Y %I:%M%p',
+ '%b %dth %Y %I:%M%p',
'%Y-%m-%d',
'%d.%m.%Y',
'%d/%m/%Y',
return upload_date
def determine_ext(url, default_ext=u'unknown_video'):
+ if url is None:
+ return default_ext
guess = url.partition(u'?')[0].rpartition(u'.')[2]
if re.match(r'^[A-Za-z0-9]+$', guess):
return guess
2: -12,
}
- def ucs2_len(s):
- return sum((2 if ord(c) > 0xffff else 1) for c in s)
-
- fileno = out.fileno()
+ try:
+ fileno = out.fileno()
+ except AttributeError:
+ # If the output stream doesn't have a fileno, it's virtual
+ return False
if fileno not in WIN_OUTPUT_IDS:
return False
if not_a_console(h):
return False
- remaining = ucs2_len(s)
- while remaining > 0:
+ def next_nonbmp_pos(s):
+ try:
+ return next(i for i, c in enumerate(s) if ord(c) > 0xffff)
+ except StopIteration:
+ return len(s)
+
+ while s:
+ count = min(next_nonbmp_pos(s), 1024)
+
ret = WriteConsoleW(
- h, s, min(remaining, 1024), ctypes.byref(written), None)
+ h, s, count if count else 2, ctypes.byref(written), None)
if ret == 0:
raise OSError('Failed to write string')
- remaining -= written.value
+ if not count: # We just wrote a non-BMP character
+ assert written.value == 2
+ s = s[1:]
+ else:
+ assert written.value > 0
+ s = s[written.value:]
return True
return u'%.2f%s' % (converted, suffix)
-def str_to_int(int_str):
- int_str = re.sub(r'[,\.]', u'', int_str)
- return int(int_str)
-
-
def get_term_width():
columns = os.environ.get('COLUMNS', None)
if columns:
return "HEAD"
-def int_or_none(v, scale=1, default=None):
- return default if v is None else (int(v) // scale)
+def int_or_none(v, scale=1, default=None, get_attr=None, invscale=1):
+ if get_attr:
+ if v is not None:
+ v = getattr(v, get_attr, None)
+ return default if v is None else (int(v) * invscale // scale)
+
+
+def str_to_int(int_str):
+ if int_str is None:
+ return None
+ int_str = re.sub(r'[,\.]', u'', int_str)
+ return int(int_str)
-def float_or_none(v, scale=1, default=None):
- return default if v is None else (float(v) / scale)
+def float_or_none(v, scale=1, invscale=1, default=None):
+ return default if v is None else (float(v) * invscale / scale)
def parse_duration(s):
def strip_jsonp(code):
- return re.sub(r'(?s)^[a-zA-Z_]+\s*\(\s*(.*)\);\s*?\s*$', r'\1', code)
+ return re.sub(r'(?s)^[a-zA-Z0-9_]+\s*\(\s*(.*)\);?\s*?\s*$', r'\1', code)
+
+
+def qualities(quality_ids):
+ """ Get a numeric quality value out of a list of possible values """
+ def q(qid):
+ try:
+ return quality_ids.index(qid)
+ except ValueError:
+ return -1
+ return q
+
+
+DEFAULT_OUTTMPL = '%(title)s-%(id)s.%(ext)s'
+
+try:
+ subprocess_check_output = subprocess.check_output
+except AttributeError:
+ def subprocess_check_output(*args, **kwargs):
+ assert 'input' not in kwargs
+ p = subprocess.Popen(*args, stdout=subprocess.PIPE, **kwargs)
+ output, _ = p.communicate()
+ ret = p.poll()
+ if ret:
+ raise subprocess.CalledProcessError(ret, p.args, output=output)
+ return output