Trixy7/trixy_core/config/validation.py

# -*- coding: utf-8 -*-
"""
Schema-basierte Validierung für Konfigurationsdateien.

Bietet Validierung von Konfigurationen gegen definierte Schemas
mit detaillierten Fehlermeldungen.
"""

from __future__ import annotations

import dataclasses
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Sequence, Type, get_type_hints


class ConfigValueType(Enum):
    """
    Unterstützte Konfigurationswert-Typen.
    """

    STRING = "string"
    INTEGER = "integer"
    FLOAT = "float"
    BOOLEAN = "boolean"
    LIST = "list"
    DICT = "dict"
    ANY = "any"


@dataclass
class ConfigFieldError:
    """
    Fehler bei der Konfigurationsvalidierung.
    """

    path: str
    """Pfad zum fehlerhaften Feld (z.B. 'network.host')."""

    message: str
    """Fehlermeldung."""

    value: Any = None
    """Der fehlerhafte Wert."""

    expected: str = ""
    """Beschreibung des erwarteten Wertes."""

    def __str__(self) -> str:
        """String-Repräsentation."""
        msg = f"{self.path}: {self.message}"
        if self.expected:
            msg += f" (erwartet: {self.expected})"
        return msg


class ConfigValidationError(Exception):
    """
    Exception für Konfigurationsvalidierungsfehler.
    """

    def __init__(
        self,
        message: str,
        errors: list[ConfigFieldError] | None = None,
    ) -> None:
        """
        Initialisiert den Fehler.

        Args:
            message: Übergreifende Fehlermeldung.
            errors: Liste von Feldfehlern.
        """
        super().__init__(message)
        self.errors = errors or []

    def __str__(self) -> str:
        """String-Repräsentation."""
        if not self.errors:
            return str(self.args[0])

        lines = [str(self.args[0])]
        for error in self.errors:
            lines.append(f"  - {error}")
        return "\n".join(lines)


@dataclass
class ConfigFieldSpec:
    """
    Spezifikation eines Konfigurationsfeldes.
    """

    name: str
    """Feldname."""

    value_type: ConfigValueType = ConfigValueType.ANY
    """Erwarteter Werttyp."""

    required: bool = True
    """Ob das Feld erforderlich ist."""

    default: Any = None
    """Standardwert."""

    min_value: float | None = None
    """Minimaler Wert (für Zahlen)."""

    max_value: float | None = None
    """Maximaler Wert (für Zahlen)."""

    min_length: int | None = None
    """Minimale Länge (für Strings/Listen)."""

    max_length: int | None = None
    """Maximale Länge (für Strings/Listen)."""

    pattern: str | None = None
    """Regex-Pattern (für Strings)."""

    choices: Sequence[Any] | None = None
    """Erlaubte Werte."""

    validator: Callable[[Any], bool] | None = None
    """Benutzerdefinierte Validierungsfunktion."""

    description: str = ""
    """Beschreibung für Dokumentation."""

    nested_schema: "ConfigSchema | None" = None
    """Schema für verschachtelte Dicts."""

    item_spec: "ConfigFieldSpec | None" = None
    """Spezifikation für Listen-Elemente."""


