forked from microsoft/ai4eutils
-
Notifications
You must be signed in to change notification settings - Fork 0
/
ai4e_web_utils.py
75 lines (56 loc) · 2.22 KB
/
ai4e_web_utils.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
#
# ai4e_web_utils.py
#
# Functions for interacting with http requests
#
#%% Imports
import os
# pip install progressbar2, not progressbar
import progressbar
import urllib
import tempfile
#%% Functions
class DownloadProgressBar():
"""
Console progress indicator for downloads.
stackoverflow.com/questions/37748105/how-to-use-progressbar-module-with-urlretrieve
"""
def __init__(self):
self.pbar = None
def __call__(self, block_num, block_size, total_size):
if not self.pbar:
self.pbar = progressbar.ProgressBar(max_value=total_size)
self.pbar.start()
downloaded = block_num * block_size
if downloaded < total_size:
self.pbar.update(downloaded)
else:
self.pbar.finish()
def download_url(url, destination_filename=None, progress_updater=None,
force_download=False, output_dir=None, verbose=False):
"""
Download a URL, optionally downloading to a temporary file
"""
# if progress_updater is None:
# progress_updater = DownloadProgressBar()
# This is not intended to guarantee uniqueness, we just know it happens to guarantee
# uniqueness for this application.
if destination_filename is None:
if output_dir is None:
output_dir = os.path.join(tempfile.gettempdir(),'ai4e')
os.makedirs(output_dir,exist_ok=True)
url_as_filename = url.replace('://', '_').replace('.', '_').replace('/', '_')
destination_filename = \
os.path.join(output_dir,url_as_filename)
if (not force_download) and (os.path.isfile(destination_filename)):
if verbose:
print('Bypassing download of already-downloaded file {}'.format(os.path.basename(url)))
return destination_filename
if verbose:
print('Downloading file {}'.format(os.path.basename(url)),end='')
urllib.request.urlretrieve(url, destination_filename, progress_updater)
assert(os.path.isfile(destination_filename))
nBytes = os.path.getsize(destination_filename)
if verbose:
print('...done, {} bytes.'.format(nBytes))
return destination_filename