You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
126 lines
4.2 KiB
126 lines
4.2 KiB
#!/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 |
|
|
|
|
|
@dataclass |
|
class Job: |
|
name: str |
|
dst_dir: str |
|
action: str |
|
active: bool = False |
|
|
|
|
|
# Добавить обозначения {warn} {good} {error} {nice}, менять их на |
|
# символы и красить вывод консоли |
|
def run_job(job: Job) -> None: |
|
"""Запусить задачу бэкапа.""" |
|
print(f"Начинаю {job.name}") |
|
|
|
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) |
|
|
|
cmd = ( |
|
job.action.replace("{name}", job.name) |
|
.replace("{date}", date) |
|
.replace("{time}", time) |
|
) |
|
os.system(cmd) |
|
|
|
|
|
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 |
|
(задача не активна). |
|
""" |
|
|
|
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]") |
|
|
|
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 |
|
|
|
|
|
if __name__ == "__main__": |
|
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() |
|
|
|
jobs = yaml_loader(os.path.abspath(args.file)) |
|
|
|
# если флаг принудительного бэкапа всего, то мы прохоидм в цикле |
|
# все джобы и ставим их active в true |
|
if not args.forceall: |
|
ask_about_backups(jobs) |
|
print("-----------------------") |
|
else: |
|
for j in jobs: |
|
j.active = True |
|
|
|
for j in jobs: |
|
if j.active: |
|
run_job(j)
|
|
|