Trixy7/trixy_core/scheduler/action/command.py

# -*- coding: utf-8 -*-
"""
Command Action - Führt Shell-Befehle aus.
"""

import asyncio
import os
import shlex
from typing import Any

from trixy_core.scheduler.action.base import Action
from trixy_core.scheduler.form_fields import FieldType, FormField, FormValidationError


class CommandAction(Action):
    """
    Action die einen Shell-Befehl ausführt.

    Beispiel:
        - CommandAction("echo 'Hello World'")
        - CommandAction(["python", "-c", "print('Hello')"])
        - CommandAction("cleanup.sh", cwd="/scripts", env={"DEBUG": "1"})

    Sicherheitshinweis:
        Diese Action führt System-Befehle aus.
        Nur für vertrauenswürdige Eingaben verwenden!
    """

    def __init__(
        self,
        command: str | list[str],
        cwd: str | None = None,
        env: dict[str, str] | None = None,
        shell: bool = False,
        capture_output: bool = True,
        name: str | None = None,
        timeout_seconds: float = 60.0,
    ):
        """
        Initialisiert die Action.

        Args:
            command: Befehl als String oder Liste
            cwd: Arbeitsverzeichnis
            env: Umgebungsvariablen (werden zu os.environ hinzugefügt)
            shell: Shell verwenden (Vorsicht bei untrusted input!)
            capture_output: stdout/stderr erfassen
            name: Action-Name
            timeout_seconds: Timeout
        """
        cmd_str = command if isinstance(command, str) else " ".join(command)
        super().__init__(name or f"command_{cmd_str[:30]}", timeout_seconds)

        self._command = command
        self._cwd = cwd
        self._env = env
        self._shell = shell
        self._capture_output = capture_output

    @property
    def command(self) -> str | list[str]:
        """Befehl."""
        return self._command

    @property
    def cwd(self) -> str | None:
        """Arbeitsverzeichnis."""
        return self._cwd

    async def _execute(self, context: dict) -> dict:
        """Führt den Befehl aus."""
        # Environment vorbereiten
        env = os.environ.copy()
        if self._env:
            env.update(self._env)

        # Kontext-Variablen als Env-Vars
        for key, value in context.items():
            if isinstance(value, (str, int, float, bool)):
                env[f"TRIXY_{key.upper()}"] = str(value)

        # Befehl ausführen
        if self._shell:
            cmd = self._command if isinstance(self._command, str) else " ".join(self._command)
            process = await asyncio.create_subprocess_shell(
                cmd,
                stdout=asyncio.subprocess.PIPE if self._capture_output else None,
                stderr=asyncio.subprocess.PIPE if self._capture_output else None,
                cwd=self._cwd,
                env=env,
            )
        else:
            cmd = self._command if isinstance(self._command, list) else shlex.split(self._command)
            process = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE if self._capture_output else None,
                stderr=asyncio.subprocess.PIPE if self._capture_output else None,
                cwd=self._cwd,
                env=env,
            )

        try:
            stdout, stderr = await asyncio.wait_for(
                process.communicate(),
                timeout=self._timeout_seconds,
            )
        except asyncio.TimeoutError:
            process.kill()
            await process.wait()
            raise TimeoutError(f"Befehl-Timeout nach {self._timeout_seconds}s")

        # Ergebnis zusammenstellen
        result = {
            "return_code": process.returncode,
            "success": process.returncode == 0,
        }

        if self._capture_output:
            result["stdout"] = stdout.decode("utf-8", errors="replace") if stdout else ""
            result["stderr"] = stderr.decode("utf-8", errors="replace") if stderr else ""

        if process.returncode != 0:
            raise RuntimeError(
                f"Befehl fehlgeschlagen (code={process.returncode}): "
                f"{result.get('stderr', 'Kein Output')}"
            )

        return result

    def to_dict(self) -> dict:
        """Konvertiert zu Dictionary."""
        return {
            **self._base_dict(),
            "command": self._command,
            "cwd": self._cwd,
            "env": self._env,
            "shell": self._shell,
            "capture_output": self._capture_output,
        }

    @classmethod
    def from_dict(cls, data: dict) -> "CommandAction":
        """Erstellt Action aus Dictionary."""
        return cls(
            command=data["command"],
            cwd=data.get("cwd"),
            env=data.get("env"),
            shell=data.get("shell", False),
            capture_output=data.get("capture_output", True),
            name=data.get("name"),
            timeout_seconds=data.get("timeout_seconds", 60.0),
        )

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

    @classmethod
    def form_fields(cls) -> list[FormField]:
        """Gibt die Formular-Felder fuer die TUI zurueck."""
        return [
            FormField(
                name="command", label="Befehl", required=True,
                placeholder="echo 'Hello World'",
                help_text="Shell-Befehl oder Pfad",
            ),
            FormField(
                name="cwd", label="Arbeitsverzeichnis",
                placeholder="/home/user",
            ),
            FormField(
                name="shell", label="Shell verwenden",
                field_type=FieldType.CHECKBOX, default=False,
                help_text="Vorsicht: Nur fuer vertrauenswuerdige Eingaben",
            ),
            FormField(
                name="capture_output", label="Ausgabe erfassen",
                field_type=FieldType.CHECKBOX, default=True,
            ),
            FormField(
                name="timeout_seconds", label="Timeout (Sek.)",
                field_type=FieldType.NUMBER,
                default=60.0, min_value=1.0, is_float=True,
            ),
        ]

    @classmethod
    def validate_form(cls, data: dict) -> list[FormValidationError]:
        """Validiert Formular-Daten. Gibt Liste von Fehlern zurueck."""
        errors: list[FormValidationError] = []
        if not data.get("command", "").strip():
            errors.append(FormValidationError("command", "Befehl ist erforderlich"))
        return errors


