Python 运维自动化入门：大专生如何用脚本拉开与同龄人的差距

“运维要学 Python 吗？”

要，但不是你想象的那种 Python。你不需要学机器学习，不需要学爬虫，不需要学 Django 建网站——你需要的是能解决运维实际问题的 Python，这部分比你想象的要简单很多。

为什么运维要学 Python

Shell 能做的事：
  简单的文件操作、进程管理、文本处理
  适合：快速一次性的任务

Shell 做起来很麻烦的事：
  HTTP 请求（调用 API、发钉钉通知）
  Excel/JSON 文件处理
  SSH 批量登录多台服务器
  复杂的逻辑判断和数据结构
  定时任务的管理和状态维护

Python 做起来很轻松：
  以上所有，加上更好的错误处理和代码复用

实际场景对比：
  用 Shell 发钉钉告警：curl + JSON 拼接，很容易出错
  用 Python 发钉钉告警：5 行代码，清晰易改

  用 Shell 批量巡检 100 台服务器：嵌套循环 + expect，脆弱难维护
  用 Python 批量巡检：paramiko 库，20 行代码，能处理超时和异常

运维 Python 只需要这几个库

不要被 Python 生态的庞大吓到。运维场景 80% 的需求，就这几个库：

# 内置库（不用安装）
import os           # 文件/目录/环境变量操作
import subprocess   # 执行系统命令，获取输出
import json         # JSON 解析
import re           # 正则表达式
import datetime     # 时间处理
import logging      # 日志记录
import argparse     # 命令行参数解析

# 需要安装的库（pip install）
import paramiko     # SSH 连接，批量操作服务器
import requests     # HTTP 请求（调用各种 API）
import schedule     # 定时任务
import yaml         # 解析 YAML 配置文件

5 个真实可用的运维脚本

脚本1：批量服务器巡检（paramiko）

#!/usr/bin/env python3
"""
批量巡检脚本：登录多台服务器，收集 CPU/内存/磁盘信息
用法：python3 check-servers.py
"""

import paramiko
import json
import datetime
from concurrent.futures import ThreadPoolExecutor

# 服务器列表（实际使用时从配置文件读取）
SERVERS = [
    {"host": "192.168.1.101", "port": 22, "user": "admin", "key": "/root/.ssh/id_rsa"},
    {"host": "192.168.1.102", "port": 22, "user": "admin", "key": "/root/.ssh/id_rsa"},
    {"host": "192.168.1.103", "port": 22, "user": "admin", "key": "/root/.ssh/id_rsa"},
]

# 要执行的巡检命令
COMMANDS = {
    "cpu_usage": "top -bn1 | grep 'Cpu(s)' | awk '{print $2}' | cut -d'%' -f1",
    "memory_usage": "free | grep Mem | awk '{printf \"%.1f\", $3/$2*100}'",
    "disk_usage": "df -h / | awk 'NR==2{print $5}'",
    "load_avg": "cat /proc/loadavg | awk '{print $1,$2,$3}'",
    "container_count": "docker ps -q 2>/dev/null | wc -l || echo 0",
}

def check_server(server):
    """巡检单台服务器，返回结果字典"""
    result = {"host": server["host"], "status": "ok", "data": {}, "error": None}
    
    try:
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(
            hostname=server["host"],
            port=server["port"],
            username=server["user"],
            key_filename=server["key"],
            timeout=10
        )
        
        for metric, cmd in COMMANDS.items():
            stdin, stdout, stderr = ssh.exec_command(cmd, timeout=5)
            output = stdout.read().decode().strip()
            result["data"][metric] = output
        
        ssh.close()
        
    except Exception as e:
        result["status"] = "error"
        result["error"] = str(e)
    
    return result

