# -*- coding: utf-8 -*- """ Schedule-View: Tabellarische Liste aller Schedule-Jobs. Zeigt Jobs in einer DataTable mit Name, Trigger, Status, Ausfuehrungszeiten. Enter/Klick auf eine Zeile oeffnet die Job-Detailansicht. Buttons zum Aktivieren/Deaktivieren, Erstellen und Loeschen. """ from __future__ import annotations from textual.app import ComposeResult from textual.containers import Horizontal from textual.widgets import Static, Button, DataTable from trixy_core.tui.views.base import MainView class ScheduleView(MainView): """ F6: Schedule-Ansicht. Zeigt alle Schedule-Jobs tabellarisch. Zeilen-Auswahl oeffnet Details, Buttons fuer Enable/Disable/Delete/New. """ TITLE = "Schedule" VIEW_ID = "schedule" def __init__(self, **kwargs) -> None: super().__init__(**kwargs) self._jobs: list[dict] = [] self._refresh_counter: int = 0 def compose(self) -> ComposeResult: yield Static("", id="schedule-header") yield DataTable(id="schedule-table", cursor_type="row") with Horizontal(id="schedule-actions"): yield Button("Neu", id="schedule-btn-new", variant="success") yield Button("Aktivieren", id="schedule-btn-enable", variant="success") yield Button("Deaktivieren", id="schedule-btn-disable", variant="warning") yield Button("Loeschen", id="schedule-btn-delete", variant="error") yield Static("", id="schedule-status") async def on_activate(self) -> None: """Laedt die Job-Liste.""" table = self.query_one("#schedule-table", DataTable) if not table.columns: table.add_columns("Name", "Trigger", "Status", "Letzte Ausfuehrung", "Naechste") await self._load_jobs() async def refresh_data(self) -> None: """Periodisches Update (alle 10 Sekunden).""" self._refresh_counter += 1 if self._refresh_counter % 5 == 0: await self._load_jobs() async def _load_jobs(self) -> None: """Laedt die Job-Liste vom Server.""" if not self.connection: return try: response = await self.connection.request_schedule_jobs() except Exception: self._set_status("Fehler beim Laden der Jobs", error=True) return if response is None: return self._jobs = response.jobs self._update_header() self._update_table() def _update_header(self) -> None: """Aktualisiert die Kopfzeile.""" header = self.query_one("#schedule-header", Static) total = len(self._jobs) enabled = sum(1 for j in self._jobs if j.get("enabled", False)) header.update(f"[bold]Schedule:[/bold] {enabled}/{total} aktiv") def _update_table(self) -> None: """Aktualisiert die Job-Tabelle.""" table = self.query_one("#schedule-table", DataTable) table.clear() sorted_jobs = sorted(self._jobs, key=lambda j: j.get("name", "")) for job in sorted_jobs: name = job.get("name", job.get("job_id", "")) trigger_type = job.get("trigger_type", "—") enabled = job.get("enabled", False) state = job.get("state", "") if enabled: status = f"[green]Aktiv[/green] ({state})" else: status = f"[dim]Deaktiviert[/dim] ({state})" last_run = job.get("last_run", "—") or "—" next_run = job.get("next_run", "—") or "—" # Datum-Strings kuerzen if last_run != "—" and "T" in last_run: last_run = last_run.replace("T", " ")[:19] if next_run != "—" and "T" in next_run: next_run = next_run.replace("T", " ")[:19] table.add_row( name, trigger_type, status, last_run, next_run, key=job.get("job_id", name), ) def _get_selected_job(self) -> dict | None: """Gibt den aktuell markierten Job zurueck.""" try: table = self.query_one("#schedule-table", DataTable) row_key, _ = table.coordinate_to_cell_key(table.cursor_coordinate) job_id = str(row_key.value) if row_key else "" if job_id: for j in self._jobs: if j.get("job_id") == job_id: return j except Exception: pass return None async def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: """Enter auf einer Zeile: Oeffnet die Detail-Ansicht.""" row_key = str(event.row_key.value) if event.row_key else "" job_data = None for j in self._jobs: if j.get("job_id") == row_key: job_data = j break if not job_data: return try: app = self.app if hasattr(app, "open_schedule_detail"): await app.open_schedule_detail(job_data) except Exception: pass async def on_button_pressed(self, event: Button.Pressed) -> None: """Verarbeitet Button-Klicks.""" btn_id = event.button.id if not self.connection: return # "Neu" benoetigt kein ausgewaehltes Plugin if btn_id == "schedule-btn-new": try: app = self.app if hasattr(app, "open_schedule_detail"): await app.open_schedule_detail({"job_id": "", "name": "Neuer Job"}) except Exception: pass return job = self._get_selected_job() if not job: self._set_status("Kein Job ausgewaehlt", error=True) return job_id = job.get("job_id", "") action = "" if btn_id == "schedule-btn-enable": action = "enable" elif btn_id == "schedule-btn-disable": action = "disable" elif btn_id == "schedule-btn-delete": action = "delete" if action: await self._schedule_action(action, job_id) async def _schedule_action(self, action: str, job_id: str) -> None: """Fuehrt eine Schedule-Aktion aus und aktualisiert die Liste.""" try: result = await self.connection.schedule_action(action, job_id) if result and result.success: self._set_status(result.message) await self._load_jobs() else: error = result.error if result else "Unbekannter Fehler" self._set_status(f"Fehler: {error}", error=True) except Exception as e: self._set_status(f"Fehler: {e}", error=True) def _set_status(self, text: str, error: bool = False) -> None: """Setzt die Statuszeile.""" status = self.query_one("#schedule-status", Static) if error: status.update(f"[red]{text}[/red]") else: status.update(f"[green]{text}[/green]")