class ConfigSchema:
    """
    Schema für Konfigurationsvalidierung.

    Definiert die erwartete Struktur und Regeln für eine
    Konfiguration.

    Example:
        schema = ConfigSchema("ServerConfig")
        schema.field("host", ConfigValueType.STRING, required=True)
        schema.field("port", ConfigValueType.INTEGER, min_value=1, max_value=65535)
        schema.field("debug", ConfigValueType.BOOLEAN, default=False)

        # Validieren
        schema.validate(config_data)
    """

    def __init__(
        self,
        name: str = "",
        strict: bool = False,
    ) -> None:
        """
        Initialisiert das Schema.

        Args:
            name: Name des Schemas.
            strict: Ob unbekannte Felder einen Fehler verursachen.
        """
        self.name = name
        self.strict = strict
        self._fields: dict[str, ConfigFieldSpec] = {}

    def field(
        self,
        name: str,
        value_type: ConfigValueType = ConfigValueType.ANY,
        required: bool = True,
        default: Any = None,
        min_value: float | None = None,
        max_value: float | None = None,
        min_length: int | None = None,
        max_length: int | None = None,
        pattern: str | None = None,
        choices: Sequence[Any] | None = None,
        validator: Callable[[Any], bool] | None = None,
        description: str = "",
        nested_schema: "ConfigSchema | None" = None,
        item_spec: ConfigFieldSpec | None = None,
    ) -> "ConfigSchema":
        """
        Fügt eine Feldspezifikation hinzu.

        Args:
            name: Feldname.
            value_type: Erwarteter Typ.
            required: Ob erforderlich.
            default: Standardwert.
            min_value: Minimaler Wert.
            max_value: Maximaler Wert.
            min_length: Minimale Länge.
            max_length: Maximale Länge.
            pattern: Regex-Pattern.
            choices: Erlaubte Werte.
            validator: Benutzerdefinierte Prüffunktion.
            description: Beschreibung.
            nested_schema: Schema für verschachtelte Dicts.
            item_spec: Spezifikation für Listen-Elemente.

        Returns:
            Self für Method-Chaining.
        """
        self._fields[name] = ConfigFieldSpec(
            name=name,
            value_type=value_type,
            required=required,
            default=default,
            min_value=min_value,
            max_value=max_value,
            min_length=min_length,
            max_length=max_length,
            pattern=pattern,
            choices=choices,
            validator=validator,
            description=description,
            nested_schema=nested_schema,
            item_spec=item_spec,
        )
        return self

    def nested(
        self,
        name: str,
        schema: "ConfigSchema",
        required: bool = True,
        description: str = "",
    ) -> "ConfigSchema":
        """
        Fügt ein verschachteltes Schema hinzu.

        Args:
            name: Feldname.
            schema: Das verschachtelte Schema.
            required: Ob erforderlich.
            description: Beschreibung.

        Returns:
            Self für Method-Chaining.
        """
        self._fields[name] = ConfigFieldSpec(
            name=name,
            value_type=ConfigValueType.DICT,
            required=required,
            description=description,
            nested_schema=schema,
        )
        return self

    def _validate_type(
        self,
        value: Any,
        spec: ConfigFieldSpec,
        path: str,
    ) -> list[ConfigFieldError]:
        """Validiert den Typ eines Wertes."""
        errors: list[ConfigFieldError] = []

        if spec.value_type == ConfigValueType.ANY:
            return errors

        type_map = {
            ConfigValueType.STRING: str,
            ConfigValueType.INTEGER: int,
            ConfigValueType.FLOAT: (int, float),
            ConfigValueType.BOOLEAN: bool,
            ConfigValueType.LIST: (list, tuple),
            ConfigValueType.DICT: dict,
        }

        expected_type = type_map.get(spec.value_type)
        if expected_type and not isinstance(value, expected_type):
            errors.append(ConfigFieldError(
                path=path,
                message=f"Ungültiger Typ: {type(value).__name__}",
                value=value,
                expected=spec.value_type.value,
            ))

        return errors

    def _validate_constraints(
        self,
        value: Any,
        spec: ConfigFieldSpec,
        path: str,
    ) -> list[ConfigFieldError]:
        """Validiert Constraints für einen Wert."""
        errors: list[ConfigFieldError] = []

        # Bereichsprüfung
        if spec.min_value is not None and isinstance(value, (int, float)):
            if value < spec.min_value:
                errors.append(ConfigFieldError(
                    path=path,
                    message=f"Wert {value} ist kleiner als Minimum",
                    value=value,
                    expected=f">= {spec.min_value}",
                ))

        if spec.max_value is not None and isinstance(value, (int, float)):
            if value > spec.max_value:
                errors.append(ConfigFieldError(
                    path=path,
                    message=f"Wert {value} ist größer als Maximum",
                    value=value,
                    expected=f"<= {spec.max_value}",
                ))

        # Längenprüfung
        if hasattr(value, "__len__"):
            if spec.min_length is not None and len(value) < spec.min_length:
                errors.append(ConfigFieldError(
                    path=path,
                    message=f"Länge {len(value)} ist zu kurz",
                    value=value,
                    expected=f">= {spec.min_length}",
                ))

            if spec.max_length is not None and len(value) > spec.max_length:
                errors.append(ConfigFieldError(
                    path=path,
                    message=f"Länge {len(value)} ist zu lang",
                    value=value,
                    expected=f"<= {spec.max_length}",
                ))

        # Pattern-Prüfung
        if spec.pattern and isinstance(value, str):
            import re
            if not re.match(spec.pattern, value):
                errors.append(ConfigFieldError(
                    path=path,
                    message="Wert entspricht nicht dem erwarteten Muster",
                    value=value,
                    expected=f"Pattern: {spec.pattern}",
                ))

        # Choices-Prüfung
        if spec.choices is not None:
            if value not in spec.choices:
                errors.append(ConfigFieldError(
                    path=path,
                    message=f"Ungültiger Wert",
                    value=value,
                    expected=f"einer von {list(spec.choices)}",
                ))

        # Custom Validator
        if spec.validator:
            try:
                if not spec.validator(value):
                    errors.append(ConfigFieldError(
                        path=path,
                        message="Benutzerdefinierte Validierung fehlgeschlagen",
                        value=value,
                    ))
            except Exception as e:
                errors.append(ConfigFieldError(
                    path=path,
                    message=f"Validierungsfehler: {e}",
                    value=value,
                ))

        return errors

    def _validate_field(
        self,
        data: dict,
        spec: ConfigFieldSpec,
        path_prefix: str,
    ) -> list[ConfigFieldError]:
        """Validiert ein einzelnes Feld."""
        errors: list[ConfigFieldError] = []
        path = f"{path_prefix}.{spec.name}" if path_prefix else spec.name

        # Existenz prüfen
        if spec.name not in data:
            if spec.required:
                errors.append(ConfigFieldError(
                    path=path,
                    message="Erforderliches Feld fehlt",
                ))
            return errors

        value = data[spec.name]

        # None-Wert
        if value is None:
            if spec.required:
                errors.append(ConfigFieldError(
                    path=path,
                    message="Erforderliches Feld darf nicht null sein",
                    value=value,
                ))
            return errors

        # Typ validieren
        errors.extend(self._validate_type(value, spec, path))

        # Constraints validieren
        errors.extend(self._validate_constraints(value, spec, path))

        # Verschachteltes Schema
        if spec.nested_schema and isinstance(value, dict):
            errors.extend(spec.nested_schema.validate_and_collect(value, path))

        # Listen-Elemente
        if spec.item_spec and isinstance(value, (list, tuple)):
            for i, item in enumerate(value):
                item_path = f"{path}[{i}]"
                errors.extend(self._validate_type(item, spec.item_spec, item_path))
                errors.extend(self._validate_constraints(item, spec.item_spec, item_path))

                if spec.item_spec.nested_schema and isinstance(item, dict):
                    errors.extend(
                        spec.item_spec.nested_schema.validate_and_collect(item, item_path)
                    )

        return errors

    def validate_and_collect(
        self,
        data: dict,
        path_prefix: str = "",
    ) -> list[ConfigFieldError]:
        """
        Validiert Daten und sammelt alle Fehler.

        Args:
            data: Zu validierende Daten.
            path_prefix: Pfad-Präfix für Fehler.

        Returns:
            Liste von Fehlern.
        """
        errors: list[ConfigFieldError] = []

        if not isinstance(data, dict):
            errors.append(ConfigFieldError(
                path=path_prefix or "(root)",
                message="Konfiguration muss ein Dictionary sein",
                value=data,
            ))
            return errors

        # Bekannte Felder validieren
        for spec in self._fields.values():
            errors.extend(self._validate_field(data, spec, path_prefix))

        # Unbekannte Felder prüfen
        if self.strict:
            known_fields = set(self._fields.keys())
            for key in data:
                if key not in known_fields:
                    path = f"{path_prefix}.{key}" if path_prefix else key
                    errors.append(ConfigFieldError(
                        path=path,
                        message="Unbekanntes Feld",
                        value=data[key],
                    ))

        return errors

    def validate(self, data: dict) -> dict:
        """
        Validiert Daten und gibt sie mit Defaults zurück.

        Args:
            data: Zu validierende Daten.

        Returns:
            Validierte Daten mit Defaults.

        Raises:
            ConfigValidationError: Wenn Validierung fehlschlägt.
        """
        errors = self.validate_and_collect(data)

        if errors:
            raise ConfigValidationError(
                f"Konfigurationsvalidierung für '{self.name}' fehlgeschlagen",
                errors=errors,
            )

        # Defaults anwenden
        result = dict(data)
        for name, spec in self._fields.items():
            if name not in result and spec.default is not None:
                result[name] = spec.default

        return result

    def is_valid(self, data: dict) -> bool:
        """
        Prüft, ob Daten valide sind.

        Args:
            data: Zu prüfende Daten.

        Returns:
            True wenn valide.
        """
        errors = self.validate_and_collect(data)
        return len(errors) == 0


