paase
d15c927f55
- Implement modular Textual TUI (main, app, widgets, config, api_client) - Integrate IPv64.net REST API client using httpx - Support record listing, creation, in-place edit, and deletion - Add flexible token resolution (.env, env export, interactive prompt) - Add bilingual README (DE/EN) and project configuration
275 lines
9.1 KiB
Python
Executable File
275 lines
9.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""IPv64 DNS TUI Client
|
|
Author: Pascal Bouquet <pascal@ptbos.de>
|
|
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)
|