py-backup/pyback.py

#!/usr/local/bin/python3.8
import argparse
import os
from dataclasses import dataclass
from datetime import datetime
from typing import List

import yaml
from loguru import logger
from rich import box, print
from rich.console import Console
from rich.padding import Padding
from rich.panel import Panel
from rich.prompt import Prompt
from rich.table import Table
from rich.text import Text


@dataclass(frozen=True)
class Action:
    message: str
    cmd: str

    @property
    def pretty_message(self) -> str:
        _m = self.message
        _m = _m.replace("{warn}", "[bold yellow]\[!][/bold yellow]")
        _m = _m.replace("{process}", "[bold orange3]\[..][/bold orange3]")

        return _m

    @property
    def pretty_cmd(self) -> str:
        _date: str = datetime.today().strftime("%Y-%m-%d")
        _time: str = datetime.now().strftime("%H%M")
        return self.cmd.replace("{date}", _date).replace("{time}", _time)


@dataclass()
class Job:
    name: str
    dst_dir: str
    actions: List[Action]
    active: bool = False


def run_job(job: Job, console: Console) -> None:
    """Запусить задачу бэкапа.

    Проходит по всем вложенным `actions` и, выводя сообщение
    `pretty_message`, выполняет задачу из `pretyy_cmd`.

    Notes
    -----
    Для удобства использования даты и времени, все это пишется с
    помощью плейсхолдеров в скрипте, и преобразуется при вызове
    `pretty_cmd`.
    """

    date = datetime.today().strftime("%Y-%m-%d")
    time = datetime.now().strftime("%H%M")

    path = job.dst_dir.replace("{date}", date).replace("{time}", time)

    if not os.path.isdir(path):
        os.makedirs(path)

    print(f"[bold yellow]\[..][/bold yellow] [bold]{job.name}[/bold] начинаю ...")

    for action in job.actions:
        with console.status(action.pretty_message):
            os.system(action.pretty_cmd)

    print(f"[bold green]\[OK][/bold green] [bold]{job.name}[/bold] выполнено")


def ask_about_backups(jobs: List[Job], force_all: bool) -> List[Job]:
    """Проверить список задач.

    Проходит в цикле по всем задачам, спрашивая нужно ли ее выполнять.
    Если нужно, ставит флаг `job.active` в True.

    Parameters
    ----------
    jobs : List[Job]
        Список задач. Состоит из объектов Job.
    force_all : bool
        Если True, то всем задачам будет установлен флаг
        `active`=True. Вопросы пользователю заданы не будут.
    Returns
    -------
    jobs : List[Job]
        Список задач после опроса пользователя по каждой из них. В
        соответствии с решением пользователя поле `active` будет либо
        изменено на True (задача активна), либо установленно False
        (задача не активна).
    """
    if force_all:
        for job in jobs:
            job.active = True
        return jobs

    print(Panel(Text("Составляю список задач", justify="center")))
    for job in jobs:
        answer = Prompt.ask(
            f"Архивируем [bold]{job.name}[/bold]?",
            choices=["y", "n"],
            default="y",
        ).lower()
        if answer == "y":
            job.active = True
            print("[bold green]\[+] Задача добавлена[/bold green]")
        elif answer == "n":
            job.active = False
            print("[bold red]\[-] Пропуск задачи[/bold red]")

    print(Panel(Text("Список задач составлен, начинаю бэкап", justify="center")))
    return jobs


def create_actions_from_list(actions: list) -> List[Action]:
    a: List[Action] = []

    for com in actions:
        a.append(Action(com["com"]["message"], com["com"]["cmd"]))

    return a


def yaml_loader(file: str) -> List[Job]:
    """Парсинг yaml файла с задачами бэкапа.


    See Also
    --------
    yamlchecker.com - сервис проверки yaml
    """
    jobs: list = []
    with open(file, "r") as conf:
        for task in yaml.safe_load_all(conf):
            if task:
                jobs.append(
                    Job(
                        name=task["name"],
                        dst_dir=task["dst_dir"],
                        actions=create_actions_from_list(task["actions"]),
                    )
                )

    return jobs


def main() -> None:
    parser = argparse.ArgumentParser(description="Скрипт для бэкапа")
    required_args = parser.add_argument_group("Обязательные аргументы")
    required_args.add_argument(
        "--file", type=str, required=True, help="файл с задачами в формате YAML"
    )
    optional_args = parser.add_argument_group("Необязательные аргументы")
    optional_args.add_argument(
        "--forceall",
        action="store_true",
        help="выполнить все задачи из списка без запроса подтверждения",
    )
    args = parser.parse_args()
    console = Console()
    jobs = yaml_loader(os.path.abspath(args.file))

    # logger.debug(jobs)
    jobs = ask_about_backups(jobs, args.forceall)

    for j in jobs:
        if j.active:
            run_job(j, console)

    # table = Table(title="Результаты бэкапов", box=box.SIMPLE)
    # table.add_column("Задача", justify="right", style="cyan", no_wrap=True)
    # table.add_column("Архив", style="magenta")
    # table.add_column("Статус", justify="right", style="green")

    # table.add_row(
    #     "Archives криппи и прочее",
    #     "/back/Archives/Archives-{date}-{time}",
    #     "ОК",
    # )
    # table.add_row(
    #     "Archivebox-emacs",
    #     "/back/Archives/Archivebox-emacs",
    #     "Пропуск",
    # )

    # console.print(table, justify="center")


if __name__ == "__main__":
    main()