40 lines
1.7 KiB
Python
40 lines
1.7 KiB
Python
# shell_runner.py
|
|
import subprocess, shutil
|
|
from pathlib import Path
|
|
from typing import Callable, Optional
|
|
|
|
TIMEOUT_SECONDS = 60
|
|
|
|
def run_powershell(script: str, base_dir: str, qa_callback: Optional[Callable[[str], str]] = None) -> str:
|
|
"""
|
|
Run a PowerShell script with working directory set to base_dir.
|
|
Returns a string combining stdout, stderr, and exit code.
|
|
If qa_callback is provided and the command fails or has stderr,
|
|
the callback is called with the stderr content and its result is appended.
|
|
"""
|
|
safe_dir = str(base_dir).replace("'", "''")
|
|
full_script = f"Set-Location -LiteralPath '{safe_dir}'\n{script}"
|
|
# Try common executable names
|
|
exe = next((x for x in ["powershell.exe", "pwsh.exe", "powershell", "pwsh"] if shutil.which(x)), None)
|
|
if not exe: return "ERROR: Neither powershell nor pwsh found in PATH"
|
|
|
|
try:
|
|
r = subprocess.run(
|
|
[exe, "-NoProfile", "-NonInteractive", "-Command", full_script],
|
|
capture_output=True, text=True, timeout=TIMEOUT_SECONDS, cwd=base_dir
|
|
)
|
|
parts = []
|
|
if r.stdout.strip(): parts.append(f"STDOUT:\n{r.stdout.strip()}")
|
|
if r.stderr.strip(): parts.append(f"STDERR:\n{r.stderr.strip()}")
|
|
parts.append(f"EXIT CODE: {r.returncode}")
|
|
|
|
# QA Interceptor logic
|
|
if (r.returncode != 0 or r.stderr.strip()) and qa_callback:
|
|
qa_analysis = qa_callback(r.stderr.strip())
|
|
if qa_analysis:
|
|
parts.append(f"\nQA ANALYSIS:\n{qa_analysis}")
|
|
|
|
return "\n".join(parts)
|
|
except subprocess.TimeoutExpired: return f"ERROR: timed out after {TIMEOUT_SECONDS}s"
|
|
except Exception as e: return f"ERROR: {e}"
|