def main():
    print(f"=== 服务器巡检报告 {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ===\n")
    
    # 并发巡检（最多 10 个并发）
    with ThreadPoolExecutor(max_workers=10) as executor:
        results = list(executor.map(check_server, SERVERS))
    
    # 打印结果
    alerts = []
    for r in results:
        if r["status"] == "error":
            print(f"❌ {r['host']}: {r['error']}")
            alerts.append(f"{r['host']} 连接失败")
            continue
        
        d = r["data"]
        cpu = float(d.get("cpu_usage", 0))
        mem = float(d.get("memory_usage", 0))
        disk = d.get("disk_usage", "0%").replace("%", "")
        
        status = "✅"
        issues = []
        
        if cpu > 80:
            issues.append(f"CPU {cpu:.1f}%")
            status = "⚠️ "
        if mem > 85:
            issues.append(f"内存 {mem:.1f}%")
            status = "⚠️ "
        if int(disk) > 85:
            issues.append(f"磁盘 {disk}%")
            status = "⚠️ "
        
        issue_str = "  [" + ", ".join(issues) + "]" if issues else ""
        print(f"{status} {r['host']}: CPU {cpu:.1f}%  内存 {mem:.1f}%  磁盘 {disk}%  "
              f"容器 {d.get('container_count', 0)} 个{issue_str}")
        
        if issues:
            alerts.append(f"{r['host']}: {', '.join(issues)}")
    
    # 汇总
    print(f"\n共巡检 {len(results)} 台，{len(alerts)} 台异常")
    
    return alerts

if __name__ == "__main__":
    main()

脚本2：发送钉钉告警（requests）

#!/usr/bin/env python3
"""
钉钉 Webhook 告警发送工具
用法：python3 dingtalk-alert.py "告警标题" "告警内容"
"""

import requests
import json
import hashlib
import hmac
import base64
import time
import sys

WEBHOOK_URL = "https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN"
SECRET = "YOUR_SECRET"  # 如果设置了加签，填入 Secret

def get_sign():
    """计算钉钉加签"""
    timestamp = str(round(time.time() * 1000))
    string_to_sign = f"{timestamp}\n{SECRET}"
    hmac_code = hmac.new(
        SECRET.encode('utf-8'),
        string_to_sign.encode('utf-8'),
        digestmod=hashlib.sha256
    ).digest()
    sign = base64.b64encode(hmac_code).decode('utf-8')
    return timestamp, sign

def send_alert(title, content, at_all=False):
    """发送 Markdown 格式的告警消息"""
    
    timestamp, sign = get_sign()
    url = f"{WEBHOOK_URL}&timestamp={timestamp}&sign={sign}"
    
    data = {
        "msgtype": "markdown",
        "markdown": {
            "title": title,
            "text": f"## {title}\n\n{content}\n\n> 发送时间：{time.strftime('%Y-%m-%d %H:%M:%S')}"
        },
        "at": {
            "isAtAll": at_all
        }
    }
    
    response = requests.post(
        url,
        data=json.dumps(data),
        headers={"Content-Type": "application/json"},
        timeout=10
    )
    
    result = response.json()
    if result.get("errcode") == 0:
        print("✅ 钉钉告警发送成功")
        return True
    else:
        print(f"❌ 发送失败：{result}")
        return False

if __name__ == "__main__":
    if len(sys.argv) < 3:
        print("用法: python3 dingtalk-alert.py <标题> <内容>")
        sys.exit(1)
    
    send_alert(sys.argv[1], sys.argv[2])

脚本3：日志分析——统计错误频率

#!/usr/bin/env python3
"""
Nginx 访问日志分析：统计状态码分布、慢请求、TOP URL
用法：python3 log-analyze.py /var/log/nginx/access.log
"""

import re
import sys
from collections import Counter, defaultdict
from datetime import datetime

# Nginx 日志格式（默认格式）
LOG_PATTERN = re.compile(
    r'(?P<ip>[\d.]+) .+ \[(?P<time>[^\]]+)\] '
    r'"(?P<method>\w+) (?P<url>\S+) [^"]+" '
    r'(?P<status>\d+) \d+ "[^"]*" "[^"]*" (?P<duration>[\d.]+)'
)

