jeka/kiauh
1
0
Fork 0
mirror of https://github.com/dw-0/kiauh.git synced 2026-05-08 10:18:18 +05:00
Code Issues Packages Projects Releases Wiki Activity

refactor: replace backup_manager with backup_service

Browse Source
This commit is contained in:
dw-0
2025-09-28 09:41:07 +02:00
parent 1ed1e0fc4c
commit aff63665de
24 changed files with 252 additions and 256 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kiauh/core/backup_manager/__init__.py
-12
View File
@@ -1,12 +0,0 @@
# ======================================================================= #
# Copyright (C) 2020 - 2025 Dominik Willner <th33xitus@gmail.com> #
# #
# This file is part of KIAUH - Klipper Installation And Update Helper #
# https://github.com/dw-0/kiauh #
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from pathlib import Path
BACKUP_ROOT_DIR = Path.home().joinpath("kiauh-backups")
kiauh/core/backup_manager/backup_manager.py
-108
View File
@@ -1,108 +0,0 @@
# ======================================================================= #
# Copyright (C) 2020 - 2025 Dominik Willner <th33xitus@gmail.com> #
# #
# This file is part of KIAUH - Klipper Installation And Update Helper #
# https://github.com/dw-0/kiauh #
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import shutil
from pathlib import Path
from typing import List
from core.backup_manager import BACKUP_ROOT_DIR
from core.logger import Logger
from utils.common import get_current_date
class BackupManagerException(Exception):
pass
# noinspection PyUnusedLocal
# noinspection PyMethodMayBeStatic
class BackupManager:
def __init__(self, backup_root_dir: Path = BACKUP_ROOT_DIR):
self._backup_root_dir: Path = backup_root_dir
self._ignore_folders: List[str] = []
@property
def backup_root_dir(self) -> Path:
return self._backup_root_dir
@backup_root_dir.setter
def backup_root_dir(self, value: Path):
self._backup_root_dir = value
@property
def ignore_folders(self) -> List[str]:
return self._ignore_folders
@ignore_folders.setter
def ignore_folders(self, value: List[str]):
self._ignore_folders = value
def backup_file(
self, file: Path, target: Path | None = None, custom_filename=None
) -> bool:
Logger.print_status(f"Creating backup of {file} ...")
if not file.exists():
Logger.print_info("File does not exist! Skipping ...")
return False
target = self.backup_root_dir if target is None else target
if Path(file).is_file():
date = get_current_date().get("date")
time = get_current_date().get("time")
filename = f"{file.stem}-{date}-{time}{file.suffix}"
filename = custom_filename if custom_filename is not None else filename
try:
Path(target).mkdir(exist_ok=True)
shutil.copyfile(file, target.joinpath(filename))
Logger.print_ok("Backup successful!")
return True
except OSError as e:
Logger.print_error(f"Unable to backup '{file}':\n{e}")
return False
else:
Logger.print_info(f"File '{file}' not found ...")
return False
def backup_directory(
self, name: str, source: Path, target: Path | None = None
) -> Path | None:
Logger.print_status(f"Creating backup of {name} in {target} ...")
if source is None or not Path(source).exists():
Logger.print_info("Source directory does not exist! Skipping ...")
return None
target = self.backup_root_dir if target is None else target
try:
date = get_current_date().get("date")
time = get_current_date().get("time")
backup_target = target.joinpath(f"{name.lower()}-{date}-{time}")
shutil.copytree(
source,
backup_target,
ignore=self.ignore_folders_func,
ignore_dangling_symlinks=True,
)
Logger.print_ok("Backup successful!")
return backup_target
except OSError as e:
Logger.print_error(f"Unable to backup directory '{source}':\n{e}")
raise BackupManagerException(f"Unable to backup directory '{source}':\n{e}")
def ignore_folders_func(self, dirpath, filenames) -> List[str]:
return (
[f for f in filenames if f in self._ignore_folders]
if self._ignore_folders
else []
)
kiauh/core/constants.py
-3
View File
@@ -11,8 +11,6 @@ import os
import pwd
from pathlib import Path
from core.backup_manager import BACKUP_ROOT_DIR
# global dependencies
GLOBAL_DEPS = ["git", "wget", "curl", "unzip", "dfu-util", "python3-virtualenv"]
@@ -24,7 +22,6 @@ CURRENT_USER = pwd.getpwuid(os.getuid())[0]
# dirs
SYSTEMD = Path("/etc/systemd/system")
PRINTER_DATA_BACKUP_DIR = BACKUP_ROOT_DIR.joinpath("printer-data-backups")
NGINX_SITES_AVAILABLE = Path("/etc/nginx/sites-available")
NGINX_SITES_ENABLED = Path("/etc/nginx/sites-enabled")
NGINX_CONFD = Path("/etc/nginx/conf.d")
kiauh/core/services/backup_service.py
+111
View File
@@ -0,0 +1,111 @@
# ======================================================================= #
# Copyright (C) 2020 - 2025 Dominik Willner <th33xitus@gmail.com> #
# #
# This file is part of KIAUH - Klipper Installation And Update Helper #
# https://github.com/dw-0/kiauh #
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import shutil
from datetime import datetime
from pathlib import Path
from typing import Optional
from core.logger import Logger
class BackupService:
def __init__(self):
self._backup_root = Path.home().joinpath("kiauh_backups")
@property
def backup_root(self) -> Path:
return self._backup_root
def backup_file(
self,
source_path: Path,
target_path: Optional[Path | str] = None,
target_name: Optional[str] = None,
) -> bool:
source_path = Path(source_path)
Logger.print_status(f"Creating backup of {source_path} ...")
if not source_path.exists():
Logger.print_info(
f"File '{source_path}' does not exist! Skipping backup..."
)
return False
if not source_path.is_file():
Logger.print_info(f"'{source_path}' is not a file! Skipping backup...")
return False
try:
self._backup_root.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
filename = (
target_name or f"{source_path.stem}_{timestamp}{source_path.suffix}"
)
if target_path is not None:
backup_path = self._backup_root.joinpath(target_path, filename)
else:
backup_path = self._backup_root.joinpath(filename)
backup_path.mkdir(parents=True, exist_ok=True)
shutil.copy2(source_path, backup_path)
Logger.print_ok(
f"Successfully backed up '{source_path}' to '{backup_path}'"
)
return True
except Exception as e:
Logger.print_error(f"Failed to backup '{source_path}': {e}")
return False
def backup_directory(
self,
source_path: Path,
backup_name: str,
target_path: Optional[Path | str] = None,
) -> Optional[Path]:
source_path = Path(source_path)
Logger.print_status(f"Creating backup of {source_path} ...")
if not source_path.exists():
Logger.print_info(
f"Directory '{source_path}' does not exist! Skipping backup..."
)
return None
if not source_path.is_dir():
Logger.print_info(f"'{source_path}' is not a directory! Skipping backup...")
return None
try:
self._backup_root.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
backup_dir_name = f"{backup_name}_{timestamp}"
if target_path is not None:
backup_path = self._backup_root.joinpath(target_path, backup_dir_name)
else:
backup_path = self._backup_root.joinpath(backup_dir_name)
shutil.copytree(source_path, backup_path)
Logger.print_ok(
f"Successfully backed up '{source_path}' to '{backup_path}'"
)
return backup_path
except Exception as e:
Logger.print_error(f"Failed to backup directory '{source_path}': {e}")
return None
kiauh/core/settings/kiauh_settings.py
+3 -3
View File
@@ -14,8 +14,8 @@ from typing import Any, Callable, List, TypeVar
from components.klipper import KLIPPER_REPO_URL
from components.moonraker import MOONRAKER_REPO_URL
from core.backup_manager.backup_manager import BackupManager
from core.logger import DialogType, Logger
from core.services.backup_service import BackupService
from core.submodules.simple_config_parser.src.simple_config_parser.simple_config_parser import (
SimpleConfigParser,
)
@@ -374,8 +374,8 @@ class KiauhSettings:
kill()
def _migrate_repo_config(self) -> None:
bm = BackupManager()
if not bm.backup_file(CUSTOM_CFG):
svc = BackupService()
if not svc.backup_file(CUSTOM_CFG):
Logger.print_dialog(
DialogType.ERROR,
[