Source code for langchain_experimental.llm_bash.bash

"""封装了对子进程的操作，用于运行命令。"""
from __future__ import annotations

import platform
import re
import subprocess
from typing import TYPE_CHECKING, List, Union
from uuid import uuid4

if TYPE_CHECKING:
    import pexpect


class BashProcess:
    """包装器，用于启动子进程。

使用Python内置的subprocess.run()
在Windows系统上**不**支持持久进程，因为pexpect使用Unix伪终端（ptys）。MacOS和Linux可以。

示例：
.. code-block:: python

    from langchain_community.utilities.bash import BashProcess

    bash = BashProcess(
        strip_newlines = False,
        return_err_output = False,
        persistent = False
    )
    bash.run('echo 'hello world'')


"""

    strip_newlines: bool = False
    """是否在输出上运行.strip()函数"""
    return_err_output: bool = False
    """是否返回失败命令的输出，或者只返回错误消息和堆栈跟踪"""
    persistent: bool = False
    """是否生成持久会话
    注意：不适用于Windows环境"""

    def __init__(
        self,
        strip_newlines: bool = False,
        return_err_output: bool = False,
        persistent: bool = False,
    ):
        """
        使用默认设置进行初始化
        """
        self.strip_newlines = strip_newlines
        self.return_err_output = return_err_output
        self.prompt = ""
        self.process = None
        if persistent:
            self.prompt = str(uuid4())
            self.process = self._initialize_persistent_process(self, self.prompt)

    @staticmethod
    def _lazy_import_pexpect() -> pexpect:
        """仅在需要时导入pexpect。"""
        if platform.system() == "Windows":
            raise ValueError(
                "Persistent bash processes are not yet supported on Windows."
            )
        try:
            import pexpect

        except ImportError:
            raise ImportError(
                "pexpect required for persistent bash processes."
                " To install, run `pip install pexpect`."
            )
        return pexpect

    @staticmethod
    def _initialize_persistent_process(self: BashProcess, prompt: str) -> pexpect.spawn:
        # Start bash in a clean environment
        # Doesn't work on windows
        """初始化一个在干净环境中持久的bash设置。
注意：在Windows上不可用

参数:
    Prompt(str): 要执行的bash命令
"""  # noqa: E501
        pexpect = self._lazy_import_pexpect()
        process = pexpect.spawn(
            "env", ["-i", "bash", "--norc", "--noprofile"], encoding="utf-8"
        )
        # Set the custom prompt
        process.sendline("PS1=" + prompt)

        process.expect_exact(prompt, timeout=10)
        return process

    def run(self, commands: Union[str, List[str]]) -> str:
        """在现有的持久子进程中运行命令，或者在新的子进程环境中运行命令。

参数：
    commands(List[str]): 要在会话中执行的命令列表
"""  # noqa: E501
        if isinstance(commands, str):
            commands = [commands]
        commands = ";".join(commands)
        if self.process is not None:
            return self._run_persistent(
                commands,
            )
        else:
            return self._run(commands)

    def _run(self, command: str) -> str:
        """在子进程中运行命令并返回输出。

参数:
    command: 要运行的命令
"""  # noqa: E501
        try:
            output = subprocess.run(
                command,
                shell=True,
                check=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
            ).stdout.decode()
        except subprocess.CalledProcessError as error:
            if self.return_err_output:
                return error.stdout.decode()
            return str(error)
        if self.strip_newlines:
            output = output.strip()
        return output

    def process_output(self, output: str, command: str) -> str:
        """使用正则表达式从输出中删除命令

参数:
    output: 进程的输出字符串
    command: 执行的命令
"""  # noqa: E501
        pattern = re.escape(command) + r"\s*\n"
        output = re.sub(pattern, "", output, count=1)
        return output.strip()

    def _run_persistent(self, command: str) -> str:
        """在持久环境中运行命令并返回输出。

参数:
    command: 要执行的命令
"""  # noqa: E501
        pexpect = self._lazy_import_pexpect()
        if self.process is None:
            raise ValueError("Process not initialized")
        self.process.sendline(command)

        # Clear the output with an empty string
        self.process.expect(self.prompt, timeout=10)
        self.process.sendline("")

        try:
            self.process.expect([self.prompt, pexpect.EOF], timeout=10)
        except pexpect.TIMEOUT:
            return f"Timeout error while executing command {command}"
        if self.process.after == pexpect.EOF:
            return f"Exited with error status: {self.process.exitstatus}"
        output = self.process.before
        output = self.process_output(output, command)
        if self.strip_newlines:
            return output.strip()
        return output