diff --git a/aider/run_cmd.py b/aider/run_cmd.py index e69de29bb..72c8c6287 100644 --- a/aider/run_cmd.py +++ b/aider/run_cmd.py @@ -0,0 +1,72 @@ +import os +import subprocess +import sys +from io import BytesIO + + +def run_cmd(command): + import sys + + if not sys.stdin.isatty(): + return run_cmd_subprocess(command) + + try: + import pexpect # noqa: F401 + except ImportError: + return run_cmd_subprocess(command) + + return run_cmd_pexpect(command) + + +def run_cmd_subprocess(command): + try: + result = subprocess.run( + command, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + shell=True, + encoding=sys.stdout.encoding, + errors="replace", + ) + return result.returncode, result.stdout + except Exception as e: + return 1, str(e) + + +def run_cmd_pexpect(command): + """ + Run a shell command interactively using pexpect, capturing all output. + + :param command: The command to run as a string. + :return: A tuple containing (exit_status, output) + """ + import pexpect + + output = BytesIO() + + def output_callback(b): + output.write(b) + return b + + try: + # Use the SHELL environment variable, falling back to /bin/sh if not set + shell = os.environ.get("SHELL", "/bin/sh") + + if os.path.exists(shell): + # Use the shell from SHELL environment variable + child = pexpect.spawn(shell, args=["-c", command], encoding="utf-8") + else: + # Fall back to spawning the command directly + child = pexpect.spawn(command, encoding="utf-8") + + # Transfer control to the user, capturing output + child.interact(output_filter=output_callback) + + # Wait for the command to finish and get the exit status + child.close() + return child.exitstatus, output.getvalue().decode("utf-8", errors="replace") + + except pexpect.ExceptionPexpect as e: + error_msg = f"Error running command: {e}" + return 1, error_msg diff --git a/aider/utils.py b/aider/utils.py index 0bd37c0b2..c21d41022 100644 --- a/aider/utils.py +++ b/aider/utils.py @@ -1,87 +1,18 @@ import itertools import os -import subprocess import sys import tempfile import time -from io import BytesIO from pathlib import Path import git from aider.dump import dump # noqa: F401 +from aider.run_cmd import run_cmd IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".bmp", ".tiff", ".webp"} -def run_interactive_command(command): - import sys - - if not sys.stdin.isatty(): - return run_interactive_command_subprocess(command) - - try: - import pexpect # noqa: F401 - except ImportError: - return run_interactive_command_subprocess(command) - - return run_interactive_command_pexpect(command) - - -def run_interactive_command_subprocess(command): - try: - result = subprocess.run( - command, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - shell=True, - encoding=sys.stdout.encoding, - errors="replace", - ) - return result.returncode, result.stdout - except Exception as e: - return 1, str(e) - - -def run_interactive_command_pexpect(command): - """ - Run a shell command interactively using pexpect, capturing all output. - - :param command: The command to run as a string. - :return: A tuple containing (exit_status, output) - """ - import pexpect - - output = BytesIO() - - def output_callback(b): - output.write(b) - return b - - try: - # Use the SHELL environment variable, falling back to /bin/sh if not set - shell = os.environ.get("SHELL", "/bin/sh") - - if os.path.exists(shell): - # Use the shell from SHELL environment variable - child = pexpect.spawn(shell, args=["-c", command], encoding="utf-8") - else: - # Fall back to spawning the command directly - child = pexpect.spawn(command, encoding="utf-8") - - # Transfer control to the user, capturing output - child.interact(output_filter=output_callback) - - # Wait for the command to finish and get the exit status - child.close() - return child.exitstatus, output.getvalue().decode("utf-8", errors="replace") - - except pexpect.ExceptionPexpect as e: - error_msg = f"Error running command: {e}" - return 1, error_msg - - class IgnorantTemporaryDirectory: def __init__(self): if sys.version_info >= (3, 10): @@ -395,7 +326,7 @@ def check_pip_install_extra(io, module, prompt, pip_install_cmd): if __name__ == "__main__": if len(sys.argv) > 1: command = " ".join(sys.argv[1:]) - exit_status, output = run_interactive_command(command) + exit_status, output = run_cmd(command) dump(exit_status) dump(output) else: