| import os |
| import tqdm |
| import requests |
|
|
| try: |
| import wget |
| except: |
| wget = None |
|
|
| def HF_download_file(url, output_path=None): |
| url = url.replace("/blob/", "/resolve/").replace("?download=true", "").strip() |
| output_path = ( |
| os.path.basename(url) |
| ) if output_path is None else ( |
| os.path.join(output_path, os.path.basename(url)) if os.path.isdir(output_path) else output_path |
| ) |
|
|
| if wget != None: |
| wget.download( |
| url, |
| out=output_path |
| ) |
| else: |
| response = requests.get(url, stream=True, timeout=300) |
|
|
| if response.status_code == 200: |
| progress_bar = tqdm.tqdm( |
| total=int(response.headers.get("content-length", 0)), |
| desc=os.path.basename(url), |
| ncols=100, |
| unit="byte", |
| leave=False |
| ) |
|
|
| with open(output_path, "wb") as f: |
| for chunk in response.iter_content(chunk_size=10 * 1024 * 1024): |
| progress_bar.update(len(chunk)) |
| f.write(chunk) |
|
|
| progress_bar.close() |
| else: raise ValueError(response.status_code) |
|
|
| return output_path |
