如何在Python中与Shell交互并获取每条命令的返回码

直接用Python内置的subprocess模块就能实现需求，核心思路是让Shell在每条命令执行完成后主动输出唯一结束标记+返回码，Python端通过监听这个标记来判断命令完成，同时分离命令输出和返回码。以下是针对PowerShell的完整实现：

import subprocess
import time

class ShellInteractor:
    def __init__(self, shell_cmd=["powershell", "-NoProfile", "-Command", "-"]):
        # 启动持久化Shell进程：-NoProfile避免加载用户配置，-Command -表示从stdin接收命令
        self.process = subprocess.Popen(
            shell_cmd,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,  # 合并stderr到stdout，统一处理输出
            text=True,
            bufsize=1,  # 开启行缓冲，实时读取输出
            universal_newlines=True
        )
        # 读取并跳过初始的PowerShell提示符（如"PS C:\Users\xxx> "）
        self._skip_prompt()

    def _skip_prompt(self):
        prompt_end = "> "
        buffer = ""
        while True:
            char = self.process.stdout.read(1)
            if not char:
                break
            buffer += char
            if buffer.endswith(prompt_end):
                break

    def run_command(self, cmd, timeout=None):
        # 定义唯一结束标记，确保不会出现在命令输出中
        end_marker = "===CMD_FINISHED==="
        # 构造完整命令：执行原命令后，输出结束标记+返回码
        full_cmd = f"{cmd}; echo '{end_marker}$LASTEXITCODE'\n"
        
        # 发送命令到Shell
        self.process.stdin.write(full_cmd)
        self.process.stdin.flush()

        output_lines = []
        return_code = None
        start_time = time.time()

        while True:
            # 处理超时逻辑
            if timeout and (time.time() - start_time) > timeout:
                self.process.kill()
                raise TimeoutError(f"命令 '{cmd}' 超时（{timeout}秒）")
            
            # 逐行读取输出
            line = self.process.stdout.readline()
            if not line:
                raise RuntimeError("Shell进程意外退出")
            
            # 检测到结束标记，解析返回码
            if end_marker in line:
                return_code = int(line.split(end_marker)[1].strip())
                break
            
            # 收集命令输出（跳过空行）
            if line.strip():
                output_lines.append(line.rstrip("\n"))
        
        # 跳过后续的Shell提示符，为下一次命令做准备
        self._skip_prompt()
        
        return "\n".join(output_lines), return_code

# 使用示例
if __name__ == "__main__":
    interactor = ShellInteractor()
    
    # 执行目录查看命令
    output, code = interactor.run_command("Get-ChildItem .")
    print(f"命令输出:\n{output}")
    print(f"返回码: {code}")
    
    # 执行耗时命令（休眠3秒后输出内容）
    output, code = interactor.run_command("Start-Sleep -Seconds 3; echo '休眠完成'")
    print(f"\n命令输出:\n{output}")
    print(f"返回码: {code}")
    
    # 执行错误命令（读取不存在的文件）
    output, code = interactor.run_command("Get-Item NonExistentFile.txt")
    print(f"\n命令输出:\n{output}")
    print(f"返回码: {code}")
    
    # 关闭Shell进程
    interactor.process.stdin.write("exit\n")
    interactor.process.stdin.flush()
    interactor.process.wait()

核心逻辑说明 #

1. 持久化进程启动：通过subprocess.Popen启动长期运行的Shell进程，配置stdin/stdout为管道，合并标准错误到标准输出，简化输出处理流程。
2. 结束标记与返回码：每次发送命令时附加输出逻辑，让Shell在命令执行完成后输出唯一标记+返回码（PowerShell用$LASTEXITCODE，CMD用%errorlevel%，Bash用$?），Python端通过扫描标记精准判断命令结束时机。
3. 提示符处理：每次命令执行完成后，读取并跳过Shell的提示符，避免干扰下一次命令的输出捕获。
4. 超时控制：可选的超时参数，防止长时间运行的命令阻塞程序。

适配其他Shell的调整 #

• Windows CMD：启动命令改为["cmd", "/k"]，构造命令时替换为full_cmd = f"{cmd}\necho {end_marker}%errorlevel%\n"
• Linux/macOS Bash：启动命令改为["bash"]，构造命令时替换为full_cmd = f"{cmd}; echo '{end_marker}$?'\n"

注意事项 #

• 结束标记需选择在命令输出中不可能出现的字符串，避免误判命令结束。
• 开启行缓冲（bufsize=1）确保实时读取输出，避免缓冲区积压导致延迟。
• 使用完成后务必发送exit命令并调用wait()，防止产生僵尸进程。

内容的提问来源于stack exchange，提问作者SoreDakeNoKoto