Python高级技术之：如何利用`Python`的`subprocess`模块，进行`Shell`命令的自动化。 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

各位观众老爷，大家好！欢迎来到“Python高级技术之subprocess模块玩转Shell命令自动化”讲座现场。今天咱们就来聊聊如何让Python化身你的自动化运维小助手，轻松驾驭Shell命令。

开场白：为啥要用subprocess？

想象一下，你辛辛苦苦写了个Python脚本，需要调用一些Linux命令，比如查看磁盘空间、重启服务、或者执行一些复杂的系统管理操作。难道要每次都手动打开终端，输入命令吗？当然不用！Python的subprocess模块就是来拯救你的。它允许你在Python脚本中启动新的进程，连接到它们的输入/输出/错误管道，并获取它们的返回码。简单来说，就是让你用Python代码控制Shell命令，实现自动化。

第一幕：subprocess模块初体验

subprocess模块的核心在于它的几个主要函数，其中最常用的是subprocess.run()。

import subprocess

# 执行一个简单的命令：列出当前目录的文件
result = subprocess.run(['ls', '-l'], capture_output=True, text=True)

# 打印命令的输出
print("命令的输出：")
print(result.stdout)

# 打印命令的错误输出（如果有的话）
print("命令的错误输出：")
print(result.stderr)

# 打印命令的返回码
print("命令的返回码：")
print(result.returncode)

这个例子中，我们使用subprocess.run()执行了ls -l命令。capture_output=True告诉Python捕获命令的标准输出和标准错误输出。text=True则确保输出以文本形式返回，而不是字节流。

result对象包含了命令执行的所有信息，包括标准输出(stdout)、标准错误输出(stderr)和返回码(returncode)。返回码0表示命令执行成功，非0值表示出错。

关键参数解析：

参数	作用
`args`	要执行的命令，可以是一个字符串或者一个列表。如果是列表，每个元素都是命令的一个参数。例如：`['ls', '-l']`。
`capture_output`	是否捕获命令的标准输出和标准错误输出。默认为`False`。如果设置为`True`，则可以通过`result.stdout`和`result.stderr`访问输出。
`text`	是否以文本模式打开标准输出和标准错误输出。默认为`False`，表示以字节流模式打开。如果设置为`True`，则以文本模式打开，返回的输出是字符串。
`shell`	是否通过shell执行命令。默认为`False`。如果设置为`True`，则`args`参数可以是一个包含整个命令的字符串。但要注意，使用`shell=True`存在安全风险，特别是当命令字符串来自用户输入时。
`cwd`	设置命令执行的当前工作目录。
`env`	设置命令执行的环境变量。
`timeout`	设置命令执行的超时时间（秒）。如果命令执行时间超过`timeout`，则抛出`subprocess.TimeoutExpired`异常。
`check`	如果命令的返回码不是0，则抛出`subprocess.CalledProcessError`异常。默认为`False`。

第二幕：更高级的用法

使用shell=True的便利与风险：

shell=True允许你直接执行包含管道、重定向等复杂Shell语法的命令。
```
import subprocess

# 使用shell=True执行一个包含管道的命令
result = subprocess.run('ls -l | grep "python"', capture_output=True, text=True, shell=True)

print(result.stdout)
```
警告： 尽量避免使用shell=True，特别是当命令字符串包含用户输入时。因为这可能导致Shell注入漏洞。比如，如果用户输入'; rm -rf /'，那么这个命令就会被执行，导致系统数据丢失。

安全替代方案： 尽量将命令拆分成参数列表，避免使用shell=True。
```
import subprocess

# 安全的替代方案
command = ['ls', '-l']
grep_pattern = 'python'
pipe = subprocess.Popen(command, stdout=subprocess.PIPE)
result = subprocess.run(['grep', grep_pattern], stdin=pipe.stdout, capture_output=True, text=True)
pipe.stdout.close()  # Allow p1 to receive a SIGPIPE if p2 exits.

print(result.stdout)
```
这段代码使用 subprocess.Popen 创建了一个管道，首先执行 ls -l 并将其输出传递给 grep 命令进行过滤。这种方式更安全，因为它避免了将整个命令交给 shell 解析，减少了潜在的注入风险。

