yt_dlp/downloader/fragment.py

   1 from __future__ import division, unicode_literals
   2
   3 import http.client
   4 import json
   5 import math
   6 import os
   7 import time
   8
   9 try:
  10     import concurrent.futures
  11     can_threaded_download = True
  12 except ImportError:
  13     can_threaded_download = False
  14
  15 from .common import FileDownloader
  16 from .http import HttpFD
  17 from ..aes import aes_cbc_decrypt_bytes, unpad_pkcs7
  18 from ..compat import (
  19     compat_os_name,
  20     compat_urllib_error,
  21     compat_struct_pack,
  22 )
  23 from ..utils import (
  24     DownloadError,
  25     error_to_compat_str,
  26     encodeFilename,
  27     sanitized_Request,
  28     traverse_obj,
  29 )
  30
  31
  32 class HttpQuietDownloader(HttpFD):
  33     def to_screen(self, *args, **kargs):
  34         pass
  35
  36     def report_retry(self, err, count, retries):
  37         super().to_screen(
  38             f'[download] Got server HTTP error: {err}. Retrying (attempt {count} of {self.format_retries(retries)}) ...')
  39
  40
  41 class FragmentFD(FileDownloader):
  42     """
  43     A base file downloader class for fragmented media (e.g. f4m/m3u8 manifests).
  44
  45     Available options:
  46
  47     fragment_retries:   Number of times to retry a fragment for HTTP error (DASH
  48                         and hlsnative only)
  49     skip_unavailable_fragments:
  50                         Skip unavailable fragments (DASH and hlsnative only)
  51     keep_fragments:     Keep downloaded fragments on disk after downloading is
  52                         finished
  53     concurrent_fragment_downloads:  The number of threads to use for native hls and dash downloads
  54     _no_ytdl_file:      Don't use .ytdl file
  55
  56     For each incomplete fragment download yt-dlp keeps on disk a special
  57     bookkeeping file with download state and metadata (in future such files will
  58     be used for any incomplete download handled by yt-dlp). This file is
  59     used to properly handle resuming, check download file consistency and detect
  60     potential errors. The file has a .ytdl extension and represents a standard
  61     JSON file of the following format:
  62
  63     extractor:
  64         Dictionary of extractor related data. TBD.
  65
  66     downloader:
  67         Dictionary of downloader related data. May contain following data:
  68             current_fragment:
  69                 Dictionary with current (being downloaded) fragment data:
  70                 index:  0-based index of current fragment among all fragments
  71             fragment_count:
  72                 Total count of fragments
  73
  74     This feature is experimental and file format may change in future.
  75     """
  76
  77     def report_retry_fragment(self, err, frag_index, count, retries):
  78         self.to_screen(
  79             '\r[download] Got server HTTP error: %s. Retrying fragment %d (attempt %d of %s) ...'
  80             % (error_to_compat_str(err), frag_index, count, self.format_retries(retries)))
  81
  82     def report_skip_fragment(self, frag_index, err=None):
  83         err = f' {err};' if err else ''
  84         self.to_screen(f'[download]{err} Skipping fragment {frag_index:d} ...')
  85
  86     def _prepare_url(self, info_dict, url):
  87         headers = info_dict.get('http_headers')
  88         return sanitized_Request(url, None, headers) if headers else url
  89
  90     def _prepare_and_start_frag_download(self, ctx, info_dict):
  91         self._prepare_frag_download(ctx)
  92         self._start_frag_download(ctx, info_dict)
  93
  94     def __do_ytdl_file(self, ctx):
  95         return ctx['live'] is not True and ctx['tmpfilename'] != '-' and not self.params.get('_no_ytdl_file')
  96
  97     def _read_ytdl_file(self, ctx):
  98         assert 'ytdl_corrupt' not in ctx
  99         stream, _ = self.sanitize_open(self.ytdl_filename(ctx['filename']), 'r')
 100         try:
 101             ytdl_data = json.loads(stream.read())
 102             ctx['fragment_index'] = ytdl_data['downloader']['current_fragment']['index']
 103             if 'extra_state' in ytdl_data['downloader']:
 104                 ctx['extra_state'] = ytdl_data['downloader']['extra_state']
 105         except Exception:
 106             ctx['ytdl_corrupt'] = True
 107         finally:
 108             stream.close()
 109
 110     def _write_ytdl_file(self, ctx):
 111         frag_index_stream, _ = self.sanitize_open(self.ytdl_filename(ctx['filename']), 'w')
 112         try:
 113             downloader = {
 114                 'current_fragment': {
 115                     'index': ctx['fragment_index'],
 116                 },
 117             }
 118             if 'extra_state' in ctx:
 119                 downloader['extra_state'] = ctx['extra_state']
 120             if ctx.get('fragment_count') is not None:
 121                 downloader['fragment_count'] = ctx['fragment_count']
 122             frag_index_stream.write(json.dumps({'downloader': downloader}))
 123         finally:
 124             frag_index_stream.close()
 125
 126     def _download_fragment(self, ctx, frag_url, info_dict, headers=None, request_data=None):
 127         fragment_filename = '%s-Frag%d' % (ctx['tmpfilename'], ctx['fragment_index'])
 128         fragment_info_dict = {
 129             'url': frag_url,
 130             'http_headers': headers or info_dict.get('http_headers'),
 131             'request_data': request_data,
 132             'ctx_id': ctx.get('ctx_id'),
 133         }
 134         success = ctx['dl'].download(fragment_filename, fragment_info_dict)
 135         if not success:
 136             return False, None
 137         if fragment_info_dict.get('filetime'):
 138             ctx['fragment_filetime'] = fragment_info_dict.get('filetime')
 139         ctx['fragment_filename_sanitized'] = fragment_filename
 140         try:
 141             return True, self._read_fragment(ctx)
 142         except FileNotFoundError:
 143             if not info_dict.get('is_live'):
 144                 raise
 145             return False, None
 146
 147     def _read_fragment(self, ctx):
 148         down, frag_sanitized = self.sanitize_open(ctx['fragment_filename_sanitized'], 'rb')
 149         ctx['fragment_filename_sanitized'] = frag_sanitized
 150         frag_content = down.read()
 151         down.close()
 152         return frag_content
 153
 154     def _append_fragment(self, ctx, frag_content):
 155         try:
 156             ctx['dest_stream'].write(frag_content)
 157             ctx['dest_stream'].flush()
 158         finally:
 159             if self.__do_ytdl_file(ctx):
 160                 self._write_ytdl_file(ctx)
 161             if not self.params.get('keep_fragments', False):
 162                 self.try_remove(encodeFilename(ctx['fragment_filename_sanitized']))
 163             del ctx['fragment_filename_sanitized']
 164
 165     def _prepare_frag_download(self, ctx):
 166         if 'live' not in ctx:
 167             ctx['live'] = False
 168         if not ctx['live']:
 169             total_frags_str = '%d' % ctx['total_frags']
 170             ad_frags = ctx.get('ad_frags', 0)
 171             if ad_frags:
 172                 total_frags_str += ' (not including %d ad)' % ad_frags
 173         else:
 174             total_frags_str = 'unknown (live)'
 175         self.to_screen(
 176             '[%s] Total fragments: %s' % (self.FD_NAME, total_frags_str))
 177         self.report_destination(ctx['filename'])
 178         dl = HttpQuietDownloader(
 179             self.ydl,
 180             {
 181                 'continuedl': self.params.get('continuedl', True),
 182                 'quiet': self.params.get('quiet'),
 183                 'noprogress': True,
 184                 'ratelimit': self.params.get('ratelimit'),
 185                 'retries': self.params.get('retries', 0),
 186                 'nopart': self.params.get('nopart', False),
 187                 'test': self.params.get('test', False),
 188             }
 189         )
 190         tmpfilename = self.temp_name(ctx['filename'])
 191         open_mode = 'wb'
 192         resume_len = 0
 193
 194         # Establish possible resume length
 195         if os.path.isfile(encodeFilename(tmpfilename)):
 196             open_mode = 'ab'
 197             resume_len = os.path.getsize(encodeFilename(tmpfilename))
 198
 199         # Should be initialized before ytdl file check
 200         ctx.update({
 201             'tmpfilename': tmpfilename,
 202             'fragment_index': 0,
 203         })
 204
 205         if self.__do_ytdl_file(ctx):
 206             if os.path.isfile(encodeFilename(self.ytdl_filename(ctx['filename']))):
 207                 self._read_ytdl_file(ctx)
 208                 is_corrupt = ctx.get('ytdl_corrupt') is True
 209                 is_inconsistent = ctx['fragment_index'] > 0 and resume_len == 0
 210                 if is_corrupt or is_inconsistent:
 211                     message = (
 212                         '.ytdl file is corrupt' if is_corrupt else
 213                         'Inconsistent state of incomplete fragment download')
 214                     self.report_warning(
 215                         '%s. Restarting from the beginning ...' % message)
 216                     ctx['fragment_index'] = resume_len = 0
 217                     if 'ytdl_corrupt' in ctx:
 218                         del ctx['ytdl_corrupt']
 219                     self._write_ytdl_file(ctx)
 220             else:
 221                 self._write_ytdl_file(ctx)
 222                 assert ctx['fragment_index'] == 0
 223
 224         dest_stream, tmpfilename = self.sanitize_open(tmpfilename, open_mode)
 225
 226         ctx.update({
 227             'dl': dl,
 228             'dest_stream': dest_stream,
 229             'tmpfilename': tmpfilename,
 230             # Total complete fragments downloaded so far in bytes
 231             'complete_frags_downloaded_bytes': resume_len,
 232         })
 233
 234     def _start_frag_download(self, ctx, info_dict):
 235         resume_len = ctx['complete_frags_downloaded_bytes']
 236         total_frags = ctx['total_frags']
 237         ctx_id = ctx.get('ctx_id')
 238         # This dict stores the download progress, it's updated by the progress
 239         # hook
 240         state = {
 241             'status': 'downloading',
 242             'downloaded_bytes': resume_len,
 243             'fragment_index': ctx['fragment_index'],
 244             'fragment_count': total_frags,
 245             'filename': ctx['filename'],
 246             'tmpfilename': ctx['tmpfilename'],
 247         }
 248
 249         start = time.time()
 250         ctx.update({
 251             'started': start,
 252             'fragment_started': start,
 253             # Amount of fragment's bytes downloaded by the time of the previous
 254             # frag progress hook invocation
 255             'prev_frag_downloaded_bytes': 0,
 256         })
 257
 258         def frag_progress_hook(s):
 259             if s['status'] not in ('downloading', 'finished'):
 260                 return
 261
 262             if ctx_id is not None and s.get('ctx_id') != ctx_id:
 263                 return
 264
 265             state['max_progress'] = ctx.get('max_progress')
 266             state['progress_idx'] = ctx.get('progress_idx')
 267
 268             time_now = time.time()
 269             state['elapsed'] = time_now - start
 270             frag_total_bytes = s.get('total_bytes') or 0
 271             s['fragment_info_dict'] = s.pop('info_dict', {})
 272             if not ctx['live']:
 273                 estimated_size = (
 274                     (ctx['complete_frags_downloaded_bytes'] + frag_total_bytes)
 275                     / (state['fragment_index'] + 1) * total_frags)
 276                 state['total_bytes_estimate'] = estimated_size
 277
 278             if s['status'] == 'finished':
 279                 state['fragment_index'] += 1
 280                 ctx['fragment_index'] = state['fragment_index']
 281                 state['downloaded_bytes'] += frag_total_bytes - ctx['prev_frag_downloaded_bytes']
 282                 ctx['complete_frags_downloaded_bytes'] = state['downloaded_bytes']
 283                 ctx['speed'] = state['speed'] = self.calc_speed(
 284                     ctx['fragment_started'], time_now, frag_total_bytes)
 285                 ctx['fragment_started'] = time.time()
 286                 ctx['prev_frag_downloaded_bytes'] = 0
 287             else:
 288                 frag_downloaded_bytes = s['downloaded_bytes']
 289                 state['downloaded_bytes'] += frag_downloaded_bytes - ctx['prev_frag_downloaded_bytes']
 290                 if not ctx['live']:
 291                     state['eta'] = self.calc_eta(
 292                         start, time_now, estimated_size - resume_len,
 293                         state['downloaded_bytes'] - resume_len)
 294                 ctx['speed'] = state['speed'] = self.calc_speed(
 295                     ctx['fragment_started'], time_now, frag_downloaded_bytes)
 296                 ctx['prev_frag_downloaded_bytes'] = frag_downloaded_bytes
 297             self._hook_progress(state, info_dict)
 298
 299         ctx['dl'].add_progress_hook(frag_progress_hook)
 300
 301         return start
 302
 303     def _finish_frag_download(self, ctx, info_dict):
 304         ctx['dest_stream'].close()
 305         if self.__do_ytdl_file(ctx):
 306             ytdl_filename = encodeFilename(self.ytdl_filename(ctx['filename']))
 307             if os.path.isfile(ytdl_filename):
 308                 self.try_remove(ytdl_filename)
 309         elapsed = time.time() - ctx['started']
 310
 311         if ctx['tmpfilename'] == '-':
 312             downloaded_bytes = ctx['complete_frags_downloaded_bytes']
 313         else:
 314             self.try_rename(ctx['tmpfilename'], ctx['filename'])
 315             if self.params.get('updatetime', True):
 316                 filetime = ctx.get('fragment_filetime')
 317                 if filetime:
 318                     try:
 319                         os.utime(ctx['filename'], (time.time(), filetime))
 320                     except Exception:
 321                         pass
 322             downloaded_bytes = os.path.getsize(encodeFilename(ctx['filename']))
 323
 324         self._hook_progress({
 325             'downloaded_bytes': downloaded_bytes,
 326             'total_bytes': downloaded_bytes,
 327             'filename': ctx['filename'],
 328             'status': 'finished',
 329             'elapsed': elapsed,
 330             'ctx_id': ctx.get('ctx_id'),
 331             'max_progress': ctx.get('max_progress'),
 332             'progress_idx': ctx.get('progress_idx'),
 333         }, info_dict)
 334
 335     def _prepare_external_frag_download(self, ctx):
 336         if 'live' not in ctx:
 337             ctx['live'] = False
 338         if not ctx['live']:
 339             total_frags_str = '%d' % ctx['total_frags']
 340             ad_frags = ctx.get('ad_frags', 0)
 341             if ad_frags:
 342                 total_frags_str += ' (not including %d ad)' % ad_frags
 343         else:
 344             total_frags_str = 'unknown (live)'
 345         self.to_screen(
 346             '[%s] Total fragments: %s' % (self.FD_NAME, total_frags_str))
 347
 348         tmpfilename = self.temp_name(ctx['filename'])
 349
 350         # Should be initialized before ytdl file check
 351         ctx.update({
 352             'tmpfilename': tmpfilename,
 353             'fragment_index': 0,
 354         })
 355
 356     def decrypter(self, info_dict):
 357         _key_cache = {}
 358
 359         def _get_key(url):
 360             if url not in _key_cache:
 361                 _key_cache[url] = self.ydl.urlopen(self._prepare_url(info_dict, url)).read()
 362             return _key_cache[url]
 363
 364         def decrypt_fragment(fragment, frag_content):
 365             decrypt_info = fragment.get('decrypt_info')
 366             if not decrypt_info or decrypt_info['METHOD'] != 'AES-128':
 367                 return frag_content
 368             iv = decrypt_info.get('IV') or compat_struct_pack('>8xq', fragment['media_sequence'])
 369             decrypt_info['KEY'] = decrypt_info.get('KEY') or _get_key(info_dict.get('_decryption_key_url') or decrypt_info['URI'])
 370             # Don't decrypt the content in tests since the data is explicitly truncated and it's not to a valid block
 371             # size (see https://github.com/ytdl-org/youtube-dl/pull/27660). Tests only care that the correct data downloaded,
 372             # not what it decrypts to.
 373             if self.params.get('test', False):
 374                 return frag_content
 375             return unpad_pkcs7(aes_cbc_decrypt_bytes(frag_content, decrypt_info['KEY'], iv))
 376
 377         return decrypt_fragment
 378
 379     def download_and_append_fragments_multiple(self, *args, pack_func=None, finish_func=None):
 380         '''
 381         @params (ctx1, fragments1, info_dict1), (ctx2, fragments2, info_dict2), ...
 382                 all args must be either tuple or list
 383         '''
 384         interrupt_trigger = [True]
 385         max_progress = len(args)
 386         if max_progress == 1:
 387             return self.download_and_append_fragments(*args[0], pack_func=pack_func, finish_func=finish_func)
 388         max_workers = self.params.get('concurrent_fragment_downloads', 1)
 389         if max_progress > 1:
 390             self._prepare_multiline_status(max_progress)
 391         is_live = any(traverse_obj(args, (..., 2, 'is_live'), default=[]))
 392
 393         def thread_func(idx, ctx, fragments, info_dict, tpe):
 394             ctx['max_progress'] = max_progress
 395             ctx['progress_idx'] = idx
 396             return self.download_and_append_fragments(
 397                 ctx, fragments, info_dict, pack_func=pack_func, finish_func=finish_func,
 398                 tpe=tpe, interrupt_trigger=interrupt_trigger)
 399
 400         class FTPE(concurrent.futures.ThreadPoolExecutor):
 401             # has to stop this or it's going to wait on the worker thread itself
 402             def __exit__(self, exc_type, exc_val, exc_tb):
 403                 pass
 404
 405         if compat_os_name == 'nt':
 406             def bindoj_result(future):
 407                 while True:
 408                     try:
 409                         return future.result(0.1)
 410                     except KeyboardInterrupt:
 411                         raise
 412                     except concurrent.futures.TimeoutError:
 413                         continue
 414         else:
 415             def bindoj_result(future):
 416                 return future.result()
 417
 418         def interrupt_trigger_iter(fg):
 419             for f in fg:
 420                 if not interrupt_trigger[0]:
 421                     break
 422                 yield f
 423
 424         spins = []
 425         for idx, (ctx, fragments, info_dict) in enumerate(args):
 426             tpe = FTPE(math.ceil(max_workers / max_progress))
 427             job = tpe.submit(thread_func, idx, ctx, interrupt_trigger_iter(fragments), info_dict, tpe)
 428             spins.append((tpe, job))
 429
 430         result = True
 431         for tpe, job in spins:
 432             try:
 433                 result = result and bindoj_result(job)
 434             except KeyboardInterrupt:
 435                 interrupt_trigger[0] = False
 436             finally:
 437                 tpe.shutdown(wait=True)
 438         if not interrupt_trigger[0] and not is_live:
 439             raise KeyboardInterrupt()
 440         # we expect the user wants to stop and DO WANT the preceding postprocessors to run;
 441         # so returning a intermediate result here instead of KeyboardInterrupt on live
 442         return result
 443
 444     def download_and_append_fragments(
 445             self, ctx, fragments, info_dict, *, pack_func=None, finish_func=None,
 446             tpe=None, interrupt_trigger=None):
 447         if not interrupt_trigger:
 448             interrupt_trigger = (True, )
 449
 450         fragment_retries = self.params.get('fragment_retries', 0)
 451         is_fatal = (
 452             ((lambda _: False) if info_dict.get('is_live') else (lambda idx: idx == 0))
 453             if self.params.get('skip_unavailable_fragments', True) else (lambda _: True))
 454
 455         if not pack_func:
 456             pack_func = lambda frag_content, _: frag_content
 457
 458         def download_fragment(fragment, ctx):
 459             if not interrupt_trigger[0]:
 460                 return False, fragment['frag_index']
 461
 462             frag_index = ctx['fragment_index'] = fragment['frag_index']
 463             ctx['last_error'] = None
 464             headers = info_dict.get('http_headers', {}).copy()
 465             byte_range = fragment.get('byte_range')
 466             if byte_range:
 467                 headers['Range'] = 'bytes=%d-%d' % (byte_range['start'], byte_range['end'] - 1)
 468
 469             # Never skip the first fragment
 470             fatal = is_fatal(fragment.get('index') or (frag_index - 1))
 471             count, frag_content = 0, None
 472             while count <= fragment_retries:
 473                 try:
 474                     success, frag_content = self._download_fragment(ctx, fragment['url'], info_dict, headers)
 475                     if not success:
 476                         return False, frag_index
 477                     break
 478                 except (compat_urllib_error.HTTPError, http.client.IncompleteRead) as err:
 479                     # Unavailable (possibly temporary) fragments may be served.
 480                     # First we try to retry then either skip or abort.
 481                     # See https://github.com/ytdl-org/youtube-dl/issues/10165,
 482                     # https://github.com/ytdl-org/youtube-dl/issues/10448).
 483                     count += 1
 484                     ctx['last_error'] = err
 485                     if count <= fragment_retries:
 486                         self.report_retry_fragment(err, frag_index, count, fragment_retries)
 487                 except DownloadError:
 488                     # Don't retry fragment if error occurred during HTTP downloading
 489                     # itself since it has own retry settings
 490                     if not fatal:
 491                         break
 492                     raise
 493
 494             if count > fragment_retries:
 495                 if not fatal:
 496                     return False, frag_index
 497                 ctx['dest_stream'].close()
 498                 self.report_error('Giving up after %s fragment retries' % fragment_retries)
 499                 return False, frag_index
 500             return frag_content, frag_index
 501
 502         def append_fragment(frag_content, frag_index, ctx):
 503             if not frag_content:
 504                 if not is_fatal(frag_index - 1):
 505                     self.report_skip_fragment(frag_index, 'fragment not found')
 506                     return True
 507                 else:
 508                     ctx['dest_stream'].close()
 509                     self.report_error(
 510                         'fragment %s not found, unable to continue' % frag_index)
 511                     return False
 512             self._append_fragment(ctx, pack_func(frag_content, frag_index))
 513             return True
 514
 515         decrypt_fragment = self.decrypter(info_dict)
 516
 517         max_workers = math.ceil(
 518             self.params.get('concurrent_fragment_downloads', 1) / ctx.get('max_progress', 1))
 519         if can_threaded_download and max_workers > 1:
 520
 521             def _download_fragment(fragment):
 522                 ctx_copy = ctx.copy()
 523                 frag_content, frag_index = download_fragment(fragment, ctx_copy)
 524                 return fragment, frag_content, frag_index, ctx_copy.get('fragment_filename_sanitized')
 525
 526             self.report_warning('The download speed shown is only of one thread. This is a known issue and patches are welcome')
 527             with tpe or concurrent.futures.ThreadPoolExecutor(max_workers) as pool:
 528                 for fragment, frag_content, frag_index, frag_filename in pool.map(_download_fragment, fragments):
 529                     ctx['fragment_filename_sanitized'] = frag_filename
 530                     ctx['fragment_index'] = frag_index
 531                     result = append_fragment(decrypt_fragment(fragment, frag_content), frag_index, ctx)
 532                     if not result:
 533                         return False
 534         else:
 535             for fragment in fragments:
 536                 if not interrupt_trigger[0]:
 537                     break
 538                 frag_content, frag_index = download_fragment(fragment, ctx)
 539                 result = append_fragment(decrypt_fragment(fragment, frag_content), frag_index, ctx)
 540                 if not result:
 541                     return False
 542
 543         if finish_func is not None:
 544             ctx['dest_stream'].write(finish_func())
 545             ctx['dest_stream'].flush()
 546         self._finish_frag_download(ctx, info_dict)
 547         return True