# -*- coding: utf-8 -*- """ File Value Conditions - Prüft Werte in JSON/INI/YAML-Dateien. """ import json import os from configparser import ConfigParser from enum import Enum from typing import Any from trixy_core.scheduler.condition.base import Condition from trixy_core.scheduler.condition.sensor import ComparisonOperator from trixy_core.scheduler.form_fields import FieldType, FormField, FormValidationError class FileValueCondition(Condition): """ Basis-Klasse für Datei-Wert-Conditions. Liest einen Wert aus einer Konfigurationsdatei und prüft ihn. """ def __init__( self, file_path: str, key_path: str, operator: ComparisonOperator = ComparisonOperator.EQUAL, expected: Any = None, default: Any = None, name: str | None = None, invert: bool = False, ): """ Initialisiert die Condition. Args: file_path: Pfad zur Datei key_path: Pfad zum Wert (z.B. "section.key" oder "nested.deep.value") operator: Vergleichsoperator expected: Erwarteter Wert default: Default-Wert wenn Key nicht existiert name: Condition-Name invert: Ergebnis invertieren """ super().__init__(name, invert) self._file_path = file_path self._key_path = key_path self._operator = operator self._expected = expected self._default = default @property def file_path(self) -> str: """Dateipfad.""" return self._file_path @property def key_path(self) -> str: """Schlüsselpfad.""" return self._key_path @property def operator(self) -> ComparisonOperator: """Vergleichsoperator.""" return self._operator @property def expected(self) -> Any: """Erwarteter Wert.""" return self._expected def _read_file(self) -> dict: """Liest Datei und gibt Dictionary zurück. Muss überschrieben werden.""" raise NotImplementedError def _get_value(self, data: dict, key_path: str) -> Any: """ Extrahiert Wert aus verschachteltem Dictionary. Args: data: Dictionary mit Daten key_path: Pfad wie "section.key.subkey" Returns: Gefundener Wert oder self._default """ keys = key_path.split(".") current = data for key in keys: if isinstance(current, dict): if key in current: current = current[key] else: return self._default elif isinstance(current, list): try: index = int(key) current = current[index] except (ValueError, IndexError): return self._default else: return self._default return current def _compare(self, value: Any) -> bool: """Vergleicht Wert mit expected.""" try: if self._operator == ComparisonOperator.EQUAL: return value == self._expected elif self._operator == ComparisonOperator.NOT_EQUAL: return value != self._expected elif self._operator == ComparisonOperator.GREATER: return value > self._expected elif self._operator == ComparisonOperator.GREATER_EQUAL: return value >= self._expected elif self._operator == ComparisonOperator.LESS: return value < self._expected elif self._operator == ComparisonOperator.LESS_EQUAL: return value <= self._expected elif self._operator == ComparisonOperator.BETWEEN: min_val, max_val = self._expected return min_val <= value <= max_val elif self._operator == ComparisonOperator.IN: return value in self._expected elif self._operator == ComparisonOperator.NOT_IN: return value not in self._expected except (TypeError, ValueError): return False return False def _evaluate(self, context: dict) -> bool: """Prüft Datei-Wert.""" try: # Datei lesen data = self._read_file() # Wert extrahieren value = self._get_value(data, self._key_path) # Vergleichen return self._compare(value) except FileNotFoundError: from trixy_core.utils.debug import pwarn pwarn(f"Datei nicht gefunden: {self._file_path}") return False except Exception as e: from trixy_core.utils.debug import perror perror(f"Fehler beim Lesen von {self._file_path}: {e}") return False def _base_file_dict(self) -> dict: """Basis-Dictionary für File-Conditions.""" return { **self._base_dict(), "file_path": self._file_path, "key_path": self._key_path, "operator": self._operator.value, "expected": self._expected, "default": self._default, } @classmethod def _operator_choices(cls) -> list[tuple[str, str]]: """Operator-Auswahl fuer Formular-Felder.""" return [ ("eq", "Gleich (==)"), ("ne", "Ungleich (!=)"), ("gt", "Groesser (>)"), ("gte", "Groesser/Gleich (>=)"), ("lt", "Kleiner (<)"), ("lte", "Kleiner/Gleich (<=)"), ] @classmethod def _base_form_fields(cls) -> list[FormField]: """Basis-Formular-Felder fuer alle Datei-Wert-Conditions.""" return [ FormField( name="file_path", label="Dateipfad", required=True, placeholder="/path/to/config.json", help_text="Pfad zur Konfigurationsdatei", ), FormField( name="key_path", label="Schluessel-Pfad", required=True, placeholder="section.key", help_text="Pfad zum Wert (z.B. 'settings.enabled')", ), FormField( name="expected_value", label="Erwarteter Wert", placeholder="true", ), FormField( name="operator", label="Operator", field_type=FieldType.SELECT, default="eq", choices=cls._operator_choices(), ), FormField( name="invert", label="Invertieren", field_type=FieldType.CHECKBOX, default=False, ), ] @classmethod def _base_validate_form(cls, data: dict) -> list[FormValidationError]: """Basis-Validierung fuer alle Datei-Wert-Conditions.""" errors: list[FormValidationError] = [] if not data.get("file_path", "").strip(): errors.append(FormValidationError("file_path", "Dateipfad ist erforderlich")) if not data.get("key_path", "").strip(): errors.append(FormValidationError("key_path", "Schluessel-Pfad ist erforderlich")) return errors class JsonValueCondition(FileValueCondition): """ Prüft einen Wert in einer JSON-Datei. Beispiel: ```json // config.json { "settings": { "enabled": true, "threshold": 25 } } ``` JsonValueCondition("config.json", "settings.enabled", expected=True) JsonValueCondition("config.json", "settings.threshold", ComparisonOperator.GREATER, 20) """ def _read_file(self) -> dict: """Liest JSON-Datei.""" with open(self._file_path, "r", encoding="utf-8") as f: return json.load(f) def to_dict(self) -> dict: """Konvertiert zu Dictionary.""" return self._base_file_dict() @classmethod def from_dict(cls, data: dict) -> "JsonValueCondition": """Erstellt Condition aus Dictionary.""" return cls( file_path=data["file_path"], key_path=data["key_path"], operator=ComparisonOperator(data.get("operator", "eq")), expected=data.get("expected"), default=data.get("default"), name=data.get("name"), invert=data.get("invert", False), ) @classmethod def display_name(cls) -> str: """Anzeigename fuer die TUI.""" return "JSON-Wert" @classmethod def form_fields(cls) -> list[FormField]: """Gibt die Formular-Felder fuer die TUI zurueck.""" fields = cls._base_form_fields() fields[0] = FormField( name="file_path", label="JSON-Dateipfad", required=True, placeholder="/path/to/config.json", help_text="Pfad zur JSON-Datei", ) return fields @classmethod def validate_form(cls, data: dict) -> list[FormValidationError]: """Validiert Formular-Daten. Gibt Liste von Fehlern zurueck.""" return cls._base_validate_form(data) class IniValueCondition(FileValueCondition): """ Prüft einen Wert in einer INI-Datei. Beispiel: ```ini ; config.ini [settings] enabled = true threshold = 25 ``` IniValueCondition("config.ini", "settings.enabled", expected="true") IniValueCondition("config.ini", "settings.threshold", ComparisonOperator.GREATER, 20, value_type=int) """ def __init__( self, file_path: str, key_path: str, operator: ComparisonOperator = ComparisonOperator.EQUAL, expected: Any = None, default: Any = None, value_type: type | None = None, name: str | None = None, invert: bool = False, ): """ Initialisiert die Condition. Args: file_path: Pfad zur INI-Datei key_path: Pfad als "section.key" operator: Vergleichsoperator expected: Erwarteter Wert default: Default-Wert value_type: Typ für Konvertierung (int, float, bool) name: Condition-Name invert: Ergebnis invertieren """ super().__init__( file_path, key_path, operator, expected, default, name, invert ) self._value_type = value_type def _read_file(self) -> dict: """Liest INI-Datei als Dictionary.""" config = ConfigParser() config.read(self._file_path, encoding="utf-8") # Konvertiere zu Dictionary result: dict[str, Any] = {} for section in config.sections(): result[section] = dict(config[section]) # DEFAULT-Sektion hinzufügen if config.defaults(): result["DEFAULT"] = dict(config.defaults()) return result def _get_value(self, data: dict, key_path: str) -> Any: """Extrahiert Wert mit optionaler Typkonvertierung.""" value = super()._get_value(data, key_path) if value is None or value == self._default: return value # Typkonvertierung if self._value_type: try: if self._value_type == bool: # Spezielle Boolean-Behandlung if isinstance(value, str): return value.lower() in ("true", "1", "yes", "on") return bool(value) return self._value_type(value) except (TypeError, ValueError): return value return value def to_dict(self) -> dict: """Konvertiert zu Dictionary.""" data = self._base_file_dict() if self._value_type: data["value_type"] = self._value_type.__name__ return data @classmethod def from_dict(cls, data: dict) -> "IniValueCondition": """Erstellt Condition aus Dictionary.""" value_type = None if data.get("value_type"): type_map = {"int": int, "float": float, "bool": bool, "str": str} value_type = type_map.get(data["value_type"]) return cls( file_path=data["file_path"], key_path=data["key_path"], operator=ComparisonOperator(data.get("operator", "eq")), expected=data.get("expected"), default=data.get("default"), value_type=value_type, name=data.get("name"), invert=data.get("invert", False), ) @classmethod def display_name(cls) -> str: """Anzeigename fuer die TUI.""" return "INI-Wert" @classmethod def form_fields(cls) -> list[FormField]: """Gibt die Formular-Felder fuer die TUI zurueck.""" fields = cls._base_form_fields() fields[0] = FormField( name="file_path", label="INI-Dateipfad", required=True, placeholder="/path/to/config.ini", help_text="Pfad zur INI-Datei", ) fields[1] = FormField( name="key_path", label="Schluessel-Pfad", required=True, placeholder="section.key", help_text="Format: Sektion.Schluessel", ) return fields @classmethod def validate_form(cls, data: dict) -> list[FormValidationError]: """Validiert Formular-Daten. Gibt Liste von Fehlern zurueck.""" return cls._base_validate_form(data) class YamlValueCondition(FileValueCondition): """ Prüft einen Wert in einer YAML-Datei. Beispiel: ```yaml # config.yaml settings: enabled: true threshold: 25 tags: - important - production ``` YamlValueCondition("config.yaml", "settings.enabled", expected=True) YamlValueCondition("config.yaml", "settings.tags.0", expected="important") """ def _read_file(self) -> dict: """Liest YAML-Datei.""" try: import yaml except ImportError: raise ImportError( "PyYAML ist nicht installiert. " "Installieren mit: pip install pyyaml" ) with open(self._file_path, "r", encoding="utf-8") as f: return yaml.safe_load(f) or {} def to_dict(self) -> dict: """Konvertiert zu Dictionary.""" return self._base_file_dict() @classmethod def from_dict(cls, data: dict) -> "YamlValueCondition": """Erstellt Condition aus Dictionary.""" return cls( file_path=data["file_path"], key_path=data["key_path"], operator=ComparisonOperator(data.get("operator", "eq")), expected=data.get("expected"), default=data.get("default"), name=data.get("name"), invert=data.get("invert", False), ) @classmethod def display_name(cls) -> str: """Anzeigename fuer die TUI.""" return "YAML-Wert" @classmethod def form_fields(cls) -> list[FormField]: """Gibt die Formular-Felder fuer die TUI zurueck.""" fields = cls._base_form_fields() fields[0] = FormField( name="file_path", label="YAML-Dateipfad", required=True, placeholder="/path/to/config.yaml", help_text="Pfad zur YAML-Datei", ) return fields @classmethod def validate_form(cls, data: dict) -> list[FormValidationError]: """Validiert Formular-Daten. Gibt Liste von Fehlern zurueck.""" return cls._base_validate_form(data) class TomlValueCondition(FileValueCondition): """ Prüft einen Wert in einer TOML-Datei. Beispiel: ```toml # config.toml [settings] enabled = true threshold = 25 ``` TomlValueCondition("config.toml", "settings.enabled", expected=True) """ def _read_file(self) -> dict: """Liest TOML-Datei.""" try: import tomllib except ImportError: try: import tomli as tomllib # type: ignore except ImportError: raise ImportError( "tomli ist nicht installiert (Python < 3.11). " "Installieren mit: pip install tomli" ) with open(self._file_path, "rb") as f: return tomllib.load(f) def to_dict(self) -> dict: """Konvertiert zu Dictionary.""" return self._base_file_dict() @classmethod def from_dict(cls, data: dict) -> "TomlValueCondition": """Erstellt Condition aus Dictionary.""" return cls( file_path=data["file_path"], key_path=data["key_path"], operator=ComparisonOperator(data.get("operator", "eq")), expected=data.get("expected"), default=data.get("default"), name=data.get("name"), invert=data.get("invert", False), ) @classmethod def display_name(cls) -> str: """Anzeigename fuer die TUI.""" return "TOML-Wert" @classmethod def form_fields(cls) -> list[FormField]: """Gibt die Formular-Felder fuer die TUI zurueck.""" fields = cls._base_form_fields() fields[0] = FormField( name="file_path", label="TOML-Dateipfad", required=True, placeholder="/path/to/config.toml", help_text="Pfad zur TOML-Datei", ) return fields @classmethod def validate_form(cls, data: dict) -> list[FormValidationError]: """Validiert Formular-Daten. Gibt Liste von Fehlern zurueck.""" return cls._base_validate_form(data) class EnvFileValueCondition(FileValueCondition): """ Prüft einen Wert in einer .env-Datei. Beispiel: ```env # .env DEBUG=true API_KEY=secret123 MAX_CONNECTIONS=10 ``` EnvFileValueCondition(".env", "DEBUG", expected="true") EnvFileValueCondition(".env", "MAX_CONNECTIONS", ComparisonOperator.GREATER, 5, value_type=int) """ def __init__( self, file_path: str, key: str, operator: ComparisonOperator = ComparisonOperator.EQUAL, expected: Any = None, default: Any = None, value_type: type | None = None, name: str | None = None, invert: bool = False, ): """ Initialisiert die Condition. Args: file_path: Pfad zur .env-Datei key: Variablenname operator: Vergleichsoperator expected: Erwarteter Wert default: Default-Wert value_type: Typ für Konvertierung name: Condition-Name invert: Ergebnis invertieren """ super().__init__( file_path, key, operator, expected, default, name, invert ) self._value_type = value_type def _read_file(self) -> dict: """Liest .env-Datei.""" result: dict[str, str] = {} with open(self._file_path, "r", encoding="utf-8") as f: for line in f: line = line.strip() # Kommentare und leere Zeilen überspringen if not line or line.startswith("#"): continue # KEY=VALUE parsen if "=" in line: key, _, value = line.partition("=") key = key.strip() value = value.strip() # Quotes entfernen if (value.startswith('"') and value.endswith('"')) or \ (value.startswith("'") and value.endswith("'")): value = value[1:-1] result[key] = value return result def _get_value(self, data: dict, key_path: str) -> Any: """Extrahiert Wert mit optionaler Typkonvertierung.""" # Bei .env ist key_path einfach der Key (kein Nesting) value = data.get(key_path, self._default) if value is None or value == self._default: return value # Typkonvertierung if self._value_type: try: if self._value_type == bool: if isinstance(value, str): return value.lower() in ("true", "1", "yes", "on") return bool(value) return self._value_type(value) except (TypeError, ValueError): return value return value def to_dict(self) -> dict: """Konvertiert zu Dictionary.""" data = self._base_file_dict() if self._value_type: data["value_type"] = self._value_type.__name__ return data @classmethod def from_dict(cls, data: dict) -> "EnvFileValueCondition": """Erstellt Condition aus Dictionary.""" value_type = None if data.get("value_type"): type_map = {"int": int, "float": float, "bool": bool, "str": str} value_type = type_map.get(data["value_type"]) return cls( file_path=data["file_path"], key=data["key_path"], operator=ComparisonOperator(data.get("operator", "eq")), expected=data.get("expected"), default=data.get("default"), value_type=value_type, name=data.get("name"), invert=data.get("invert", False), ) @classmethod def display_name(cls) -> str: """Anzeigename fuer die TUI.""" return "Env-Datei-Wert" @classmethod def form_fields(cls) -> list[FormField]: """Gibt die Formular-Felder fuer die TUI zurueck.""" return [ FormField( name="file_path", label="Env-Dateipfad", required=True, placeholder="/path/to/.env", help_text="Pfad zur .env-Datei", ), FormField( name="key_path", label="Variablenname", required=True, placeholder="DEBUG", help_text="Name der Umgebungsvariable", ), FormField( name="expected_value", label="Erwarteter Wert", placeholder="true", ), FormField( name="operator", label="Operator", field_type=FieldType.SELECT, default="eq", choices=cls._operator_choices(), ), 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("file_path", "").strip(): errors.append(FormValidationError("file_path", "Dateipfad ist erforderlich")) if not data.get("key_path", "").strip(): errors.append(FormValidationError("key_path", "Variablenname ist erforderlich")) return errors