# -*- coding: utf-8 -*- """ Cron Trigger - Zeitbasierter Trigger mit Cron-Ausdrücken. """ 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 CronField: """Parser für ein Cron-Feld.""" def __init__( self, expression: str, min_val: int, max_val: int, aliases: dict[str, int] | None = None, ): """ Initialisiert das Cron-Feld. Args: expression: Cron-Ausdruck (z.B. "*/5", "1-10", "1,5,10") min_val: Minimaler Wert max_val: Maximaler Wert aliases: Aliase (z.B. {"MON": 1, "TUE": 2}) """ self._expression = expression self._min_val = min_val self._max_val = max_val self._aliases = aliases or {} self._values: set[int] = set() self._parse() def _parse(self) -> None: """Parsed den Cron-Ausdruck.""" expr = self._expression.upper() # Aliase ersetzen for alias, val in self._aliases.items(): expr = expr.replace(alias, str(val)) # Wildcard if expr == "*": self._values = set(range(self._min_val, self._max_val + 1)) return # Mehrere Werte for part in expr.split(","): part = part.strip() # Range mit Step: */5, 1-10/2 if "/" in part: range_part, step_part = part.split("/", 1) step = int(step_part) if range_part == "*": start, end = self._min_val, self._max_val elif "-" in range_part: start, end = map(int, range_part.split("-", 1)) else: start = int(range_part) end = self._max_val self._values.update(range(start, end + 1, step)) # Range: 1-10 elif "-" in part: start, end = map(int, part.split("-", 1)) self._values.update(range(start, end + 1)) # Einzelwert else: self._values.add(int(part)) # Werte validieren self._values = { v for v in self._values if self._min_val <= v <= self._max_val } def matches(self, value: int) -> bool: """Prüft ob Wert matched.""" return value in self._values @property def values(self) -> set[int]: """Erlaubte Werte.""" return self._values.copy() class CronExpression: """Parser für komplette Cron-Ausdrücke.""" # Aliase für Wochentage (0=Sonntag in cron) DAY_ALIASES = { "SUN": 0, "MON": 1, "TUE": 2, "WED": 3, "THU": 4, "FRI": 5, "SAT": 6, } # Aliase für Monate MONTH_ALIASES = { "JAN": 1, "FEB": 2, "MAR": 3, "APR": 4, "MAY": 5, "JUN": 6, "JUL": 7, "AUG": 8, "SEP": 9, "OCT": 10, "NOV": 11, "DEC": 12, } def __init__(self, expression: str): """ Initialisiert den Cron-Ausdruck. Format: minute hour day_of_month month day_of_week Erweitert: second minute hour day_of_month month day_of_week Args: expression: Cron-Ausdruck """ self._expression = expression self._parse() def _parse(self) -> None: """Parsed den Cron-Ausdruck.""" parts = self._expression.split() # 5 Felder: Standard Cron (ohne Sekunden) if len(parts) == 5: self._second = CronField("0", 0, 59) self._minute = CronField(parts[0], 0, 59) self._hour = CronField(parts[1], 0, 23) self._day = CronField(parts[2], 1, 31) self._month = CronField(parts[3], 1, 12, self.MONTH_ALIASES) self._weekday = CronField(parts[4], 0, 6, self.DAY_ALIASES) # 6 Felder: Mit Sekunden elif len(parts) == 6: self._second = CronField(parts[0], 0, 59) self._minute = CronField(parts[1], 0, 59) self._hour = CronField(parts[2], 0, 23) self._day = CronField(parts[3], 1, 31) self._month = CronField(parts[4], 1, 12, self.MONTH_ALIASES) self._weekday = CronField(parts[5], 0, 6, self.DAY_ALIASES) else: raise ValueError( f"Ungültiger Cron-Ausdruck: {self._expression}. " "Erwartet 5 oder 6 Felder." ) def matches(self, dt: datetime) -> bool: """Prüft ob Zeitpunkt matched.""" # Wochentag: Python=0-6 (Mo-So), Cron=0-6 (So-Sa) cron_weekday = (dt.weekday() + 1) % 7 return ( self._second.matches(dt.second) and self._minute.matches(dt.minute) and self._hour.matches(dt.hour) and self._day.matches(dt.day) and self._month.matches(dt.month) and self._weekday.matches(cron_weekday) ) def get_next(self, after: datetime) -> datetime: """ Berechnet nächsten Auslösezeitpunkt. Args: after: Zeitpunkt ab dem gesucht wird Returns: Nächster passender Zeitpunkt """ from datetime import timedelta # Zur nächsten Sekunde current = after.replace(microsecond=0) + timedelta(seconds=1) # Maximale Suche: 4 Jahre (für Schaltjahre) max_iterations = 366 * 24 * 60 * 60 * 4 for _ in range(max_iterations): if self.matches(current): return current # Inkrementiere - optimiert für häufige Fälle if not self._second.matches(current.second): # Nächste Sekunde current = current + timedelta(seconds=1) elif not self._minute.matches(current.minute): # Nächste Minute current = current.replace(second=0) + timedelta(minutes=1) elif not self._hour.matches(current.hour): # Nächste Stunde current = current.replace(second=0, minute=0) + timedelta(hours=1) else: # Nächster Tag current = (current + timedelta(days=1)).replace( hour=0, minute=0, second=0 ) raise ValueError( f"Kein nächster Zeitpunkt gefunden für: {self._expression}" ) @property def expression(self) -> str: """Original-Ausdruck.""" return self._expression class CronTrigger(Trigger): """ Trigger basierend auf Cron-Ausdrücken. Beispiele: - "* * * * *" - Jede Minute - "0 * * * *" - Jede Stunde - "0 0 * * *" - Jeden Tag um Mitternacht - "0 0 * * 1" - Jeden Montag um Mitternacht - "*/5 * * * *" - Alle 5 Minuten - "0 9-17 * * 1-5" - Stündlich 9-17 Uhr, Mo-Fr """ def __init__( self, expression: str, name: str | None = None, timezone: str | None = None, ): """ Initialisiert den Cron-Trigger. Args: expression: Cron-Ausdruck name: Trigger-Name timezone: Zeitzone (optional, default: lokal) """ super().__init__(name) self._expression = expression self._timezone = timezone self._cron = CronExpression(expression) @property def expression(self) -> str: """Cron-Ausdruck.""" return self._expression @property def timezone(self) -> str | None: """Zeitzone.""" return self._timezone def should_fire(self, context: TriggerContext) -> bool: """Prüft ob Trigger auslösen soll.""" if not self.is_active: return False if not self._cron.matches(context.current_time): return False # Verhindern, dass der Trigger mehrfach in derselben Sekunde feuert if context.last_fire_time: last = context.last_fire_time.replace(microsecond=0) now = context.current_time.replace(microsecond=0) if now <= last: return False return True 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 reference = last_fire_time or datetime.now() return self._cron.get_next(reference) def to_dict(self) -> dict: """Konvertiert zu Dictionary.""" return { **self._base_dict(), "expression": self._expression, "timezone": self._timezone, } @classmethod def from_dict(cls, data: dict) -> "CronTrigger": """Erstellt Trigger aus Dictionary.""" return cls( expression=data["expression"], name=data.get("name"), timezone=data.get("timezone"), ) @classmethod def display_name(cls) -> str: """Anzeigename fuer die TUI.""" return "Cron" @classmethod def form_fields(cls) -> list[FormField]: """Gibt die Formular-Felder fuer die TUI zurueck.""" return [ FormField( name="expression", label="Cron-Ausdruck", required=True, placeholder="*/5 * * * *", help_text="Format: Minute Stunde Tag Monat Wochentag", ), FormField( name="timezone", label="Zeitzone", placeholder="Europe/Berlin", ), ] @classmethod def validate_form(cls, data: dict) -> list[FormValidationError]: """Validiert Formular-Daten. Gibt Liste von Fehlern zurueck.""" errors: list[FormValidationError] = [] expr = data.get("expression", "").strip() if not expr: errors.append(FormValidationError("expression", "Cron-Ausdruck ist erforderlich")) else: parts = expr.split() if len(parts) not in (5, 6): errors.append(FormValidationError("expression", "Erwartet 5 oder 6 Felder")) return errors def __repr__(self) -> str: return f"CronTrigger(expression={self._expression})" # Vordefinierte Cron-Ausdrücke class CronPresets: """Häufig verwendete Cron-Ausdrücke.""" EVERY_MINUTE = "* * * * *" EVERY_5_MINUTES = "*/5 * * * *" EVERY_10_MINUTES = "*/10 * * * *" EVERY_15_MINUTES = "*/15 * * * *" EVERY_30_MINUTES = "*/30 * * * *" EVERY_HOUR = "0 * * * *" EVERY_2_HOURS = "0 */2 * * *" EVERY_6_HOURS = "0 */6 * * *" EVERY_12_HOURS = "0 */12 * * *" DAILY = "0 0 * * *" DAILY_NOON = "0 12 * * *" WEEKLY = "0 0 * * 0" MONTHLY = "0 0 1 * *" YEARLY = "0 0 1 1 *" WEEKDAYS = "0 0 * * 1-5" WEEKENDS = "0 0 * * 0,6"