#!/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()