Server IP : 85.214.239.14 / Your IP : 3.145.37.211 Web Server : Apache/2.4.62 (Debian) System : Linux h2886529.stratoserver.net 4.9.0 #1 SMP Tue Jan 9 19:45:01 MSK 2024 x86_64 User : www-data ( 33) PHP Version : 7.4.18 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, MySQL : OFF | cURL : OFF | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : OFF Directory : /usr/share/doc/python3-rich/examples/ |
Upload File : |
""" A rudimentary URL downloader (like wget or curl) to demonstrate Rich progress bars. """ import os.path import sys from concurrent.futures import ThreadPoolExecutor import signal from functools import partial from threading import Event from typing import Iterable from urllib.request import urlopen from rich.progress import ( BarColumn, DownloadColumn, Progress, TaskID, TextColumn, TimeRemainingColumn, TransferSpeedColumn, ) progress = Progress( TextColumn("[bold blue]{task.fields[filename]}", justify="right"), BarColumn(bar_width=None), "[progress.percentage]{task.percentage:>3.1f}%", "•", DownloadColumn(), "•", TransferSpeedColumn(), "•", TimeRemainingColumn(), ) done_event = Event() def handle_sigint(signum, frame): done_event.set() signal.signal(signal.SIGINT, handle_sigint) def copy_url(task_id: TaskID, url: str, path: str) -> None: """Copy data from a url to a local file.""" progress.console.log(f"Requesting {url}") 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: progress.start_task(task_id) for data in iter(partial(response.read, 32768), b""): dest_file.write(data) progress.update(task_id, advance=len(data)) if done_event.is_set(): return progress.console.log(f"Downloaded {path}") def download(urls: Iterable[str], dest_dir: str): """Download multiple 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, start=False) pool.submit(copy_url, task_id, url, dest_path) if __name__ == "__main__": # Try with https://releases.ubuntu.com/20.04/ubuntu-20.04.3-desktop-amd64.iso if sys.argv[1:]: download(sys.argv[1:], "./") else: print("Usage:\n\tpython downloader.py URL1 URL2 URL3 (etc)")