Trixy7/trixy_core/scheduler/trigger/cron.py

# -*- 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"