#!/usr/bin/env python3 """IPv64 DNS TUI Client Author: Pascal Bouquet License: GNU General Public License v3.0 (GPL-3.0) Year: 2026 """ from textual.app import App, ComposeResult from textual.containers import Horizontal, Vertical from textual.widgets import ( Button, DataTable, Footer, Header, Label, Select, Static, ) from api_client import IPv64Client from widgets import RecordForm class IPv64TUI(App): CSS = """ Screen { layout: vertical; } #top-bar { height: 3; margin: 1 1 0 1; } #main-container { height: 1fr; margin: 0 1; } #left-pane { width: 30%; border: round $primary; padding: 0 1; } #right-pane { width: 70%; border: round $primary; } #form-pane { height: auto; border: round $accent; margin: 0 1 1 1; padding: 1; } .form-row { height: 3; align-vertical: middle; margin-bottom: 1; } .form-row Input, .form-row Select { width: 1fr; } .form-label { width: 12; } .hidden { display: none; } #status { height: 1; margin: 0 1; color: $text-muted; } """ BINDINGS = [ ("r", "refresh_data", "Aktualisieren"), ("e", "edit_selected", "Bearbeiten"), ("d", "delete_selected", "Löschen"), ("q", "quit", "Beenden"), ] def __init__(self, api_key: str): super().__init__() self.client = IPv64Client(api_key) self.domains_cache = {} self.current_domain = None def compose(self) -> ComposeResult: yield Header(show_clock=True) yield Static("IPv64.net DNS Manager", id="top-bar") with Horizontal(id="main-container"): with Vertical(id="left-pane"): yield Label("Domain auswählen:") yield Select([], id="domain-select", prompt="Wähle Domain...") yield Button("Domains neu laden", id="btn-refresh", variant="primary") with Vertical(id="right-pane"): yield DataTable(id="records-table") yield RecordForm() yield Static("Bereit.", id="status") yield Footer() async def on_mount(self) -> None: table = self.query_one("#records-table", DataTable) table.cursor_type = "row" table.add_columns("ID", "Typ", "Subdomain", "Inhalt", "Prio") await self.load_domains() async def load_domains(self) -> None: self.set_status("Lade Domains von IPv64...") try: res = await self.client.get_domains() domains_sub = res.get("subdomains", {}) self.domains_cache = domains_sub options = [(dom, dom) for dom in domains_sub.keys()] dom_select = self.query_one("#domain-select", Select) # Verhindert, dass während des Updates unerwünschte Events feuern with self.prevent(Select.Changed): dom_select.set_options(options) # Vorherige Auswahl beibehalten oder Fallback auf erste Domain if self.current_domain and self.current_domain in self.domains_cache: dom_select.value = self.current_domain elif options: self.current_domain = options[0][0] dom_select.value = self.current_domain else: self.current_domain = None dom_select.value = Select.BLANK if self.current_domain: self.update_table(self.current_domain) else: self.query_one("#records-table", DataTable).clear() self.set_status(f"{len(options)} Domains geladen.") except Exception as e: self.set_status(f"Fehler: {e}") def update_table(self, domain_name: str) -> None: table = self.query_one("#records-table", DataTable) table.clear() domain_data = self.domains_cache.get(domain_name, {}) records = domain_data.get("records", []) for rec in records: rec_id = str(rec.get("record_id", "")) table.add_row( rec_id, str(rec.get("type", "-")), str(rec.get("praefix", "@")), str(rec.get("content", "-")), str(rec.get("prio", "-")), key=rec_id, ) def on_select_changed(self, event: Select.Changed) -> None: if event.select.id == "domain-select": if event.value != Select.BLANK: self.current_domain = str(event.value) self.update_table(self.current_domain) # Formular zurücksetzen beim Domain-Wechsel self.query_one(RecordForm).reset_mode() def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: """Öffnet den Bearbeitungsmodus auch bei Doppel-/Enter-Klick auf eine Zeile.""" self.action_edit_selected() async def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "btn-refresh": await self.load_domains() elif event.button.id == "btn-submit": await self.submit_record() elif event.button.id == "btn-cancel": self.query_one(RecordForm).reset_mode() self.set_status("Bearbeitung abgebrochen.") def action_edit_selected(self) -> None: table = self.query_one("#records-table", DataTable) if table.row_count == 0 or table.cursor_row is None: self.set_status("Kein Eintrag zum Bearbeiten ausgewählt.") return row_data = table.get_row_at(table.cursor_row) rec_id, typ, subdomain, content, prio = ( str(row_data[0]), str(row_data[1]), str(row_data[2]), str(row_data[3]), str(row_data[4]), ) form = self.query_one(RecordForm) form.set_edit_mode(rec_id, subdomain, typ, content, prio) self.set_status(f"Bearbeite Record ID {rec_id}. Daten im Formular anpassen.") async def submit_record(self) -> None: if not self.current_domain: self.set_status("Bitte zuerst eine Domain auswählen.") return form = self.query_one(RecordForm) data = form.get_data() if not data["content"]: self.set_status("Content darf nicht leer sein.") return old_record_id = data["record_id"] try: # Wenn wir im Bearbeitungsmodus sind: Alten Eintrag zuerst löschen if old_record_id: self.set_status(f"Aktualisiere Record {old_record_id}...") del_res = await self.client.delete_record_by_id(old_record_id) if del_res.get("status") not in ("success", None) and del_res.get("info") != "success": self.set_status(f"Fehler beim Ersetzen des Records: {del_res}") return else: self.set_status("Erstelle neuen Record...") res = await self.client.add_record( self.current_domain, data["subdomain"], data["type"], data["content"], data["prio"], ) if res.get("status") == "success" or res.get("info") == "success": msg = "Record aktualisiert." if old_record_id else "Record angelegt." self.set_status(msg) form.reset_mode() await self.load_domains() else: self.set_status(f"API-Rückmeldung: {res}") except Exception as e: self.set_status(f"Fehler: {e}") async def action_delete_selected(self) -> None: table = self.query_one("#records-table", DataTable) if table.row_count == 0 or table.cursor_row is None: self.set_status("Keine Einträge vorhanden.") return current_row_idx = table.cursor_row row_data = table.get_row_at(current_row_idx) record_id = str(row_data[0]) if not record_id or record_id == "-": self.set_status("Keine gültige Record-ID gefunden.") return self.set_status(f"Lösche Record ID {record_id}...") try: res = await self.client.delete_record_by_id(record_id) if res.get("status") == "success" or res.get("info") == "success": self.set_status(f"Record {record_id} gelöscht.") # Falls wir diesen Record gerade im Formular bearbeitet hatten: Formular leeren form = self.query_one(RecordForm) if form.edit_record_id == record_id: form.reset_mode() await self.load_domains() else: self.set_status(f"API-Rückmeldung: {res}") except Exception as e: self.set_status(f"Fehler beim Löschen: {e}") def action_refresh_data(self) -> None: self.run_worker(self.load_domains()) def set_status(self, text: str) -> None: self.query_one("#status", Static).update(text)