"""Subprocess runner for deploy.sh and deploy.ps1 action calls.""" from __future__ import annotations import subprocess from pathlib import Path from typing import Any from .models import ActionResult from .output_parser import parse_script_result class ScriptActionRunner: def __init__(self, script_base_dir: str | Path = "doc_scripts") -> None: self.script_base_dir = Path(script_base_dir) def run( self, action: str, *, params: dict[str, Any], script_entry: str, config_path: str, ip: str | None = None, hash_code: str | None = None, stop_first: bool = False, trace_file_path: str | None = None, timeout_sec: int | None = None, ) -> ActionResult: command = self.build_command( action, script_entry=script_entry, config_path=config_path, ip=ip, hash_code=hash_code, stop_first=stop_first, trace_file_path=trace_file_path, ) completed = subprocess.run( command, cwd=str(self.script_base_dir), capture_output=True, text=True, timeout=timeout_sec, check=False, ) return parse_script_result( action=action, stdout=completed.stdout, stderr=completed.stderr, exit_code=completed.returncode, backend="script", tool_name=script_entry, ) def build_command( self, action: str, *, script_entry: str, config_path: str, ip: str | None = None, hash_code: str | None = None, stop_first: bool = False, trace_file_path: str | None = None, ) -> list[str]: if script_entry == "deploy.sh": command = [ "bash", "./deploy.sh", "--config", config_path, "--action", action, ] if ip: command.extend(["--ip", ip]) if hash_code: command.extend(["--hash-code", hash_code]) if stop_first: command.append("--stop-first") if trace_file_path: command.extend(["--trace-file", trace_file_path]) return command if script_entry == "deploy.ps1": command = [ "powershell", "-File", ".\\deploy.ps1", "-ConfigPath", config_path, "-Action", action, ] if ip: command.extend(["-Ip", ip]) if hash_code: command.extend(["-HashCode", hash_code]) if stop_first: command.append("-RollbackStopFirst") return command raise ValueError(f"Unsupported script entry: {script_entry}") def select_script_entry(os_name: str | None = None) -> str: import platform name = (os_name or platform.system()).lower() if "windows" in name: return "deploy.ps1" return "deploy.sh"