BW3-Core/install_service.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
r"""!
    ____  ____  ______       __      __       __       _____
   / __ )/ __ \/ ___/ |     / /___ _/ /______/ /_     |__  /
  / __  / / / /\__ \| | /| / / __ `/ __/ ___/ __ \     /_ <
 / /_/ / /_/ /___/ /| |/ |/ / /_/ / /_/ /__/ / / /   ___/ /
/_____/\____//____/ |__/|__/\__,_/\__/\___/_/ /_/   /____/
                German BOS Information Script
                     by Bastian Schroll

@file:        install_service.py
@date:        15.11.2025
@author:      Claus Schichl
@description: Install Service File with argparse CLI
"""

import os
import subprocess
import sys
import logging
import argparse
import yaml
from pathlib import Path

#  === constants for directories and files ===
BASE_DIR = Path(__file__).resolve().parent
BW_DIR = '/opt/boswatch3'
SERVICE_DIR = Path('/etc/systemd/system')
CONFIG_DIR = (BASE_DIR / 'config').resolve()
LOG_FILE = (BASE_DIR / 'log' / 'install' / 'service_install.log').resolve()
os.makedirs(LOG_FILE.parent, exist_ok=True)


#  === language management (default german)===
_lang = 'de'


def get_lang():
    return _lang


def set_lang(lang):
    global _lang
    _lang = lang


