# -*- coding: utf-8 -*- """ F6: Installation — ProgressBar, Log, Start/Abbrechen/Deinstallieren. Zeigt kontextabhaengige Buttons: - Neuinstallation: "Installation starten" - Gleiche Version: "Erneut installieren" + "Deinstallieren" - Neuere Version: "Update auf {version}" + "Deinstallieren" - Aeltere Version: "Downgrade auf {version}" + "Deinstallieren" """ from __future__ import annotations import os from typing import TYPE_CHECKING from textual.containers import Horizontal from textual.widgets import Button, Label, RichLog, Static from .base import InstallerView if TYPE_CHECKING: from ..detect import ExistingInstallation from ..runner import InstallRunner class InstallView(InstallerView): """Installations-Ansicht (F6).""" TITLE = "Installation" VIEW_ID = "install" MODES = ["server", "client", "standalone", "config"] def __init__(self, installer_config, existing: "ExistingInstallation | None" = None, **kwargs) -> None: super().__init__(installer_config, **kwargs) self._existing = existing self._runner: InstallRunner | None = None self._running = False def compose(self): yield Static("── Installation ──", classes="section-header") # Status-Info basierend auf bestehender Installation yield Label("", id="install-info") yield Label("", id="install-status") yield Label(self._render_bar(0.0), id="install-progress-bar") yield Label("") # Buttons — nebeneinander, kontextabhaengig from ..detect import InstallState has_existing = self._existing and self._existing.found with Horizontal(id="install-buttons"): if not has_existing: yield Button("Installation starten", variant="success", id="btn-start") else: state = self._existing.state if state == InstallState.SAME_VERSION: yield Button("Erneut installieren", variant="warning", id="btn-start") elif state == InstallState.NEWER_AVAILABLE: yield Button( f"Update auf {self._config.version}", variant="success", id="btn-start", ) elif state == InstallState.OLDER_AVAILABLE: yield Button( f"Downgrade auf {self._config.version}", variant="warning", id="btn-start", ) yield Button("Deinstallieren", variant="error", id="btn-uninstall") yield Button("Abbrechen", variant="default", id="btn-cancel", disabled=True) yield Label("") yield Static("── Log ──", classes="section-header") yield RichLog(highlight=True, markup=True, id="install-log", min_width=60) async def on_activate(self) -> None: """Zeigt Info ueber bestehende Installation.""" info_label = self.query_one("#install-info", Label) from ..detect import InstallState if not self._existing or not self._existing.found: info_label.update("Keine bestehende Installation erkannt — Neuinstallation") return ex = self._existing lines = [f"Bestehende Installation erkannt:"] lines.append(f" Version: {ex.installed_version}") lines.append(f" Pfad: {ex.source_path}") lines.append(f" Service: {ex.service_name} ({'aktiv' if ex.service_active else 'inaktiv'})") if ex.state == InstallState.SAME_VERSION: lines.append(f"\n Gleiche Version — Erneut installieren aktualisiert Einstellungen") elif ex.state == InstallState.NEWER_AVAILABLE: lines.append(f"\n Update verfuegbar: {ex.installed_version} -> {self._config.version}") elif ex.state == InstallState.OLDER_AVAILABLE: lines.append(f"\n Downgrade: {ex.installed_version} -> {self._config.version}") info_label.update("\n".join(lines)) def _log(self, message: str) -> None: """Schreibt eine Nachricht ins Log.""" try: log = self.query_one("#install-log", RichLog) log.write(message) except Exception: pass @staticmethod def _render_bar(fraction: float, width: int = 40) -> str: """Rendert eine klassische ASCII-ProgressBar.""" pct = int(fraction * 100) filled = int(width * fraction) bar = "\u2588" * filled + "\u2591" * (width - filled) return f"[{bar}] {pct}%" def _update_progress(self, fraction: float, step_name: str) -> None: """Aktualisiert ProgressBar und Status-Label.""" try: bar_label = self.query_one("#install-progress-bar", Label) bar_label.update(self._render_bar(fraction)) status = self.query_one("#install-status", Label) if fraction >= 1.0: status.update("Abgeschlossen!") else: status.update(step_name) except Exception: pass def _set_buttons_running(self, running: bool) -> None: """Schaltet Buttons fuer laufende Operation um.""" for btn_id in ("btn-start", "btn-uninstall"): try: self.query_one(f"#{btn_id}", Button).disabled = running except Exception: pass try: self.query_one("#btn-cancel", Button).disabled = not running except Exception: pass async def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "btn-start": await self._start_installation() elif event.button.id == "btn-uninstall": await self._start_uninstall() elif event.button.id == "btn-cancel": self._cancel() async def _start_installation(self) -> None: """Startet Installation/Reinstall/Update.""" # 1. Validierung aller Views — Fehler sammeln, nicht springen app = self.app errors: list[tuple[str, str]] = [] for view in getattr(app, "_views", []): try: error = view.validate() if error: errors.append((view.TITLE, error)) except Exception as e: errors.append((view.TITLE, f"Validierung fehlgeschlagen: {e}")) if errors: self._log("Validierungsfehler:\n") for tab_name, error_msg in errors: self._log(f" [{tab_name}] {error_msg}") self._log(f"\nBitte die markierten Tabs korrigieren.") return # 2. Preflight-Check try: from ..preflight import run_preflight, format_preflight self._log("Preflight-Check...") self._log("") preflight = run_preflight(self._config) self._log(format_preflight(preflight)) self._log("") if not preflight.passed: self._log("Abgebrochen — Preflight-Check fehlgeschlagen.") return if preflight.warnings: self._log("Warnungen vorhanden — wird fortgesetzt.\n") except Exception as e: self._log(f"Preflight-Fehler: {e}") return # 3. Service stoppen falls aktiv if self._existing and self._existing.service_active: self._log("Stoppe laufenden Service...") from ..runner import InstallRunner temp_runner = InstallRunner(self._config, self._log, self._update_progress) await temp_runner._run_sudo( "systemctl", "stop", f"{self._existing.service_name}.service", ) self._log(" -> Service gestoppt\n") # 4. Runner starten self._set_buttons_running(True) self._running = True from ..runner import InstallRunner self._runner = InstallRunner( config=self._config, log_callback=self._log, progress_callback=self._update_progress, ) success = await self._runner.run() self._running = False self._set_buttons_running(False) if success: # Service starten falls gewuenscht if self._config.install_service and self._config.autostart: self._log("\nStarte Service...") await self._runner._run_sudo( "systemctl", "start", f"trixy-{self._config.mode}.service", ) self._log(" -> Service gestartet") self._log("\nInstallation erfolgreich!") else: self._log("\nInstallation fehlgeschlagen oder abgebrochen.") async def _start_uninstall(self) -> None: """Deinstalliert Trixy komplett inkl. aller System-Konfigurationen.""" if not self._existing or not self._existing.found: self._log("Keine Installation zum Deinstallieren gefunden.") return self._log("=== Trixy Deinstallation ===\n") self._set_buttons_running(True) self._running = True from ..runner import InstallRunner self._runner = InstallRunner(self._config, self._log, self._update_progress) runner = self._runner ex = self._existing total_steps = 6 step = 0 # 1. Service stoppen und entfernen step += 1 self._update_progress(step / total_steps, "Service entfernen") self._log(f"[{step}/{total_steps}] Service stoppen und entfernen...") if ex.service_active: await runner._run_sudo("systemctl", "stop", f"{ex.service_name}.service") self._log(" -> Service gestoppt") service_file = f"/etc/systemd/system/{ex.service_name}.service" if os.path.isfile(service_file): await runner._run_sudo( "systemctl", "disable", f"{ex.service_name}.service", check=False, ) await runner._run_sudo("rm", "-f", service_file) await runner._run_sudo("systemctl", "daemon-reload") self._log(" -> Service-Datei entfernt") else: self._log(" -> Keine Service-Datei vorhanden") if self._cancelled(): return # 2. WiFi-Hotspot entfernen (hostapd + dnsmasq) step += 1 self._update_progress(step / total_steps, "WiFi-Konfiguration") self._log(f"\n[{step}/{total_steps}] WiFi-Hotspot Konfiguration entfernen...") if ex.wifi_enabled: await runner._run_sudo("systemctl", "stop", "hostapd", check=False) await runner._run_sudo("systemctl", "disable", "hostapd", check=False) if os.path.isfile("/etc/hostapd/hostapd.conf"): await runner._run_sudo("rm", "-f", "/etc/hostapd/hostapd.conf") self._log(" -> /etc/hostapd/hostapd.conf entfernt") await runner._run_sudo("systemctl", "stop", "dnsmasq", check=False) await runner._run_sudo("systemctl", "disable", "dnsmasq", check=False) if os.path.isfile("/etc/dnsmasq.conf.orig"): await runner._run_sudo( "mv", "/etc/dnsmasq.conf.orig", "/etc/dnsmasq.conf", ) self._log(" -> /etc/dnsmasq.conf wiederhergestellt (Backup)") elif os.path.isfile("/etc/dnsmasq.conf"): await runner._run_sudo("rm", "-f", "/etc/dnsmasq.conf") self._log(" -> /etc/dnsmasq.conf entfernt") self._log(" -> hostapd + dnsmasq deaktiviert") else: self._log(" -> Kein WiFi-Hotspot konfiguriert") if self._cancelled(): return # 3. Netzwerk-Konfiguration bereinigen step += 1 self._update_progress(step / total_steps, "Netzwerk-Konfiguration") self._log(f"\n[{step}/{total_steps}] Netzwerk-Konfiguration bereinigen...") if ex.wifi_enabled: dhcpcd = "/etc/dhcpcd.conf" if os.path.isfile(dhcpcd): await runner._run_sudo( "sed", "-i", "/^# Trixy AP/,/^$/d", dhcpcd, ) self._log(" -> Trixy-AP Eintrag aus /etc/dhcpcd.conf entfernt") await runner._run_sudo( "iptables", "-t", "nat", "-D", "POSTROUTING", "-o", "eth0", "-j", "MASQUERADE", check=False, ) if os.path.isfile("/etc/iptables.ipv4.nat"): await runner._run_sudo("rm", "-f", "/etc/iptables.ipv4.nat") self._log(" -> NAT-Regeln entfernt") await runner._run_sudo( "sed", "-i", "s/^net.ipv4.ip_forward=1/#net.ipv4.ip_forward=1/", "/etc/sysctl.conf", check=False, ) await runner._run_sudo( "sysctl", "-w", "net.ipv4.ip_forward=0", check=False, ) self._log(" -> IP-Forwarding deaktiviert") else: self._log(" -> Keine Netzwerk-Aenderungen vorhanden") if self._cancelled(): return # 4. Hostname zuruecksetzen step += 1 self._update_progress(step / total_steps, "Hostname") self._log(f"\n[{step}/{total_steps}] Hostname...") current_hostname = "" try: current_hostname = os.uname().nodename except Exception: pass if current_hostname: default_names = {"trixyone", "trixy", "trixysat", "trixycfg"} if current_hostname.lower() in default_names: await runner._run_sudo("hostnamectl", "set-hostname", "raspberrypi") self._log(" -> Hostname auf 'raspberrypi' zurueckgesetzt") else: self._log(f" -> Hostname '{current_hostname}' beibehalten (kein Trixy-Name)") else: self._log(f" -> Hostname unveraendert") if self._cancelled(): return # 5. Installationsverzeichnis entfernen step += 1 self._update_progress(step / total_steps, "Dateien entfernen") self._log(f"\n[{step}/{total_steps}] Installationsverzeichnis entfernen...") install_path = ex.install_path if install_path and os.path.isdir(install_path): self._log(f" Entferne {install_path}...") await runner._run_sudo("rm", "-rf", install_path) self._log(" -> Verzeichnis entfernt") else: self._log(f" -> Verzeichnis nicht gefunden: {install_path}") if self._cancelled(): return # 6. Zusammenfassung step += 1 self._update_progress(1.0, "Fertig") self._log(f"\n[{step}/{total_steps}] Deinstallation abgeschlossen:") self._log(f" Service: {ex.service_name} entfernt") self._log(f" Verzeichnis: {install_path} entfernt") if ex.wifi_enabled: self._log(f" WiFi-Hotspot: hostapd + dnsmasq deaktiviert") self._log(f" Netzwerk: NAT + IP-Forwarding entfernt") self._log(f"\n Hinweis: apt-Pakete (ffmpeg, portaudio, etc.) wurden") self._log(f" nicht entfernt — diese werden ggf. von anderen") self._log(f" Programmen benoetigt.") self._log("\n=== Deinstallation abgeschlossen ===") self._running = False self._set_buttons_running(False) def _cancelled(self) -> bool: """Prueft ob abgebrochen wurde.""" if self._runner and self._runner._cancelled: self._log("\n[ABBRUCH] Deinstallation abgebrochen.") self._running = False self._set_buttons_running(False) return True return False def _cancel(self) -> None: """Bricht die laufende Operation ab.""" if self._runner: self._runner.cancel() self._log("\nAbbruch angefordert...")