py-backup/pyback.py

#!/usr/local/bin/python3.8
import os
from dataclasses import dataclass
from datetime import datetime
import yaml
import argparse
from typing import List
from rich import print
from rich.prompt import Prompt
from rich.padding import Padding
from rich.panel import Panel
from rich.text import Text
from rich.table import Table
from rich.console import Console
from rich import box


@dataclass
class Job:
    name: str
    dst_dir: str
    action: str
    active: bool = False


# Добавить обозначения {warn} {good} {error} {nice}, менять их на
# символы и красить вывод консоли
def run_job(job: Job, console: Console) -> None:
    """Запусить задачу бэкапа."""

    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)
    #  BOLDYELLOW="\e[1;33m[!]\e[0m"
    #  ENDCOLOR="\e[0m"
    cmd = (
        job.action.replace("{name}", job.name)
        .replace("{date}", date)
        .replace("{time}", time)
        .replace("{warn}", "\033[33m[!]\033[0m")
    )
    with console.status(
        f"[bold yellow]\[..][/bold yellow] [bold]{job.name}[/bold] начинаю ..."
    ):
        os.system(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 yaml_loader(file: str) -> List[Job]:
    jobs: list = []
    with open(file, "r") as conf:
        for task in yaml.safe_load_all(conf):
            try:
                jobs.append(
                    Job(
                        name=task["name"],
                        dst_dir=task["dst_dir"],
                        action=task["action"],
                    )
                )
            except AttributeError:
                print(
                    "Ваш список задач говно, попробуйте не ставить --- после последнего элемента"
                )
    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))

    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()