FILE_FLAGS: dict[FlagName, FlagsInt] = {
FlagName('do not sync'): FlagsInt(1 << 62)
}
+TOK_FO_ID = 'format_id'
+TOK_FS_AP = 'filesize_approx'
+TOK_FRAG_C = 'fragment_count'
+TOK_FRAG_I = 'fragment_index'
+TOK_LOADED = 'downloaded_bytes'
MILLION = 1000 * 1000
MEGA = 1024 * 1024
def _download_next(self) -> None:
if self._to_download:
downloaded_before: int = 0
- estimated_total: int = 0
- estimated_total_mb: float
+ sizes: dict[str, list[int]] = {}
def hook(d) -> None:
nonlocal downloaded_before
if d['status'] in {'downloading', 'finished'}:
- downloaded_i = d['downloaded_bytes']
+ downloaded_i = d[TOK_LOADED]
downloaded_mb = (downloaded_i + downloaded_before) / MEGA
- msg = f'{int(100 * downloaded_mb/estimated_total_mb)}%: '\
- f'{downloaded_mb:5.1f}/{estimated_total_mb:.1f}'
+ size = sizes[d['info_dict'][TOK_FO_ID]]
+ if (not size[0]) and TOK_FRAG_C in d and TOK_FRAG_I in d:
+ progress = min(d[TOK_FRAG_I] + 1, d[TOK_FRAG_C]
+ ) / d[TOK_FRAG_C]
+ size[1] = d[TOK_LOADED] / progress
+ guess_total_mb = sum(t[1] for t in sizes.values()) / MEGA
+ msg = f'{int(100 * downloaded_mb/guess_total_mb)}%: '\
+ f'{downloaded_mb:5.1f}/{guess_total_mb:.1f}'
self._update_status(video_id, f'downloading: {msg} MB')
if d['status'] == 'finished':
downloaded_before += downloaded_i
with YoutubeDL(YT_DL_PARAMS | {'progress_hooks': [hook]}) as ydl:
self._update_status(video_id, 'preparing download')
info = ydl.sanitize_info(ydl.extract_info(url, download=False))
- for requested in info['requested_formats']:
- estimated_total += requested['filesize_approx']
- estimated_total_mb = estimated_total / MEGA
+ for f in info['requested_formats']:
+ sizes[f[TOK_FO_ID]] = [False, 1]
+ if TOK_FS_AP in f:
+ sizes[f[TOK_FO_ID]] = [True, f[TOK_FS_AP]]
ydl.download(url)
self._sync_db()