# -*- coding: utf-8 -*- """ Activity Conditions - Aktivitätsbasierte Bedingungen. """ from datetime import datetime, timedelta from typing import Any from trixy_core.scheduler.condition.base import Condition from trixy_core.scheduler.form_fields import FieldType, FormField, FormValidationError class RecentActivityCondition(Condition): """ Prüft ob kürzlich Aktivität stattfand. Beispiel: - RecentActivityCondition("last_voice_input", minutes=10) → Nur wenn in letzten 10 Min. Voice-Input war - RecentActivityCondition("last_motion", seconds=30) → Nur wenn in letzten 30 Sek. Bewegung war """ def __init__( self, activity_key: str, seconds: int = 0, minutes: int = 0, hours: int = 0, name: str | None = None, invert: bool = False, ): """ Initialisiert die Condition. Args: activity_key: Schlüssel für Aktivitäts-Timestamp im Kontext seconds: Zeitfenster in Sekunden minutes: Zeitfenster in Minuten hours: Zeitfenster in Stunden name: Condition-Name invert: Ergebnis invertieren """ super().__init__(name, invert) self._activity_key = activity_key self._window = timedelta(seconds=seconds, minutes=minutes, hours=hours) if self._window.total_seconds() <= 0: raise ValueError("Zeitfenster muss größer als 0 sein") @property def activity_key(self) -> str: """Aktivitäts-Schlüssel.""" return self._activity_key @property def window(self) -> timedelta: """Zeitfenster.""" return self._window def _evaluate(self, context: dict) -> bool: """Prüft ob Aktivität kürzlich war.""" last_activity = context.get(self._activity_key) if not last_activity: return False # Timestamp parsen if isinstance(last_activity, str): try: last_activity = datetime.fromisoformat(last_activity) except ValueError: return False elif not isinstance(last_activity, datetime): return False now = context.get("current_time", datetime.now()) if isinstance(now, str): now = datetime.fromisoformat(now) elapsed = now - last_activity return elapsed <= self._window def to_dict(self) -> dict: """Konvertiert zu Dictionary.""" return { **self._base_dict(), "activity_key": self._activity_key, "window_seconds": self._window.total_seconds(), } @classmethod def from_dict(cls, data: dict) -> "RecentActivityCondition": """Erstellt Condition aus Dictionary.""" return cls( activity_key=data["activity_key"], seconds=int(data.get("window_seconds", 0)), name=data.get("name"), invert=data.get("invert", False), ) @classmethod def display_name(cls) -> str: """Anzeigename fuer die TUI.""" return "Letzte Aktivitaet" @classmethod def form_fields(cls) -> list[FormField]: """Gibt die Formular-Felder fuer die TUI zurueck.""" return [ FormField( name="activity_key", label="Aktivitaets-Schluessel", required=True, placeholder="last_voice_input", help_text="Schluessel fuer den Aktivitaets-Timestamp im Kontext", ), FormField( name="within_seconds", label="Zeitfenster (Sek.)", field_type=FieldType.NUMBER, required=True, min_value=1, is_float=True, help_text="Maximales Alter der Aktivitaet in Sekunden", ), FormField( name="invert", label="Invertieren", field_type=FieldType.CHECKBOX, default=False, ), ] @classmethod def validate_form(cls, data: dict) -> list[FormValidationError]: """Validiert Formular-Daten. Gibt Liste von Fehlern zurueck.""" errors: list[FormValidationError] = [] if not data.get("activity_key", "").strip(): errors.append(FormValidationError("activity_key", "Aktivitaets-Schluessel ist erforderlich")) within = data.get("within_seconds", 0) if not within or within <= 0: errors.append(FormValidationError("within_seconds", "Zeitfenster muss groesser als 0 sein")) return errors class ActivityCountCondition(Condition): """ Prüft ob Aktivitäts-Zähler einen Schwellenwert erreicht. Beispiel: - ActivityCountCondition("error_count", min_count=5) → Nur wenn mindestens 5 Fehler - ActivityCountCondition("login_attempts", max_count=3) → Nur wenn max. 3 Login-Versuche """ def __init__( self, count_key: str, min_count: int | None = None, max_count: int | None = None, name: str | None = None, invert: bool = False, ): """ Initialisiert die Condition. Args: count_key: Schlüssel für Zähler im Kontext min_count: Mindestanzahl (optional) max_count: Maximalanzahl (optional) name: Condition-Name invert: Ergebnis invertieren """ super().__init__(name, invert) self._count_key = count_key self._min_count = min_count self._max_count = max_count if min_count is None and max_count is None: raise ValueError("min_count oder max_count muss gesetzt sein") @property def count_key(self) -> str: """Zähler-Schlüssel.""" return self._count_key @property def min_count(self) -> int | None: """Mindestanzahl.""" return self._min_count @property def max_count(self) -> int | None: """Maximalanzahl.""" return self._max_count def _evaluate(self, context: dict) -> bool: """Prüft ob Zähler Bedingung erfüllt.""" count = context.get(self._count_key, 0) try: count = int(count) except (TypeError, ValueError): return False if self._min_count is not None and count < self._min_count: return False if self._max_count is not None and count > self._max_count: return False return True def to_dict(self) -> dict: """Konvertiert zu Dictionary.""" return { **self._base_dict(), "count_key": self._count_key, "min_count": self._min_count, "max_count": self._max_count, } @classmethod def from_dict(cls, data: dict) -> "ActivityCountCondition": """Erstellt Condition aus Dictionary.""" return cls( count_key=data["count_key"], min_count=data.get("min_count"), max_count=data.get("max_count"), name=data.get("name"), invert=data.get("invert", False), ) @classmethod def display_name(cls) -> str: """Anzeigename fuer die TUI.""" return "Aktivitaets-Anzahl" @classmethod def form_fields(cls) -> list[FormField]: """Gibt die Formular-Felder fuer die TUI zurueck.""" return [ FormField( name="activity_key", label="Zaehler-Schluessel", required=True, placeholder="error_count", help_text="Schluessel fuer den Aktivitaets-Zaehler im Kontext", ), FormField( name="min_count", label="Mindestanzahl", field_type=FieldType.NUMBER, min_value=0, help_text="Mindestanzahl fuer Ausloesung", ), FormField( name="time_window_seconds", label="Zeitfenster (Sek.)", field_type=FieldType.NUMBER, min_value=0, is_float=True, help_text="Optionales Zeitfenster fuer Zaehlung", ), FormField( name="invert", label="Invertieren", field_type=FieldType.CHECKBOX, default=False, ), ] @classmethod def validate_form(cls, data: dict) -> list[FormValidationError]: """Validiert Formular-Daten. Gibt Liste von Fehlern zurueck.""" errors: list[FormValidationError] = [] if not data.get("activity_key", "").strip(): errors.append(FormValidationError("activity_key", "Zaehler-Schluessel ist erforderlich")) return errors