# -*- coding: utf-8 -*- """ Schedule SubView: Trigger (F2). Interaktive Bearbeitung der Trigger-Konfiguration eines Schedule-Jobs. Bietet eine Typ-Auswahl und dynamisch gerenderte Formularfelder basierend auf der FormMeta-Definition vom Server. """ from __future__ import annotations from typing import Any from textual.app import ComposeResult from textual.containers import Vertical, Horizontal from textual.widgets import Static, Input, Switch, Select, TextArea from trixy_core.tui.views.base import MainView # Wochentage fuer den Weekdays-Feldtyp (Montag bis Sonntag) _WEEKDAY_LABELS: list[tuple[str, str]] = [ ("mo", "Mo"), ("di", "Di"), ("mi", "Mi"), ("do", "Do"), ("fr", "Fr"), ("sa", "Sa"), ("so", "So"), ] # Sonderwert fuer "Kein Trigger" _NO_TRIGGER = "__none__" def _resolve_field_type(field_type: Any) -> str: """ Loest den Feldtyp auf. FieldType kommt manchmal als dict mit 'value'-Key (nach Deserialisierung) und manchmal direkt als String. """ if isinstance(field_type, dict): return str(field_type.get("value", "text")) return str(field_type) if field_type else "text" class ScheduleTriggerView(MainView): """F2 im Schedule-Kontext: Interaktive Trigger-Konfiguration.""" TITLE = "Trigger" VIEW_ID = "sched_trigger" def __init__(self, **kwargs) -> None: super().__init__(**kwargs) self._job_data: dict = {} self._form_meta: dict[str, Any] = {} self._detail_data: object | None = None self._current_type: str | None = None self._field_widgets: dict[str, Any] = {} # ------------------------------------------------------------------ # Oeffentliche API # ------------------------------------------------------------------ def set_job(self, job_data: dict) -> None: """Setzt die Job-Basisdaten aus der Listenansicht.""" self._job_data = job_data self._detail_data = None def set_form_meta(self, meta: dict[str, Any]) -> None: """Setzt die FormMeta-Definition fuer Trigger-Typen vom Server.""" self._form_meta = meta or {} def collect_data(self) -> dict[str, Any]: """ Sammelt die aktuellen Formularwerte und gibt ein Trigger-Dict zurueck. Gibt ein leeres Dict zurueck wenn kein Trigger ausgewaehlt ist. Format: {"type": "cron", "expression": "...", ...} """ if not self._current_type or self._current_type == _NO_TRIGGER: return {} result: dict[str, Any] = {"type": self._current_type} meta_entry = self._form_meta.get(self._current_type, {}) fields: list[dict] = meta_entry.get("fields", []) for field_def in fields: name = field_def.get("name", "") if not name: continue ft = _resolve_field_type(field_def.get("field_type")) value = self._read_field_value(name, ft, field_def) if value is not None: result[name] = value return result # ------------------------------------------------------------------ # Compose & Lifecycle # ------------------------------------------------------------------ def compose(self) -> ComposeResult: with Vertical(classes="section-box"): yield Static("[bold]Trigger-Typ[/bold]", classes="section-header") yield Select( [], id="sched-trigger-type", prompt="Trigger-Typ waehlen...", allow_blank=True, ) with Vertical(id="sched-trigger-fields", classes="section-box"): yield Static( "[dim]Bitte Trigger-Typ auswaehlen[/dim]", id="sched-trigger-placeholder", ) async def on_activate(self) -> None: """Laedt die Detail-Daten und befuellt Typ-Selektor und Felder.""" self._populate_type_selector() await self._load_detail() async def refresh_data(self) -> None: """Kein periodisches Auto-Refresh bei editierbarer Ansicht.""" pass # ------------------------------------------------------------------ # Event-Handler # ------------------------------------------------------------------ async def on_select_changed(self, event: Select.Changed) -> None: """Reagiert auf Aenderung des Trigger-Typ-Selektors.""" if event.select.id != "sched-trigger-type": return selected = event.value if selected is Select.BLANK: self._current_type = None await self._rebuild_fields(None) elif str(selected) == _NO_TRIGGER: self._current_type = _NO_TRIGGER await self._rebuild_fields(None) else: self._current_type = str(selected) await self._rebuild_fields(self._current_type) # ------------------------------------------------------------------ # Internes: Typ-Selektor befuellen # ------------------------------------------------------------------ def _populate_type_selector(self) -> None: """Befuellt den Typ-Selektor mit verfuegbaren Trigger-Typen aus FormMeta.""" selector = self.query_one("#sched-trigger-type", Select) options: list[tuple[str, str]] = [("Kein Trigger", _NO_TRIGGER)] for type_key, type_meta in self._form_meta.items(): display_name = type_meta.get("display_name", type_key) options.append((display_name, type_key)) selector.set_options(options) # Vorauswahl wenn Detail-Daten einen Trigger haben if self._detail_data and self._detail_data.trigger: trigger_type = self._detail_data.trigger.get("type", "") if trigger_type and trigger_type in self._form_meta: selector.value = trigger_type # ------------------------------------------------------------------ # Internes: Detail-Daten laden # ------------------------------------------------------------------ async def _load_detail(self) -> None: """Laedt detaillierte Job-Daten vom Server und setzt Formularwerte.""" 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 await self._apply_detail_to_form() async def _apply_detail_to_form(self) -> None: """Wendet die geladenen Detail-Daten auf das Formular an.""" d = self._detail_data if d is None: return selector = self.query_one("#sched-trigger-type", Select) if not d.trigger: selector.value = _NO_TRIGGER self._current_type = _NO_TRIGGER await self._rebuild_fields(None) return trigger = d.trigger trigger_type = trigger.get("type", "") if trigger_type and trigger_type in self._form_meta: selector.value = trigger_type self._current_type = trigger_type await self._rebuild_fields(trigger_type, prefill=trigger) else: selector.value = _NO_TRIGGER self._current_type = _NO_TRIGGER await self._rebuild_fields(None) # ------------------------------------------------------------------ # Internes: Dynamisches Formular rendern # ------------------------------------------------------------------ async def _rebuild_fields( self, trigger_type: str | None, prefill: dict[str, Any] | None = None ) -> None: """ Entfernt alte Formular-Widgets und erstellt neue fuer den gewaehlten Typ. Args: trigger_type: Der gewaehlte Trigger-Typ oder None fuer keinen Trigger. prefill: Optionale Vorbelegung der Felder mit vorhandenen Werten. """ # Alten Container komplett ersetzen um DuplicateIds zu vermeiden old_container = self.query_one("#sched-trigger-fields", Vertical) container = Vertical(id="sched-trigger-fields", classes="section-box") await old_container.remove() await self.mount(container) self._field_widgets.clear() if not trigger_type or trigger_type not in self._form_meta: await container.mount( Static( "[dim]Kein Trigger konfiguriert[/dim]", id="sched-trigger-placeholder", ) ) return meta_entry = self._form_meta[trigger_type] fields: list[dict] = meta_entry.get("fields", []) if not fields: await container.mount( Static( "[dim]Keine Felder fuer diesen Trigger-Typ[/dim]", id="sched-trigger-placeholder", ) ) return prefill = prefill or {} for field_def in fields: name = field_def.get("name", "") if not name: continue label = field_def.get("label", name) ft = _resolve_field_type(field_def.get("field_type")) help_text = field_def.get("help_text", "") required = field_def.get("required", False) default = field_def.get("default") prefill_value = prefill.get(name, default) # Label mit Pflichtfeld-Markierung label_text = f"[bold]{label}[/bold]" if required: label_text += " [red]*[/red]" if help_text: label_text += f" [dim]{help_text}[/dim]" widget_id = f"sched-tfield-{name}" if ft == "weekdays": await container.mount(Static(label_text)) await self._mount_weekdays(container, widget_id, prefill_value) self._field_widgets[name] = (widget_id, ft) else: widget = self._create_field_widget( ft, widget_id, field_def, prefill_value ) if widget is not None: await container.mount(Static(label_text)) await container.mount(widget) self._field_widgets[name] = (widget_id, ft) def _create_field_widget( self, field_type: str, widget_id: str, field_def: dict, prefill_value: Any, ) -> Any: """ Erstellt das passende Textual-Widget fuer einen Feldtyp. Args: field_type: Der Typ des Feldes (text, number, checkbox, etc.) widget_id: Die CSS-ID fuer das Widget. field_def: Die komplette Feld-Definition aus FormMeta. prefill_value: Vorbelegungswert oder None. Returns: Das erstellte Widget oder bei weekdays ein Horizontal-Container. """ placeholder = field_def.get("placeholder", "") max_length = field_def.get("max_length") if field_type == "text": value = str(prefill_value) if prefill_value is not None else "" return Input( value=value, placeholder=placeholder or "", id=widget_id, max_length=max_length or 0, ) if field_type == "number": value = str(prefill_value) if prefill_value is not None else "" is_float = field_def.get("is_float", False) ph = placeholder or ("Dezimalzahl" if is_float else "Ganzzahl") return Input( value=value, placeholder=ph, id=widget_id, type="number", ) if field_type == "checkbox": checked = bool(prefill_value) if prefill_value is not None else False return Switch(value=checked, id=widget_id) if field_type == "textarea": value = str(prefill_value) if prefill_value is not None else "" lines = field_def.get("lines", 4) area = TextArea(id=widget_id) area.text = value area.styles.height = lines + 2 return area if field_type == "datetime": value = str(prefill_value) if prefill_value is not None else "" ph = placeholder or "YYYY-MM-DD HH:MM:SS" return Input( value=value, placeholder=ph, id=widget_id, ) if field_type == "tags": # Tags als kommaseparierter String if isinstance(prefill_value, list): value = ", ".join(str(t) for t in prefill_value) elif prefill_value is not None: value = str(prefill_value) else: value = "" return Input( value=value, placeholder=placeholder or "Tag1, Tag2, Tag3", id=widget_id, ) if field_type == "weekdays": # Weekdays werden direkt in _rebuild_fields behandelt return None if field_type == "select": choices = field_def.get("choices", []) options: list[tuple[str, str]] = [] for choice in choices: if isinstance(choice, dict): options.append(( str(choice.get("label", choice.get("value", ""))), str(choice.get("value", "")), )) else: options.append((str(choice), str(choice))) sel = Select(options, id=widget_id, allow_blank=True, prompt="Auswahl...") if prefill_value is not None: sel.value = str(prefill_value) return sel # Fallback: einfaches Textfeld fuer unbekannte Typen value = str(prefill_value) if prefill_value is not None else "" return Input( value=value, placeholder=placeholder or "", id=widget_id, ) async def _mount_weekdays( self, container: Vertical, widget_id: str, prefill_value: Any ) -> None: """ Erstellt 7 beschriftete Switches fuer die Wochentag-Auswahl. Args: container: Der Container in den die Switches gemountet werden. widget_id: Basis-ID fuer die einzelnen Switches. prefill_value: Liste aktiver Tage oder Dict mit Tag-Booleans. """ # Prefill normalisieren active_days: set[str] = set() if isinstance(prefill_value, list): active_days = {str(d).lower() for d in prefill_value} elif isinstance(prefill_value, dict): active_days = {k for k, v in prefill_value.items() if v} for day_key, day_label in _WEEKDAY_LABELS: checked = day_key in active_days row = Horizontal(classes="weekday-row") await container.mount(row) await row.mount(Static(f"[bold]{day_label}[/bold]")) await row.mount(Switch( value=checked, id=f"{widget_id}-{day_key}", )) # ------------------------------------------------------------------ # Internes: Feldwerte auslesen # ------------------------------------------------------------------ def _read_field_value( self, name: str, field_type: str, field_def: dict ) -> Any: """ Liest den aktuellen Wert eines Formularfelds. Args: name: Der Feldname. field_type: Der Feldtyp. field_def: Die Feld-Definition. Returns: Den Feldwert im passenden Python-Typ. """ widget_id = f"sched-tfield-{name}" try: if field_type in ("text", "datetime"): widget = self.query_one(f"#{widget_id}", Input) return widget.value if field_type == "number": widget = self.query_one(f"#{widget_id}", Input) raw = widget.value.strip() if not raw: return None is_float = field_def.get("is_float", False) return float(raw) if is_float else int(raw) if field_type == "checkbox": widget = self.query_one(f"#{widget_id}", Switch) return widget.value if field_type == "textarea": widget = self.query_one(f"#{widget_id}", TextArea) return widget.text if field_type == "tags": widget = self.query_one(f"#{widget_id}", Input) raw = widget.value.strip() if not raw: return [] return [t.strip() for t in raw.split(",") if t.strip()] if field_type == "weekdays": active: list[str] = [] for day_key, _ in _WEEKDAY_LABELS: try: sw = self.query_one( f"#{widget_id}-{day_key}", Switch ) if sw.value: active.append(day_key) except Exception: continue return active if field_type == "select": widget = self.query_one(f"#{widget_id}", Select) if widget.value is Select.BLANK: return None return str(widget.value) # Fallback widget = self.query_one(f"#{widget_id}", Input) return widget.value except Exception: return None