- 什么是
subprocess？为什么用它？ - 核心函数：
run(),call(),check_output()(推荐使用的方式) - 进阶用法：
Popen类 (功能最全、最灵活的方式) - 安全注意事项：Shell 注入
- 实用代码示例
什么是 subprocess？为什么用它？
subprocess 模块允许你从 Python 脚本中创建新的进程，可以连接到它们的输入/输出/错误管道,并获取它们的返回码。
（图片来源网络，侵删）
就是让你在 Python 代码里能像在终端（命令行）里一样运行其他程序或命令。
为什么需要它？
- 自动化任务：编写 Python 脚本来批量处理文件，比如用
ffmpeg转换视频格式，用git管理代码仓库。 - 调用外部工具：利用系统上已有的强大工具，
curl访问网络、grep搜索文本、aws调用云服务 API。 - 系统集成：将 Python 脚本无缝集成到现有的基于命令行的工作流中。
核心函数 (现代 Python 推荐方式)
Python 3.5+ 引入了 subprocess.run()，它是一个更统一、更推荐的接口，可以替代许多旧的函数,我们先重点掌握它。
subprocess.run()
这是最通用、最推荐的函数,它的基本语法是：
（图片来源网络，侵删）
subprocess.run(args, *, stdin=None, input=None, stdout=None, stderr=None, capture_output=False, shell=False, text=False, encoding=None, errors=None, ...) # 最简单的用法 subprocess.run(['ls', '-l'])
关键参数解析：
args: 这是最重要的参数。shell=False(默认值)，args必须是一个列表，列表的每个元素是命令的一个部分。['ls', '-l', '/home']。shell=True，args可以是一个字符串，Shell 会负责解析这个字符串。'ls -l /home'。
capture_output=True: 如果设置为True，程序的标准输出和标准错误会被捕获，而不是直接打印在终端上，可以通过result.stdout和result.stderr访问。text=True或universal_newlines=True: 如果设置为True，捕获的输出 (stdout,stderr) 会以字符串形式返回，否则，默认是字节形式 (bytes)。check=True: 如果子进程返回了非零的退出码（表示出错），会抛出CalledProcessError异常,这对于判断命令是否成功执行非常有用。input: 提供一个字符串或字节串作为子进程的标准输入。
返回值：
run() 函数返回一个 CompletedProcess 对象,这个对象包含了丰富的信息：
returncode: 子进程的退出码。0表示成功，非0表示失败。stdout: 捕获到的标准输出（字符串或字节）。stderr: 捕获到的标准错误（字符串或字节）。args: 传递给子进程的命令。
进阶用法：Popen 类
Popen (process open) 是 subprocess 模块中最底层的、功能最全的类，当你需要更精细的控制时（比如与子进程进行实时交互、并行执行多个命令等）,就需要使用它。
（图片来源网络，侵删）
Popen 的核心特点：
- 非阻塞启动：
Popen启动子进程后会立即返回，不会等待子进程结束,你需要手动去检查子进程的状态。 - 管道控制：可以灵活地打开、关闭子进程的
stdin,stdout,stderr管道。
基本用法：
import subprocess
# 启动一个进程，但不等待它结束
p = subprocess.Popen(['ls', '-l'])
# 你可以做其他事情...
# print("主进程继续执行...")
# 检查进程是否结束
# p.poll() 返回 None 表示进程仍在运行，返回码表示已结束
# p.wait() 会阻塞，直到进程结束并返回返回码
# 等待进程结束
p.wait()
print(f"进程 {p.pid} 已结束，返回码: {p.returncode}")
与子进程交互：
import subprocess
# 启动一个可以交互的进程，'bc' (计算器)
p = subprocess.Popen(['bc'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)
# 通过 stdin 发送数据
p.stdin.write("2 + 2\n")
p.stdin.flush() # 确保数据被发送
# 从 stdout 读取结果
output = p.stdout.readline()
print(f"计算结果: {output.strip()}") # 输出: 计算结果: 4
# 发送退出命令
p.stdin.write("quit\n")
p.stdin.flush()
# 等待进程完全退出
p.wait()
安全注意事项：Shell 注入
永远不要轻易使用 shell=True，除非你完全信任输入的参数。
当 shell=True 时，你传递的命令字符串会通过系统的 shell (如 /bin/sh 或 cmd.exe) 来执行，这会带来严重的安全风险，即 Shell 注入。
危险示例：
import subprocess
user_input = "malicious; rm -rf /" # 用户输入了恶意命令
# 危险！不要这样做！
# 这行命令会被 shell 解释为：'echo "malicious; rm -rf /"'
# 实际上会执行两条命令：echo ... 和 rm -rf /
subprocess.run(f'echo "{user_input}"', shell=True)
安全做法：
始终使用列表形式传递参数，并保持 shell=False (默认)。subprocess 会负责安全地将参数传递给目标程序，而不会经过 shell 解析。
import subprocess user_input = "malicious; rm -rf /" # 安全的做法 # subprocess 会将 'echo' 作为第一个参数，user_input 作为第二个参数 # 不会执行 'rm -rf /' subprocess.run(['echo', user_input])
实用代码示例
示例 1：获取命令输出
获取当前目录下所有 .py 文件的个数。
import subprocess
# 使用 check_output 捕获输出
# 返回的是字节，需要解码
output_bytes = subprocess.check_output(['ls', '*.py'])
# 解码为字符串
output_str = output_bytes.decode('utf-8')
file_list = output_str.strip().split('\n')
print(f"找到 {len(file_list)} 个 Python 文件:")
print(file_list)
示例 2：检查命令是否成功执行
尝试创建一个目录，如果目录已存在，mkdir 会报错。
import subprocess
import os
dir_name = "my_new_dir"
# 使用 check=True，如果命令失败会抛出异常
try:
# -p 参数可以创建多级目录，且如果目录已存在不会报错
subprocess.run(['mkdir', '-p', dir_name], check=True)
print(f"目录 '{dir_name}' 创建成功或已存在。")
except subprocess.CalledProcessError as e:
print(f"创建目录失败，返回码: {e.returncode}")
except FileNotFoundError:
print("错误: 'mkdir' 命令未找到，请确保它在系统路径中。")
# 使用 run 的返回码检查
result = subprocess.run(['ls', dir_name])
if result.returncode == 0:
print(f"目录 '{dir_name}' 确实存在。")
示例 3：使用 Popen 实时读取输出
执行一个需要时间的命令（ping）,并实时打印其输出。
import subprocess
print("开始 ping 8.8.8.8...")
# 使用 Popen，并实时读取 stdout
# bufsize=1 表示行缓冲
# universal_newlines=True 表示以文本模式读取
with subprocess.Popen(['ping', '-c', '5', '8.8.8.8'], stdout=subprocess.PIPE, text=True, bufsize=1) as p:
for line in p.stdout: # 逐行读取
print(line, end='') # end='' 因为 line 中已经包含了换行符
print("\nPing 结束。")
if p.returncode == 0:
print("所有包都成功接收。")
else:
print("有包丢失或 ping 失败。")
| 需求场景 | 推