#  === text-dictionary ===
TEXT = {
    "de": {
        "script_title": "🛠️ BOSWatch Service Manager",
        "mode_dry": "DRY-RUN (nur Vorschau)",
        "mode_live": "LIVE",
        "no_yaml": "❌ Keine .yaml-Dateien im config-Verzeichnis gefunden.",
        "found_yaml": "🔍 Gefundene YAML-Dateien: {}",
        "action_prompt": "\nWas möchtest du tun? (i=installieren, r=entfernen, e=beenden): ",
        "edited_prompt": "Wurden die YAML-Dateien korrekt bearbeitet? (y/n): ",
        "edit_abort": "⚠️ Bitte YAML-Dateien zuerst bearbeiten. Vorgang abgebrochen.",
        "install_confirm": "Service für '{}' installieren? (y/n): ",
        "skip_invalid_yaml": "⏭ Überspringe fehlerhafte YAML: {}",
        "install_done": "✅ Installation abgeschlossen. Services installiert: {}, übersprungen: {}",
        "invalid_input": "Ungültige Eingabe. Erlaubt sind: {}",
        "no_services": "Keine bw3-Services gefunden.",
        "available_services": "\nVerfügbare bw3-Services:",
        "remove_prompt": "Was soll deinstalliert werden? ",
        "invalid_choice": "Ungültige Auswahl.",
        "not_root": "🛑 Dieses Skript muss mit Root-Rechten ausgeführt werden (sudo).",
        "help_dry_run": "Nur anzeigen, nicht ausführen",
        "help_verbose": "Ausführliche Ausgabe",
        "help_quiet": "Weniger Ausgabe",
        "help_lang": "Sprache für alle Ausgaben [de/en] (Standard: de)",
        "creating_service_file": "📄 Erstelle Service-Datei für {} → {}",
        "removing_service": "\n🗑 Entferne Service: {}",
        "service_deleted": "{} gelöscht.",
        "service_not_found": "{} nicht gefunden oder im Dry-Run-Modus.",
        "yaml_error": "⚠ Fehler in YAML {}: {}",
        "yaml_read_error": "⚠ Fehler beim Lesen der YAML-Datei {}: {}",
        "unknown_yaml_type": "⚠ YAML-Typ für {} nicht erkannt. Service wird übersprungen.",
        "verify_warn": "⚠ Warnung bei systemd-analyze verify:\n{}",
        "verify_ok": "{} erfolgreich verifiziert.",
        "install_skipped": "⏭ Installation für '{}' übersprungen",
        "file_write_error": "⚠ Fehler beim Schreiben der Datei {}: {}",
        "all": "[a] Alle deinstallieren",
        "exit": "[e] Beenden",
        "service_active": "✅ Service {0} läuft erfolgreich.",
        "service_inactive": "⚠  Service {0} ist **nicht aktiv** – bitte prüfen.",
        "dryrun_status_check": "🧪 [Dry-Run] Service-Status von {0} würde jetzt geprüft.",
        "max_attempts_exceeded": "❌ Maximale Anzahl an Eingabeversuchen überschritten. Das Menü wird beendet.",
        "user_interrupt": "\nAbbruch durch Benutzer.",
        "unhandled_error": "Unbehandelter Fehler: {}",
        "max_retries_skip": "Maximale Anzahl Eingabeversuche überschritten. Überspringe Service.",
        "max_retries_exit": "Maximale Anzahl Eingabeversuche überschritten. Beende Programm.",
        "colorama_missing": "⚠️ Colorama nicht installiert – versuche automatische Installation...",
        "colorama_install": "➡️ Installiere Colorama...",
        "colorama_install_ok": "✅ Colorama erfolgreich installiert.",
        "colorama_install_fail": "❌ Colorama konnte nicht automatisch installiert werden.",
        "verify_timeout": "⚠ Timeout bei systemd-analyze verify für: {}",
        "status_timeout": "⚠ Timeout beim Prüfen des Service-Status: {}"

    },
    "en": {
        "script_title": "🛠️ BOSWatch Service Manager",
        "mode_dry": "DRY-RUN (preview only)",
        "mode_live": "LIVE",
        "no_yaml": "❌ No .yaml files found in config directory.",
        "found_yaml": "🔍 YAML files found: {}",
        "action_prompt": "\nWhat would you like to do? (i=install, r=remove, e=exit): ",
        "edited_prompt": "Have the YAML files been edited correctly? (y/n): ",
        "edit_abort": "⚠ Please edit the YAML files first. Aborting.",
        "install_confirm": "Install service for '{}' ? (y/n): ",
        "skip_invalid_yaml": "⏭ Skipping invalid YAML: {}",
        "install_done": "✅ Installation complete. Services installed: {}, skipped: {}",
        "invalid_input": "Invalid input. Allowed: {}",
        "no_services": "No bw3 services found.",
        "available_services": "\nAvailable bw3 services:",
        "remove_prompt": "What should be removed? ",
        "invalid_choice": "Invalid choice.",
        "not_root": "🛑 This script must be run as root (sudo).",
        "help_dry_run": "Show actions only, do not execute",
        "help_verbose": "Show detailed output",
        "help_quiet": "Reduce output verbosity",
        "help_lang": "Language for all output [de/en] (default: de)",
        "creating_service_file": "📄 Creating service file for {} → {}",
        "removing_service": "\n🗑 Removing service: {}",
        "service_deleted": "{} deleted.",
        "service_not_found": "{} not found or in dry-run mode.",
        "yaml_error": "⚠ YAML error in {}: {}",
        "yaml_read_error": "⚠ Error reading YAML file {}: {}",
        "unknown_yaml_type": "⚠ Unknown YAML type for {}. Skipping service.",
        "verify_warn": "⚠ Warning in systemd-analyze verify:\n{}",
        "verify_ok": "{} verified successfully.",
        "install_skipped": "⏭ Installation skipped for '{}'",
        "file_write_error": "⚠ Error writing file {}: {}",
        "all": "[a] Remove all",
        "exit": "[e] Exit",
        "service_active": "✅ Service {0} is running successfully.",
        "service_inactive": "⚠ Service {0} is **not active** – please check.",
        "dryrun_status_check": "🧪 [Dry-Run] Service status of {0} would be checked now.",
        "max_attempts_exceeded": "❌ Maximum number of input attempts exceeded. Exiting menu.",
        "user_interrupt": "\nInterrupted by user.",
        "unhandled_error": "Unhandled error: {}",
        "max_retries_skip": "Maximum input attempts exceeded. Skipping service.",
        "max_retries_exit": "Maximum input attempts exceeded. Exiting program.",
        "colorama_missing": "⚠️ Colorama not installed – attempting automatic installation...",
        "colorama_install": "➡️ Installing Colorama...",
        "colorama_install_ok": "✅ Colorama installed successfully.",
        "colorama_install_fail": "❌ Colorama could not be installed automatically.",
        "verify_timeout": "⚠ Timeout during systemd-analyze verify for: {}",
        "status_timeout": "⚠ Timeout while checking service status: {}"
    }
}


