【摸鱼笔记】subprocess，用子进程访问外部命令

subprocess是一个非常强大的模块，它能通过启动新进程的方式执行外部命令、访问系统Shell、管道等。

同时也能捕获这些新进程的输出和错误信息；

subprocess模块是Python的标准库中的一部分，支持 Python 3.5 及 3.5 以上版本；

3.5 以下版本需要额外安装；

执行外部命令

subprocess.run()

subprocess.run() 用于运行外部命令并等待其完成；

capture_output: 抓取输出；capture_output 参数与 stdout，stderr 相关，

在不传入capture_output 和 stdout、stderr 的默认情况下，subprocess不会抓取 stdout 和 stderr，并且返回值对象里这两个属性为
None。

传入stdout=PIPE 和 stderr=PIPE 或者 stdout=PIPE、stderr=PIPE 其中一个，subprocess会对应着抓取它们；

如果希望抓取并将两个流合并在一起，可以使用 stdout=PIPE 和 stderr=STDOUT；

直接传入 capture_output=True，subprocess会stdout、stderr两个都抓取；

因为在capture_output=True时，内置的 Popen 对象将自动用 stdout=PIPE 和 stderr=PIPE 创建管道。

所以stdout 和 stderr 参数不应当与 capture_output 同时传入。

check: 返回状态检查；check 参数和子进程运行退出状态码有关

当子进程退出时 return code（状态码）不为 0，则会触发 CalledProcessError异常；

退出的 return code 会保存到 CalledProcessError.returncode 属性中，

如果 stdout、stderr有被抓取，则 stdout 和 stderr 也会在 CalledProcessError.stdout 和 CalledProcessError.stderr属性里;

timeout ：超时时间，如果传入 timeout，并且进程花费的时间超出 timeout，则会触发TimeoutExpired异常;

text: 功能类似 universal_newlines 参数,

默认不传入的情况下，subprocess.run() 返回的stdout、stderr是字节类型,

当text或者universal_newlines为True，stdout、stderr输出是字符串类型；

cwd: 指定 subprocess 子进程工作路径；

input ：向子进程传入的参数，可选参数，以下是官方文档的解释

input 参数将被传递给 Popen.communicate() 以及子进程的 stdin。它必须是一个字节序列；

如果指定了 encoding 或 errors 或者将 text 设置为 True，那么也可以是一个字符串。

当使用此参数时，在创建内部 Popen 对象时将自动带上 stdin=PIPE，并且不能再手动指定 stdin 参数。

一般使用

python">import subprocesscomplete = subprocess.run(["ipconfig", "/all"],capture_output=True,text=True)
print(complete.stdout)

拼接参数

如果命令需要接受参数，可以将它们作为列表的一部分传递给subprocess.run()或subprocess.Popen()。

例如，要将文件名作为参数传递给命令，可以这样做：

python">import subprocesssub_path = "data_lake"
complete = subprocess.run(["dir", sub_path],shell=True,capture_output=True,text=True,cwd="E:\\VsCodeProjects\\")
print(complete.stdout)

指定路径

使用cwd参数来指定执行外部命令的工作目录。例如，要在特定目录中执行命令，可以这样做：

python">import subprocesscomplete = subprocess.run(["dir"],shell=True,capture_output=True,text=True,cwd="E:\\VsCodeProjects\\")
print(complete.stdout)

使用Shell命令

默认情况下，subprocess不会使用Shell来执行命令。这是出于安全考虑，以防止潜在的Shell注入攻击。但有些情况下，可能需要使用Shell来执行命令，可以将shell参数设置为True。

python">import subprocesscomplete = subprocess.run(["dir", "nonexistent"],shell=True,capture_output=True,text=True)
print("标准输出:")
print(complete.stdout)
print("标准错误:")
print(complete.stderr)

输入输出

标准输出

python">import subprocess# 使用 capture_output
complete = subprocess.run(["ipconfig", "/all"],capture_output=True,text=True)
print(complete.stdout)# 单独使用 stdout
complete = subprocess.run(["ipconfig", "/all"],stdout=subprocess.PIPE,text=True)
print(complete.stdout)# 单独使用 stdout 输出到txt
with open("stdout_tmp.txt", mode="w", encoding="utf8") as fw:complete = subprocess.run(["ipconfig", "/all"],stdout=fw,text=True)print(complete.stdout)# stdout 与 stderr 合并
complete = subprocess.run(["ipconfig", "/all"],stdout=subprocess.PIPE,stderr=subprocess.STDOUT,text=True)
print(complete.stdout)

标准输入

subprocess模块还可以将数据传递给外部命令的标准输入。要实现这一点，可以使用stdin参数，并将其设置为一个文件对象或一个字节串。

