Trixy7/make_installer/installer/detect.py

# -*- coding: utf-8 -*-
"""
Erkennung bestehender Trixy-Installationen.

Liest Version, Modus, Service-Status und Einstellungen
einer bereits installierten Trixy-Instanz.
"""

from __future__ import annotations

import json
import os
import re
import subprocess
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path


class InstallState(Enum):
    """Zustand der bestehenden Installation."""
    NOT_FOUND = "not_found"
    SAME_VERSION = "same_version"
    NEWER_AVAILABLE = "newer_available"     # Installer ist neuer → Update
    OLDER_AVAILABLE = "older_available"     # Installer ist aelter → Downgrade


@dataclass
class ExistingInstallation:
    """Informationen ueber eine bestehende Installation."""
    found: bool = False
    state: InstallState = InstallState.NOT_FOUND
    installed_version: str = ""
    install_path: str = ""
    source_path: str = ""
    service_name: str = ""
    service_active: bool = False
    service_enabled: bool = False

    # Erkannte Einstellungen
    hostname: str = ""
    username: str = ""
    autostart: bool = False
    install_service: bool = False
    auto_regist: bool = False
    nlp_backend: str = ""
    ollama_host: str = ""
    wifi_enabled: bool = False
    wifi_ssid: str = ""
    wifi_password: str = ""
    wifi_channel: int = 6
    ap_ip: str = "10.10.10.1"
    dhcp_range_start: str = "10.10.10.10"
    dhcp_range_end: str = "10.10.10.50"
    server_host: str = ""
    server_port: int = 2101
    room: str = ""
    alias: str = ""
    selected_plugins: list[str] = field(default_factory=list)


def detect_installation(install_path: str, mode: str, installer_version: str) -> ExistingInstallation:
    """
    Erkennt eine bestehende Trixy-Installation.

    Args:
        install_path: Erwarteter Installationspfad (z.B. /home/pi/trixy)
        mode: Installationsmodus (server/client/standalone/config)
        installer_version: Version des Installers

    Returns:
        ExistingInstallation mit allen erkannten Informationen.
    """
    info = ExistingInstallation()
    source = f"{install_path}/source"

    # Pruefen ob source-Verzeichnis existiert
    if not os.path.isdir(source):
        return info

    # Version auslesen
    version_file = f"{source}/trixy_core/utils/version.py"
    if not os.path.isfile(version_file):
        # Eventuell alte Struktur (ohne source/ Unterverzeichnis)
        return info

    info.found = True
    info.install_path = install_path
    info.source_path = source
    info.installed_version = _parse_version_file(version_file)

    # Versionsvergleich
    info.state = _compare_versions(info.installed_version, installer_version)

    # Service-Status
    info.service_name = f"trixy-{mode}"
    info.service_active = _is_service_active(info.service_name)
    info.service_enabled = _is_service_enabled(info.service_name)
    info.install_service = info.service_enabled or info.service_active
    info.autostart = info.service_enabled

    # Auto-Regist aus Service-Datei lesen
    service_file = f"/etc/systemd/system/{info.service_name}.service"
    if os.path.isfile(service_file):
        _parse_service_file(service_file, info)

    # Hostname
    try:
        info.hostname = os.uname().nodename
    except Exception:
        pass

    # Username
    info.username = os.environ.get("SUDO_USER", os.environ.get("USER", "pi"))

    # Installierte Plugins erkennen
    plugins_dir = f"{source}/plugins"
    if os.path.isdir(plugins_dir):
        for entry in sorted(os.listdir(plugins_dir)):
            plugin_path = f"{plugins_dir}/{entry}"
            if os.path.isdir(plugin_path) and os.path.isfile(f"{plugin_path}/main.py"):
                # Plugin-Config lesen um enabled-Status zu pruefen
                config_file = f"{plugin_path}/config.json"
                if os.path.isfile(config_file):
                    try:
                        with open(config_file) as f:
                            cfg = json.load(f)
                        if cfg.get("enabled", False):
                            info.selected_plugins.append(entry)
                    except (json.JSONDecodeError, OSError):
                        pass

    # WiFi-Konfiguration aus hostapd lesen
    _parse_hostapd(info)

    # DHCP-Konfiguration aus dnsmasq lesen
    _parse_dnsmasq(info)

    # Client-Konfiguration aus Config-Datei lesen
    if mode == "client":
        _parse_client_config(source, info)

    # NLP-Backend aus Plugin-Config lesen
    nlp_config = f"{source}/plugins/nlp_llm/config.json"
    if os.path.isfile(nlp_config):
        try:
            with open(nlp_config) as f:
                cfg = json.load(f)
            info.nlp_backend = cfg.get("backend", "")
            info.ollama_host = cfg.get("ollama_host", "")
            # Wenn ollama mit nicht-localhost Host → ollama_remote
            if info.nlp_backend == "ollama" and info.ollama_host and "localhost" not in info.ollama_host and "127.0.0.1" not in info.ollama_host:
                info.nlp_backend = "ollama_remote"
        except (json.JSONDecodeError, OSError):
            pass

    return info


def _parse_version_file(path: str) -> str:
    """Liest die Version aus version.py."""
    try:
        with open(path) as f:
            content = f.read()
        major = re.search(r'^MAJOR:\s*int\s*=\s*(\d+)', content, re.MULTILINE)
        minor = re.search(r'^MINOR:\s*int\s*=\s*(\d+)', content, re.MULTILINE)
        revision = re.search(r'^REVISION:\s*int\s*=\s*(\d+)', content, re.MULTILINE)
        if major and minor and revision:
            return f"{major.group(1)}.{minor.group(1)}.{revision.group(1)}"
    except OSError:
        pass
    return ""


