4.12 Python 编写运维自动化脚本

预计阅读时间：15 分钟

学习目标

• 掌握 Python 运维脚本的核心库：argparse、subprocess、requests、paramiko、psutil、logging
• 理解何时该从 Shell 切换到 Python，以及两者的权衡
• 能用 Python 编写实用的运维自动化脚本（命令包装、SSH 批量操作、系统健康检查）
• 学会使用 virtualenv / Poetry 管理项目依赖和环境隔离

核心知识

常用库速查表

知识关联

• 前置知识：2.1：Shell 脚本入门 Shell 脚本入门、2.8：Shell 脚本进阶 Shell 脚本进阶、2.11：SSH 深入 SSH 深入（paramiko 依赖 SSH 协议理解）
• 后续影响：4.1：Ansible 自动化 Ansible 自动化运维（Python 是 Ansible 底层语言）、4.9：CI/CD 持续部署 CI/CD 持续部署（Python 编写流水线脚本）
• 配套技术：Python + Fabric + Invoke 替代 Shell 做远程运维，Poetry/virtualenv 管理项目依赖隔离

原理讲解

当 Shell 不够用时——何时该换 Python 写运维脚本？

Shell 脚本（2.1：Shell 脚本入门、2.8：Shell 脚本进阶）擅长文件操作、管道组合和快速原型，但在以下场景力不从心：

• 复杂逻辑：多层循环、条件分支、数据结构（字典/列表）
• 异常处理：网络超时、SSH 断开、JSON/XML 解析失败
• 跨平台：Shell 命令因发行版/系统而异，Python 标准库统一
• 依赖管理：多脚本共享函数库时，Shell 的 source 机制脆弱
• 性能：大量文本解析或 HTTP 调用时 Python 更高效稳定

决策原则：单次手动的任务用 Shell；需要重试、告警、持久化、结构化数据的自动化任务用 Python。

Python 运维场景清单

实践中哪些任务值得用 Python 而不是 Shell？按出现频率排序的运维场景清单：

Python vs Shell 选型对比

经验法则：脚本 30 行以内且一次性使用 → Shell；需要长期维护、会重试/告警/解析结构化数据 → Python。两者也常组合：Shell 管道里用 python3 -c 处理中间数据。

subprocess 详解：run / Popen / shell 安全

subprocess 是 Python 运维脚本的"shell"，掌握三个层次即可覆盖绝大多数场景：

import subprocess

# run()：一次性执行（90% 的场景用这个）
r = subprocess.run(["systemctl", "is-active", "nginx"],
                   capture_output=True, text=True)
print(r.returncode, r.stdout.strip())   # 输出: 0 active

# Popen()：实时读日志流（等价于 tail -f 并过滤）
p = subprocess.Popen(["journalctl", "-f", "-u", "nginx"],
                     stdout=subprocess.PIPE, text=True)
for line in p.stdout:
    if "error" in line:
        print("发现错误行:", line.strip())
        break
p.terminate()

# shell=True 的危害演示：命令注入
filename = "foo; rm -rf /tmp/evil"
subprocess.run(f"ls {filename}", shell=True)    # 危险！文件名里的分号会被 shell 执行
subprocess.run(["ls", filename])                # 安全：列表参数不经 shell 解释

shell=True 安全警告：一旦用了 shell=True，字符串里的 ;、|、>、$() 都会被 shell 解释。脚本输入来自外部（文件名、URL 参数、API 返回值）时，必须用列表参数形式让 Python 直接 exec。真需要管道时用 Popen(stdout=PIPE) 串联两个进程，而不是拼 shell 字符串——后者既是安全漏洞也是引号地狱。

为什么用 Python 而不是 Bash 写运维脚本