处理输入和输出：

除了捕获输出之外，你还可以向命令传递输入。

import subprocess

# 向命令传递输入
input_data = "This is some input for the command."
result = subprocess.run(['wc', '-w'], capture_output=True, text=True, input=input_data)

print(result.stdout) #输出单词数量

input参数用于向命令的标准输入传递数据。

实时输出：

有时候，你希望实时查看命令的输出，而不是等待命令执行完毕。这时可以使用subprocess.Popen()和stdout=subprocess.PIPE。

import subprocess
import shlex

command = "ping 8.8.8.8" # 使用 shlex.split 处理命令字符串
process = subprocess.Popen(shlex.split(command), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)

while True:
    output = process.stdout.readline()
    if output:
        print(output.strip())
    return_code = process.poll()
    if return_code is not None:
        # Process has finished, read any remaining output and exit
        for output in process.stdout.readlines():
            print(output.strip())
        for error in process.stderr.readlines():
            print(error.strip())
        print('RETURN CODE', return_code)
        break

subprocess.Popen()启动一个新进程，stdout=subprocess.PIPE告诉Python连接到命令的标准输出管道。然后，你可以循环读取管道中的数据，并实时打印出来。
shlex.split用于处理包含空格和特殊字符的命令字符串，将其分割成参数列表。

处理错误：

如果命令执行出错，subprocess.run()会抛出subprocess.CalledProcessError异常（如果check=True）。

import subprocess

try:
    result = subprocess.run(['rm', 'nonexistent_file'], check=True, capture_output=True, text=True)
except subprocess.CalledProcessError as e:
    print(f"命令执行出错：{e}")
    print(f"错误输出：{e.stderr}")
    print(f"返回码：{e.returncode}")

try...except块用于捕获异常，并打印错误信息。

设置超时时间：

为了防止命令执行时间过长，可以使用timeout参数设置超时时间。

import subprocess

try:
    result = subprocess.run(['sleep', '10'], timeout=5, check=True, capture_output=True, text=True)
except subprocess.TimeoutExpired as e:
    print(f"命令执行超时：{e}")

如果命令执行时间超过5秒，就会抛出subprocess.TimeoutExpired异常。

第三幕：实际应用场景

自动化部署：

可以使用subprocess模块编写自动化部署脚本，例如：

import subprocess

def deploy_app(app_name, version):
    """自动化部署应用"""
    try:
        # 停止应用
        subprocess.run(['systemctl', 'stop', app_name], check=True)

        # 更新代码
        subprocess.run(['git', 'pull'], cwd=f'/opt/{app_name}', check=True)

        # 构建应用
        subprocess.run(['mvn', 'clean', 'install'], cwd=f'/opt/{app_name}', check=True)

        # 启动应用
        subprocess.run(['systemctl', 'start', app_name], check=True)

        print(f"应用 {app_name} 部署成功，版本：{version}")

    except subprocess.CalledProcessError as e:
        print(f"部署失败：{e}")

# 调用部署函数
deploy_app('my_app', '1.0.0')

这个脚本可以自动停止应用、更新代码、构建应用和启动应用。

系统监控：

可以使用subprocess模块编写系统监控脚本，例如：

import subprocess

def get_cpu_usage():
    """获取CPU使用率"""
    result = subprocess.run(['top', '-bn1'], capture_output=True, text=True)
    lines = result.stdout.splitlines()
    for line in lines:
        if line.startswith("%Cpu(s)"):
            cpu_usage = line.split(":")[1].split(",")[0].strip()
            return cpu_usage
    return None

def get_memory_usage():
    """获取内存使用率"""
    result = subprocess.run(['free', '-m'], capture_output=True, text=True)
    lines = result.stdout.splitlines()
    total_memory = int(lines[1].split()[1])
    used_memory = int(lines[1].split()[2])
    memory_usage = round(used_memory / total_memory * 100, 2)
    return memory_usage

