# -*- coding: utf-8 -*- """ Schedule SubView: Conditions (F3). Interaktive Verwaltung der Bedingungen (Conditions) eines Schedule-Jobs. Ermoeglicht Hinzufuegen, Bearbeiten und Entfernen von Bedingungen mit dynamischen Formularfeldern je nach Bedingungstyp. """ from __future__ import annotations from textual.app import ComposeResult from textual.containers import Vertical, Horizontal from textual.widgets import Static, Input, Switch, Button, Select, DataTable, TextArea from trixy_core.tui.views.base import MainView # Wochentags-Labels fuer den "weekdays"-Feldtyp _WEEKDAY_LABELS: list[str] = ["Mo", "Di", "Mi", "Do", "Fr", "Sa", "So"] def _resolve_field_type(raw: str | dict) -> str: """Extrahiert den FieldType-String aus einem Wert oder Dict.""" if isinstance(raw, dict): return str(raw.get("value", "text")) return str(raw) class ScheduleConditionView(MainView): """F3 im Schedule-Kontext: Bedingungen des Jobs (interaktiv).""" TITLE = "Conditions" VIEW_ID = "sched_conditions" def __init__(self, **kwargs) -> None: super().__init__(**kwargs) self._job_data: dict = {} self._form_meta: dict = {} self._conditions: list[dict] = [] self._editing_index: int | None = None self._selected_row_index: int | None = None # ------------------------------------------------------------------ # Oeffentliche API # ------------------------------------------------------------------ def set_job(self, job_data: dict) -> None: """Setzt die Job-Basisdaten.""" self._job_data = job_data self._conditions = [] self._editing_index = None self._selected_row_index = None def set_form_meta(self, meta: dict) -> None: """Setzt die FormMeta-Definition fuer Condition-Typen.""" self._form_meta = meta or {} def collect_data(self) -> list[dict]: """Gibt die aktuelle Liste aller Conditions zurueck.""" return list(self._conditions) # ------------------------------------------------------------------ # Compose # ------------------------------------------------------------------ def compose(self) -> ComposeResult: yield Static("[bold]Bedingungen[/bold]", classes="section-header") yield DataTable(id="sched-cond-table", cursor_type="row") yield Static("", id="sched-cond-info") with Horizontal(classes="button-bar"): yield Button("Hinzufuegen", id="sched-cond-add", variant="success") yield Button("Entfernen", id="sched-cond-remove", variant="error") # Bearbeitungsbereich — wird dynamisch befuellt with Vertical(id="sched-cond-edit-section"): yield Static("", id="sched-cond-edit-title") yield Static("Typ:", classes="field-label") yield Select( [], id="sched-cond-type-select", prompt="Bedingungstyp waehlen...", ) yield Vertical(id="sched-cond-fields") yield Button( "Uebernehmen", id="sched-cond-apply", variant="primary", ) yield Static("", id="sched-cond-status") # ------------------------------------------------------------------ # Lifecycle # ------------------------------------------------------------------ async def on_activate(self) -> None: """Laedt die bestehenden Conditions beim Aktivieren.""" table = self.query_one("#sched-cond-table", DataTable) if not table.columns: table.add_columns("Typ", "Name", "Aktiv") # Detail-Daten laden falls Job-ID vorhanden await self._load_detail() # Typ-Optionen aus FormMeta befuellen self._populate_type_select() # Bearbeitungsbereich initial verstecken self._hide_edit_section() async def _load_detail(self) -> None: """Laedt detaillierte Job-Daten vom Server und befuellt _conditions.""" 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 if detail.conditions: self._conditions = list(detail.conditions) self._refresh_table() # ------------------------------------------------------------------ # Tabelle # ------------------------------------------------------------------ def _refresh_table(self) -> None: """Aktualisiert die Conditions-Tabelle und die Info-Zeile.""" table = self.query_one("#sched-cond-table", DataTable) table.clear() info = self.query_one("#sched-cond-info", Static) if not self._conditions: info.update( "[dim]Keine Bedingungen — Job wird immer ausgefuehrt " "wenn Trigger feuert[/dim]" ) else: info.update( f"[bold]{len(self._conditions)}[/bold] Bedingung(en) " "— alle muessen erfuellt sein" ) for cond in self._conditions: cond_type = cond.get("type", "Unbekannt") display_name = self._display_name_for_type(cond_type) name = cond.get("name", "—") enabled = "Ja" if cond.get("enabled", True) else "Nein" table.add_row(display_name, name, enabled) def _display_name_for_type(self, cond_type: str) -> str: """Gibt den Anzeigenamen eines Condition-Typs zurueck.""" meta = self._form_meta.get(cond_type) if meta and isinstance(meta, dict): return meta.get("display_name", cond_type) return cond_type # ------------------------------------------------------------------ # Bearbeitungsbereich # ------------------------------------------------------------------ def _show_edit_section(self, title: str) -> None: """Macht den Bearbeitungsbereich sichtbar.""" section = self.query_one("#sched-cond-edit-section") section.display = True title_widget = self.query_one("#sched-cond-edit-title", Static) title_widget.update(f"[bold]{title}[/bold]") def _hide_edit_section(self) -> None: """Versteckt den Bearbeitungsbereich.""" section = self.query_one("#sched-cond-edit-section") section.display = False self._clear_dynamic_fields() def _populate_type_select(self) -> None: """Befuellt das Typ-Dropdown mit den verfuegbaren Condition-Typen.""" select = self.query_one("#sched-cond-type-select", Select) options: list[tuple[str, str]] = [] for type_key, type_meta in self._form_meta.items(): if isinstance(type_meta, dict): label = type_meta.get("display_name", type_key) else: label = type_key options.append((label, type_key)) select.set_options(options) def _clear_dynamic_fields(self) -> None: """Entfernt alle dynamisch erzeugten Formularfelder.""" container = self.query_one("#sched-cond-fields", Vertical) container.remove_children() async def _render_fields_for_type( self, cond_type: str, values: dict | None = None ) -> None: """Erzeugt dynamische Formularfelder fuer den gewaehlten Typ.""" # Alten Container ersetzen um DuplicateIds zu vermeiden old = self.query_one("#sched-cond-fields", Vertical) container = Vertical(id="sched-cond-fields") edit_section = self.query_one("#sched-cond-edit-section", Vertical) apply_btn = self.query_one("#sched-cond-apply", Button) await old.remove() await edit_section.mount(container, before=apply_btn) type_meta = self._form_meta.get(cond_type) if not type_meta or not isinstance(type_meta, dict): await container.mount( Static("[dim]Keine Felder fuer diesen Typ definiert[/dim]") ) return fields = type_meta.get("fields", []) if values is None: values = {} for field_def in fields: field_name: str = field_def.get("name", "") field_label: str = field_def.get("label", field_name) field_type_raw = field_def.get("field_type", "text") field_type: str = _resolve_field_type(field_type_raw) default = field_def.get("default") current = values.get(field_name, default) required: bool = field_def.get("required", False) # Pflichtfeld-Markierung req_mark = " *" if required else "" widget_id = f"sched-cond-field-{field_name}" if field_type == "checkbox": val = bool(current) if current is not None else False await container.mount( Static(f"{field_label}{req_mark}:", classes="field-label") ) await container.mount(Switch(value=val, id=widget_id)) elif field_type == "textarea": text = str(current) if current is not None else "" await container.mount( Static(f"{field_label}{req_mark}:", classes="field-label") ) await container.mount(TextArea(text, id=widget_id)) elif field_type == "weekdays": await container.mount( Static(f"{field_label}{req_mark}:", classes="field-label") ) # Aktuellen Wert als Liste parsen active_days: list[bool] = [False] * 7 if isinstance(current, list) and len(current) == 7: active_days = [bool(d) for d in current] elif isinstance(current, str) and current: # Kommaseparierte Indizes for part in current.split(","): part = part.strip() if part.isdigit() and 0 <= int(part) < 7: active_days[int(part)] = True for i, day_label in enumerate(_WEEKDAY_LABELS): day_id = f"{widget_id}-{i}" row = Horizontal(classes="weekday-row") await container.mount(row) await row.mount(Static(f" {day_label}", classes="weekday-label")) await row.mount(Switch(value=active_days[i], id=day_id)) elif field_type == "select": choices_raw = field_def.get("choices", []) options: list[tuple[str, str]] = [] for c in choices_raw: if isinstance(c, dict): options.append( (c.get("label", c.get("value", "")), c.get("value", "")) ) elif isinstance(c, (list, tuple)) and len(c) >= 2: options.append((str(c[0]), str(c[1]))) else: options.append((str(c), str(c))) await container.mount( Static(f"{field_label}{req_mark}:", classes="field-label") ) sel = Select(options, id=widget_id, prompt="Bitte waehlen...") if current is not None: sel.value = str(current) await container.mount(sel) elif field_type == "number": text = str(current) if current is not None else "" await container.mount( Static(f"{field_label}{req_mark}:", classes="field-label") ) await container.mount( Input( value=text, placeholder=field_def.get("placeholder", ""), id=widget_id, ) ) elif field_type == "tags": # Kommaseparierte Tags if isinstance(current, list): text = ", ".join(str(t) for t in current) else: text = str(current) if current is not None else "" await container.mount( Static( f"{field_label}{req_mark} (Komma-getrennt):", classes="field-label", ) ) await container.mount( Input( value=text, placeholder=field_def.get("placeholder", "z.B. tag1, tag2"), id=widget_id, ) ) else: # text, datetime und Fallback text = str(current) if current is not None else "" placeholder = field_def.get("placeholder", "") if field_type == "datetime" and not placeholder: placeholder = "YYYY-MM-DD HH:MM:SS" await container.mount( Static(f"{field_label}{req_mark}:", classes="field-label") ) await container.mount( Input(value=text, placeholder=placeholder, id=widget_id) ) def _collect_field_values(self, cond_type: str) -> dict: """Sammelt die aktuellen Werte aller dynamischen Felder.""" type_meta = self._form_meta.get(cond_type, {}) fields = type_meta.get("fields", []) if isinstance(type_meta, dict) else [] values: dict = {"type": cond_type} for field_def in fields: field_name: str = field_def.get("name", "") field_type_raw = field_def.get("field_type", "text") field_type: str = _resolve_field_type(field_type_raw) widget_id = f"sched-cond-field-{field_name}" try: if field_type == "checkbox": widget = self.query_one(f"#{widget_id}", Switch) values[field_name] = widget.value elif field_type == "textarea": widget = self.query_one(f"#{widget_id}", TextArea) values[field_name] = widget.text elif field_type == "weekdays": days: list[bool] = [] for i in range(7): day_id = f"{widget_id}-{i}" sw = self.query_one(f"#{day_id}", Switch) days.append(sw.value) values[field_name] = days elif field_type == "select": widget = self.query_one(f"#{widget_id}", Select) val = widget.value values[field_name] = val if val != Select.BLANK else "" elif field_type == "number": widget = self.query_one(f"#{widget_id}", Input) raw = widget.value.strip() if raw: # Ganzzahl oder Fliesskomma try: values[field_name] = int(raw) except ValueError: try: values[field_name] = float(raw) except ValueError: values[field_name] = raw else: values[field_name] = None elif field_type == "tags": widget = self.query_one(f"#{widget_id}", Input) raw = widget.value.strip() if raw: values[field_name] = [ t.strip() for t in raw.split(",") if t.strip() ] else: values[field_name] = [] else: # text, datetime, Fallback widget = self.query_one(f"#{widget_id}", Input) values[field_name] = widget.value except Exception: # Widget nicht gefunden — Standardwert values[field_name] = field_def.get("default") return values # ------------------------------------------------------------------ # Event-Handler: Buttons # ------------------------------------------------------------------ async def on_button_pressed(self, event: Button.Pressed) -> None: """Verarbeitet Button-Klicks.""" button_id = event.button.id if button_id == "sched-cond-add": await self._handle_add() elif button_id == "sched-cond-remove": self._handle_remove() elif button_id == "sched-cond-apply": await self._handle_apply() async def _handle_add(self) -> None: """Oeffnet den Bearbeitungsbereich fuer eine neue Bedingung.""" self._editing_index = None self._show_edit_section("Neue Bedingung hinzufuegen") # Typ-Selektor zuruecksetzen select = self.query_one("#sched-cond-type-select", Select) select.clear() # Felder leeren self._clear_dynamic_fields() self._set_status("") def _handle_remove(self) -> None: """Entfernt die aktuell selektierte Bedingung.""" if self._selected_row_index is None: self._set_status("Keine Bedingung ausgewaehlt", error=True) return idx = self._selected_row_index if 0 <= idx < len(self._conditions): removed = self._conditions.pop(idx) removed_type = removed.get("type", "Unbekannt") self._selected_row_index = None # Falls gerade diese Bedingung bearbeitet wird, Bereich schliessen if self._editing_index == idx: self._hide_edit_section() self._editing_index = None elif self._editing_index is not None and self._editing_index > idx: # Index nach unten korrigieren self._editing_index -= 1 self._refresh_table() self._set_status( f"Bedingung '{self._display_name_for_type(removed_type)}' entfernt" ) else: self._set_status("Ungueltiger Index", error=True) async def _handle_apply(self) -> None: """Uebernimmt die Werte aus dem Bearbeitungsbereich.""" select = self.query_one("#sched-cond-type-select", Select) cond_type = select.value if cond_type == Select.BLANK or not cond_type: self._set_status("Bitte einen Bedingungstyp waehlen", error=True) return values = self._collect_field_values(str(cond_type)) if self._editing_index is not None: # Bestehende Bedingung aktualisieren if 0 <= self._editing_index < len(self._conditions): self._conditions[self._editing_index] = values self._set_status( f"Bedingung {self._editing_index + 1} aktualisiert" ) else: self._set_status("Ungueltiger Index", error=True) return else: # Neue Bedingung anfuegen self._conditions.append(values) self._set_status( f"Bedingung '{self._display_name_for_type(str(cond_type))}' " "hinzugefuegt" ) self._refresh_table() self._hide_edit_section() self._editing_index = None # ------------------------------------------------------------------ # Event-Handler: Tabellenzeile selektiert # ------------------------------------------------------------------ async def on_data_table_row_selected( self, event: DataTable.RowSelected ) -> None: """Wird aufgerufen wenn eine Tabellenzeile angeklickt wird.""" table = self.query_one("#sched-cond-table", DataTable) # Zeilenindex ermitteln row_index: int | None = None for i, row_key in enumerate(table.rows): if row_key == event.row_key: row_index = i break if row_index is None or row_index >= len(self._conditions): return self._selected_row_index = row_index self._editing_index = row_index cond = self._conditions[row_index] cond_type = cond.get("type", "") self._show_edit_section( f"Bedingung {row_index + 1} bearbeiten" ) # Typ im Selektor setzen select = self.query_one("#sched-cond-type-select", Select) if cond_type and cond_type in self._form_meta: select.value = cond_type else: select.clear() # Felder rendern mit bestehenden Werten if cond_type: await self._render_fields_for_type(cond_type, cond) self._set_status(f"Bedingung {row_index + 1} zum Bearbeiten geladen") # ------------------------------------------------------------------ # Event-Handler: Typ-Auswahl geaendert # ------------------------------------------------------------------ async def on_select_changed(self, event: Select.Changed) -> None: """Reagiert auf Aenderungen im Typ-Selektor.""" if event.select.id != "sched-cond-type-select": return cond_type = event.value if cond_type == Select.BLANK or not cond_type: old = self.query_one("#sched-cond-fields", Vertical) new_container = Vertical(id="sched-cond-fields") edit_section = self.query_one("#sched-cond-edit-section", Vertical) apply_btn = self.query_one("#sched-cond-apply", Button) await old.remove() await edit_section.mount(new_container, before=apply_btn) return # Bestehende Werte laden falls Bearbeitung existing_values: dict | None = None if ( self._editing_index is not None and 0 <= self._editing_index < len(self._conditions) ): existing = self._conditions[self._editing_index] if existing.get("type") == str(cond_type): existing_values = existing await self._render_fields_for_type(str(cond_type), existing_values) # ------------------------------------------------------------------ # Status # ------------------------------------------------------------------ def _set_status(self, text: str, error: bool = False) -> None: """Setzt die Statuszeile.""" status = self.query_one("#sched-cond-status", Static) if error: status.update(f"[red]{text}[/red]") elif text: status.update(f"[green]{text}[/green]") else: status.update("")