# -*- coding: utf-8 -*- """ Schedule SubView: Info (F1). Interaktive Ansicht zum Bearbeiten der allgemeinen Job-Informationen: Name, Beschreibung, Aktiviert-Status und erweiterte Konfiguration. Bei bestehenden Jobs werden zusaetzlich Laufzeitstatistiken angezeigt. """ from __future__ import annotations from textual.app import ComposeResult from textual.containers import Horizontal, Vertical from textual.widgets import Button, Input, Static, Switch from trixy_core.tui.views.base import MainView from trixy_core.tui.widgets.status_table import StatusTable class ScheduleInfoView(MainView): """F1 im Schedule-Kontext: Allgemeine Job-Informationen (editierbar).""" TITLE = "Info" VIEW_ID = "sched_info" DEFAULT_CSS = """ ScheduleInfoView { padding: 1 2; } ScheduleInfoView .section-box { margin-bottom: 1; } ScheduleInfoView .section-header { margin-bottom: 1; } ScheduleInfoView .field-label { margin-top: 1; height: 1; } ScheduleInfoView .field-row { height: 3; align: left middle; } ScheduleInfoView .field-row .field-label { width: auto; margin-right: 2; margin-top: 0; } ScheduleInfoView #sched-save-btn { margin-top: 1; } ScheduleInfoView #sched-save-status { margin-top: 1; height: 1; } """ def __init__(self, **kwargs) -> None: super().__init__(**kwargs) self._job_data: dict = {} self._detail_data: object | None = None def set_job(self, job_data: dict) -> None: """Setzt die Job-Basisdaten und verwirft gecachte Details.""" self._job_data = job_data self._detail_data = None def compose(self) -> ComposeResult: # --- Allgemein --- with Vertical(classes="section-box"): yield Static("[bold]Allgemein[/bold]", classes="section-header") yield Static("Name", classes="field-label") yield Input(id="sched-info-name", placeholder="Job-Name") yield Static("Beschreibung", classes="field-label") yield Input(id="sched-info-desc", placeholder="Beschreibung") with Horizontal(classes="field-row"): yield Static("Aktiviert", classes="field-label") yield Switch(id="sched-info-enabled", value=True) # --- Konfiguration --- with Vertical(classes="section-box"): yield Static("[bold]Konfiguration[/bold]", classes="section-header") yield Static("Max. Ausfuehrungen (0 = unbegrenzt)", classes="field-label") yield Input( id="sched-info-max-runs", placeholder="0", type="integer", ) yield Static("Max. Fehler", classes="field-label") yield Input( id="sched-info-max-failures", placeholder="3", type="integer", ) yield Static("Timeout (Sek.)", classes="field-label") yield Input( id="sched-info-timeout", placeholder="60.0", type="number", ) yield Static("Prioritaet", classes="field-label") yield Input( id="sched-info-priority", placeholder="0", type="integer", ) with Horizontal(classes="field-row"): yield Static("Bei Start ausfuehren", classes="field-label") yield Switch(id="sched-info-run-on-startup", value=False) yield Static("Tags (kommagetrennt)", classes="field-label") yield Input(id="sched-info-tags", placeholder="tag1, tag2, ...") # --- Statistik (nur bei bestehendem Job sichtbar) --- with Vertical(id="sched-info-stats-section", classes="section-box"): yield Static("[bold]Statistik[/bold]", classes="section-header") yield StatusTable(id="sched-info-stats") # --- Speichern --- yield Button("Speichern", id="sched-save-btn", variant="success") yield Static("", id="sched-save-status") async def on_activate(self) -> None: """Laedt Detail-Daten und befuellt die Eingabefelder.""" job_id = self._job_data.get("job_id", "") # Statistik-Bereich nur anzeigen wenn bestehender Job stats_section = self.query_one("#sched-info-stats-section", Vertical) stats_section.display = bool(job_id) if job_id: await self._load_detail() else: # Neuer Job: Felder mit Standardwerten befuellen self._populate_defaults() async def refresh_data(self) -> None: """Periodisches Update – nur Statistik aktualisieren bei bestehendem Job.""" job_id = self._job_data.get("job_id", "") if job_id and self._detail_data is not None: await self._load_detail_stats_only() async def _load_detail(self) -> None: """Laedt detaillierte Job-Daten vom Server und befuellt alle Felder.""" if not self.connection or not self._job_data: return job_id = self._job_data.get("job_id", "") if not job_id: return try: detail = await self.connection.request_schedule_detail(job_id) except Exception: return if detail is None or detail.error: return self._detail_data = detail self._populate_from_detail(detail) def _populate_defaults(self) -> None: """Befuellt Felder mit Standardwerten fuer einen neuen Job.""" name = self._job_data.get("name", "") self.query_one("#sched-info-name", Input).value = name self.query_one("#sched-info-desc", Input).value = "" self.query_one("#sched-info-enabled", Switch).value = True self.query_one("#sched-info-max-runs", Input).value = "0" self.query_one("#sched-info-max-failures", Input).value = "3" self.query_one("#sched-info-timeout", Input).value = "60.0" self.query_one("#sched-info-priority", Input).value = "0" self.query_one("#sched-info-run-on-startup", Switch).value = False self.query_one("#sched-info-tags", Input).value = "" def _populate_from_detail(self, detail: object) -> None: """Befuellt alle Felder aus den geladenen Detail-Daten.""" # Allgemein self.query_one("#sched-info-name", Input).value = detail.name or "" self.query_one("#sched-info-desc", Input).value = detail.description or "" self.query_one("#sched-info-enabled", Switch).value = detail.enabled # Konfiguration cfg = detail.config or {} self.query_one("#sched-info-max-runs", Input).value = str( cfg.get("max_runs", 0) ) self.query_one("#sched-info-max-failures", Input).value = str( cfg.get("max_failures", 3) ) self.query_one("#sched-info-timeout", Input).value = str( cfg.get("timeout_seconds", 60.0) ) self.query_one("#sched-info-priority", Input).value = str( cfg.get("priority", 0) ) self.query_one("#sched-info-run-on-startup", Switch).value = bool( cfg.get("run_on_startup", False) ) tags = cfg.get("tags", []) self.query_one("#sched-info-tags", Input).value = ", ".join(tags) # Statistik self._update_stats(detail) def _update_stats(self, detail: object) -> None: """Aktualisiert die Statistik-Tabelle (nur-lesend).""" last_run = ( detail.last_run.replace("T", " ")[:19] if detail.last_run else "—" ) next_run = ( detail.next_run.replace("T", " ")[:19] if detail.next_run else "—" ) stats = self.query_one("#sched-info-stats", StatusTable) stats.update_data([ ("Ausfuehrungen", str(detail.run_count)), ("Fehler", str(detail.failure_count)), ("Letzte Ausfuehrung", last_run), ("Naechste Ausfuehrung", next_run), ]) async def _load_detail_stats_only(self) -> None: """Laedt nur die Statistik-Daten neu (ohne Formularfelder zu ueberschreiben).""" if not self.connection or not self._job_data: return job_id = self._job_data.get("job_id", "") if not job_id: return try: detail = await self.connection.request_schedule_detail(job_id) except Exception: return if detail is None or detail.error: return self._detail_data = detail self._update_stats(detail) def collect_data(self) -> dict: """ Sammelt die eingegebenen Daten aus allen Formularfeldern. Returns: dict mit Schluesseln: name, description, enabled, config. config enthaelt: max_runs, max_failures, timeout_seconds, priority, run_on_startup, tags. """ name = self.query_one("#sched-info-name", Input).value.strip() description = self.query_one("#sched-info-desc", Input).value.strip() enabled = self.query_one("#sched-info-enabled", Switch).value # Konfigurations-Felder auslesen und konvertieren try: max_runs = int(self.query_one("#sched-info-max-runs", Input).value) except (ValueError, TypeError): max_runs = 0 try: max_failures = int( self.query_one("#sched-info-max-failures", Input).value ) except (ValueError, TypeError): max_failures = 3 try: timeout_seconds = float( self.query_one("#sched-info-timeout", Input).value ) except (ValueError, TypeError): timeout_seconds = 60.0 try: priority = int(self.query_one("#sched-info-priority", Input).value) except (ValueError, TypeError): priority = 0 run_on_startup = self.query_one("#sched-info-run-on-startup", Switch).value tags_raw = self.query_one("#sched-info-tags", Input).value.strip() tags = [t.strip() for t in tags_raw.split(",") if t.strip()] if tags_raw else [] return { "name": name, "description": description, "enabled": enabled, "config": { "max_runs": max_runs, "max_failures": max_failures, "timeout_seconds": timeout_seconds, "priority": priority, "run_on_startup": run_on_startup, "tags": tags, }, } async def on_button_pressed(self, event: Button.Pressed) -> None: """Handler fuer Button-Klicks.""" if event.button.id == "sched-save-btn": await self._save() async def _save(self) -> None: """Aggregiert Daten aller Schedule-SubViews und speichert den Job.""" status = self.query_one("#sched-save-status", Static) if not self.connection: status.update("[red]Keine Verbindung zum Server.[/red]") return # Daten aus diesem View sammeln info_data = self.collect_data() if not info_data["name"]: status.update("[red]Name darf nicht leer sein.[/red]") return # Daten aus Geschwister-Views sammeln (Trigger, Conditions, Actions) trigger: dict = {} conditions: list[dict] = [] actions: list[dict] = [] view_map = self.app._view_map # type: ignore[attr-defined] trigger_view = view_map.get("sched_trigger") if trigger_view and hasattr(trigger_view, "collect_data"): trigger = trigger_view.collect_data() conditions_view = view_map.get("sched_conditions") if conditions_view and hasattr(conditions_view, "collect_data"): conditions = conditions_view.collect_data() actions_view = view_map.get("sched_actions") if actions_view and hasattr(actions_view, "collect_data"): actions = actions_view.collect_data() job_id = self._job_data.get("job_id", "") try: status.update("[yellow]Speichern...[/yellow]") result = await self.connection.save_schedule_job( job_id=job_id, name=info_data["name"], description=info_data["description"], trigger=trigger, conditions=conditions, actions=actions, config=info_data["config"], ) except Exception as exc: status.update(f"[red]Fehler: {exc}[/red]") return if result is None or (hasattr(result, "error") and result.error): error_msg = getattr(result, "error", "Unbekannter Fehler") status.update(f"[red]Fehler: {error_msg}[/red]") return # Bei neuem Job die zurueckgegebene ID uebernehmen if not job_id and hasattr(result, "job_id") and result.job_id: self._job_data["job_id"] = result.job_id # Statistik-Bereich sichtbar machen stats_section = self.query_one("#sched-info-stats-section", Vertical) stats_section.display = True status.update("[green]Gespeichert.[/green]")