From acde067e685d8868c92c49be4e82a66c8bba18da Mon Sep 17 00:00:00 2001 From: dw-0 Date: Sun, 4 Aug 2024 18:36:04 +0200 Subject: [PATCH] refactor(extensions): refactor telegram bot extension Signed-off-by: Dominik Willner --- kiauh/extensions/telegram_bot/__init__.py | 28 ++++ .../telegram_bot/moonraker_telegram_bot.py | 153 ++++++++---------- .../moonraker_telegram_bot_extension.py | 44 +++-- 3 files changed, 114 insertions(+), 111 deletions(-) diff --git a/kiauh/extensions/telegram_bot/__init__.py b/kiauh/extensions/telegram_bot/__init__.py index e69de29..4dd030f 100644 --- a/kiauh/extensions/telegram_bot/__init__.py +++ b/kiauh/extensions/telegram_bot/__init__.py @@ -0,0 +1,28 @@ +# ======================================================================= # +# Copyright (C) 2020 - 2024 Dominik Willner # +# # +# 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 + +MODULE_PATH = Path(__file__).resolve().parent + +# repo +TG_BOT_REPO = "https://github.com/nlef/moonraker-telegram-bot.git" + +# names +TG_BOT_CFG_NAME = "telegram.conf" +TG_BOT_LOG_NAME = "telegram.log" +TG_BOT_SERVICE_NAME = "moonraker-telegram-bot.service" +TG_BOT_ENV_FILE_NAME = "moonraker-telegram-bot.env" + +# directories +TG_BOT_DIR = Path.home().joinpath("moonraker-telegram-bot") +TG_BOT_ENV = Path.home().joinpath("moonraker-telegram-bot-env") + +# files +TG_BOT_SERVICE_TEMPLATE = MODULE_PATH.joinpath(f"assets/{TG_BOT_SERVICE_NAME}") +TG_BOT_ENV_FILE_TEMPLATE = MODULE_PATH.joinpath(f"assets/{TG_BOT_ENV_FILE_NAME}") diff --git a/kiauh/extensions/telegram_bot/moonraker_telegram_bot.py b/kiauh/extensions/telegram_bot/moonraker_telegram_bot.py index cd113d5..47d7177 100644 --- a/kiauh/extensions/telegram_bot/moonraker_telegram_bot.py +++ b/kiauh/extensions/telegram_bot/moonraker_telegram_bot.py @@ -6,149 +6,126 @@ # # # This file may be distributed under the terms of the GNU GPLv3 license # # ======================================================================= # +from __future__ import annotations +from dataclasses import dataclass from pathlib import Path -from subprocess import DEVNULL, CalledProcessError, run -from typing import List +from subprocess import CalledProcessError, run from core.instance_manager.base_instance import BaseInstance -from utils.constants import SYSTEMD +from extensions.telegram_bot import ( + TG_BOT_CFG_NAME, + TG_BOT_DIR, + TG_BOT_ENV, + TG_BOT_ENV_FILE_NAME, + TG_BOT_ENV_FILE_TEMPLATE, + TG_BOT_LOG_NAME, + TG_BOT_SERVICE_TEMPLATE, +) from utils.logger import Logger -MODULE_PATH = Path(__file__).resolve().parent - -TELEGRAM_BOT_DIR = Path.home().joinpath("moonraker-telegram-bot") -TELEGRAM_BOT_ENV = Path.home().joinpath("moonraker-telegram-bot-env") -TELEGRAM_BOT_REPO = "https://github.com/nlef/moonraker-telegram-bot.git" - # noinspection PyMethodMayBeStatic -# todo: make this to a dataclass +@dataclass class MoonrakerTelegramBot(BaseInstance): - @classmethod - def blacklist(cls) -> List[str]: - return ["None", "mcu"] + bot_dir: Path = TG_BOT_DIR + env_dir: Path = TG_BOT_ENV + cfg_file: Path | None = None + log: Path | None = None def __init__(self, suffix: str = ""): - super().__init__(instance_type=self, suffix=suffix) - self.bot_dir: Path = TELEGRAM_BOT_DIR - self.env_dir: Path = TELEGRAM_BOT_ENV - self._cfg_file = self.cfg_dir.joinpath("telegram.conf") - self._log = self.log_dir.joinpath("telegram.log") - self._assets_dir = MODULE_PATH.joinpath("assets") + super().__init__(suffix=suffix) - @property - def cfg_file(self) -> Path: - return self._cfg_file - - @property - def log(self) -> Path: - return self._log + def __post_init__(self): + super().__post_init__() + self.cfg_file = self.cfg_dir.joinpath(TG_BOT_CFG_NAME) + self.log = self.log_dir.joinpath(TG_BOT_LOG_NAME) def create(self) -> None: + from utils.sys_utils import create_env_file, create_service_file + Logger.print_status("Creating new Moonraker Telegram Bot Instance ...") - service_template_path = MODULE_PATH.joinpath( - "assets/moonraker-telegram-bot.service" - ) - service_file_name = self.get_service_file_name(extension=True) - service_file_target = SYSTEMD.joinpath(service_file_name) - env_template_file_path = MODULE_PATH.joinpath( - "assets/moonraker-telegram-bot.env" - ) - env_file_target = self.sysd_dir.joinpath("moonraker-telegram-bot.env") try: self.create_folders() - self.write_service_file( - service_template_path, service_file_target, env_file_target + create_service_file( + name=self.get_service_file_name(extension=True), + content=self._prep_service_file_content(), + ) + create_env_file( + path=self.sysd_dir.joinpath(TG_BOT_ENV_FILE_NAME), + content=self._prep_env_file_content(), ) - self.write_env_file(env_template_file_path, env_file_target) except CalledProcessError as e: - Logger.print_error( - f"Error creating service file {service_file_target}: {e}" - ) + Logger.print_error(f"Error creating instance: {e}") raise except OSError as e: - Logger.print_error(f"Error creating env file {env_file_target}: {e}") + Logger.print_error(f"Error creating env file: {e}") raise def delete(self) -> None: - service_file = self.get_service_file_name(extension=True) - service_file_path = self.get_service_file_path() + service_file: str = self.get_service_file_name(extension=True) + service_file_path: Path = self.get_service_file_path() Logger.print_status(f"Deleting Moonraker Telegram Bot Instance: {service_file}") try: - command = ["sudo", "rm", "-f", service_file_path] + command = ["sudo", "rm", "-f", service_file_path.as_posix()] run(command, check=True) Logger.print_ok(f"Service file deleted: {service_file_path}") except CalledProcessError as e: Logger.print_error(f"Error deleting service file: {e}") raise - def write_service_file( - self, - service_template_path: Path, - service_file_target: Path, - env_file_target: Path, - ) -> None: - service_content = self._prep_service_file( - service_template_path, env_file_target - ) - command = ["sudo", "tee", service_file_target.as_posix()] - run( - command, - input=service_content.encode(), - stdout=DEVNULL, - check=True, - ) - Logger.print_ok(f"Service file created: {service_file_target}") + def _prep_service_file_content(self) -> str: + template = TG_BOT_SERVICE_TEMPLATE - def write_env_file( - self, env_template_file_path: Path, env_file_target: Path - ) -> None: - env_file_content = self._prep_env_file(env_template_file_path) - with open(env_file_target, "w") as env_file: - env_file.write(env_file_content) - Logger.print_ok(f"Env file created: {env_file_target}") - - def _prep_service_file( - self, service_template_path: Path, env_file_path: Path - ) -> str: try: - with open(service_template_path, "r") as template_file: + with open(template, "r") as template_file: template_content = template_file.read() except FileNotFoundError: - Logger.print_error( - f"Unable to open {service_template_path} - File not found" - ) + Logger.print_error(f"Unable to open {template} - File not found") raise - service_content = template_content.replace("%USER%", self.user) + + service_content = template_content.replace( + "%USER%", + self.user, + ) service_content = service_content.replace( "%TELEGRAM_BOT_DIR%", - str(self.bot_dir), + self.bot_dir.as_posix(), + ) + service_content = service_content.replace( + "%ENV%", + self.env_dir.as_posix(), + ) + service_content = service_content.replace( + "%ENV_FILE%", + self.sysd_dir.joinpath(TG_BOT_ENV_FILE_NAME).as_posix(), ) - service_content = service_content.replace("%ENV%", str(self.env_dir)) - service_content = service_content.replace("%ENV_FILE%", str(env_file_path)) return service_content - def _prep_env_file(self, env_template_file_path: Path) -> str: + def _prep_env_file_content(self) -> str: + template = TG_BOT_ENV_FILE_TEMPLATE + try: - with open(env_template_file_path, "r") as env_file: + with open(template, "r") as env_file: env_template_file_content = env_file.read() except FileNotFoundError: - Logger.print_error( - f"Unable to open {env_template_file_path} - File not found" - ) + Logger.print_error(f"Unable to open {template} - File not found") raise + env_file_content = env_template_file_content.replace( "%TELEGRAM_BOT_DIR%", - str(self.bot_dir), + self.bot_dir.as_posix(), ) env_file_content = env_file_content.replace( "%CFG%", f"{self.cfg_dir}/printer.cfg", ) - env_file_content = env_file_content.replace("%LOG%", str(self.log)) + env_file_content = env_file_content.replace( + "%LOG%", + self.log.as_posix() if self.log else "", + ) return env_file_content diff --git a/kiauh/extensions/telegram_bot/moonraker_telegram_bot_extension.py b/kiauh/extensions/telegram_bot/moonraker_telegram_bot_extension.py index f2a900d..030a82f 100644 --- a/kiauh/extensions/telegram_bot/moonraker_telegram_bot_extension.py +++ b/kiauh/extensions/telegram_bot/moonraker_telegram_bot_extension.py @@ -13,10 +13,10 @@ from typing import List from components.moonraker.moonraker import Moonraker from core.instance_manager.instance_manager import InstanceManager from extensions.base_extension import BaseExtension +from extensions.telegram_bot import TG_BOT_REPO from extensions.telegram_bot.moonraker_telegram_bot import ( - TELEGRAM_BOT_DIR, - TELEGRAM_BOT_ENV, - TELEGRAM_BOT_REPO, + TG_BOT_DIR, + TG_BOT_ENV, MoonrakerTelegramBot, ) from utils.common import check_install_dependencies @@ -70,7 +70,7 @@ class TelegramBotExtension(BaseExtension): create_example_cfg = get_confirm("Create example telegram.conf?") try: - git_clone_wrapper(TELEGRAM_BOT_REPO, TELEGRAM_BOT_DIR) + git_clone_wrapper(TG_BOT_REPO, TG_BOT_DIR) self._install_dependencies() # create and start services / create bot configs @@ -88,9 +88,7 @@ class TelegramBotExtension(BaseExtension): Logger.print_status( f"Creating Telegram Bot config in {current_instance.cfg_dir} ..." ) - template = TELEGRAM_BOT_DIR.joinpath( - "scripts/base_install_template" - ) + template = TG_BOT_DIR.joinpath("scripts/base_install_template") target_file = current_instance.cfg_file if not target_file.exists(): show_config_dialog = True @@ -133,7 +131,7 @@ class TelegramBotExtension(BaseExtension): tb_im = InstanceManager(MoonrakerTelegramBot) tb_im.stop_all_instance() - git_pull_wrapper(TELEGRAM_BOT_REPO, TELEGRAM_BOT_DIR) + git_pull_wrapper(TG_BOT_REPO, TG_BOT_DIR) self._install_dependencies() tb_im.start_all_instance() @@ -158,24 +156,24 @@ class TelegramBotExtension(BaseExtension): def _install_dependencies(self) -> None: # install dependencies - script = TELEGRAM_BOT_DIR.joinpath("scripts/install.sh") + script = TG_BOT_DIR.joinpath("scripts/install.sh") package_list = parse_packages_from_file(script) check_install_dependencies(package_list) # create virtualenv - create_python_venv(TELEGRAM_BOT_ENV) - requirements = TELEGRAM_BOT_DIR.joinpath("scripts/requirements.txt") - install_python_requirements(TELEGRAM_BOT_ENV, requirements) + create_python_venv(TG_BOT_ENV) + requirements = TG_BOT_DIR.joinpath("scripts/requirements.txt") + install_python_requirements(TG_BOT_ENV, requirements) def _patch_bot_update_manager(self, instances: List[Moonraker]) -> None: - env_py = f"{TELEGRAM_BOT_ENV}/bin/python" + env_py = f"{TG_BOT_ENV}/bin/python" add_config_section( section="update_manager moonraker-telegram-bot", instances=instances, options=[ ("type", "git_repo"), - ("path", str(TELEGRAM_BOT_DIR)), - ("orgin", TELEGRAM_BOT_REPO), + ("path", str(TG_BOT_DIR)), + ("orgin", TG_BOT_REPO), ("env", env_py), ("requirements", "scripts/requirements.txt"), ("install_script", "scripts/install.sh"), @@ -199,24 +197,24 @@ class TelegramBotExtension(BaseExtension): cmd_sysctl_manage("daemon-reload") def _remove_bot_dir(self) -> None: - if not TELEGRAM_BOT_DIR.exists(): - Logger.print_info(f"'{TELEGRAM_BOT_DIR}' does not exist. Skipped ...") + if not TG_BOT_DIR.exists(): + Logger.print_info(f"'{TG_BOT_DIR}' does not exist. Skipped ...") return try: - shutil.rmtree(TELEGRAM_BOT_DIR) + shutil.rmtree(TG_BOT_DIR) except OSError as e: - Logger.print_error(f"Unable to delete '{TELEGRAM_BOT_DIR}':\n{e}") + Logger.print_error(f"Unable to delete '{TG_BOT_DIR}':\n{e}") def _remove_bot_env(self) -> None: - if not TELEGRAM_BOT_ENV.exists(): - Logger.print_info(f"'{TELEGRAM_BOT_ENV}' does not exist. Skipped ...") + if not TG_BOT_ENV.exists(): + Logger.print_info(f"'{TG_BOT_ENV}' does not exist. Skipped ...") return try: - shutil.rmtree(TELEGRAM_BOT_ENV) + shutil.rmtree(TG_BOT_ENV) except OSError as e: - Logger.print_error(f"Unable to delete '{TELEGRAM_BOT_ENV}':\n{e}") + Logger.print_error(f"Unable to delete '{TG_BOT_ENV}':\n{e}") def _delete_bot_logs(self, instances: List[MoonrakerTelegramBot]) -> None: all_logfiles = []