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.

268 lines
10 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. assert 'ytdl_corrupt' not in ctx
  60. stream, _ = sanitize_open(self.ytdl_filename(ctx['filename']), 'r')
  61. try:
  62. ctx['fragment_index'] = json.loads(stream.read())['downloader']['current_fragment']['index']
  63. except Exception:
  64. ctx['ytdl_corrupt'] = True
  65. finally:
  66. stream.close()
  67. def _write_ytdl_file(self, ctx):
  68. frag_index_stream, _ = sanitize_open(self.ytdl_filename(ctx['filename']), 'w')
  69. downloader = {
  70. 'current_fragment': {
  71. 'index': ctx['fragment_index'],
  72. },
  73. }
  74. if ctx.get('fragment_count') is not None:
  75. downloader['fragment_count'] = ctx['fragment_count']
  76. frag_index_stream.write(json.dumps({'downloader': downloader}))
  77. frag_index_stream.close()
  78. def _download_fragment(self, ctx, frag_url, info_dict, headers=None):
  79. fragment_filename = '%s-Frag%d' % (ctx['tmpfilename'], ctx['fragment_index'])
  80. success = ctx['dl'].download(fragment_filename, {
  81. 'url': frag_url,
  82. 'http_headers': headers or info_dict.get('http_headers'),
  83. })
  84. if not success:
  85. return False, None
  86. down, frag_sanitized = sanitize_open(fragment_filename, 'rb')
  87. ctx['fragment_filename_sanitized'] = frag_sanitized
  88. frag_content = down.read()
  89. down.close()
  90. return True, frag_content
  91. def _append_fragment(self, ctx, frag_content):
  92. try:
  93. ctx['dest_stream'].write(frag_content)
  94. ctx['dest_stream'].flush()
  95. finally:
  96. if self.__do_ytdl_file(ctx):
  97. self._write_ytdl_file(ctx)
  98. if not self.params.get('keep_fragments', False):
  99. os.remove(encodeFilename(ctx['fragment_filename_sanitized']))
  100. del ctx['fragment_filename_sanitized']
  101. def _prepare_frag_download(self, ctx):
  102. if 'live' not in ctx:
  103. ctx['live'] = False
  104. if not ctx['live']:
  105. total_frags_str = '%d' % ctx['total_frags']
  106. ad_frags = ctx.get('ad_frags', 0)
  107. if ad_frags:
  108. total_frags_str += ' (not including %d ad)' % ad_frags
  109. else:
  110. total_frags_str = 'unknown (live)'
  111. self.to_screen(
  112. '[%s] Total fragments: %s' % (self.FD_NAME, total_frags_str))
  113. self.report_destination(ctx['filename'])
  114. dl = HttpQuietDownloader(
  115. self.ydl,
  116. {
  117. 'continuedl': True,
  118. 'quiet': True,
  119. 'noprogress': True,
  120. 'ratelimit': self.params.get('ratelimit'),
  121. 'retries': self.params.get('retries', 0),
  122. 'nopart': self.params.get('nopart', False),
  123. 'test': self.params.get('test', False),
  124. }
  125. )
  126. tmpfilename = self.temp_name(ctx['filename'])
  127. open_mode = 'wb'
  128. resume_len = 0
  129. # Establish possible resume length
  130. if os.path.isfile(encodeFilename(tmpfilename)):
  131. open_mode = 'ab'
  132. resume_len = os.path.getsize(encodeFilename(tmpfilename))
  133. # Should be initialized before ytdl file check
  134. ctx.update({
  135. 'tmpfilename': tmpfilename,
  136. 'fragment_index': 0,
  137. })
  138. if self.__do_ytdl_file(ctx):
  139. if os.path.isfile(encodeFilename(self.ytdl_filename(ctx['filename']))):
  140. self._read_ytdl_file(ctx)
  141. is_corrupt = ctx.get('ytdl_corrupt') is True
  142. is_inconsistent = ctx['fragment_index'] > 0 and resume_len == 0
  143. if is_corrupt or is_inconsistent:
  144. message = (
  145. '.ytdl file is corrupt' if is_corrupt else
  146. 'Inconsistent state of incomplete fragment download')
  147. self.report_warning(
  148. '%s. Restarting from the beginning...' % message)
  149. ctx['fragment_index'] = resume_len = 0
  150. if 'ytdl_corrupt' in ctx:
  151. del ctx['ytdl_corrupt']
  152. self._write_ytdl_file(ctx)
  153. else:
  154. self._write_ytdl_file(ctx)
  155. assert ctx['fragment_index'] == 0
  156. dest_stream, tmpfilename = sanitize_open(tmpfilename, open_mode)
  157. ctx.update({
  158. 'dl': dl,
  159. 'dest_stream': dest_stream,
  160. 'tmpfilename': tmpfilename,
  161. # Total complete fragments downloaded so far in bytes
  162. 'complete_frags_downloaded_bytes': resume_len,
  163. })
  164. def _start_frag_download(self, ctx):
  165. total_frags = ctx['total_frags']
  166. # This dict stores the download progress, it's updated by the progress
  167. # hook
  168. state = {
  169. 'status': 'downloading',
  170. 'downloaded_bytes': ctx['complete_frags_downloaded_bytes'],
  171. 'fragment_index': ctx['fragment_index'],
  172. 'fragment_count': total_frags,
  173. 'filename': ctx['filename'],
  174. 'tmpfilename': ctx['tmpfilename'],
  175. }
  176. start = time.time()
  177. ctx.update({
  178. 'started': start,
  179. # Amount of fragment's bytes downloaded by the time of the previous
  180. # frag progress hook invocation
  181. 'prev_frag_downloaded_bytes': 0,
  182. })
  183. def frag_progress_hook(s):
  184. if s['status'] not in ('downloading', 'finished'):
  185. return
  186. time_now = time.time()
  187. state['elapsed'] = time_now - start
  188. frag_total_bytes = s.get('total_bytes') or 0
  189. if not ctx['live']:
  190. estimated_size = (
  191. (ctx['complete_frags_downloaded_bytes'] + frag_total_bytes) /
  192. (state['fragment_index'] + 1) * total_frags)
  193. state['total_bytes_estimate'] = estimated_size
  194. if s['status'] == 'finished':
  195. state['fragment_index'] += 1
  196. ctx['fragment_index'] = state['fragment_index']
  197. state['downloaded_bytes'] += frag_total_bytes - ctx['prev_frag_downloaded_bytes']
  198. ctx['complete_frags_downloaded_bytes'] = state['downloaded_bytes']
  199. ctx['prev_frag_downloaded_bytes'] = 0
  200. else:
  201. frag_downloaded_bytes = s['downloaded_bytes']
  202. state['downloaded_bytes'] += frag_downloaded_bytes - ctx['prev_frag_downloaded_bytes']
  203. if not ctx['live']:
  204. state['eta'] = self.calc_eta(
  205. start, time_now, estimated_size,
  206. state['downloaded_bytes'])
  207. state['speed'] = s.get('speed') or ctx.get('speed')
  208. ctx['speed'] = state['speed']
  209. ctx['prev_frag_downloaded_bytes'] = frag_downloaded_bytes
  210. self._hook_progress(state)
  211. ctx['dl'].add_progress_hook(frag_progress_hook)
  212. return start
  213. def _finish_frag_download(self, ctx):
  214. ctx['dest_stream'].close()
  215. if self.__do_ytdl_file(ctx):
  216. ytdl_filename = encodeFilename(self.ytdl_filename(ctx['filename']))
  217. if os.path.isfile(ytdl_filename):
  218. os.remove(ytdl_filename)
  219. elapsed = time.time() - ctx['started']
  220. if ctx['tmpfilename'] == '-':
  221. downloaded_bytes = ctx['complete_frags_downloaded_bytes']
  222. else:
  223. self.try_rename(ctx['tmpfilename'], ctx['filename'])
  224. downloaded_bytes = os.path.getsize(encodeFilename(ctx['filename']))
  225. self._hook_progress({
  226. 'downloaded_bytes': downloaded_bytes,
  227. 'total_bytes': downloaded_bytes,
  228. 'filename': ctx['filename'],
  229. 'status': 'finished',
  230. 'elapsed': elapsed,
  231. })