在 Python 中，可以通过 `subprocess` 模块将多个 CLI 命令串联起来执行。常见的串联方式包括： 1. **管道（Pipe）**：将一个命令的输出作为另一个命令的输入。 2. **顺序执行**：按顺序依次执行多个命令。 3. **条件执行**：根据前一个命令的返回值决定是否执行下一个命令。 以下分别展示这三种方式的实现。 --- ### 1. 使用管道（Pipe）串联命令 #### 示例代码 ```python import subprocess def run_piped_commands(commands): """ 使用管道串联多个 CLI 命令。 参数: commands (list): 包含多个命令的列表，每个命令是一个字符串 返回: str: 最终命令的输出结果 """ try: # 创建第一个子进程 process = subprocess.Popen( commands[0], stdout=subprocess.PIPE, # 将标准输出传递给下一个命令 shell=True, text=True ) # 遍历后续命令并创建管道 for cmd in commands[1:]: process = subprocess.Popen( cmd, stdin=process.stdout, # 当前命令的输入为上一个命令的输出 stdout=subprocess.PIPE, shell=True, text=True ) # 等待最后一个命令执行完成并获取输出 output, _ = process.communicate() return output.strip() except Exception as e: print(f"执行命令时发生错误: {e}") return None # 示例调用 if __name__ == "__main__": commands = [ "echo 'Hello World'", # 第一个命令：生成字符串 "grep 'World'", # 第二个命令：过滤包含 'World' 的行 "tr 'a-z' 'A-Z'" # 第三个命令：将小写字母转换为大写 ] result = run_piped_commands(commands) print(f"最终输出: {result}") ``` #### 解释 1. **管道原理**：通过 `stdin` 和 `stdout` 将多个命令串联起来。 2. **代码逻辑**： - 第一个命令的输出通过 `stdout=subprocess.PIPE` 传递给下一个命令。 - 后续命令通过 `stdin=process.stdout` 接收上一个命令的输出。 - 最后一个命令的输出通过 `process.communicate()` 获取。 3. **示例流程**： - `echo 'Hello World'` 生成字符串。 - `grep 'World'` 过滤包含 `'World'` 的行。 - `tr 'a-z' 'A-Z'` 将所有小写字母转换为大写。 --- ### 2. 按顺序执行多个命令 #### 示例代码 ```python import subprocess def run_sequential_commands(commands): """ 按顺序执行多个 CLI 命令。 参数: commands (list): 包含多个命令的列表，每个命令是一个字符串 返回: list: 每个命令的返回码列表 """ return_codes = [] for cmd in commands: try: process = subprocess.run( cmd, shell=True, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) print(f"命令: {cmd}\n输出:\n{process.stdout}") return_codes.append(process.returncode) except subprocess.CalledProcessError as e: print(f"命令执行失败: {cmd}, 错误信息: {e.stderr}") return_codes.append(e.returncode) return return_codes # 示例调用 if __name__ == "__main__": commands = [ "ls -l", # 列出当前目录内容 "pwd", # 打印当前工作目录 "whoami" # 显示当前用户 ] return_codes = run_sequential_commands(commands) print(f"所有命令的返回码: {return_codes}") ``` #### 解释 1. **顺序执行**：每个命令独立执行，不会影响其他命令。 2. **代码逻辑**： - 使用 `subprocess.run()` 执行每个命令。 - 如果命令执行失败，捕获 `subprocess.CalledProcessError` 并记录错误信息。 3. **示例流程**： - `ls -l` 列出当前目录的内容。 - `pwd` 打印当前工作目录。 - `whoami` 显示当前用户。 --- ### 3. 条件执行命令 #### 示例代码 ```python import subprocess def run_conditional_commands(commands): """ 根据前一个命令的返回值决定是否执行下一个命令。 参数: commands (list): 包含多个命令的列表，每个命令是一个字符串 返回: list: 每个命令的返回码列表 """ return_codes = [] for i, cmd in enumerate(commands): try: process = subprocess.run( cmd, shell=True, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) print(f"命令: {cmd}\n输出:\n{process.stdout}") return_codes.append(process.returncode) except subprocess.CalledProcessError as e: print(f"命令执行失败: {cmd}, 错误信息: {e.stderr}") return_codes.append(e.returncode) break # 如果某个命令失败，则停止执行后续命令 return return_codes # 示例调用 if __name__ == "__main__": commands = [ "mkdir test_dir", # 创建目录 "cd test_dir && touch file.txt", # 进入目录并创建文件 "ls" # 列出目录内容 ] return_codes = run_conditional_commands(commands) print(f"所有命令的返回码: {return_codes}") ``` #### 解释 1. **条件执行**：如果某个命令执行失败，则停止执行后续命令。 2. **代码逻辑**： - 使用 `subprocess.run()` 执行每个命令。 - 如果命令失败，捕获异常并停止执行后续命令。 3. **示例流程**： - `mkdir test_dir` 创建一个测试目录。 - `cd test_dir && touch file.txt` 进入目录并创建文件。 - `ls` 列出目录内容。 --- ###