# === COLORAMA AUTO-INSTALL (dual language) ===
def colorama_auto_install():
    r"""
    Auto-installs colorama if missing.
    Note: Language detection happens before colorama is available.
    """
    # recognize language early (before colorama installation)
    import argparse
    early_parser = argparse.ArgumentParser(add_help=False)
    early_parser.add_argument('--lang', '-l', choices=['de', 'en'], default='de')
    early_args, _ = early_parser.parse_known_args()
    lang = early_args.lang

    # use text from global TEXT dictionary
    txt = TEXT[lang]

    try:
        from colorama import init as colorama_init, Fore, Style
        colorama_init(autoreset=True)
        return True, Fore, Style
    except ImportError:
        print(txt["colorama_missing"])

        # install Colorama
        print(txt["colorama_install"])
        subprocess.run(["sudo", "apt", "install", "-y", "python3-colorama"], check=False)

        # retry importing Colorama
        try:
            from colorama import init as colorama_init, Fore, Style
            colorama_init(autoreset=True)
            print(txt["colorama_install_ok"])
            return True, Fore, Style
        except ImportError:
            print(txt["colorama_install_fail"])
            return False, None, None


# === import / install colorama ===
colorama_available, Fore, Style = colorama_auto_install()

if not colorama_available:
    # provides dummy classes if colorama is not available (no crash)
    class DummyStyle:
        RESET_ALL = ""
        BRIGHT = ""

    class DummyFore:
        RED = GREEN = YELLOW = BLUE = CYAN = MAGENTA = WHITE = RESET = ""

    Fore = DummyFore()
    Style = DummyStyle()


# === logging Setup ===
def setup_logging(verbose=False, quiet=False):
    r"""
    Setup logging to file and console with colorized output.
    """
    log_level = logging.INFO
    if quiet:
        log_level = logging.WARNING
    elif verbose:
        log_level = logging.DEBUG

    logger = logging.getLogger()
    logger.setLevel(log_level)

    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')

    # File Handler (plain)
    fh = logging.FileHandler(LOG_FILE)
    fh.setFormatter(formatter)
    logger.addHandler(fh)

    # Console Handler (colorized)
    class ColorFormatter(logging.Formatter):
        COLORS = {
            logging.DEBUG: Fore.CYAN,
            logging.INFO: Fore.GREEN,
            logging.WARNING: Fore.YELLOW,
            logging.ERROR: Fore.RED,
            logging.CRITICAL: Fore.RED + Style.BRIGHT,
        }

        def format(self, record):
            color = self.COLORS.get(record.levelno, Fore.RESET)
            message = super().format(record)
            return f"{color}{message}{Style.RESET_ALL}"

    ch = logging.StreamHandler(sys.stdout)
    ch.setFormatter(ColorFormatter('%(levelname)s: %(message)s'))
    logger.addHandler(ch)

    return logger


def t(key):
    r"""
    Translation helper: returns the localized string for the given key.
    """
    lang = get_lang()
    return TEXT.get(lang, TEXT['de']).get(key, key)


def get_user_input(prompt, valid_inputs, max_attempts=3):
    r"""
    Prompt user for input until a valid input from valid_inputs is entered or max_attempts exceeded.
    Raises RuntimeError on failure.
    """
    attempts = 0
    while attempts < max_attempts:
        value = input(prompt).strip().lower()
        if value in valid_inputs:
            return value
        logging.warning(t("invalid_input").format(", ".join(valid_inputs)))
        attempts += 1
    raise RuntimeError(t("max_attempts_exceeded"))


def list_yaml_files():
    r"""
    Returns a list of .yaml or .yml files in the config directory.
    """
    return sorted([f.name for f in CONFIG_DIR.glob("*.y*ml")])


def test_yaml_file(file_path):
    r"""
    Tests if YAML file can be loaded without error.
    """
    try:
        content = file_path.read_text(encoding='utf-8')
        yaml.safe_load(content)
        return True
    except Exception as e:
        logging.error(t("yaml_error").format(file_path, e))
        return False


def detect_yaml_type(file_path):
    r"""
    Detects if YAML config is 'client' or 'server' type.
    """
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            data = yaml.safe_load(f)
        if 'client' in data:
            return 'client'
        elif 'server' in data:
            return 'server'
        else:
            logging.error(t("unknown_yaml_type").format(os.path.basename(file_path)))
            return None
    except Exception as e:
        logging.error(t("yaml_read_error").format(file_path, e))
        return None


