jeka/kiauh
1
0
Fork 0
mirror of https://github.com/dw-0/kiauh.git synced 2025-12-25 08:43:36 +05:00
Code Issues Packages Projects Releases Wiki Activity

refactor(extensions): refactor telegram bot extension

Browse Source 
Signed-off-by: Dominik Willner <th33xitus@gmail.com>
This commit is contained in:
dw-0
2024-08-04 18:36:04 +02:00
parent 96daf966ee
commit acde067e68
3 changed files with 114 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
kiauh/extensions/telegram_bot/__init__.py
View File

@@ -0,0 +1,28 @@
# ======================================================================= #
# Copyright (C) 2020 - 2024 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
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}")

153
kiauh/extensions/telegram_bot/moonraker_telegram_bot.py
View File

@@ -6,149 +6,126 @@
# # # #
# This file may be distributed under the terms of the GNU GPLv3 license # # 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 pathlib import Path
from subprocess import DEVNULL, CalledProcessError, run from subprocess import CalledProcessError, run
from typing import List
from core.instance_manager.base_instance import BaseInstance 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 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 # noinspection PyMethodMayBeStatic
# todo: make this to a dataclass @dataclass
class MoonrakerTelegramBot(BaseInstance): class MoonrakerTelegramBot(BaseInstance):
@classmethod bot_dir: Path = TG_BOT_DIR
def blacklist(cls) -> List[str]: env_dir: Path = TG_BOT_ENV
return ["None", "mcu"] cfg_file: Path | None = None
log: Path | None = None
def __init__(self, suffix: str = ""): def __init__(self, suffix: str = ""):
super().__init__(instance_type=self, suffix=suffix) super().__init__(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")
@property def __post_init__(self):
def cfg_file(self) -> Path: super().__post_init__()
return self._cfg_file self.cfg_file = self.cfg_dir.joinpath(TG_BOT_CFG_NAME)
self.log = self.log_dir.joinpath(TG_BOT_LOG_NAME)
@property
def log(self) -> Path:
return self._log
def create(self) -> None: def create(self) -> None:
from utils.sys_utils import create_env_file, create_service_file
Logger.print_status("Creating new Moonraker Telegram Bot Instance ...") 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: try:
self.create_folders() self.create_folders()
self.write_service_file( create_service_file(
service_template_path, service_file_target, env_file_target 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: except CalledProcessError as e:
Logger.print_error( Logger.print_error(f"Error creating instance: {e}")
f"Error creating service file {service_file_target}: {e}"
)
raise raise
except OSError as e: 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 raise
def delete(self) -> None: def delete(self) -> None:
service_file = self.get_service_file_name(extension=True) service_file: str = self.get_service_file_name(extension=True)
service_file_path = self.get_service_file_path() service_file_path: Path = self.get_service_file_path()
Logger.print_status(f"Deleting Moonraker Telegram Bot Instance: {service_file}") Logger.print_status(f"Deleting Moonraker Telegram Bot Instance: {service_file}")
try: try:
command = ["sudo", "rm", "-f", service_file_path] command = ["sudo", "rm", "-f", service_file_path.as_posix()]
run(command, check=True) run(command, check=True)
Logger.print_ok(f"Service file deleted: {service_file_path}") Logger.print_ok(f"Service file deleted: {service_file_path}")
except CalledProcessError as e: except CalledProcessError as e:
Logger.print_error(f"Error deleting service file: {e}") Logger.print_error(f"Error deleting service file: {e}")
raise raise
def write_service_file( def _prep_service_file_content(self) -> str:
self, template = TG_BOT_SERVICE_TEMPLATE
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 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: try:
with open(service_template_path, "r") as template_file: with open(template, "r") as template_file:
template_content = template_file.read() template_content = template_file.read()
except FileNotFoundError: except FileNotFoundError:
Logger.print_error( Logger.print_error(f"Unable to open {template} - File not found")
f"Unable to open {service_template_path} - File not found"
)
raise raise
service_content = template_content.replace("%USER%", self.user)
service_content = template_content.replace(
"%USER%",
self.user,
)
service_content = service_content.replace( service_content = service_content.replace(
"%TELEGRAM_BOT_DIR%", "%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 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: 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() env_template_file_content = env_file.read()
except FileNotFoundError: except FileNotFoundError:
Logger.print_error( Logger.print_error(f"Unable to open {template} - File not found")
f"Unable to open {env_template_file_path} - File not found"
)
raise raise
env_file_content = env_template_file_content.replace( env_file_content = env_template_file_content.replace(
"%TELEGRAM_BOT_DIR%", "%TELEGRAM_BOT_DIR%",
str(self.bot_dir), self.bot_dir.as_posix(),
) )
env_file_content = env_file_content.replace( env_file_content = env_file_content.replace(
"%CFG%", "%CFG%",
f"{self.cfg_dir}/printer.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 return env_file_content

44
kiauh/extensions/telegram_bot/moonraker_telegram_bot_extension.py
View File

@@ -13,10 +13,10 @@ from typing import List
from components.moonraker.moonraker import Moonraker from components.moonraker.moonraker import Moonraker
from core.instance_manager.instance_manager import InstanceManager from core.instance_manager.instance_manager import InstanceManager
from extensions.base_extension import BaseExtension from extensions.base_extension import BaseExtension
from extensions.telegram_bot import TG_BOT_REPO
from extensions.telegram_bot.moonraker_telegram_bot import ( from extensions.telegram_bot.moonraker_telegram_bot import (
TELEGRAM_BOT_DIR, TG_BOT_DIR,
TELEGRAM_BOT_ENV, TG_BOT_ENV,
TELEGRAM_BOT_REPO,
MoonrakerTelegramBot, MoonrakerTelegramBot,
) )
from utils.common import check_install_dependencies from utils.common import check_install_dependencies
@@ -70,7 +70,7 @@ class TelegramBotExtension(BaseExtension):
create_example_cfg = get_confirm("Create example telegram.conf?") create_example_cfg = get_confirm("Create example telegram.conf?")
try: try:
git_clone_wrapper(TELEGRAM_BOT_REPO, TELEGRAM_BOT_DIR) git_clone_wrapper(TG_BOT_REPO, TG_BOT_DIR)
self._install_dependencies() self._install_dependencies()
# create and start services / create bot configs # create and start services / create bot configs
@@ -88,9 +88,7 @@ class TelegramBotExtension(BaseExtension):
Logger.print_status( Logger.print_status(
f"Creating Telegram Bot config in {current_instance.cfg_dir} ..." f"Creating Telegram Bot config in {current_instance.cfg_dir} ..."
) )
template = TELEGRAM_BOT_DIR.joinpath( template = TG_BOT_DIR.joinpath("scripts/base_install_template")
"scripts/base_install_template"
)
target_file = current_instance.cfg_file target_file = current_instance.cfg_file
if not target_file.exists(): if not target_file.exists():
show_config_dialog = True show_config_dialog = True
@@ -133,7 +131,7 @@ class TelegramBotExtension(BaseExtension):
tb_im = InstanceManager(MoonrakerTelegramBot) tb_im = InstanceManager(MoonrakerTelegramBot)
tb_im.stop_all_instance() 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() self._install_dependencies()
tb_im.start_all_instance() tb_im.start_all_instance()
@@ -158,24 +156,24 @@ class TelegramBotExtension(BaseExtension):
def _install_dependencies(self) -> None: def _install_dependencies(self) -> None:
# install dependencies # 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) package_list = parse_packages_from_file(script)
check_install_dependencies(package_list) check_install_dependencies(package_list)
# create virtualenv # create virtualenv
create_python_venv(TELEGRAM_BOT_ENV) create_python_venv(TG_BOT_ENV)
requirements = TELEGRAM_BOT_DIR.joinpath("scripts/requirements.txt") requirements = TG_BOT_DIR.joinpath("scripts/requirements.txt")
install_python_requirements(TELEGRAM_BOT_ENV, requirements) install_python_requirements(TG_BOT_ENV, requirements)
def _patch_bot_update_manager(self, instances: List[Moonraker]) -> None: 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( add_config_section(
section="update_manager moonraker-telegram-bot", section="update_manager moonraker-telegram-bot",
instances=instances, instances=instances,
options=[ options=[
("type", "git_repo"), ("type", "git_repo"),
("path", str(TELEGRAM_BOT_DIR)), ("path", str(TG_BOT_DIR)),
("orgin", TELEGRAM_BOT_REPO), ("orgin", TG_BOT_REPO),
("env", env_py), ("env", env_py),
("requirements", "scripts/requirements.txt"), ("requirements", "scripts/requirements.txt"),
("install_script", "scripts/install.sh"), ("install_script", "scripts/install.sh"),
@@ -199,24 +197,24 @@ class TelegramBotExtension(BaseExtension):
cmd_sysctl_manage("daemon-reload") cmd_sysctl_manage("daemon-reload")
def _remove_bot_dir(self) -> None: def _remove_bot_dir(self) -> None:
if not TELEGRAM_BOT_DIR.exists(): if not TG_BOT_DIR.exists():
Logger.print_info(f"'{TELEGRAM_BOT_DIR}' does not exist. Skipped ...") Logger.print_info(f"'{TG_BOT_DIR}' does not exist. Skipped ...")
return return
try: try:
shutil.rmtree(TELEGRAM_BOT_DIR) shutil.rmtree(TG_BOT_DIR)
except OSError as e: 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: def _remove_bot_env(self) -> None:
if not TELEGRAM_BOT_ENV.exists(): if not TG_BOT_ENV.exists():
Logger.print_info(f"'{TELEGRAM_BOT_ENV}' does not exist. Skipped ...") Logger.print_info(f"'{TG_BOT_ENV}' does not exist. Skipped ...")
return return
try: try:
shutil.rmtree(TELEGRAM_BOT_ENV) shutil.rmtree(TG_BOT_ENV)
except OSError as e: 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: def _delete_bot_logs(self, instances: List[MoonrakerTelegramBot]) -> None:
all_logfiles = [] all_logfiles = []