def _compare_versions(installed: str, installer: str) -> InstallState:
    """Vergleicht zwei Versionsstrings (semver)."""
    if not installed:
        return InstallState.NOT_FOUND

    try:
        inst_parts = tuple(int(x) for x in installed.split("."))
        new_parts = tuple(int(x) for x in installer.split("."))
    except ValueError:
        return InstallState.SAME_VERSION

    if inst_parts == new_parts:
        return InstallState.SAME_VERSION
    elif inst_parts < new_parts:
        return InstallState.NEWER_AVAILABLE
    else:
        return InstallState.OLDER_AVAILABLE


def _is_service_active(name: str) -> bool:
    """Prueft ob ein systemd-Service laeuft."""
    try:
        result = subprocess.run(
            ["systemctl", "is-active", f"{name}.service"],
            capture_output=True, text=True, timeout=5,
        )
        return result.stdout.strip() == "active"
    except Exception:
        return False


def _is_service_enabled(name: str) -> bool:
    """Prueft ob ein systemd-Service beim Boot startet."""
    try:
        result = subprocess.run(
            ["systemctl", "is-enabled", f"{name}.service"],
            capture_output=True, text=True, timeout=5,
        )
        return result.stdout.strip() == "enabled"
    except Exception:
        return False


def _parse_service_file(path: str, info: ExistingInstallation) -> None:
    """Liest Einstellungen aus der systemd-Service-Datei."""
    try:
        with open(path) as f:
            content = f.read()
        if "--auto-regist" in content:
            info.auto_regist = True
    except OSError:
        pass


def _parse_hostapd(info: ExistingInstallation) -> None:
    """Liest WiFi-Einstellungen aus hostapd.conf."""
    path = "/etc/hostapd/hostapd.conf"
    if not os.path.isfile(path):
        return

    info.wifi_enabled = True
    try:
        with open(path) as f:
            for line in f:
                line = line.strip()
                if line.startswith("ssid="):
                    info.wifi_ssid = line.split("=", 1)[1]
                elif line.startswith("wpa_passphrase="):
                    info.wifi_password = line.split("=", 1)[1]
                elif line.startswith("channel="):
                    try:
                        info.wifi_channel = int(line.split("=", 1)[1])
                    except ValueError:
                        pass
    except OSError:
        pass


def _parse_dnsmasq(info: ExistingInstallation) -> None:
    """Liest DHCP-Einstellungen aus dnsmasq.conf."""
    path = "/etc/dnsmasq.conf"
    if not os.path.isfile(path):
        return

    try:
        with open(path) as f:
            for line in f:
                line = line.strip()
                if line.startswith("dhcp-range="):
                    # dhcp-range=10.10.10.10,10.10.10.50,255.255.255.0,12h
                    parts = line.split("=", 1)[1].split(",")
                    if len(parts) >= 2:
                        info.dhcp_range_start = parts[0]
                        info.dhcp_range_end = parts[1]
                elif line.startswith("address="):
                    # address=/trixyone_srv/10.10.10.1
                    parts = line.split("/")
                    if len(parts) >= 3:
                        info.ap_ip = parts[-1]
    except OSError:
        pass


def _parse_client_config(source: str, info: ExistingInstallation) -> None:
    """Liest Client-Einstellungen aus der Config-Datei."""
    config_file = f"{source}/config/client_config.json"
    if not os.path.isfile(config_file):
        return

    try:
        with open(config_file) as f:
            cfg = json.load(f)
        info.server_host = cfg.get("server_host", info.server_host)
        info.server_port = cfg.get("server_port", info.server_port)
        info.room = cfg.get("room", info.room)
        info.alias = cfg.get("alias", info.alias)
    except (json.JSONDecodeError, OSError):
        pass


def apply_existing_to_config(existing: ExistingInstallation, config: "InstallerConfig") -> None:
    """
    Uebertraegt erkannte Einstellungen auf die InstallerConfig.

    So werden die TUI-Felder mit den aktuellen Werten vorbelegt.
    """
    from .config import InstallerConfig

    config.hostname = existing.hostname or config.hostname
    config.username = existing.username or config.username
    config.install_path = existing.install_path or config.install_path
    config.autostart = existing.autostart
    config.install_service = existing.install_service
    config.auto_regist = existing.auto_regist

    # WiFi
    if existing.wifi_enabled:
        config.wifi_enabled = True
        config.wifi_ssid = existing.wifi_ssid or config.wifi_ssid
        config.wifi_password = existing.wifi_password or config.wifi_password
        config.wifi_channel = existing.wifi_channel
        config.ap_ip = existing.ap_ip
        config.dhcp_range_start = existing.dhcp_range_start
        config.dhcp_range_end = existing.dhcp_range_end

    # Client
    if existing.server_host:
        config.server_host = existing.server_host
    if existing.server_port:
        config.server_port = existing.server_port
    if existing.room:
        config.room = existing.room
    if existing.alias:
        config.alias = existing.alias

    # Plugins
    # Plugins nur fuer server/standalone uebernehmen
    if existing.selected_plugins and config.mode in ("server", "standalone"):
        config.selected_plugins = list(existing.selected_plugins)

    # NLP Backend
    if existing.nlp_backend:
        config.nlp_backend = existing.nlp_backend
    if existing.ollama_host:
        config.ollama_host = existing.ollama_host