def execute(command, dry_run=False):
    r"""
    Executes shell command unless dry_run is True.
    """
    logging.debug(f"→ {command}")
    if not dry_run:
        subprocess.run(command, shell=True, check=False)


def verify_service(service_path):
    r"""
    Runs 'systemd-analyze verify' on the service file and logs warnings/errors.
    """
    try:
        result = subprocess.run(
            ['systemd-analyze', 'verify', service_path],
            capture_output=True,
            text=True,
            timeout=10
        )
        if result.returncode != 0 or result.stderr:
            logging.warning(t("verify_warn").format(result.stderr.strip()))
        else:
            logging.debug(t("verify_ok").format(os.path.basename(service_path)))
    except subprocess.TimeoutExpired:
        logging.warning(t("verify_timeout").format(os.path.basename(service_path)))
    except Exception as e:
        logging.error(t("yaml_error").format(service_path, e))


def install_service(yaml_file, dry_run=False):
    r"""
    Creates and installs systemd service based on YAML config.
    """
    yaml_path = CONFIG_DIR / yaml_file
    yaml_type = detect_yaml_type(yaml_path)
    if yaml_type == 'server':
        is_server = True
    elif yaml_type == 'client':
        is_server = False
    else:
        logging.error(t("unknown_yaml_type").format(yaml_file))
        return

    service_name = f"bw3_{Path(yaml_file).stem}.service"
    service_path = SERVICE_DIR / service_name

    if is_server:
        exec_line = f"/usr/bin/python3 {BW_DIR}/bw_server.py -c {yaml_file}"
        description = "BOSWatch Server"
        after = "network-online.target"
        wants = "Wants=network-online.target"
    else:
        exec_line = f"/usr/bin/python3 {BW_DIR}/bw_client.py -c {yaml_file}"
        description = "BOSWatch Client"
        after = "network.target"
        wants = ""

    service_content = f"""[Unit]
Description={description}
After={after}
{wants}

[Service]
Type=simple
WorkingDirectory={BW_DIR}
ExecStart={exec_line}
Restart=on-abort

[Install]
WantedBy=multi-user.target
"""

    logging.info(t("creating_service_file").format(yaml_file, service_name))

    if not dry_run:
        try:
            with open(service_path, 'w', encoding='utf-8') as f:
                f.write(service_content)
        except IOError as e:
            logging.error(t("file_write_error").format(service_path, e))
            return
        verify_service(service_path)

    execute("systemctl daemon-reload", dry_run=dry_run)
    execute(f"systemctl enable {service_name}", dry_run=dry_run)
    execute(f"systemctl start {service_name}", dry_run=dry_run)

    if not dry_run:
        try:
            subprocess.run(
                ["systemctl", "is-active", "--quiet", service_name],
                check=True,
                timeout=5
            )
            logging.info(t("service_active").format(service_name))
        except subprocess.CalledProcessError:
            logging.warning(t("service_inactive").format(service_name))
        except subprocess.TimeoutExpired:
            logging.warning(t("status_timeout").format(service_name))
    else:
        logging.info(t("dryrun_status_check").format(service_name))


def remove_service(service_name, dry_run=False):
    r"""
    Stops, disables and removes the given systemd service.
    """
    logging.warning(t("removing_service").format(service_name))
    execute(f"systemctl stop {service_name}", dry_run=dry_run)
    execute(f"systemctl disable {service_name}", dry_run=dry_run)

    service_path = SERVICE_DIR / service_name
    if not dry_run and service_path.exists():
        try:
            service_path.unlink()
            logging.info(t("service_deleted").format(service_name))
        except Exception as e:
            logging.error(t("file_write_error").format(service_path, e))
    else:
        logging.warning(t("service_not_found").format(service_name))

    execute("systemctl daemon-reload", dry_run=dry_run)


