Compare commits

..

3 Commits

Author SHA1 Message Date
dw-0
1f9d4c823a fix(settings_menu): fix regression by checking for the correct condition (#773) 2026-01-31 11:13:37 +01:00
dw-0
c8df9427b3 fix(backup): improve reusability of backup service and enhance file handling
- Refactor `BackupService` instance management for better reuse across methods.
- Avoid redundant file backups by checking for existing files.
- Enhance directory backup logic to handle nested files and directories more efficiently.
- Standardize timestamp initialization for consistent time-based backup operations.
2026-01-31 10:59:53 +01:00
Théo Gaillard
5414aba299 fix(backup_service): backup methods use proper paths (#769) (#770)
* fix(backup_service): streamline backup methods for proper paths and add docstring

* fix(backup_service): add proper destination_path in verbose output

* fix(backup_service): replaces html headers with { } to avoid rendering

* nitpick(backup_service): correct variable name for backup destination path in logging
2026-01-29 18:52:48 +01:00
4 changed files with 49 additions and 25 deletions

View File

@@ -8,7 +8,7 @@
# ======================================================================= # # ======================================================================= #
from typing import List from typing import List, Optional
from components.klipper.klipper import Klipper from components.klipper.klipper import Klipper
from components.moonraker.moonraker import Moonraker from components.moonraker.moonraker import Moonraker
@@ -27,6 +27,7 @@ def run_client_config_removal(
client_config: BaseWebClientConfig, client_config: BaseWebClientConfig,
kl_instances: List[Klipper], kl_instances: List[Klipper],
mr_instances: List[Moonraker], mr_instances: List[Moonraker],
svc: Optional[BackupService] = None,
) -> Message: ) -> Message:
completion_msg = Message( completion_msg = Message(
title=f"{client_config.display_name} Removal Process completed", title=f"{client_config.display_name} Removal Process completed",
@@ -36,12 +37,15 @@ def run_client_config_removal(
if run_remove_routines(client_config.config_dir): if run_remove_routines(client_config.config_dir):
completion_msg.text.append(f"{client_config.display_name} removed") completion_msg.text.append(f"{client_config.display_name} removed")
BackupService().backup_printer_config_dir() if svc is None:
svc = BackupService()
svc.backup_moonraker_conf()
completion_msg = remove_moonraker_config_section( completion_msg = remove_moonraker_config_section(
completion_msg, client_config, mr_instances completion_msg, client_config, mr_instances
) )
svc.backup_printer_cfg()
completion_msg = remove_printer_config_section( completion_msg = remove_printer_config_section(
completion_msg, client_config, kl_instances completion_msg, client_config, kl_instances
) )

View File

@@ -41,6 +41,7 @@ def run_client_removal(
) )
mr_instances: List[Moonraker] = get_instances(Moonraker) mr_instances: List[Moonraker] = get_instances(Moonraker)
kl_instances: List[Klipper] = get_instances(Klipper) kl_instances: List[Klipper] = get_instances(Klipper)
svc = BackupService()
if backup_config: if backup_config:
version = "" version = ""
@@ -49,7 +50,6 @@ def run_client_removal(
with open(src.joinpath(".version"), "r") as v: with open(src.joinpath(".version"), "r") as v:
version = v.readlines()[0] version = v.readlines()[0]
svc = BackupService()
target_path = svc.backup_root.joinpath(f"{client.client_dir.name}_{version}") target_path = svc.backup_root.joinpath(f"{client.client_dir.name}_{version}")
success = svc.backup_file( success = svc.backup_file(
source_path=client.config_file, source_path=client.config_file,
@@ -67,7 +67,7 @@ def run_client_removal(
if remove_client_nginx_logs(client, kl_instances): if remove_client_nginx_logs(client, kl_instances):
completion_msg.text.append("● NGINX logs removed") completion_msg.text.append("● NGINX logs removed")
BackupService().backup_moonraker_conf() svc.backup_moonraker_conf()
section = f"update_manager {client_name}" section = f"update_manager {client_name}"
handled_instances: List[Moonraker] = remove_config_section( handled_instances: List[Moonraker] = remove_config_section(
section, mr_instances section, mr_instances
@@ -83,6 +83,7 @@ def run_client_removal(
client.client_config, client.client_config,
kl_instances, kl_instances,
mr_instances, mr_instances,
svc,
) )
if cfg_completion_msg.color == Color.GREEN: if cfg_completion_msg.color == Color.GREEN:
completion_msg.text.extend(cfg_completion_msg.text[1:]) completion_msg.text.extend(cfg_completion_msg.text[1:])

View File

@@ -100,11 +100,11 @@ class SettingsMenu(BaseMenu):
def trim_repo_url(repo: str) -> str: def trim_repo_url(repo: str) -> str:
return repo.replace(".git", "").replace("https://", "").replace("git@", "") return repo.replace(".git", "").replace("https://", "").replace("git@", "")
if not klipper_status.repo == "-": if klipper_status.repo:
url = trim_repo_url(klipper_status.repo_url) url = trim_repo_url(klipper_status.repo_url)
self.kl_repo_url = Color.apply(url, Color.CYAN) self.kl_repo_url = Color.apply(url, Color.CYAN)
self.kl_branch = Color.apply(klipper_status.branch, Color.CYAN) self.kl_branch = Color.apply(klipper_status.branch, Color.CYAN)
if not moonraker_status.repo == "-": if moonraker_status.repo:
url = trim_repo_url(moonraker_status.repo_url) url = trim_repo_url(moonraker_status.repo_url)
self.mr_repo_url = Color.apply(url, Color.CYAN) self.mr_repo_url = Color.apply(url, Color.CYAN)
self.mr_branch = Color.apply(moonraker_status.branch, Color.CYAN) self.mr_branch = Color.apply(moonraker_status.branch, Color.CYAN)

View File

@@ -22,6 +22,7 @@ from utils.instance_utils import get_instances
class BackupService: class BackupService:
def __init__(self): def __init__(self):
self._backup_root = Path.home().joinpath("kiauh_backups") self._backup_root = Path.home().joinpath("kiauh_backups")
self._timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
@property @property
def backup_root(self) -> Path: def backup_root(self) -> Path:
@@ -29,7 +30,7 @@ class BackupService:
@property @property
def timestamp(self) -> str: def timestamp(self) -> str:
return datetime.now().strftime("%Y%m%d-%H%M%S") return self._timestamp
################################################ ################################################
# GENERIC BACKUP METHODS # GENERIC BACKUP METHODS
@@ -68,10 +69,15 @@ class BackupService:
backup_dir = self._backup_root.joinpath(target_path) backup_dir = self._backup_root.joinpath(target_path)
backup_dir.mkdir(parents=True, exist_ok=True) backup_dir.mkdir(parents=True, exist_ok=True)
shutil.copy2(source_path, backup_dir.joinpath(filename)) target_path = backup_dir.joinpath(filename)
if target_path.exists():
Logger.print_info(f"File '{target_path}' already exists. Skipping ...")
return True
shutil.copy2(source_path, target_path)
Logger.print_ok( Logger.print_ok(
f"Successfully backed up '{source_path}' to '{backup_dir}'" f"Successfully backed up '{source_path}' to '{target_path}'"
) )
return True return True
@@ -111,14 +117,25 @@ class BackupService:
if backup_path.exists(): if backup_path.exists():
Logger.print_info(f"Reusing existing backup directory '{backup_path}'") Logger.print_info(f"Reusing existing backup directory '{backup_path}'")
for item in source_path.rglob("*"):
shutil.copytree( relative_path = item.relative_to(source_path)
source_path, target_item = backup_path.joinpath(relative_path)
backup_path, if item.is_file():
dirs_exist_ok=True, if not target_item.exists():
symlinks=True, target_item.parent.mkdir(parents=True, exist_ok=True)
ignore_dangling_symlinks=True, shutil.copy2(item, target_item)
) else:
Logger.print_info(f"File '{target_item}' already exists. Skipping...")
elif item.is_dir():
target_item.mkdir(parents=True, exist_ok=True)
else:
shutil.copytree(
source_path,
backup_path,
dirs_exist_ok=True,
symlinks=True,
ignore_dangling_symlinks=True,
)
Logger.print_ok( Logger.print_ok(
f"Successfully backed up '{source_path}' to '{backup_path}'" f"Successfully backed up '{source_path}' to '{backup_path}'"
@@ -134,27 +151,29 @@ class BackupService:
################################################ ################################################
def backup_printer_cfg(self): def backup_printer_cfg(self):
"""Backup printer.cfg files of all Klipper instances.
Files are backed up to:
{backup_root}/{instance_data_dir_name}/printer_{timestamp}.cfg
"""
klipper_instances: List[Klipper] = get_instances(Klipper) klipper_instances: List[Klipper] = get_instances(Klipper)
for instance in klipper_instances: for instance in klipper_instances:
target_path: Path = self._backup_root.joinpath( target_path: Path = self._backup_root.joinpath(instance.data_dir.name)
instance.data_dir.name, f"config_{self.timestamp}"
)
self.backup_file( self.backup_file(
source_path=instance.cfg_file, source_path=instance.cfg_file,
target_path=target_path, target_path=target_path,
target_name=instance.cfg_file.name,
) )
def backup_moonraker_conf(self): def backup_moonraker_conf(self):
"""Backup moonraker.conf files of all Moonraker instances.
Files are backed up to:
{backup_root}/{instance_data_dir_name}/moonraker_{timestamp}.conf
"""
moonraker_instances: List[Moonraker] = get_instances(Moonraker) moonraker_instances: List[Moonraker] = get_instances(Moonraker)
for instance in moonraker_instances: for instance in moonraker_instances:
target_path: Path = self._backup_root.joinpath( target_path: Path = self._backup_root.joinpath(instance.data_dir.name)
instance.data_dir.name, f"config_{self.timestamp}"
)
self.backup_file( self.backup_file(
source_path=instance.cfg_file, source_path=instance.cfg_file,
target_path=target_path, target_path=target_path,
target_name=instance.cfg_file.name,
) )
def backup_printer_config_dir(self) -> None: def backup_printer_config_dir(self) -> None: