43 lines
1.2 KiB
Python
43 lines
1.2 KiB
Python
"""Common utilities"""
|
|
from asyncio import gather
|
|
from os.path import expandvars
|
|
from pathlib import Path
|
|
from re import compile as re_compile
|
|
from typing import Awaitable, cast
|
|
|
|
from click import echo, progressbar
|
|
from xdg import xdg_config_home
|
|
|
|
|
|
TaskListType = list[tuple[str, Awaitable[None]]]
|
|
|
|
|
|
async def report_progress(
|
|
label: str, task_list: TaskListType, nb_workers: int = 5
|
|
) -> None:
|
|
"""Show a progress bar for a long running task list"""
|
|
iterator = iter(task_list)
|
|
|
|
with progressbar(
|
|
length=len(task_list),
|
|
label=label,
|
|
item_show_func=lambda it: cast(str, it),
|
|
) as progress:
|
|
|
|
async def worker() -> None:
|
|
try:
|
|
while True:
|
|
item_name, task = next(iterator)
|
|
progress.update(1, current_item=item_name) # type: ignore
|
|
await task
|
|
except StopIteration:
|
|
pass
|
|
|
|
workers = [worker() for _ in range(0, nb_workers)]
|
|
await gather(*workers)
|
|
|
|
|
|
def get_url_slug(url: str) -> str:
|
|
"""Return an unique slug usable as a path name for a given url."""
|
|
return url.replace("_", "___").replace("/", "__").replace(":", "_")
|