You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

257 lines
9.7 KiB

  1. from __future__ import division, unicode_literals
  2. import os
  3. import time
  4. import json
  5. from .common import FileDownloader
  6. from .http import HttpFD
  7. from ..utils import (
  8. error_to_compat_str,
  9. encodeFilename,
  10. sanitize_open,
  11. sanitized_Request,
  12. )
  13. class HttpQuietDownloader(HttpFD):
  14. def to_screen(self, *args, **kargs):
  15. pass
  16. class FragmentFD(FileDownloader):
  17. """
  18. A base file downloader class for fragmented media (e.g. f4m/m3u8 manifests).
  19. Available options:
  20. fragment_retries: Number of times to retry a fragment for HTTP error (DASH
  21. and hlsnative only)
  22. skip_unavailable_fragments:
  23. Skip unavailable fragments (DASH and hlsnative only)
  24. keep_fragments: Keep downloaded fragments on disk after downloading is
  25. finished
  26. For each incomplete fragment download youtube-dl keeps on disk a special
  27. bookkeeping file with download state and metadata (in future such files will
  28. be used for any incomplete download handled by youtube-dl). This file is
  29. used to properly handle resuming, check download file consistency and detect
  30. potential errors. The file has a .ytdl extension and represents a standard
  31. JSON file of the following format:
  32. extractor:
  33. Dictionary of extractor related data. TBD.
  34. downloader:
  35. Dictionary of downloader related data. May contain following data:
  36. current_fragment:
  37. Dictionary with current (being downloaded) fragment data:
  38. index: 0-based index of current fragment among all fragments
  39. fragment_count:
  40. Total count of fragments
  41. This feature is experimental and file format may change in future.
  42. """
  43. def report_retry_fragment(self, err, frag_index, count, retries):
  44. self.to_screen(
  45. '[download] Got server HTTP error: %s. Retrying fragment %d (attempt %d of %s)...'
  46. % (error_to_compat_str(err), frag_index, count, self.format_retries(retries)))
  47. def report_skip_fragment(self, frag_index):
  48. self.to_screen('[download] Skipping fragment %d...' % frag_index)
  49. def _prepare_url(self, info_dict, url):
  50. headers = info_dict.get('http_headers')
  51. return sanitized_Request(url, None, headers) if headers else url
  52. def _prepare_and_start_frag_download(self, ctx):
  53. self._prepare_frag_download(ctx)
  54. self._start_frag_download(ctx)
  55. @staticmethod
  56. def __do_ytdl_file(ctx):
  57. return not ctx['live'] and not ctx['tmpfilename'] == '-'
  58. def _read_ytdl_file(self, ctx):
  59. stream, _ = sanitize_open(self.ytdl_filename(ctx['filename']), 'r')
  60. ctx['fragment_index'] = json.loads(stream.read())['downloader']['current_fragment']['index']
  61. stream.close()
  62. def _write_ytdl_file(self, ctx):
  63. frag_index_stream, _ = sanitize_open(self.ytdl_filename(ctx['filename']), 'w')
  64. downloader = {
  65. 'current_fragment': {
  66. 'index': ctx['fragment_index'],
  67. },
  68. }
  69. if ctx.get('fragment_count') is not None:
  70. downloader['fragment_count'] = ctx['fragment_count']
  71. frag_index_stream.write(json.dumps({'downloader': downloader}))
  72. frag_index_stream.close()
  73. def _download_fragment(self, ctx, frag_url, info_dict, headers=None):
  74. fragment_filename = '%s-Frag%d' % (ctx['tmpfilename'], ctx['fragment_index'])
  75. success = ctx['dl'].download(fragment_filename, {
  76. 'url': frag_url,
  77. 'http_headers': headers or info_dict.get('http_headers'),
  78. })
  79. if not success:
  80. return False, None
  81. down, frag_sanitized = sanitize_open(fragment_filename, 'rb')
  82. ctx['fragment_filename_sanitized'] = frag_sanitized
  83. frag_content = down.read()
  84. down.close()
  85. return True, frag_content
  86. def _append_fragment(self, ctx, frag_content):
  87. try:
  88. ctx['dest_stream'].write(frag_content)
  89. ctx['dest_stream'].flush()
  90. finally:
  91. if self.__do_ytdl_file(ctx):
  92. self._write_ytdl_file(ctx)
  93. if not self.params.get('keep_fragments', False):
  94. os.remove(encodeFilename(ctx['fragment_filename_sanitized']))
  95. del ctx['fragment_filename_sanitized']
  96. def _prepare_frag_download(self, ctx):
  97. if 'live' not in ctx:
  98. ctx['live'] = False
  99. if not ctx['live']:
  100. total_frags_str = '%d' % ctx['total_frags']
  101. ad_frags = ctx.get('ad_frags', 0)
  102. if ad_frags:
  103. total_frags_str += ' (not including %d ad)' % ad_frags
  104. else:
  105. total_frags_str = 'unknown (live)'
  106. self.to_screen(
  107. '[%s] Total fragments: %s' % (self.FD_NAME, total_frags_str))
  108. self.report_destination(ctx['filename'])
  109. dl = HttpQuietDownloader(
  110. self.ydl,
  111. {
  112. 'continuedl': True,
  113. 'quiet': True,
  114. 'noprogress': True,
  115. 'ratelimit': self.params.get('ratelimit'),
  116. 'retries': self.params.get('retries', 0),
  117. 'nopart': self.params.get('nopart', False),
  118. 'test': self.params.get('test', False),
  119. }
  120. )
  121. tmpfilename = self.temp_name(ctx['filename'])
  122. open_mode = 'wb'
  123. resume_len = 0
  124. # Establish possible resume length
  125. if os.path.isfile(encodeFilename(tmpfilename)):
  126. open_mode = 'ab'
  127. resume_len = os.path.getsize(encodeFilename(tmpfilename))
  128. # Should be initialized before ytdl file check
  129. ctx.update({
  130. 'tmpfilename': tmpfilename,
  131. 'fragment_index': 0,
  132. })
  133. if self.__do_ytdl_file(ctx):
  134. if os.path.isfile(encodeFilename(self.ytdl_filename(ctx['filename']))):
  135. self._read_ytdl_file(ctx)
  136. if ctx['fragment_index'] > 0 and resume_len == 0:
  137. self.report_warning(
  138. 'Inconsistent state of incomplete fragment download. '
  139. 'Restarting from the beginning...')
  140. ctx['fragment_index'] = resume_len = 0
  141. self._write_ytdl_file(ctx)
  142. else:
  143. self._write_ytdl_file(ctx)
  144. assert ctx['fragment_index'] == 0
  145. dest_stream, tmpfilename = sanitize_open(tmpfilename, open_mode)
  146. ctx.update({
  147. 'dl': dl,
  148. 'dest_stream': dest_stream,
  149. 'tmpfilename': tmpfilename,
  150. # Total complete fragments downloaded so far in bytes
  151. 'complete_frags_downloaded_bytes': resume_len,
  152. })
  153. def _start_frag_download(self, ctx):
  154. total_frags = ctx['total_frags']
  155. # This dict stores the download progress, it's updated by the progress
  156. # hook
  157. state = {
  158. 'status': 'downloading',
  159. 'downloaded_bytes': ctx['complete_frags_downloaded_bytes'],
  160. 'fragment_index': ctx['fragment_index'],
  161. 'fragment_count': total_frags,
  162. 'filename': ctx['filename'],
  163. 'tmpfilename': ctx['tmpfilename'],
  164. }
  165. start = time.time()
  166. ctx.update({
  167. 'started': start,
  168. # Amount of fragment's bytes downloaded by the time of the previous
  169. # frag progress hook invocation
  170. 'prev_frag_downloaded_bytes': 0,
  171. })
  172. def frag_progress_hook(s):
  173. if s['status'] not in ('downloading', 'finished'):
  174. return
  175. time_now = time.time()
  176. state['elapsed'] = time_now - start
  177. frag_total_bytes = s.get('total_bytes') or 0
  178. if not ctx['live']:
  179. estimated_size = (
  180. (ctx['complete_frags_downloaded_bytes'] + frag_total_bytes) /
  181. (state['fragment_index'] + 1) * total_frags)
  182. state['total_bytes_estimate'] = estimated_size
  183. if s['status'] == 'finished':
  184. state['fragment_index'] += 1
  185. ctx['fragment_index'] = state['fragment_index']
  186. state['downloaded_bytes'] += frag_total_bytes - ctx['prev_frag_downloaded_bytes']
  187. ctx['complete_frags_downloaded_bytes'] = state['downloaded_bytes']
  188. ctx['prev_frag_downloaded_bytes'] = 0
  189. else:
  190. frag_downloaded_bytes = s['downloaded_bytes']
  191. state['downloaded_bytes'] += frag_downloaded_bytes - ctx['prev_frag_downloaded_bytes']
  192. if not ctx['live']:
  193. state['eta'] = self.calc_eta(
  194. start, time_now, estimated_size,
  195. state['downloaded_bytes'])
  196. state['speed'] = s.get('speed') or ctx.get('speed')
  197. ctx['speed'] = state['speed']
  198. ctx['prev_frag_downloaded_bytes'] = frag_downloaded_bytes
  199. self._hook_progress(state)
  200. ctx['dl'].add_progress_hook(frag_progress_hook)
  201. return start
  202. def _finish_frag_download(self, ctx):
  203. ctx['dest_stream'].close()
  204. if self.__do_ytdl_file(ctx):
  205. ytdl_filename = encodeFilename(self.ytdl_filename(ctx['filename']))
  206. if os.path.isfile(ytdl_filename):
  207. os.remove(ytdl_filename)
  208. elapsed = time.time() - ctx['started']
  209. if ctx['tmpfilename'] == '-':
  210. downloaded_bytes = ctx['complete_frags_downloaded_bytes']
  211. else:
  212. self.try_rename(ctx['tmpfilename'], ctx['filename'])
  213. downloaded_bytes = os.path.getsize(encodeFilename(ctx['filename']))
  214. self._hook_progress({
  215. 'downloaded_bytes': downloaded_bytes,
  216. 'total_bytes': downloaded_bytes,
  217. 'filename': ctx['filename'],
  218. 'status': 'finished',
  219. 'elapsed': elapsed,
  220. })