yt_dlp/downloader/fragment.py

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