python">import subprocessinput_data = "Hello, subprocess!"
complete = subprocess.run(["findstr", "subprocess"],input=input_data,capture_output=True,text=True)
print(complete.stdout)

标准错误

与标准输出类似，subprocess还可以捕获标准错误信息。要捕获标准错误，请使用stderr参数。

python">import subprocesscomplete = subprocess.run(["dir", "nonexistent"],shell=True,capture_output=True,text=True)
print("标准输出:")
print(complete.stdout)
print("标准错误:")
print(complete.stderr)

子进程运行状态检查

执行外部命令时，通常需要处理错误。以下是一些处理错误的常用方法

抓取stderr内容

python">import subprocesscomplete = subprocess.run(["dir", "nonexistent"],shell=True,capture_output=True,text=True)
if complete.returncode != 0:print("stderr:", complete.stderr)

触发 CalledProcessError

python">import subprocesstry:complete = subprocess.run(["dir", "nonexistent"],shell=True,check=True,capture_output=True,text=True)print("stdout:", complete.stdout)
except subprocess.CalledProcessError as e:print("CalledProcessError:")print("CalledProcessError.returncode:", e.returncode)print("CalledProcessError.stdout:", e.stdout)print("CalledProcessError.stderr:", e.stderr)

超时处理

subprocess还允许您设置执行命令的超时时间，以防止命令运行时间过长。要实现这一点，您可以使用timeout参数。

python">import subprocesstry:complete = subprocess.run(["pause"],shell=True,timeout=12,capture_output=True,text=True)print(complete.stdout)
except subprocess.TimeoutExpired as e:print("TimeoutExpired")print("TimeoutExpired.stdout", e.stdout)print("TimeoutExpired.stderr", e.stderr)

subprocess.Popen()

subprocess.Popen()提供了更多的灵活性，允许与进程进行交互，而不仅仅是等待它完成。

args 应当是一个程序参数的序列或者是一个单独的字符串或 path-like object。

默认情况下，如果 args 是序列则要运行的程序为 args 中的第一项。

stdin, stdout 和 stderr 分别指定被执行程序的标准输入、标准输出和标准错误文件句柄；

可以传入的值有 None, PIPE, DEVNULL, 或者文件对象；

当使用默认设置 None 时，将不会进行任何重定向；

PIPE 表示应当新建一个连接子进程的管道；

DEVNULL 表示将使用特殊文件 os.devnull；

stderr 还可以为 STDOUT，stdout 和 stderr 合并；

shell：默认为 False, 指定是否使用 shell 执行程序；

如果 cwd 不为 None，此函数在执行子进程前会将当前工作目录改为 cwd，cwd 可以是一个字符串、字节串或路径类对象。

text: 功能类似 universal_newlines 参数,

默认不传入的情况下，subprocess.Popen() 返回的stdout、stderr是字节类型,

当text或者universal_newlines为True，stdout、stderr输出是字符串类型；

encoding,errors: 在默认情况下，文件对象将以二进制模式打开。

如果指定了 encoding 或 errors，或者如果 text 为真值，

则文件对象 stdin, stdout 和 stderr 将使用指定的 encoding 和 errors 以文本模式打开。

python">import subprocess# 执行命令
process = subprocess.Popen(["ls", "-l"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)# 读取标准输出和错误
out, err = process.communicate()print("标准输出:")
print(out)print("标准错误:")
print(err)

管道和重定向

subprocess还可以创建管道，将一个命令的输出连接到另一个命令的输入。这在处理复杂的数据处理任务时非常有用。

例如，要将一个命令的输出传递给另一个命令，

python">import subprocess# 创建第一个命令的进程
subprocess1 = subprocess.Popen(["dir", "E:\\PyCharmProjects\\nicole\\"],shell=True,stdout=subprocess.PIPE,text=True)# 创建第二个命令的进程，将第一个命令的输出连接到它的输入
subprocess2 = subprocess.Popen(["findstr", "favicon"],shell=True,stdin=subprocess1.stdout,stdout=subprocess.PIPE,text=True)# 从第二个命令的标准输出中读取结果
complete = subprocess2.communicate()[0]
print(complete)

子进程通信

subprocess模块同时读取和写入标准输入和输出。这对于与外部进程进行双向通信非常有用。

以下是一个示例，演示如何使用subprocess进行双向通信：

python">import subprocess# 创建命令进程
process = subprocess.Popen(["python", "-u"],stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)# 写入数据到标准输入
with process.stdin as std_in:std_in.write("print('Hello from child process')\n")# 读取并打印标准输出
output, errors = process.communicate()
print("输出:", output, errors)