# -*- 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