# 获取CPU和内存使用率
cpu_usage = get_cpu_usage()
memory_usage = get_memory_usage()

print(f"CPU使用率：{cpu_usage}%")
print(f"内存使用率：{memory_usage}%")

这个脚本可以获取CPU和内存使用率，并打印出来。

第四幕：subprocess.Popen() 的进阶用法

subprocess.Popen() 是一个更底层的函数，它允许你更精细地控制子进程的创建和交互。它返回一个 Popen 对象，你可以使用这个对象来与子进程进行通信。

非阻塞模式： Popen 默认是阻塞模式，即会等待子进程执行完毕。可以通过 Popen.poll() 方法检查子进程是否结束，或者使用 Popen.wait() 方法等待子进程结束。

import subprocess
import time

command = ["sleep", "5"]  # 模拟一个耗时较长的命令
process = subprocess.Popen(command)

print("开始执行命令...")
while process.poll() is None:
    print("命令仍在执行...")
    time.sleep(1)  # 每隔1秒检查一次

print("命令执行完毕，返回码:", process.returncode)

使用 communicate() 方法： Popen.communicate() 方法用于向子进程发送数据，并从子进程接收数据。它会阻塞直到子进程结束。

import subprocess

command = ["wc", "-w"]
input_data = "This is a test string.nThis is another line."

process = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
stdout, stderr = process.communicate(input=input_data)

print("标准输出:", stdout)
print("标准错误:", stderr)
print("返回码:", process.returncode)

在这个例子中，我们使用 communicate() 方法向 wc -w 命令传递输入数据，并获取其输出。

第五幕：错误处理的最佳实践

在使用 subprocess 模块时，错误处理至关重要。以下是一些最佳实践：

使用 check=True 参数： subprocess.run() 和 subprocess.check_call() 函数的 check=True 参数可以自动检查子进程的返回码。如果返回码非零，则会抛出 subprocess.CalledProcessError 异常。
捕获标准错误： 始终捕获子进程的标准错误输出，以便诊断问题。
使用 try...except 块： 使用 try...except 块来处理可能发生的异常，例如 subprocess.CalledProcessError 和 subprocess.TimeoutExpired。
记录日志： 记录子进程的输入、输出和错误，以便进行调试和审计。

import subprocess
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def run_command(command):
    """运行命令并记录日志"""
    try:
        logging.info(f"运行命令: {command}")
        result = subprocess.run(command, check=True, capture_output=True, text=True)
        logging.info(f"命令输出:n{result.stdout}")
        return result
    except subprocess.CalledProcessError as e:
        logging.error(f"命令执行失败: {e}")
        logging.error(f"错误输出:n{e.stderr}")
        raise  # 重新抛出异常，以便上层处理
    except Exception as e:
        logging.exception(f"发生未知错误: {e}")
        raise

try:
    run_command(["ls", "-l", "/nonexistent_dir"])
except Exception:
    print("命令执行失败，请查看日志")

第六幕：安全注意事项

避免使用 shell=True： 尽量避免使用 shell=True，特别是当命令字符串来自用户输入时。这可以防止 Shell 注入攻击。
验证用户输入： 如果命令字符串包含用户输入，请务必验证输入，以确保其不包含恶意代码。
使用白名单： 对于允许执行的命令，最好使用白名单机制，只允许执行预定义的命令。
限制权限： 以最小权限运行子进程，以减少潜在的风险。

总结：

subprocess模块是Python中进行Shell命令自动化的强大工具。通过它，你可以轻松地在Python脚本中执行Shell命令，并获取命令的输出和返回码。但是，在使用subprocess模块时，要注意安全问题，避免使用shell=True，并对用户输入进行验证。掌握了这些技巧，你就可以用Python编写出更加高效、可靠的自动化运维脚本啦！

这次讲座就到这里，感谢大家的观看！希望大家以后可以灵活运用 subprocess 模块，解放双手，拥抱自动化！

发表回复 取消回复

发表回复取消回复