def analyze_log(filepath, lines=50000):
    """分析日志文件，返回统计结果"""
    
    status_counter = Counter()
    url_counter = Counter()
    slow_requests = []  # 响应 > 1s 的请求
    error_ips = Counter()
    
    with open(filepath, 'r', errors='ignore') as f:
        for i, line in enumerate(f):
            if i >= lines:
                break
            
            m = LOG_PATTERN.match(line)
            if not m:
                continue
            
            status = m.group('status')
            url = m.group('url').split('?')[0]  # 去掉查询参数
            duration = float(m.group('duration') or 0)
            ip = m.group('ip')
            
            status_counter[status] += 1
            url_counter[url] += 1
            
            # 记录慢请求
            if duration > 1.0:
                slow_requests.append({
                    "url": m.group('url'),
                    "duration": duration,
                    "time": m.group('time')
                })
            
            # 记录 5xx 错误的 IP
            if status.startswith('5'):
                error_ips[ip] += 1
    
    return {
        "status": status_counter,
        "top_urls": url_counter.most_common(10),
        "slow_requests": sorted(slow_requests, key=lambda x: x['duration'], reverse=True)[:10],
        "error_ips": error_ips.most_common(5),
    }

def main():
    filepath = sys.argv[1] if len(sys.argv) > 1 else "/var/log/nginx/access.log"
    
    print(f"分析日志：{filepath}\n")
    result = analyze_log(filepath)
    
    # 状态码分布
    total = sum(result["status"].values())
    print("=== 状态码分布 ===")
    for code, count in sorted(result["status"].items()):
        pct = count / total * 100
        bar = "█" * int(pct / 2)
        print(f"  {code}: {count:6d} ({pct:4.1f}%) {bar}")
    
    # 错误率
    error_5xx = sum(v for k, v in result["status"].items() if k.startswith('5'))
    print(f"\n5xx 错误率：{error_5xx/total*100:.2f}%  （{error_5xx}/{total}）")
    
    # TOP 10 URL
    print("\n=== TOP 10 访问 URL ===")
    for url, count in result["top_urls"]:
        print(f"  {count:6d}  {url}")
    
    # 慢请求
    if result["slow_requests"]:
        print(f"\n=== 慢请求 TOP 10（>1s）===")
        for req in result["slow_requests"]:
            print(f"  {req['duration']:.2f}s  {req['url'][:80]}")

if __name__ == "__main__":
    main()

脚本4：Docker 容器健康巡检

#!/usr/bin/env python3
"""
Docker 容器健康状态检查，异常时发送钉钉告警
"""

import subprocess
import json
import sys

ALERT_WEBHOOK = "https://oapi.dingtalk.com/robot/send?access_token=xxx"

def get_containers():
    """获取所有容器状态"""
    result = subprocess.run(
        ["docker", "ps", "-a", "--format", "{{json .}}"],
        capture_output=True, text=True
    )
    
    containers = []
    for line in result.stdout.strip().split('\n'):
        if line:
            containers.append(json.loads(line))
    return containers

def check_container_health(name):
    """检查特定容器的健康状态"""
    result = subprocess.run(
        ["docker", "inspect", name, "--format", 
         "{{.State.Status}} {{.State.Health.Status}} {{.RestartCount}}"],
        capture_output=True, text=True
    )
    
    if result.returncode != 0:
        return None
    
    parts = result.stdout.strip().split()
    return {
        "status": parts[0] if parts else "unknown",
        "health": parts[1] if len(parts) > 1 else "none",
        "restarts": int(parts[2]) if len(parts) > 2 else 0,
    }

