# -*- coding: utf-8 -*- """ DateTime Trigger - Einmaliger Trigger für spezifischen Zeitpunkt. """ from datetime import datetime from typing import Any from trixy_core.scheduler.form_fields import FieldType, FormField, FormValidationError from trixy_core.scheduler.trigger.base import Trigger, TriggerContext, TriggerState class DateTimeTrigger(Trigger): """ Trigger für einen spezifischen Zeitpunkt. Einmaliger Trigger - nach Auslösung ist er beendet. Beispiele: - DateTimeTrigger(datetime(2024, 12, 31, 23, 59)) - Silvester - DateTimeTrigger.in_minutes(30) - In 30 Minuten - DateTimeTrigger.at_time(14, 30) - Heute um 14:30 """ def __init__( self, run_at: datetime, tolerance_seconds: float = 60.0, name: str | None = None, ): """ Initialisiert den DateTime-Trigger. Args: run_at: Zeitpunkt für Auslösung tolerance_seconds: Toleranz in Sekunden (Default: 60) name: Trigger-Name """ super().__init__(name) self._run_at = run_at self._tolerance_seconds = tolerance_seconds @property def run_at(self) -> datetime: """Geplanter Zeitpunkt.""" return self._run_at @property def tolerance_seconds(self) -> float: """Toleranz in Sekunden.""" return self._tolerance_seconds @classmethod def in_seconds( cls, seconds: int, name: str | None = None, ) -> "DateTimeTrigger": """Erstellt Trigger für X Sekunden in der Zukunft.""" from datetime import timedelta return cls( run_at=datetime.now() + timedelta(seconds=seconds), name=name, ) @classmethod def in_minutes( cls, minutes: int, name: str | None = None, ) -> "DateTimeTrigger": """Erstellt Trigger für X Minuten in der Zukunft.""" from datetime import timedelta return cls( run_at=datetime.now() + timedelta(minutes=minutes), name=name, ) @classmethod def in_hours( cls, hours: int, name: str | None = None, ) -> "DateTimeTrigger": """Erstellt Trigger für X Stunden in der Zukunft.""" from datetime import timedelta return cls( run_at=datetime.now() + timedelta(hours=hours), name=name, ) @classmethod def at_time( cls, hour: int, minute: int = 0, second: int = 0, name: str | None = None, ) -> "DateTimeTrigger": """ Erstellt Trigger für heute zu bestimmter Uhrzeit. Falls Uhrzeit schon vorbei, wird morgen verwendet. """ from datetime import timedelta now = datetime.now() run_at = now.replace(hour=hour, minute=minute, second=second, microsecond=0) # Wenn Zeit schon vorbei, auf morgen setzen if run_at <= now: run_at = run_at + timedelta(days=1) return cls(run_at=run_at, name=name) @classmethod def tomorrow_at( cls, hour: int, minute: int = 0, second: int = 0, name: str | None = None, ) -> "DateTimeTrigger": """Erstellt Trigger für morgen zu bestimmter Uhrzeit.""" from datetime import timedelta now = datetime.now() run_at = now.replace(hour=hour, minute=minute, second=second, microsecond=0) run_at = run_at + timedelta(days=1) return cls(run_at=run_at, name=name) def should_fire(self, context: TriggerContext) -> bool: """Prüft ob Trigger auslösen soll.""" if not self.is_active: return False now = context.current_time diff = abs((now - self._run_at).total_seconds()) if diff <= self._tolerance_seconds: return True # Wenn Zeitpunkt vorbei, Trigger beenden if now > self._run_at: self._state = TriggerState.FINISHED return False return False def get_next_fire_time( self, last_fire_time: datetime | None = None, ) -> datetime | None: """Berechnet nächsten Auslösezeitpunkt.""" if not self.is_active: return None # Einmaliger Trigger - wenn schon gefeuert, kein nächstes Mal if last_fire_time or self._fire_count > 0: return None now = datetime.now() # Wenn Zeitpunkt in Vergangenheit if self._run_at < now: return None return self._run_at def fire(self) -> None: """Markiert als ausgelöst und beendet Trigger.""" super().fire() self._state = TriggerState.FINISHED def to_dict(self) -> dict: """Konvertiert zu Dictionary.""" return { **self._base_dict(), "run_at": self._run_at.isoformat(), "tolerance_seconds": self._tolerance_seconds, } @classmethod def from_dict(cls, data: dict) -> "DateTimeTrigger": """Erstellt Trigger aus Dictionary.""" return cls( run_at=datetime.fromisoformat(data["run_at"]), tolerance_seconds=data.get("tolerance_seconds", 60.0), name=data.get("name"), ) @classmethod def display_name(cls) -> str: """Anzeigename fuer die TUI.""" return "Zeitpunkt" @classmethod def form_fields(cls) -> list[FormField]: """Gibt die Formular-Felder fuer die TUI zurueck.""" return [ FormField( name="run_at", label="Ausfuehrungszeitpunkt", field_type=FieldType.DATETIME, required=True, placeholder="2024-12-31T23:59:00", help_text="ISO-Format: YYYY-MM-DDTHH:MM:SS", ), FormField( name="tolerance_seconds", label="Toleranz (Sek.)", field_type=FieldType.NUMBER, default=60.0, min_value=1.0, is_float=True, ), ] @classmethod def validate_form(cls, data: dict) -> list[FormValidationError]: """Validiert Formular-Daten. Gibt Liste von Fehlern zurueck.""" errors: list[FormValidationError] = [] run_at = data.get("run_at", "") if not run_at: errors.append(FormValidationError("run_at", "Zeitpunkt ist erforderlich")) else: try: datetime.fromisoformat(run_at) except ValueError: errors.append(FormValidationError("run_at", "Ungueltiges Datum (ISO-Format erwartet)")) return errors def __repr__(self) -> str: return f"DateTimeTrigger(run_at={self._run_at.isoformat()})"