def remove_menu(dry_run=False):
    r"""
    Interactive menu to remove services.
    """
    while True:
        services = sorted([
            f for f in os.listdir(SERVICE_DIR)
            if f.startswith('bw3_') and f.endswith('.service')
        ])

        if not services:
            print(Fore.YELLOW + t("no_services") + Style.RESET_ALL)
            return

        print(Fore.CYAN + "\n" + t("available_services") + Style.RESET_ALL)
        for i, s in enumerate(services):
            print(f" [{i}] {s}")
        print(" " + t("all"))
        print(" " + t("exit"))

        try:
            auswahl = get_user_input(
                t("remove_prompt"),
                ['e', 'a'] + [str(i) for i in range(len(services))]
            )
        except RuntimeError:
            logging.error(t("max_attempts_exceeded"))
            break

        if auswahl == 'e':
            break
        elif auswahl == 'a':
            for s in services:
                remove_service(s, dry_run=dry_run)
            # directly continue to the next loop (updated list!)
            continue
        else:
            remove_service(services[int(auswahl)], dry_run=dry_run)
            # also directly continue to the next loop (updated list!)
            continue


def init_language():
    r"""
    Parses --lang/-l argument early to set language before other parsing.
    """
    lang_parser = argparse.ArgumentParser(add_help=False)
    lang_parser.add_argument(
        '--lang', '-l',
        choices=['de', 'en'],
        default='de',
        metavar='LANG',
        help=TEXT["en"]["help_lang"]
    )
    lang_args, remaining_argv = lang_parser.parse_known_args()
    set_lang(lang_args.lang)
    return lang_parser, remaining_argv


def main(dry_run=False):
    r"""
    main program: install or remove service.
    """
    print(Fore.GREEN + Style.BRIGHT + t("script_title") + Style.RESET_ALL)
    print(t('mode_dry') if dry_run else t('mode_live'))
    print()

    yaml_files = list_yaml_files()
    if not yaml_files:
        print(Fore.RED + t("no_yaml") + Style.RESET_ALL)
        sys.exit(1)

    print(Fore.GREEN + t("found_yaml").format(len(yaml_files)) + Style.RESET_ALL)
    for f in yaml_files:
        file_path = CONFIG_DIR / f
        valid = test_yaml_file(file_path)
        status = Fore.GREEN + "✅" if valid else Fore.RED + "❌"
        print(f" - {f} {status}{Style.RESET_ALL}")

    try:
        action = get_user_input(t("action_prompt"), ['i', 'r', 'e'])
    except RuntimeError:
        logging.error(t("max_retries_exit"))
        sys.exit(1)

    if action == 'e':
        sys.exit(0)
    elif action == 'r':
        remove_menu(dry_run=dry_run)
        return

    try:
        edited = get_user_input(t("edited_prompt"), ['y', 'n'])
    except RuntimeError:
        logging.error(t("max_retries_exit"))
        sys.exit(1)

    if edited == 'n':
        print(Fore.YELLOW + t("edit_abort") + Style.RESET_ALL)
        sys.exit(0)

    installed = 0
    skipped = 0

    for yaml_file in yaml_files:
        file_path = CONFIG_DIR / yaml_file
        if not test_yaml_file(file_path):
            print(Fore.RED + t("skip_invalid_yaml").format(yaml_file) + Style.RESET_ALL)
            skipped += 1
            continue

        try:
            install = get_user_input(t("install_confirm").format(yaml_file), ['y', 'n'])
        except RuntimeError:
            logging.error(t("max_retries_skip"))
            skipped += 1
            continue

        if install == 'y':
            install_service(yaml_file, dry_run=dry_run)
            installed += 1
        else:
            logging.info(t("install_skipped").format(yaml_file))
            skipped += 1

    print()
    logging.info(t("install_done").format(installed, skipped))


if __name__ == "__main__":
    lang_parser, remaining_argv = init_language()

    parser = argparse.ArgumentParser(
        description=t("script_title"),
        parents=[lang_parser]
    )
    parser.add_argument('--dry-run', action='store_true', help=t("help_dry_run"))
    parser.add_argument('--verbose', action='store_true', help=t("help_verbose"))
    parser.add_argument('--quiet', action='store_true', help=t("help_quiet"))

    args = parser.parse_args(remaining_argv)

    setup_logging(verbose=args.verbose, quiet=args.quiet)

    if os.geteuid() != 0:
        print(Fore.RED + t("not_root") + Style.RESET_ALL)
        sys.exit(1)

    try:
        main(dry_run=args.dry_run)
    except KeyboardInterrupt:
        print(t("user_interrupt"))
        sys.exit(1)
    except Exception as e:
        logging.critical(t("unhandled_error").format(e))
        sys.exit(1)