Compare commits

..

6 Commits

Author SHA1 Message Date
dw-0
5b56959293 fix(settings_menu): fix regression by checking for the correct condition 2026-01-31 11:12: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
Théo Gaillard
80948edbb4 fix(gcode_shell_cmd): update comment to clarify config directory usage (#767) 2026-01-19 17:04:28 +01:00
Théo Gaillard
a455edba93 docs(fs_utils): add the documentation for create_symlink (#768) 2026-01-19 17:03:04 +01:00
Théo Gaillard
810ab3a2fa fix(fs-utils): enhance check_file_exist to support symlink resolution (#766) 2026-01-19 17:02:18 +01:00
6 changed files with 64 additions and 28 deletions

View File

@@ -8,7 +8,7 @@
# ======================================================================= #
from typing import List
from typing import List, Optional
from components.klipper.klipper import Klipper
from components.moonraker.moonraker import Moonraker
@@ -27,6 +27,7 @@ def run_client_config_removal(
client_config: BaseWebClientConfig,
kl_instances: List[Klipper],
mr_instances: List[Moonraker],
svc: Optional[BackupService] = None,
) -> Message:
completion_msg = Message(
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):
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, client_config, mr_instances
)
svc.backup_printer_cfg()
completion_msg = remove_printer_config_section(
completion_msg, client_config, kl_instances
)

View File

@@ -41,6 +41,7 @@ def run_client_removal(
)
mr_instances: List[Moonraker] = get_instances(Moonraker)
kl_instances: List[Klipper] = get_instances(Klipper)
svc = BackupService()
if backup_config:
version = ""
@@ -49,7 +50,6 @@ def run_client_removal(
with open(src.joinpath(".version"), "r") as v:
version = v.readlines()[0]
svc = BackupService()
target_path = svc.backup_root.joinpath(f"{client.client_dir.name}_{version}")
success = svc.backup_file(
source_path=client.config_file,
@@ -67,7 +67,7 @@ def run_client_removal(
if remove_client_nginx_logs(client, kl_instances):
completion_msg.text.append("● NGINX logs removed")
BackupService().backup_moonraker_conf()
svc.backup_moonraker_conf()
section = f"update_manager {client_name}"
handled_instances: List[Moonraker] = remove_config_section(
section, mr_instances
@@ -83,6 +83,7 @@ def run_client_removal(
client.client_config,
kl_instances,
mr_instances,
svc,
)
if cfg_completion_msg.color == Color.GREEN:
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:
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)
self.kl_repo_url = Color.apply(url, 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)
self.mr_repo_url = Color.apply(url, 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:
def __init__(self):
self._backup_root = Path.home().joinpath("kiauh_backups")
self._timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
@property
def backup_root(self) -> Path:
@@ -29,7 +30,7 @@ class BackupService:
@property
def timestamp(self) -> str:
return datetime.now().strftime("%Y%m%d-%H%M%S")
return self._timestamp
################################################
# GENERIC BACKUP METHODS
@@ -68,10 +69,15 @@ class BackupService:
backup_dir = self._backup_root.joinpath(target_path)
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(
f"Successfully backed up '{source_path}' to '{backup_dir}'"
f"Successfully backed up '{source_path}' to '{target_path}'"
)
return True
@@ -111,14 +117,25 @@ class BackupService:
if backup_path.exists():
Logger.print_info(f"Reusing existing backup directory '{backup_path}'")
shutil.copytree(
source_path,
backup_path,
dirs_exist_ok=True,
symlinks=True,
ignore_dangling_symlinks=True,
)
for item in source_path.rglob("*"):
relative_path = item.relative_to(source_path)
target_item = backup_path.joinpath(relative_path)
if item.is_file():
if not target_item.exists():
target_item.parent.mkdir(parents=True, exist_ok=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(
f"Successfully backed up '{source_path}' to '{backup_path}'"
@@ -134,27 +151,29 @@ class BackupService:
################################################
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)
for instance in klipper_instances:
target_path: Path = self._backup_root.joinpath(
instance.data_dir.name, f"config_{self.timestamp}"
)
target_path: Path = self._backup_root.joinpath(instance.data_dir.name)
self.backup_file(
source_path=instance.cfg_file,
target_path=target_path,
target_name=instance.cfg_file.name,
)
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)
for instance in moonraker_instances:
target_path: Path = self._backup_root.joinpath(
instance.data_dir.name, f"config_{self.timestamp}"
)
target_path: Path = self._backup_root.joinpath(instance.data_dir.name)
self.backup_file(
source_path=instance.cfg_file,
target_path=target_path,
target_name=instance.cfg_file.name,
)
def backup_printer_config_dir(self) -> None:

View File

@@ -97,7 +97,7 @@ class GcodeShellCmdExtension(BaseExtension):
def install_example_cfg(self, instances: List[Klipper]):
cfg_dirs = [instance.base.cfg_dir for instance in instances]
# copy extension to klippy/extras
# copy extension to config directories
for cfg_dir in cfg_dirs:
Logger.print_status(f"Create shell_command.cfg in '{cfg_dir}' ...")
if check_file_exist(cfg_dir.joinpath("shell_command.cfg")):

View File

@@ -24,13 +24,16 @@ from core.logger import Logger
def check_file_exist(file_path: Path, sudo=False) -> bool:
"""
Helper function for checking the existence of a file |
Helper function for checking the existence of a file.
Also works with symlinks (returns False if broken) |
:param file_path: the absolute path of the file to check
:param sudo: use sudo if required
:return: True, if file exists, otherwise False
"""
if sudo:
command = ["sudo", "find", file_path.as_posix()]
# -L forces find to follow symlinks
# -maxdepth = 0 avoids losing time if `file_path` is a directory
command = ["sudo", "find", "-L", file_path.as_posix(), "-maxdepth", "0"]
try:
check_output(command, stderr=DEVNULL)
return True
@@ -44,7 +47,16 @@ def check_file_exist(file_path: Path, sudo=False) -> bool:
def create_symlink(source: Path, target: Path, sudo=False) -> None:
"""
Helper function to create a symlink from source to target
If the target file exists, it will be overwritten. |
:param source: the source file/directory
:param target: the target file/directory
:param sudo: use sudo if required
:return: None
"""
try:
# -f forcibly creates/overwrites the symlink
cmd = ["ln", "-sf", source.as_posix(), target.as_posix()]
if sudo:
cmd.insert(0, "sudo")