def validate_config(
    data: dict,
    schema: ConfigSchema,
) -> dict:
    """
    Validiert eine Konfiguration gegen ein Schema.

    Args:
        data: Konfigurationsdaten.
        schema: Validierungsschema.

    Returns:
        Validierte Daten mit Defaults.

    Raises:
        ConfigValidationError: Wenn Validierung fehlschlägt.
    """
    return schema.validate(data)


def schema_from_dataclass(
    dataclass_type: Type,
    strict: bool = False,
) -> ConfigSchema:
    """
    Erstellt ein Schema aus einer Dataclass.

    Args:
        dataclass_type: Die Dataclass.
        strict: Ob unbekannte Felder einen Fehler verursachen.

    Returns:
        ConfigSchema basierend auf der Dataclass.
    """
    if not dataclasses.is_dataclass(dataclass_type):
        raise ValueError("Typ muss eine Dataclass sein")

    schema = ConfigSchema(name=dataclass_type.__name__, strict=strict)
    type_hints = get_type_hints(dataclass_type)

    type_map = {
        str: ConfigValueType.STRING,
        int: ConfigValueType.INTEGER,
        float: ConfigValueType.FLOAT,
        bool: ConfigValueType.BOOLEAN,
        list: ConfigValueType.LIST,
        dict: ConfigValueType.DICT,
    }

    for dc_field in dataclasses.fields(dataclass_type):
        field_type = type_hints.get(dc_field.name, Any)
        has_default = (
            dc_field.default is not dataclasses.MISSING
            or dc_field.default_factory is not dataclasses.MISSING
        )

        value_type = type_map.get(field_type, ConfigValueType.ANY)

        default = None
        if dc_field.default is not dataclasses.MISSING:
            default = dc_field.default
        elif dc_field.default_factory is not dataclasses.MISSING:
            default = dc_field.default_factory()

        schema.field(
            name=dc_field.name,
            value_type=value_type,
            required=not has_default,
            default=default,
        )

    return schema