Skip to content

Instantly share code, notes, and snippets.

@curtis18

curtis18/downloader.py

Forked from molcay/downloader.py
Created July 1, 2022 07:09
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save curtis18/c515f6a6d3ef60bce382a23baec58e1a to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save curtis18/c515f6a6d3ef60bce382a23baec58e1a to your computer and use it in GitHub Desktop.
Download ZIP

Revisions

  1. @molcay molcay created this gist May 5, 2020.
    66 changes: 66 additions & 0 deletions downloader.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,66 @@
    """
    A rudimentary URL downloader (like wget or curl) to demonstrate Rich progress bars.
    SOURCE: https://github.com/willmcgugan/rich/blob/master/examples/downloader.py
    """

    from concurrent.futures import ThreadPoolExecutor
    from functools import partial
    import os.path
    import sys
    from typing import Iterable
    from urllib.request import urlopen

    from rich.progress import (
    BarColumn,
    DownloadColumn,
    TextColumn,
    TransferSpeedColumn,
    TimeRemainingColumn,
    Progress,
    TaskID,
    )


    progress = Progress(
    TextColumn("[bold blue]{task.fields[filename]}", justify="right"),
    BarColumn(bar_width=None),
    "[progress.percentage]{task.percentage:>3.1f}%",
    "•",
    DownloadColumn(),
    "•",
    TransferSpeedColumn(),
    "•",
    TimeRemainingColumn(),
    )


    def copy_url(task_id: TaskID, url: str, path: str) -> None:
    """Copy data from a url to a local file."""
    response = urlopen(url)
    # This will break if the response doesn't contain content length
    progress.update(task_id, total=int(response.info()["Content-length"]))
    with open(path, "wb") as dest_file:
    for data in iter(partial(response.read, 32768), b""):
    dest_file.write(data)
    progress.update(task_id, advance=len(data))


    def download(urls: Iterable[str], dest_dir: str):
    """Download multuple files to the given directory."""
    with progress:
    with ThreadPoolExecutor(max_workers=4) as pool:
    for url in urls:
    filename = url.split("/")[-1]
    dest_path = os.path.join(dest_dir, filename)
    task_id = progress.add_task("download", filename=filename)
    pool.submit(copy_url, task_id, url, dest_path)


    if __name__ == "__main__":
    # Try with https://releases.ubuntu.com/20.04/ubuntu-20.04-desktop-amd64.iso
    if sys.argv[1:]:
    download(sys.argv[1:], "./")
    else:
    print("Usage:\n\tpython downloader.py URL1 URL2 URL3 (etc)")