def main():
    # 需要监控的关键容器（名称）
    CRITICAL_CONTAINERS = ["nginx", "api", "mysql", "redis"]
    
    containers = get_containers()
    container_map = {c["Names"]: c for c in containers}
    
    alerts = []
    
    for name in CRITICAL_CONTAINERS:
        info = check_container_health(name)
        
        if info is None:
            alerts.append(f"❌ {name}：容器不存在")
            continue
        
        if info["status"] != "running":
            alerts.append(f"❌ {name}：状态为 {info['status']}")
        elif info["restarts"] > 5:
            alerts.append(f"⚠️  {name}：重启次数 {info['restarts']} 次（可能有问题）")
        elif info["health"] == "unhealthy":
            alerts.append(f"⚠️  {name}：健康检查失败")
        else:
            print(f"✅ {name}：正常运行（重启 {info['restarts']} 次）")
    
    if alerts:
        print("\n发现异常：")
        for a in alerts:
            print(f"  {a}")
        
        # 这里调用前面的钉钉告警脚本或内联发送
        # send_alert("Docker 容器异常告警", "\n".join(alerts), at_all=True)
    
    return len(alerts)

if __name__ == "__main__":
    sys.exit(main())

脚本5：配置管理——批量修改多台服务器的配置

#!/usr/bin/env python3
"""
批量推送配置文件到多台服务器
用法：python3 push-config.py --config nginx.conf --dest /etc/nginx/conf.d/app.conf
"""

import paramiko
import argparse
import sys
from pathlib import Path

SERVERS = [
    "192.168.1.101",
    "192.168.1.102",
    "192.168.1.103",
]

SSH_USER = "admin"
SSH_KEY = "/root/.ssh/id_rsa"

def push_config(server, local_file, remote_path, reload_cmd=None):
    """推送配置到单台服务器，可选重载服务"""
    try:
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(server, username=SSH_USER, key_filename=SSH_KEY, timeout=10)
        
        # 用 SFTP 传输文件
        sftp = ssh.open_sftp()
        sftp.put(local_file, remote_path)
        sftp.close()
        
        # 可选：重载服务
        if reload_cmd:
            stdin, stdout, stderr = ssh.exec_command(reload_cmd)
            exit_code = stdout.channel.recv_exit_status()
            if exit_code != 0:
                error = stderr.read().decode()
                ssh.close()
                return False, f"重载命令失败: {error}"
        
        ssh.close()
        return True, "成功"
        
    except Exception as e:
        return False, str(e)

def main():
    parser = argparse.ArgumentParser(description="批量推送配置文件")
    parser.add_argument("--config", required=True, help="本地配置文件路径")
    parser.add_argument("--dest", required=True, help="远程目标路径")
    parser.add_argument("--reload", help="推送后执行的重载命令", default=None)
    args = parser.parse_args()
    
    if not Path(args.config).exists():
        print(f"❌ 文件不存在: {args.config}")
        sys.exit(1)
    
    print(f"推送 {args.config} → {args.dest}")
    print(f"目标服务器: {len(SERVERS)} 台\n")
    
    success = 0
    for server in SERVERS:
        ok, msg = push_config(server, args.config, args.dest, args.reload)
        if ok:
            print(f"✅ {server}: {msg}")
            success += 1
        else:
            print(f"❌ {server}: {msg}")
    
    print(f"\n结果: {success}/{len(SERVERS)} 台成功")

if __name__ == "__main__":
    main()

Python 和 Shell 的选择策略

用 Shell（bash）：
  快速的一次性操作
  简单的文件处理和进程管理
  已有大量 Shell 管道的场景
  执行系统命令的小胶水脚本（< 50 行）

用 Python：
  需要 HTTP 请求（requests 比 curl 脚本好维护）
  批量 SSH 操作多台服务器（paramiko）
  需要处理 JSON/YAML/Excel 数据
  有复杂逻辑判断、需要异常处理
  脚本需要长期维护和他人协作
  超过 50 行、有多个函数的脚本

小结

运维用的 Python 不需要学算法、不需要学框架、不需要学 OOP 理论——你只需要能写出“能跑、能排查、同事能看懂”的脚本。

从上面 5 个脚本开始，改成你自己的服务器地址、你自己的业务场景，跑通了就是真实的运维自动化经验。这比背 100 道 Python 题目有价值得多。