class ScriptAction(Action):
    """
    Action die ein Script ausführt.

    Beispiel:
        - ScriptAction("/path/to/script.py")
        - ScriptAction("backup.sh", interpreter="bash")
    """

    INTERPRETERS = {
        ".py": "python3",
        ".sh": "bash",
        ".bash": "bash",
        ".zsh": "zsh",
        ".rb": "ruby",
        ".pl": "perl",
        ".js": "node",
    }

    def __init__(
        self,
        script_path: str,
        args: list[str] | None = None,
        interpreter: str | None = None,
        cwd: str | None = None,
        env: dict[str, str] | None = None,
        name: str | None = None,
        timeout_seconds: float = 300.0,
    ):
        """
        Initialisiert die Action.

        Args:
            script_path: Pfad zum Script
            args: Argumente für das Script
            interpreter: Interpreter (wird aus Extension abgeleitet wenn None)
            cwd: Arbeitsverzeichnis
            env: Umgebungsvariablen
            name: Action-Name
            timeout_seconds: Timeout
        """
        script_name = os.path.basename(script_path)
        super().__init__(name or f"script_{script_name}", timeout_seconds)

        self._script_path = script_path
        self._args = args or []
        self._interpreter = interpreter
        self._cwd = cwd
        self._env = env

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

    def _get_interpreter(self) -> str | None:
        """Ermittelt Interpreter aus Extension."""
        if self._interpreter:
            return self._interpreter

        _, ext = os.path.splitext(self._script_path)
        return self.INTERPRETERS.get(ext.lower())

    async def _execute(self, context: dict) -> dict:
        """Führt das Script aus."""
        # Prüfe ob Script existiert
        if not os.path.isfile(self._script_path):
            raise FileNotFoundError(f"Script nicht gefunden: {self._script_path}")

        # Befehl zusammenstellen
        interpreter = self._get_interpreter()

        if interpreter:
            cmd = [interpreter, self._script_path] + self._args
        else:
            # Script direkt ausführen (muss executable sein)
            cmd = [self._script_path] + self._args

        # CommandAction für Ausführung nutzen
        command_action = CommandAction(
            command=cmd,
            cwd=self._cwd or os.path.dirname(self._script_path),
            env=self._env,
            timeout_seconds=self._timeout_seconds,
        )

        return await command_action._execute(context)

    def to_dict(self) -> dict:
        """Konvertiert zu Dictionary."""
        return {
            **self._base_dict(),
            "script_path": self._script_path,
            "args": self._args,
            "interpreter": self._interpreter,
            "cwd": self._cwd,
            "env": self._env,
        }

    @classmethod
    def from_dict(cls, data: dict) -> "ScriptAction":
        """Erstellt Action aus Dictionary."""
        return cls(
            script_path=data["script_path"],
            args=data.get("args"),
            interpreter=data.get("interpreter"),
            cwd=data.get("cwd"),
            env=data.get("env"),
            name=data.get("name"),
            timeout_seconds=data.get("timeout_seconds", 300.0),
        )

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

    @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/script.py",
            ),
            FormField(
                name="args", label="Argumente",
                field_type=FieldType.TAGS,
                placeholder="--verbose, --output=/tmp",
                help_text="Komma-separierte Argumente",
            ),
            FormField(
                name="interpreter", label="Interpreter",
                placeholder="python3",
                help_text="Wird aus Dateiendung abgeleitet wenn leer",
            ),
            FormField(
                name="cwd", label="Arbeitsverzeichnis",
            ),
            FormField(
                name="timeout_seconds", label="Timeout (Sek.)",
                field_type=FieldType.NUMBER,
                default=300.0, min_value=1.0, is_float=True,
            ),
        ]

    @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