misc-csechet/frontools/utils.py

"""Common utilities"""
from asyncio import gather
from datetime import datetime
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


class ErrorSummary:
    def __init__(self) -> None:
        self._errors: list[str] = []

    def add_error(self, message: str) -> None:
        self._errors.append(message)

    def echo(self) -> None:
        if not len(self._errors):
            return
        echo("***** Error summary :")
        for error in self._errors:
            echo(error, err=True)


def get_default_screenshot_directory() -> Path:
    """Return the default folder where to store screenshot (XDG pictures dir by default)"""
    pictures_dir_re = re_compile(r'\w*XDG_PICTURES_DIR\w*\="(?P<path>.*)"\w*$')
    pictures_dir = None
    with open(
        Path(xdg_config_home()) / "user-dirs.dirs", encoding="utf-8"
    ) as user_dirs:
        for line in user_dirs.readlines():
            match = pictures_dir_re.match(line)
            if match is None:
                continue

            path_string = match.group("path")
            path_string = expandvars(path_string)
            pictures_dir = Path(path_string)
            break

    if pictures_dir is None:
        pictures_dir = Path.home()

    return pictures_dir / datetime.now().strftime("%Y-%m-%d - %H-%m-%S")


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(":", "_")