Bash 脚本在简单任务上无可替代——启动快、管道组合优雅、系统命令直接调用。但当脚本超过 100 行或需要处理复杂逻辑时，Bash 的局限暴露：数据结构缺失——只有字符串和数组，没有字典/集合/对象，解析 JSON 需要 jq 等外部工具；错误处理脆弱——set -e 只能捕获命令返回码，无法区分"命令不存在"和"命令执行失败"；跨平台差异——sed -i 在 macOS 和 Linux 上语法不同，readlink -f 在 BSD 上不存在；可测试性差——Bash 函数难以单元测试。Python 标准库统一了这些差异，subprocess 提供安全的命令调用，try/except 提供完整的异常处理链，argparse 自动生成帮助文档。经验法则：脚本 30 行以内且一次性 → Bash；需要长期维护、会重试/告警/解析结构化数据 → Python。

为什么 paramiko 而不是直接 subprocess 调用 ssh

用 subprocess.run(["ssh", "host", "cmd"]) 调用 SSH 有几个问题：输出解析困难——SSH 的连接信息、警告、错误混在 stdout/stderr 中，难以程序化分离；连接复用困难——每次 subprocess 调用都是独立的 SSH 连接，批量操作 100 台服务器需要建立 100 次 TCP 连接+密钥交换；认证管理复杂——密码认证需要交互式输入或 Expect 脚本，密钥认证需要管理 agent 转发。paramiko 作为纯 Python SSH 库解决了所有问题：原生 API 控制连接生命周期、支持连接复用（SSHClient.get_transport()）、内置多种认证方式（密钥/密码/agent）、输出以 file-like 对象返回便于程序化处理。Fabric 是 paramiko 的上层封装，进一步简化了批量操作语法。

示例代码

示例 1：subprocess 命令包装器

#!/usr/bin/env python3
"""命令安全执行包装器 —— 含超时、日志、重试"""
import subprocess, sys, logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

def run_cmd(cmd, timeout=30, retries=2):
    for attempt in range(1, retries + 2):
        try:
            r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, check=True)
            logging.info("OK: %s", " ".join(cmd))
            return r.stdout.strip()
        except subprocess.TimeoutExpired:
            logging.warning("超时 (attempt %d): %s", attempt, cmd)
        except subprocess.CalledProcessError as e:
            logging.error("失败 (attempt %d): %s\n%s", attempt, e, e.stderr)
        if attempt <= retries:
            import time; time.sleep(2 ** attempt)
    sys.exit(1)

if __name__ == "__main__":
    print(run_cmd(["df", "-h"]))  # 输出: Filesystem      Size  Used Avail Use% Mounted on ...

示例 2：paramiko SSH 批量命令

#!/usr/bin/env python3
"""批量 SSH 执行命令，收集结果"""
import paramiko, argparse
from concurrent.futures import ThreadPoolExecutor

def ssh_run(host, cmd, key_path="~/.ssh/id_ed25519"):
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(host, username="root", key_filename=key_path, timeout=10)
    _, stdout, stderr = client.exec_command(cmd, timeout=15)
    rc = stdout.channel.recv_exit_status()
    out = stdout.read().decode().strip()
    err = stderr.read().decode().strip()
    client.close()
    return host, rc, out, err

def main():
    ap = argparse.ArgumentParser(description="批量 SSH 执行命令")
    ap.add_argument("hosts", nargs="+", help="目标主机列表")
    ap.add_argument("-c", "--cmd", default="uptime", help="要执行的命令")
    ap.add_argument("-k", "--key", default="~/.ssh/id_ed25519", help="SSH 密钥路径")
    args = ap.parse_args()

    with ThreadPoolExecutor(max_workers=8) as pool:
        futures = [pool.submit(ssh_run, h, args.cmd, args.key) for h in args.hosts]
        for f in futures:
            host, rc, out, err = f.result()
            status = "OK" if rc == 0 else "FAIL"
            print(f"[{status}] {host}: {out or err}")  # 输出: [OK] 10.0.0.1: 09:30:00 up 10 days

if __name__ == "__main__":
    main()

示例 3：psutil 系统健康检查与告警

#!/usr/bin/env python3
"""采集系统指标，超阈值时输出 JSON 告警"""
import psutil, json, socket, logging
from datetime import datetime

logging.basicConfig(level=logging.WARNING, format="%(message)s")
alerts = []
host = socket.gethostname()
now = datetime.now().isoformat()

