jeka/kiauh
1
0
Fork 0
mirror of https://github.com/dw-0/kiauh.git synced 2025-12-13 02:24:27 +05:00
Code Issues Packages Projects Releases Wiki Activity

feat: rework completion message for webclient remove process

Browse Source 
Signed-off-by: Dominik Willner <th33xitus@gmail.com>
This commit is contained in:
dw-0
2024-10-27 18:41:49 +01:00
parent 12127efa21
commit 1ca1e8ff6f
7 changed files with 128 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
kiauh/components/klipper/menus/klipper_remove_menu.py
View File

@@ -106,3 +106,4 @@ class KlipperRemoveMenu(BaseMenu):
self.remove_klipper_service = False
self.remove_klipper_dir = False
self.remove_klipper_env = False
self.select_state = False

54
kiauh/components/webui_client/client_config/client_config_remove.py
View File

@@ -14,6 +14,8 @@ from components.klipper.klipper import Klipper
from components.moonraker.moonraker import Moonraker
from components.webui_client.base_data import BaseWebClientConfig
from core.logger import Logger
from core.services.message_service import Message
from core.types.color import Color
from utils.config_utils import remove_config_section
from utils.fs_utils import run_remove_routines
from utils.instance_utils import get_instances
@@ -23,21 +25,49 @@ def run_client_config_removal(
client_config: BaseWebClientConfig,
kl_instances: List[Klipper],
mr_instances: List[Moonraker],
) -> None:
remove_client_config_dir(client_config)
remove_client_config_symlink(client_config)
remove_config_section(f"update_manager {client_config.name}", mr_instances)
remove_config_section(client_config.config_section, kl_instances)
def remove_client_config_dir(client_config: BaseWebClientConfig) -> None:
) -> Message:
completion_msg = Message(
title=f"{client_config.display_name} Removal Process completed",
color=Color.GREEN,
)
Logger.print_status(f"Removing {client_config.display_name} ...")
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")
def remove_client_config_symlink(client_config: BaseWebClientConfig) -> None:
instances: List[Klipper] = get_instances(Klipper)
handled_configs = []
for instance in instances:
run_remove_routines(
if run_remove_routines(
instance.base.cfg_dir.joinpath(client_config.config_filename)
):
handled_configs.append(instance)
if handled_configs:
instance_names = [i.service_file_path.stem for i in handled_configs]
completion_msg.text.append(
f"● {client_config.display_name} removed from instances: {', '.join(instance_names)}"
)
mr_section = f"update_manager {client_config.name}"
handled_mr_instances = remove_config_section(mr_section, mr_instances)
if handled_mr_instances:
instance_names = [i.service_file_path.stem for i in handled_mr_instances]
completion_msg.text.append(
f"● Moonraker config section '{mr_section}' removed for instance: {', '.join(instance_names)}"
)
kl_section = client_config.config_section
handled_kl_instances = remove_config_section(kl_section, kl_instances)
if handled_kl_instances:
instance_names = [i.service_file_path.stem for i in handled_kl_instances]
completion_msg.text.append(
f"● Klipper config section '{mr_section}' removed for instance: {', '.join(instance_names)}"
)
if not completion_msg.text:
completion_msg.color = Color.YELLOW
completion_msg.centered = True
completion_msg.text.append("Nothing to remove.")
else:
completion_msg.text.insert(0, "The following actions were performed:")
return completion_msg

62
kiauh/components/webui_client/client_remove.py
View File

@@ -44,27 +44,36 @@ def run_client_removal(
if backup_config:
bm = BackupManager()
bm.backup_file(client.config_file)
completion_msg.text.append(f"● {client.config_file.name} backup created")
if bm.backup_file(client.config_file):
completion_msg.text.append(f"● {client.config_file.name} backup created")
if remove_client:
client_name = client.name
remove_client_dir(client)
remove_client_nginx_config(client_name)
remove_client_nginx_logs(client, kl_instances)
if remove_client_dir(client):
completion_msg.text.append(f"● {client.display_name} removed")
if remove_client_nginx_config(client_name):
completion_msg.text.append("● NGINX config removed")
if remove_client_nginx_logs(client, kl_instances):
completion_msg.text.append("● NGINX logs removed")
section = f"update_manager {client_name}"
remove_config_section(section, mr_instances)
completion_msg.text.append(f"● {client.display_name} removed")
handled_instances: List[Moonraker] = remove_config_section(
section, mr_instances
)
if handled_instances:
names = [i.service_file_path.stem for i in handled_instances]
completion_msg.text.append(
f"● Moonraker config section '{section}' removed for instance: {', '.join(names)}"
)
if remove_client_cfg:
# todo: return a Message here as well to display correct actions taken
run_client_config_removal(
cfg_completion_msg = run_client_config_removal(
client.client_config,
kl_instances,
mr_instances,
)
completion_msg.text.append(f"● {client.client_config.display_name} removed")
if cfg_completion_msg.color == Color.GREEN:
completion_msg.text.extend(cfg_completion_msg.text[1:])
if not completion_msg.text:
completion_msg.color = Color.YELLOW
@@ -76,29 +85,28 @@ def run_client_removal(
return completion_msg
def remove_client_dir(client: BaseWebClient) -> None:
def remove_client_dir(client: BaseWebClient) -> bool:
Logger.print_status(f"Removing {client.display_name} ...")
run_remove_routines(client.client_dir)
return run_remove_routines(client.client_dir)
def remove_client_nginx_config(name: str) -> None:
def remove_client_nginx_config(name: str) -> bool:
Logger.print_status(f"Removing NGINX config for {name.capitalize()} ...")
remove_with_sudo(NGINX_SITES_AVAILABLE.joinpath(name))
remove_with_sudo(NGINX_SITES_ENABLED.joinpath(name))
return remove_with_sudo(
[
NGINX_SITES_AVAILABLE.joinpath(name),
NGINX_SITES_ENABLED.joinpath(name),
]
)
def remove_client_nginx_logs(client: BaseWebClient, instances: List[Klipper]) -> None:
def remove_client_nginx_logs(client: BaseWebClient, instances: List[Klipper]) -> bool:
Logger.print_status(f"Removing NGINX logs for {client.display_name} ...")
remove_with_sudo(client.nginx_access_log)
remove_with_sudo(client.nginx_error_log)
files = [client.nginx_access_log, client.nginx_error_log]
if instances:
for instance in instances:
files.append(instance.base.log_dir.joinpath(client.nginx_access_log.name))
files.append(instance.base.log_dir.joinpath(client.nginx_error_log.name))
if not instances:
return
for instance in instances:
run_remove_routines(
instance.base.log_dir.joinpath(client.nginx_access_log.name)
)
run_remove_routines(instance.base.log_dir.joinpath(client.nginx_error_log.name))
return remove_with_sudo(files)

23
kiauh/components/webui_client/menus/client_remove_menu.py
View File

@@ -31,7 +31,7 @@ class ClientRemoveMenu(BaseMenu):
self.remove_client: bool = False
self.remove_client_cfg: bool = False
self.backup_config_json: bool = False
self.selection_state: bool = False
self.select_state: bool = False
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
from core.menus.remove_menu import RemoveMenu
@@ -57,13 +57,14 @@ class ClientRemoveMenu(BaseMenu):
o1 = checked if self.remove_client else unchecked
o2 = checked if self.remove_client_cfg else unchecked
o3 = checked if self.backup_config_json else unchecked
sel_state = f"{'Select'if not self.select_state else 'Deselect'} everything"
menu = textwrap.dedent(
f"""
╟───────────────────────────────────────────────────────╢
║ Enter a number and hit enter to select / deselect ║
║ the specific option for removal. ║
╟───────────────────────────────────────────────────────╢
║ a) {self._get_selection_state_str():37} ║
║ a) {sel_state:49} ║
╟───────────────────────────────────────────────────────╢
║ 1) {o1} Remove {client_name:16} ║
║ 2) {o2} Remove {client_config_name:24} ║
@@ -76,10 +77,10 @@ class ClientRemoveMenu(BaseMenu):
print(menu, end="")
def toggle_all(self, **kwargs) -> None:
self.selection_state = not self.selection_state
self.remove_client = self.selection_state
self.remove_client_cfg = self.selection_state
self.backup_config_json = self.selection_state
self.select_state = not self.select_state
self.remove_client = self.select_state
self.remove_client_cfg = self.select_state
self.backup_config_json = self.select_state
def toggle_rm_client(self, **kwargs) -> None:
self.remove_client = not self.remove_client
@@ -110,12 +111,4 @@ class ClientRemoveMenu(BaseMenu):
self.remove_client = False
self.remove_client_cfg = False
self.backup_config_json = False
def _get_selection_state_str(self) -> str:
return (
"Select everything" if not self.selection_state else "Deselect everything"
)
def _go_back(self, **kwargs) -> None:
if self.previous_menu is not None:
self.previous_menu().run()
self.select_state = False

9
kiauh/core/backup_manager/backup_manager.py
View File

@@ -44,12 +44,14 @@ class BackupManager:
def ignore_folders(self, value: List[str]):
self._ignore_folders = value
def backup_file(self, file: Path, target: Path | None = None, custom_filename=None):
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
return False
target = self.backup_root_dir if target is None else target
@@ -62,10 +64,13 @@ class BackupManager:
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

8
kiauh/utils/config_utils.py
View File

@@ -78,7 +78,10 @@ def add_config_section_at_top(section: str, instances: List[InstanceType]) -> No
Logger.print_ok("OK!")
def remove_config_section(section: str, instances: List[InstanceType]) -> None:
def remove_config_section(
section: str, instances: List[InstanceType]
) -> List[InstanceType]:
removed_from: List[instances] = []
for instance in instances:
cfg_file = instance.cfg_file
Logger.print_status(f"Remove section '[{section}]' from '{cfg_file}' ...")
@@ -96,4 +99,7 @@ def remove_config_section(section: str, instances: List[InstanceType]) -> None:
scp.remove_section(section)
scp.write_file(cfg_file)
removed_from.append(instance)
Logger.print_ok("OK!")
return removed_from

40
kiauh/utils/fs_utils.py
View File

@@ -13,7 +13,7 @@ from __future__ import annotations
import re
import shutil
from pathlib import Path
from subprocess import DEVNULL, PIPE, CalledProcessError, check_output, run
from subprocess import DEVNULL, PIPE, CalledProcessError, call, check_output, run
from typing import List
from zipfile import ZipFile
@@ -53,13 +53,28 @@ def create_symlink(source: Path, target: Path, sudo=False) -> None:
raise
def remove_with_sudo(file: Path) -> None:
try:
cmd = ["sudo", "rm", "-rf", file.as_posix()]
run(cmd, stderr=PIPE, check=True)
except CalledProcessError as e:
Logger.print_error(f"Failed to remove {file}: {e}")
raise
def remove_with_sudo(files: Path | List[Path]) -> bool:
_files = []
_removed = []
if isinstance(files, list):
_files = files
else:
_files.append(files)
for f in _files:
try:
cmd = ["sudo", "find", f.as_posix()]
if call(cmd, stderr=DEVNULL, stdout=DEVNULL) == 1:
Logger.print_info(f"File '{f}' does not exist. Skipped ...")
continue
cmd = ["sudo", "rm", "-rf", f.as_posix()]
run(cmd, stderr=PIPE, check=True)
Logger.print_ok(f"File '{f}' was successfully removed!")
_removed.append(f)
except CalledProcessError as e:
Logger.print_error(f"Error removing file '{f}': {e}")
return len(_removed) > 0
@deprecated(info="Use remove_with_sudo instead", replaced_by=remove_with_sudo)
@@ -84,16 +99,17 @@ def run_remove_routines(file: Path) -> bool:
elif file.is_file() or file.is_symlink():
file.unlink()
else:
raise OSError(f"File '{file}' is neither a file nor a directory!")
Logger.print_error(f"File '{file}' is neither a file nor a directory!")
return False
Logger.print_ok(f"File '{file}' was successfully removed!")
return True
except OSError as e:
Logger.print_error(f"Unable to delete '{file}':\n{e}")
try:
Logger.print_info("Trying to remove with sudo ...")
remove_with_sudo(file)
Logger.print_ok(f"File '{file}' was successfully removed!")
return True
if remove_with_sudo(file):
Logger.print_ok(f"File '{file}' was successfully removed!")
return True
except CalledProcessError as e:
Logger.print_error(f"Error deleting '{file}' with sudo:\n{e}")
Logger.print_error("Remove this directory manually!")