app/utils/subprocess_utils.py

from __future__ import annotations

import logging
import os
import signal
import subprocess
from typing import Dict, Any, List

_logger = logging.getLogger(__name__)


def run_subprocess(
    cmd: List[str],
    timeout: float,
    max_output: int = 65536,
) -> Dict[str, Any]:
    """Run a subprocess with timeout handling and output capture.

    Consolidates the duplicate subprocess logic from code_executor_service
    and csv_analysis_service.
    """
    proc = subprocess.Popen(
        cmd,
        stdin=subprocess.DEVNULL,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    try:
        stdout_bytes, stderr_bytes = proc.communicate(timeout=timeout)
        timed_out = False
    except subprocess.TimeoutExpired:
        try:
            if os.name == "nt":
                proc.kill()
            else:
                os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
        except Exception:
            proc.kill()
        stdout_bytes, stderr_bytes = proc.communicate()
        timed_out = True

    return {
        "stdout": (stdout_bytes.decode("utf-8", errors="replace")[:max_output] if stdout_bytes else ""),
        "stderr": (stderr_bytes.decode("utf-8", errors="replace")[:max_output] if stderr_bytes else ""),
        "exit_code": proc.returncode,
        "timed_out": timed_out,
    }