def check(metric, value, warn, crit):
    if value >= crit:
        alerts.append({"level": "CRITICAL", "host": host, "metric": metric, "value": value, "threshold": crit, "time": now})
    elif value >= warn:
        alerts.append({"level": "WARNING", "host": host, "metric": metric, "value": value, "threshold": warn, "time": now})

check("cpu_percent", psutil.cpu_percent(interval=2), 70, 90)
check("mem_percent", psutil.virtual_memory().percent, 80, 95)
check("disk_root_percent", psutil.disk_usage("/").percent, 80, 92)
check("disk_data_percent", psutil.disk_usage("/data").percent, 85, 95)

if alerts:
    for a in alerts:
        logging.warning(json.dumps(a))  # 输出: {"level": "WARNING", "host": "...", "metric": "cpu_percent", ...}
else:
    print(json.dumps({"level": "OK", "host": host, "time": now}))  # 输出: {"level": "OK", "host": "...", "time": "..."}

示例 4：日志分析脚本（正则 + 统计）

#!/usr/bin/env python3
"""Nginx 访问日志分析：状态码分布、TOP IP、总请求量"""
import re, sys, collections

LOG_PAT = re.compile(
    r'^(?P<ip>[\d.]+) .*? \[(?P<time>[^\]]+)\] '
    r'"(?P<req>[^"]*)" (?P<status>\d{3}) (?P<size>\d+) '
    r'"(?P<ref>[^"]*)" "(?P<ua>[^"]*)"'
)

def analyze(path):
    statuses, ips = collections.Counter(), collections.Counter()
    total = 0
    with open(path, errors="replace") as f:      # errors="replace" 容忍日志中的乱码字节
        for line in f:
            m = LOG_PAT.match(line)
            if not m:
                continue
            total += 1
            statuses[m.group("status")] += 1
            ips[m.group("ip")] += 1
    print(f"总请求数: {total}")
    print("状态码分布:", dict(statuses))
    print("TOP 5 IP:")
    for ip, c in ips.most_common(5):
        print(f"  {ip}: {c}")

if __name__ == "__main__":
    analyze(sys.argv[1] if len(sys.argv) > 1 else "/var/log/nginx/access.log")
    # 输出: 总请求数: 123456
    # 输出: 状态码分布: {'200': 118000, '404': 3456, '500': 12, ...}
    # 输出: TOP 5 IP: 1.2.3.4: 89012 ...

示例 5：批量重命名脚本（干跑模式）

#!/usr/bin/env python3
"""批量重命名：backup_20260101.tar.gz → 20260101.tar.gz
--dry-run 先预览，确认无误再去掉参数真正执行"""
import os, re, sys, argparse

def main():
    ap = argparse.ArgumentParser(description="批量重命名备份文件")
    ap.add_argument("dir", help="目标目录")
    ap.add_argument("--dry-run", action="store_true", help="只预览不执行")
    ap.add_argument("--pattern", default=r"^backup_(\d{8})\.tar\.gz$",
                    help="匹配正则，捕获组用于拼新文件名")
    args = ap.parse_args()

    renamed = 0
    for name in sorted(os.listdir(args.dir)):
        m = re.match(args.pattern, name)
        if not m:
            continue
        new_name = f"{m.group(1)}.tar.gz"
        old, new = os.path.join(args.dir, name), os.path.join(args.dir, new_name)
        if os.path.exists(new):
            print(f"跳过（目标已存在）: {name} -> {new_name}")
            continue
        if args.dry_run:
            print(f"[DRY-RUN] {name} -> {new_name}")
        else:
            os.rename(old, new)
            print(f"已重命名: {name} -> {new_name}")
        renamed += 1
    print(f"共处理 {renamed} 个文件")

if __name__ == "__main__":
    main()

# 用法: ./rename.py /data/backups --dry-run
# 输出: [DRY-RUN] backup_20260101.tar.gz -> 20260101.tar.gz ...

示例 6：服务器健康检查脚本（巡检 + 告警推送）

#!/usr/bin/env python3
"""巡检脚本：检查服务、端口、磁盘、负载，异常时推送到企业微信 webhook"""
import json, subprocess, socket, datetime
import psutil, requests

