# -*- coding: utf-8 -*- """ Base Trigger - Abstrakte Basis für alle Trigger. """ from abc import ABC, abstractmethod from dataclasses import dataclass from datetime import datetime from enum import Enum from typing import Any, TYPE_CHECKING if TYPE_CHECKING: from trixy_core.scheduler.form_fields import FormField, FormValidationError class TriggerState(Enum): """Zustand eines Triggers.""" ACTIVE = "active" # Trigger ist aktiv PAUSED = "paused" # Trigger ist pausiert FINISHED = "finished" # Trigger ist beendet (z.B. einmaliger Trigger) @dataclass class TriggerContext: """Kontext für Trigger-Auswertung.""" current_time: datetime last_fire_time: datetime | None = None event_name: str | None = None event_data: dict | None = None @classmethod def now(cls) -> "TriggerContext": """Erstellt Kontext mit aktueller Zeit.""" return cls(current_time=datetime.now()) class Trigger(ABC): """ Abstrakte Basis für Trigger. Ein Trigger bestimmt, wann ein Job ausgelöst wird. """ def __init__(self, name: str | None = None): """ Initialisiert den Trigger. Args: name: Trigger-Name (optional) """ self._name = name or self.__class__.__name__ self._state = TriggerState.ACTIVE self._last_fire_time: datetime | None = None self._fire_count = 0 @property def name(self) -> str: """Trigger-Name.""" return self._name @property def state(self) -> TriggerState: """Trigger-Zustand.""" return self._state @property def is_active(self) -> bool: """Prüft ob Trigger aktiv ist.""" return self._state == TriggerState.ACTIVE @property def last_fire_time(self) -> datetime | None: """Letzte Auslösung.""" return self._last_fire_time @property def fire_count(self) -> int: """Anzahl Auslösungen.""" return self._fire_count @abstractmethod def should_fire(self, context: TriggerContext) -> bool: """ Prüft ob Trigger auslösen soll. Args: context: Trigger-Kontext Returns: True wenn Trigger auslösen soll """ pass @abstractmethod def get_next_fire_time( self, last_fire_time: datetime | None = None, ) -> datetime | None: """ Berechnet nächsten Auslösezeitpunkt. Args: last_fire_time: Letzte Auslösung Returns: Nächster Zeitpunkt oder None """ pass def fire(self) -> None: """Markiert Trigger als ausgelöst.""" self._last_fire_time = datetime.now() self._fire_count += 1 def pause(self) -> None: """Pausiert den Trigger.""" self._state = TriggerState.PAUSED def resume(self) -> None: """Setzt Trigger fort.""" if self._state == TriggerState.PAUSED: self._state = TriggerState.ACTIVE def reset(self) -> None: """Setzt Trigger zurück.""" self._last_fire_time = None self._fire_count = 0 self._state = TriggerState.ACTIVE @abstractmethod def to_dict(self) -> dict: """Konvertiert zu Dictionary.""" pass def _base_dict(self) -> dict: """Basis-Dictionary für Serialisierung.""" return { "type": self.__class__.__name__, "name": self._name, "state": self._state.value, "fire_count": self._fire_count, "last_fire_time": ( self._last_fire_time.isoformat() if self._last_fire_time else None ), } @classmethod def form_fields(cls) -> list["FormField"]: """Gibt die Formular-Felder fuer die TUI zurueck.""" return [] @classmethod def validate_form(cls, data: dict) -> list["FormValidationError"]: """Validiert Formular-Daten. Gibt Liste von Fehlern zurueck.""" return [] @classmethod def display_name(cls) -> str: """Anzeigename fuer die TUI.""" return cls.__name__ def __repr__(self) -> str: return f"{self.__class__.__name__}(name={self._name})"