Trixy7/trixy_core/config/environment.py

# -*- coding: utf-8 -*-
"""
Umgebungsvariablen-Substitution für Konfigurationen.

Ermöglicht die Verwendung von ${VAR} und ${VAR:-default} Syntax
in Konfigurationsdateien.
"""

from __future__ import annotations

import os
import re
from dataclasses import dataclass
from typing import Any


@dataclass
class SubstitutionResult:
    """
    Ergebnis einer Substitution.

    Enthält den substituierten Wert und Metadaten.
    """

    original: str
    """Original-String vor Substitution."""

    result: str
    """Ergebnis nach Substitution."""

    substitutions: list[tuple[str, str, str]]
    """Liste von (Platzhalter, Variable, Wert) Tupeln."""

    unresolved: list[str]
    """Liste von Variablen, die nicht aufgelöst werden konnten."""

    @property
    def has_unresolved(self) -> bool:
        """Prüft, ob unaufgelöste Variablen existieren."""
        return len(self.unresolved) > 0

    @property
    def substitution_count(self) -> int:
        """Anzahl der durchgeführten Substitutionen."""
        return len(self.substitutions)


class EnvironmentSubstitutor:
    """
    Substituiert Umgebungsvariablen in Strings und Datenstrukturen.

    Unterstützt folgende Syntax:
    - ${VAR} - Ersetzt durch Wert von VAR
    - ${VAR:-default} - Ersetzt durch Wert von VAR oder "default"
    - ${VAR:=default} - Setzt VAR auf "default" wenn nicht definiert
    - ${VAR:+value} - Ersetzt durch "value" nur wenn VAR definiert ist
    - ${VAR:?error} - Wirft Fehler wenn VAR nicht definiert ist

    Example:
        substitutor = EnvironmentSubstitutor()

        # Einfache Substitution
        result = substitutor.substitute("Host: ${DB_HOST:-localhost}")

        # Verschachtelte Strukturen
        config = {"host": "${DB_HOST}", "port": "${DB_PORT:-5432}"}
        result = substitutor.substitute_recursive(config)
    """

    # Regex-Pattern für Variablen-Syntax
    PATTERN = re.compile(
        r"\$\{"
        r"(?P<var>[A-Za-z_][A-Za-z0-9_]*)"
        r"(?:"
        r"(?P<op>:-|:=|:\+|:\?)"
        r"(?P<default>[^}]*)"
        r")?"
        r"\}"
    )

    def __init__(
        self,
        env: dict[str, str] | None = None,
        strict: bool = False,
        expand_nested: bool = True,
    ) -> None:
        """
        Initialisiert den Substitutor.

        Args:
            env: Optionales Environment-Dictionary (Default: os.environ).
            strict: Ob bei fehlenden Variablen ein Fehler geworfen wird.
            expand_nested: Ob Variablen in Default-Werten expandiert werden.
        """
        self._env = env if env is not None else dict(os.environ)
        self._strict = strict
        self._expand_nested = expand_nested

    def get_variable(self, name: str, default: str | None = None) -> str | None:
        """
        Holt eine Umgebungsvariable.

        Args:
            name: Variablenname.
            default: Standardwert falls nicht definiert.

        Returns:
            Variablenwert oder Default.
        """
        return self._env.get(name, default)

    def set_variable(self, name: str, value: str) -> None:
        """
        Setzt eine Umgebungsvariable.

        Args:
            name: Variablenname.
            value: Wert.
        """
        self._env[name] = value

    def _replace_match(
        self,
        match: re.Match,
        substitutions: list[tuple[str, str, str]],
        unresolved: list[str],
    ) -> str:
        """Ersetzt einen einzelnen Match."""
        var_name = match.group("var")
        operator = match.group("op")
        default_value = match.group("default") or ""

        placeholder = match.group(0)
        value = self._env.get(var_name)

        if operator is None:
            # ${VAR} - Einfache Substitution
            if value is not None:
                substitutions.append((placeholder, var_name, value))
                return value
            elif self._strict:
                raise ValueError(
                    f"Umgebungsvariable '{var_name}' ist nicht definiert"
                )
            else:
                unresolved.append(var_name)
                return placeholder

        elif operator == ":-":
            # ${VAR:-default} - Default wenn nicht gesetzt oder leer
            if value:
                substitutions.append((placeholder, var_name, value))
                return value
            else:
                # Default-Wert ggf. auch expandieren
                if self._expand_nested:
                    default_value = self.substitute(default_value)
                substitutions.append((placeholder, var_name, default_value))
                return default_value

        elif operator == ":=":
            # ${VAR:=default} - Setze Variable wenn nicht definiert
            if value:
                substitutions.append((placeholder, var_name, value))
                return value
            else:
                if self._expand_nested:
                    default_value = self.substitute(default_value)
                self._env[var_name] = default_value
                substitutions.append((placeholder, var_name, default_value))
                return default_value

        elif operator == ":+":
            # ${VAR:+value} - Value nur wenn Variable definiert
            if value:
                if self._expand_nested:
                    default_value = self.substitute(default_value)
                substitutions.append((placeholder, var_name, default_value))
                return default_value
            else:
                substitutions.append((placeholder, var_name, ""))
                return ""

        elif operator == ":?":
            # ${VAR:?error} - Fehler wenn nicht definiert
            if value:
                substitutions.append((placeholder, var_name, value))
                return value
            else:
                error_msg = default_value or f"Variable '{var_name}' ist nicht definiert"
                raise ValueError(error_msg)

        return placeholder

    def substitute(self, text: str) -> str:
        """
        Substituiert Variablen in einem String.

        Args:
            text: String mit Variablenreferenzen.

        Returns:
            String mit substituierten Werten.
        """
        result = self.substitute_with_info(text)
        return result.result

    def substitute_with_info(self, text: str) -> SubstitutionResult:
        """
        Substituiert Variablen und gibt detaillierte Informationen zurück.

        Args:
            text: String mit Variablenreferenzen.

        Returns:
            SubstitutionResult mit Details.
        """
        substitutions: list[tuple[str, str, str]] = []
        unresolved: list[str] = []

        def replacer(match: re.Match) -> str:
            return self._replace_match(match, substitutions, unresolved)

        result = self.PATTERN.sub(replacer, text)

        return SubstitutionResult(
            original=text,
            result=result,
            substitutions=substitutions,
            unresolved=unresolved,
        )

    def substitute_recursive(
        self,
        data: Any,
        _depth: int = 0,
        max_depth: int = 100,
    ) -> Any:
        """
        Substituiert Variablen rekursiv in einer Datenstruktur.

        Args:
            data: Datenstruktur (dict, list, str, etc.).
            _depth: Aktuelle Rekursionstiefe (intern).
            max_depth: Maximale Rekursionstiefe.

        Returns:
            Datenstruktur mit substituierten Werten.
        """
        if _depth > max_depth:
            raise RecursionError(
                f"Maximale Rekursionstiefe {max_depth} überschritten"
            )

        if isinstance(data, str):
            return self.substitute(data)

        elif isinstance(data, dict):
            return {
                self.substitute(k) if isinstance(k, str) else k: self.substitute_recursive(
                    v, _depth + 1, max_depth
                )
                for k, v in data.items()
            }

        elif isinstance(data, list):
            return [
                self.substitute_recursive(item, _depth + 1, max_depth)
                for item in data
            ]

        elif isinstance(data, tuple):
            return tuple(
                self.substitute_recursive(item, _depth + 1, max_depth)
                for item in data
            )

        else:
            return data

    def list_variables(self, text: str) -> list[str]:
        """
        Listet alle referenzierten Variablen in einem String.

        Args:
            text: String mit Variablenreferenzen.

        Returns:
            Liste der Variablennamen.
        """
        matches = self.PATTERN.findall(text)
        return [m[0] for m in matches]  # Nur den Variablennamen

    def list_variables_recursive(self, data: Any) -> set[str]:
        """
        Listet alle Variablen in einer Datenstruktur.

        Args:
            data: Datenstruktur.

        Returns:
            Set der Variablennamen.
        """
        variables: set[str] = set()

        if isinstance(data, str):
            variables.update(self.list_variables(data))

        elif isinstance(data, dict):
            for key, value in data.items():
                if isinstance(key, str):
                    variables.update(self.list_variables(key))
                variables.update(self.list_variables_recursive(value))

        elif isinstance(data, (list, tuple)):
            for item in data:
                variables.update(self.list_variables_recursive(item))

        return variables

    def validate(self, data: Any) -> list[str]:
        """
        Validiert, dass alle referenzierten Variablen definiert sind.

        Args:
            data: Datenstruktur zu validieren.

        Returns:
            Liste der fehlenden Variablen.
        """
        variables = self.list_variables_recursive(data)
        missing = [var for var in variables if var not in self._env]
        return missing


def substitute_env(
    text: str,
    strict: bool = False,
) -> str:
    """
    Convenience-Funktion für Umgebungsvariablen-Substitution.

    Args:
        text: String mit Variablenreferenzen.
        strict: Ob bei fehlenden Variablen ein Fehler geworfen wird.

    Returns:
        String mit substituierten Werten.
    """
    substitutor = EnvironmentSubstitutor(strict=strict)
    return substitutor.substitute(text)


def substitute_env_recursive(
    data: Any,
    strict: bool = False,
) -> Any:
    """
    Convenience-Funktion für rekursive Substitution.

    Args:
        data: Datenstruktur.
        strict: Ob bei fehlenden Variablen ein Fehler geworfen wird.

    Returns:
        Datenstruktur mit substituierten Werten.
    """
    substitutor = EnvironmentSubstitutor(strict=strict)
    return substitutor.substitute_recursive(data)