WEBHOOK = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=XXXX"
HOST = socket.gethostname()

def check_service(name):
    r = subprocess.run(["systemctl", "is-active", name],
                       capture_output=True, text=True)
    return r.stdout.strip() == "active"

def check_port(port, host="127.0.0.1"):
    with socket.socket() as s:
        s.settimeout(2)
        return s.connect_ex((host, port)) == 0

def main():
    problems = []
    for svc in ["nginx", "mysql", "redis-server"]:
        if not check_service(svc):
            problems.append(f"服务 {svc} 不在运行")
    if not check_port(6379):
        problems.append("Redis 端口 6379 不通")
    load = psutil.getloadavg()
    if load[0] > 4:
        problems.append(f"负载过高: {load[0]:.2f}")
    disk = psutil.disk_usage("/")
    if disk.percent > 85:
        problems.append(f"磁盘 / 使用率 {disk.percent}%")

    now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    if problems:
        msg = {"msgtype": "text",
               "text": {"content": f"[{HOST}] {now}\n" + "\n".join(problems)}}
        requests.post(WEBHOOK, json=msg, timeout=5)   # 输出: 异常项推送成功
        print("异常项已推送:", problems)
    else:
        print(f"[{now}] {HOST} 健康检查通过")   # 输出: [2026-07-31 09:00:00] web-01 健康检查通过

if __name__ == "__main__":
    main()

定时运行：cron 与 systemd timer

巡检脚本要定时跑，两个坑必须先处理：cron 的 PATH 不包含 /usr/local/bin，python3 必须写绝对路径；输出必须重定向，否则脚本里 print 的告警无人看到。

# crontab -e
SHELL=/bin/bash
PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
# 每 5 分钟健康检查，输出追加到日志
*/5 * * * * /usr/bin/python3 /opt/scripts/health_check.py >> /var/log/health_check.log 2>&1
# 每天 3 点跑日志分析
0 3 * * * /usr/bin/python3 /opt/scripts/log_analyzer.py >> /var/log/log_analyzer.log 2>&1

# 更推荐的 systemd timer 方式（输出进 journalctl，天然统一管理）
# /etc/systemd/system/health-check.service
[Unit]
Description=Health Check Script

[Service]
Type=oneshot
ExecStart=/usr/bin/python3 /opt/scripts/health_check.py

# /etc/systemd/system/health-check.timer
[Unit]
Description=Run health check every 5 minutes

[Timer]
OnCalendar=*:0/5
Persistent=true

[Install]
WantedBy=timers.target

# 启用并验证
sudo systemctl daemon-reload
sudo systemctl enable --now health-check.timer
# 输出: Created symlink /etc/systemd/system/timers.target.wants/health-check.timer
sudo systemctl list-timers health-check    # 输出: NEXT 09:05:00 5min left health-check.timer
journalctl -u health-check.service -S "10 min ago"   # 查看上次执行输出

常见错误

最佳实践

练习题

1. 日志清理器：用 subprocess 找到 /var/log 下 7 天前的 *.log 文件并压缩（gzip），保留原始文件列表到 JSON。
2. 端口扫描器：用 socket + argparse 批量扫描指定主机的 22/80/443/3306/6379 端口，输出开放列表。
3. 批量 SSH 磁盘检查：用 paramiko 对一批服务器执行 df -h，解析结果，打印出磁盘使用率超过 85% 的挂载点。
4. 进程树观察者：用 psutil 每 10 秒采集 CPU top-5 进程，当某个进程连续 3 次 CPU > 80% 时发出警告。
5. Poetry 项目打包：用 poetry new myops 创建项目，将练习 3 的脚本放入，加依赖 paramiko，编写对应单元测试。

学习检查点

学完本章后，请检验自己是否掌握以下内容：

本章总结

速查表

学习路径：2.1：Shell 脚本入门/2.8：Shell 脚本进阶 Shell 基础 → 本章 Python 运维脚本 → 4.1：Ansible 自动化 Ansible（更高层抽象）→ 4.9：CI/CD 持续部署 CI/CD 集成。

延伸阅读