Trixy7/trixy_core/scheduler/condition/script.py

# -*- coding: utf-8 -*-
"""
Script Condition - Führt Python-Script aus und prüft Rückgabewert.
"""

import asyncio
import os
import sys
from typing import Any

from trixy_core.scheduler.condition.base import Condition
from trixy_core.scheduler.form_fields import FieldType, FormField, FormValidationError


class ScriptCondition(Condition):
    """
    Führt ein Python-Script aus und prüft den Rückgabewert.

    Das Script kann:
    - Einen Boolean zurückgeben (True/False)
    - Einen Wert zurückgeben der mit expected verglichen wird
    - exit(0) für True, exit(1) für False

    Beispiel:
        ```python
        # check_temperature.py
        import requests
        temp = requests.get("http://sensor/temp").json()["value"]
        return temp > 20  # oder: exit(0 if temp > 20 else 1)
        ```

        ScriptCondition("scripts/check_temperature.py")
        ScriptCondition("scripts/get_value.py", expected=42)
    """

    def __init__(
        self,
        script_path: str | None = None,
        script_code: str | None = None,
        expected: Any = True,
        timeout_seconds: float = 30.0,
        pass_context: bool = False,
        name: str | None = None,
        invert: bool = False,
    ):
        """
        Initialisiert die Condition.

        Args:
            script_path: Pfad zum Python-Script
            script_code: Inline Python-Code (alternativ zu script_path)
            expected: Erwarteter Rückgabewert (default: True)
            timeout_seconds: Timeout für Script-Ausführung
            pass_context: Kontext als Variable 'context' verfügbar machen
            name: Condition-Name
            invert: Ergebnis invertieren
        """
        super().__init__(name, invert)

        if not script_path and not script_code:
            raise ValueError("script_path oder script_code muss angegeben werden")

        self._script_path = script_path
        self._script_code = script_code
        self._expected = expected
        self._timeout_seconds = timeout_seconds
        self._pass_context = pass_context

    @property
    def script_path(self) -> str | None:
        """Script-Pfad."""
        return self._script_path

    @property
    def expected(self) -> Any:
        """Erwarteter Wert."""
        return self._expected

    def _get_script_code(self) -> str:
        """Lädt Script-Code."""
        if self._script_code:
            return self._script_code

        if self._script_path and os.path.isfile(self._script_path):
            with open(self._script_path, "r", encoding="utf-8") as f:
                return f.read()

        raise FileNotFoundError(f"Script nicht gefunden: {self._script_path}")

    def _evaluate(self, context: dict) -> bool:
        """Führt Script aus und prüft Ergebnis."""
        try:
            code = self._get_script_code()

            # Globale Variablen für Script
            script_globals = {
                "__builtins__": __builtins__,
                "__name__": "__script__",
                "__file__": self._script_path or "<inline>",
            }

            # Kontext verfügbar machen
            if self._pass_context:
                script_globals["context"] = context

            # Lokale Variablen
            script_locals: dict[str, Any] = {}

            # Script ausführen
            exec(code, script_globals, script_locals)

            # Rückgabewert suchen
            # 1. Explizite 'result' Variable
            if "result" in script_locals:
                result = script_locals["result"]
            # 2. Funktion 'check()' oder 'evaluate()'
            elif "check" in script_locals and callable(script_locals["check"]):
                result = script_locals["check"]()
            elif "evaluate" in script_locals and callable(script_locals["evaluate"]):
                result = script_locals["evaluate"]()
            # 3. Funktion 'main()'
            elif "main" in script_locals and callable(script_locals["main"]):
                result = script_locals["main"]()
            else:
                # Kein expliziter Rückgabewert = True (Script lief durch)
                result = True

            # Vergleich
            if self._expected is True:
                return bool(result)
            elif self._expected is False:
                return not bool(result)
            else:
                return result == self._expected

        except SystemExit as e:
            # exit(0) = True, exit(1) = False
            exit_code = e.code if e.code is not None else 0
            return exit_code == 0

        except Exception as e:
            from trixy_core.utils.debug import perror
            perror(f"Script-Fehler: {e}")
            return False

    def to_dict(self) -> dict:
        """Konvertiert zu Dictionary."""
        return {
            **self._base_dict(),
            "script_path": self._script_path,
            "script_code": self._script_code,
            "expected": self._expected,
            "timeout_seconds": self._timeout_seconds,
            "pass_context": self._pass_context,
        }

    @classmethod
    def from_dict(cls, data: dict) -> "ScriptCondition":
        """Erstellt Condition aus Dictionary."""
        return cls(
            script_path=data.get("script_path"),
            script_code=data.get("script_code"),
            expected=data.get("expected", True),
            timeout_seconds=data.get("timeout_seconds", 30.0),
            pass_context=data.get("pass_context", False),
            name=data.get("name"),
            invert=data.get("invert", False),
        )

    @classmethod
    def display_name(cls) -> str:
        """Anzeigename fuer die TUI."""
        return "Script-Bedingung"

    @classmethod
    def form_fields(cls) -> list[FormField]:
        """Gibt die Formular-Felder fuer die TUI zurueck."""
        return [
            FormField(
                name="script_path", label="Script-Pfad", required=True,
                placeholder="/path/to/check_script.py",
                help_text="Pfad zum Python-Script",
            ),
            FormField(
                name="args", label="Argumente",
                field_type=FieldType.TAGS,
                placeholder="--verbose, --threshold=25",
                help_text="Komma-separierte Argumente",
            ),
            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("script_path", "").strip():
            errors.append(FormValidationError("script_path", "Script-Pfad ist erforderlich"))
        return errors


class AsyncScriptCondition(Condition):
    """
    Führt ein async Python-Script aus.

    Beispiel:
        ```python
        # async_check.py
        import aiohttp

        async def check():
            async with aiohttp.ClientSession() as session:
                async with session.get("http://api/status") as resp:
                    data = await resp.json()
                    return data["status"] == "ok"
        ```
    """

    def __init__(
        self,
        script_path: str | None = None,
        script_code: str | None = None,
        expected: Any = True,
        timeout_seconds: float = 30.0,
        name: str | None = None,
        invert: bool = False,
    ):
        """
        Initialisiert die Condition.

        Args:
            script_path: Pfad zum Python-Script
            script_code: Inline Python-Code
            expected: Erwarteter Rückgabewert
            timeout_seconds: Timeout
            name: Condition-Name
            invert: Ergebnis invertieren
        """
        super().__init__(name, invert)

        if not script_path and not script_code:
            raise ValueError("script_path oder script_code muss angegeben werden")

        self._script_path = script_path
        self._script_code = script_code
        self._expected = expected
        self._timeout_seconds = timeout_seconds

    def _evaluate(self, context: dict) -> bool:
        """Führt async Script aus."""
        try:
            # Event-Loop holen oder erstellen
            try:
                loop = asyncio.get_running_loop()
                # Wir sind bereits in einem Loop - synchron ausführen
                return self._evaluate_sync(context)
            except RuntimeError:
                # Kein laufender Loop - neuen erstellen
                return asyncio.run(self._evaluate_async(context))

        except Exception as e:
            from trixy_core.utils.debug import perror
            perror(f"Async Script-Fehler: {e}")
            return False

    def _evaluate_sync(self, context: dict) -> bool:
        """Synchrone Evaluation (Fallback)."""
        # Nutze reguläre ScriptCondition
        sync_cond = ScriptCondition(
            script_path=self._script_path,
            script_code=self._script_code,
            expected=self._expected,
        )
        return sync_cond._evaluate(context)

    async def _evaluate_async(self, context: dict) -> bool:
        """Asynchrone Evaluation."""
        code = self._get_script_code()

        script_globals = {
            "__builtins__": __builtins__,
            "asyncio": asyncio,
            "context": context,
        }
        script_locals: dict[str, Any] = {}

        exec(code, script_globals, script_locals)

        # Async Funktion suchen
        check_fn = None
        for name in ["check", "evaluate", "main"]:
            if name in script_locals and asyncio.iscoroutinefunction(script_locals[name]):
                check_fn = script_locals[name]
                break

        if check_fn:
            result = await asyncio.wait_for(
                check_fn(),
                timeout=self._timeout_seconds,
            )
        else:
            result = script_locals.get("result", True)

        if self._expected is True:
            return bool(result)
        elif self._expected is False:
            return not bool(result)
        else:
            return result == self._expected

    def _get_script_code(self) -> str:
        """Lädt Script-Code."""
        if self._script_code:
            return self._script_code

        if self._script_path and os.path.isfile(self._script_path):
            with open(self._script_path, "r", encoding="utf-8") as f:
                return f.read()

        raise FileNotFoundError(f"Script nicht gefunden: {self._script_path}")

    def to_dict(self) -> dict:
        """Konvertiert zu Dictionary."""
        return {
            **self._base_dict(),
            "script_path": self._script_path,
            "script_code": self._script_code,
            "expected": self._expected,
            "timeout_seconds": self._timeout_seconds,
        }

    @classmethod
    def from_dict(cls, data: dict) -> "AsyncScriptCondition":
        """Erstellt Condition aus Dictionary."""
        return cls(
            script_path=data.get("script_path"),
            script_code=data.get("script_code"),
            expected=data.get("expected", True),
            timeout_seconds=data.get("timeout_seconds", 30.0),
            name=data.get("name"),
            invert=data.get("invert", False),
        )

    @classmethod
    def display_name(cls) -> str:
        """Anzeigename fuer die TUI."""
        return "Async Script-Bedingung"

    @classmethod
    def form_fields(cls) -> list[FormField]:
        """Gibt die Formular-Felder fuer die TUI zurueck."""
        return [
            FormField(
                name="script_path", label="Script-Pfad", required=True,
                placeholder="/path/to/async_check.py",
                help_text="Pfad zum async Python-Script",
            ),
            FormField(
                name="args", label="Argumente",
                field_type=FieldType.TAGS,
                placeholder="--verbose, --threshold=25",
                help_text="Komma-separierte Argumente",
            ),
            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("script_path", "").strip():
            errors.append(FormValidationError("script_path", "Script-Pfad ist erforderlich"))
        return errors