运维实战脚本
前四篇把 Python 语法、内置模块、网络远程、系统进程管理都过了一遍，这篇把它们攒成四个完整的实战脚本：巡检报告、日志分析、批量分发、水位告警。每个脚本都能直接照抄运行，改改路径和主机列表就能上生产。这是系列的收官篇，附录里还有一份可复用的模板库。
环境说明
文中示例基于 Python 3.12+，运行环境为 macOS 或 Linux。涉及的第三方库一条命令装好：
python3 -m pip install psutil paramiko requests PyYAML
新版系统（如 Ubuntu 24.04、macOS 自带的 pip）报 externally-managed-environment 的话，先 python3 -m venv ~/.venv/ops && source ~/.venv/ops/bin/activate 再装。
实战一：单页巡检报告
场景：每天早上想一眼看清机器状态——主机名、CPU、内存、磁盘、最吃资源的进程，打印成一张整齐的文本报告，贴到群里或留档都方便。
#!/usr/bin/env python3
"""单页巡检报告：汇总主机名/CPU/内存/磁盘/Top5 进程"""
import platform
import shutil
from datetime import datetime
import psutil
def collect_basic():
"""基础信息：主机名、系统版本、运行时长"""
boot = datetime.fromtimestamp(psutil.boot_time())
uptime = datetime.now() - boot
return {
"hostname": platform.node(),
"os": f"{platform.system()} {platform.release()}",
"uptime": str(uptime).split(".")[0],
}
def collect_disks():
"""各挂载点磁盘使用率，跳过虚拟文件系统"""
rows = []
for part in psutil.disk_partitions():
if part.fstype in ("devfs", "tmpfs", "overlay", "autofs"):
continue
usage = shutil.disk_usage(part.mountpoint)
percent = usage.used / usage.total * 100
rows.append((part.mountpoint, usage.total, usage.used, percent))
return rows
def collect_top_processes(n=5):
"""按内存占用排序取 Top N 进程"""
procs = []
for p in psutil.process_iter(["pid", "name", "memory_percent"]):
try:
procs.append(p.info)
except (psutil.NoSuchProcess, psutil.AccessDenied):
continue
procs.sort(key=lambda x: x["memory_percent"] or 0, reverse=True)
return procs[:n]
def human(nbytes):
"""字节数转人类可读格式"""
for unit in ("B", "KB", "MB", "GB", "TB"):
if nbytes < 1024:
return f"{nbytes:.1f}{unit}"
nbytes /= 1024
return f"{nbytes:.1f}PB"
def main():
basic = collect_basic()
mem = psutil.virtual_memory()
disks = collect_disks()
top = collect_top_processes()
print("=" * 56)
print(f" 巡检报告 {basic['hostname']} {datetime.now():%Y-%m-%d %H:%M:%S}")
print("=" * 56)
print(f"系统 : {basic['os']}")
print(f"已运行 : {basic['uptime']}")
print(f"CPU 使用率: {psutil.cpu_percent(interval=1):.1f}%")
print(f"内存 : {mem.percent}% ({human(mem.used)} / {human(mem.total)})")
print("-" * 56)
print(f"{'挂载点':<20}{'总量':>10}{'已用':>10}{'使用率':>8}")
for mount, total, used, pct in disks:
print(f"{mount:<20}{human(total):>10}{human(used):>10}{pct:>7.1f}%")
print("-" * 56)
print("Top5 进程（按内存）:")
for p in top:
print(f" {p['pid']:>7} {p['memory_percent'] or 0:>5.1f}% {p['name']}")
if __name__ == "__main__":
main()
运行效果：
========================================================
巡检报告 web-01 2025-06-10 09:00:03
========================================================
系统 : Linux 5.15.0-91-generic
已运行 : 21 days, 3:14:55
CPU 使用率: 12.3%
内存 : 68.2% (5.4GB / 7.8GB)
--------------------------------------------------------
挂载点 总量 已用 使用率
/ 100.0GB 42.1GB 42.1%
/data 500.0GB 380.2GB 76.0%
--------------------------------------------------------
Top5 进程（按内存）:
1234 21.3% java
逐段要点：
- 采集和格式化分层。
collect_*函数只负责拿数据、返回干净的元组或字典，main只负责排版打印。哪天想把输出改成 JSON 或 HTML 报表，采集层一行都不用动——这是写运维脚本最值得养成的习惯。 psutil.process_iter(["pid", "name", ...])把要的字段一次性取齐，比每个进程单独调p.name()、p.memory_percent()快得多，进程多的时候差距明显。- 迭代进程的瞬间进程可能已经退出，
psutil.NoSuchProcess必须兜住；memory_percent偶尔返回None，用or 0兜底——一份报告别毁在一个短命进程上。 - 中文对齐别强求：中文字符占两个英文宽度，f-string 的对齐宽度算不准，所以示例里中文表头和数据行没严格对齐；强迫症就把表头换成英文，或输出 CSV 贴到 Excel 里看。
扩展思路：报告写文件留档（> 重定向加日期后缀）；超阈值的项目标红（终端 ANSI 颜色 \033[31m）；加 --json 参数输出机器可读格式，喂给后面的汇总脚本。
实战二：Nginx 日志分析
场景：排查问题时想快速知道 access log 里 5xx 有多少、请求量按分钟怎么分布、哪些 IP 打得最凶，不想每次都手敲一堆 awk。
#!/usr/bin/env python3
"""Nginx access log 分析：5xx 统计、按分钟请求量、Top10 客户端 IP"""
import re
import sys
from collections import Counter, defaultdict
LOG_PATTERN = re.compile(
r'(?P<ip>\S+) \S+ \S+ \[(?P<time>[^\]]+)\] '
r'"(?P<request>[^"]*)" (?P<status>\d{3}) (?P<size>\S+)'
)
def analyze(path):
total = 0
status_5xx = 0
per_minute = defaultdict(int)
ip_counter = Counter()
with open(path, encoding="utf-8", errors="replace") as f:
for line in f:
m = LOG_PATTERN.match(line)
if not m:
continue # 格式不符的行（扫描器的奇怪请求）直接跳过
total += 1
if m.group("status").startswith("5"):
status_5xx += 1
# time_local 形如 10/Oct/2025:13:55:36 +0800，切到分钟
minute = m.group("time")[:17]
per_minute[minute] += 1
ip_counter[m.group("ip")] += 1
return total, status_5xx, per_minute, ip_counter
def main():
path = sys.argv[1] if len(sys.argv) > 1 else "/var/log/nginx/access.log"
total, status_5xx, per_minute, ip_counter = analyze(path)
if total == 0:
print("没有匹配到任何日志行，检查文件路径和日志格式")
return
print(f"分析文件: {path}")
print(f"总请求数: {total}，5xx 数量: {status_5xx}"
f"（占比 {status_5xx / total * 100:.2f}%）")
print("\n按分钟请求量（最后 10 个分钟点）:")
for minute in sorted(per_minute)[-10:]:
count = per_minute[minute]
print(f" {minute} {count:>6} {'#' * min(count // 10, 40)}")
print("\nTop10 客户端 IP:")
for ip, count in ip_counter.most_common(10):
print(f" {ip:<16} {count:>8}")
if __name__ == "__main__":
main()
跑法 python3 log_analyze.py /var/log/nginx/access.log，输出示例：
总请求数: 152340，5xx 数量: 87（占比 0.06%）
按分钟请求量（最后 10 个分钟点）:
10/Oct/2025:13:50 1320 ####################
Top10 客户端 IP:
61.140.22.8 8123
逐段要点：
- 命名分组
(?P<ip>...)比m.group(1)这种数字下标可读得多，以后日志格式加了字段也好改。这个正则是按默认的 combined 格式写的，你的 nginx 自定义过log_format的话照着调分组顺序。 Counter天生为计数设计，most_common(10)一行出 Top10；defaultdict(int)省掉if key not in dict的初始化判断，按分钟聚合一行搞定。errors="replace"防止某行乱码让整个分析崩掉；逐行读文件，内存占用恒定，几个 G 的日志也能跑。- 新日志格式上手前先小范围验证：
head -100 access.log | python3 log_analyze.py /dev/stdin喂一小段，确认匹配率正常再跑全量，省得对几千万行白等。
扩展思路：加 argparse 支持 --top 20、--since "13:00" 过滤；按状态码分组统计（2xx/3xx/4xx/5xx 各多少）；UV 去重用 set 装 IP；结果输出 CSV 导入报表系统。
实战三：批量分发与执行
场景：十几台机器要下发同一个配置文件，然后执行一条命令确认生效，最后把每台的结果汇成一张表——Ansible 还没铺开的过渡期，或者就这一次性任务不想写 playbook。
#!/usr/bin/env python3
"""批量分发文件并远程执行命令，汇总各机结果"""
from concurrent.futures import ThreadPoolExecutor, as_completed
import paramiko
HOSTS = [
{"host": "192.168.1.11", "user": "deploy"},
{"host": "192.168.1.12", "user": "deploy"},
]
LOCAL_FILE = "./app.conf"
REMOTE_FILE = "/tmp/app.conf"
COMMAND = "md5sum /tmp/app.conf && uptime"
MAX_WORKERS = 10
def push_and_run(conf):
"""单机完整流程：连接 -> 传文件 -> 执行 -> 断开；失败抛异常由上层收集"""
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
client.connect(conf["host"], username=conf["user"], timeout=10)
with client.open_sftp() as sftp:
sftp.put(LOCAL_FILE, REMOTE_FILE)
_, stdout, stderr = client.exec_command(COMMAND, timeout=30)
rc = stdout.channel.recv_exit_status()
if rc != 0:
raise RuntimeError(f"命令返回码 {rc}: {stderr.read().decode().strip()}")
return stdout.read().decode().strip()
finally:
client.close()
def main():
results = []
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
futures = {pool.submit(push_and_run, conf): conf["host"] for conf in HOSTS}
for future in as_completed(futures):
host = futures[future]
try:
output = future.result()
results.append((host, "OK", output.replace("\n", "; ")))
except Exception as exc: # 单机失败不拖垮整体，记录下来继续
results.append((host, "FAIL", str(exc)))
print(f"\n{'主机':<16}{'结果':<6}输出")
print("-" * 60)
for host, status, output in sorted(results):
print(f"{host:<16}{status:<6}{output}")
failed = sum(1 for r in results if r[1] == "FAIL")
print(f"\n共 {len(results)} 台，成功 {len(results) - failed}，失败 {failed}")
if __name__ == "__main__":
main()
逐段要点：
- 并发上限要收敛。SSH 操作是 IO 密集，线程池正合适，但
max_workers别开成主机数——100 台机器开 100 个线程，对端 sshd 的MaxStartups默认只放行 10 个并发未认证连接，超了直接拒绝。10~20 是稳妥区间，机器多就分批跑。 - 单机失败不拖垮整体。
future.result()会把远端函数里抛的异常原样抛回来，必须逐台try/except收进结果表，否则一台机器 SSH 不通，整个脚本带着其他 15 台的结果一起崩掉。 futures = {pool.submit(...): host}这个「future 反查主机名」字典配合as_completed，谁先完成先处理谁，不按提交顺序干等最慢的那台。finally: client.close()保证异常路径下连接也关掉，不然线程池跑完留下一堆挂着的 SSH 会话。
host key 策略
示例里 AutoAddPolicy 自动信任未知 host key，内网首次连接方便，但有中间人风险。严格环境换成 RejectPolicy，先手动连一遍把 key 收进 known_hosts。
扩展思路：命令超时后杀远端进程；加 --limit 只跑部分机器；结果写 JSON 留档。如果这个脚本开始长出「配置文件改了才重启服务」「按角色分组跑不同任务」这类需求，就是该上 Ansible 的信号了。连接层疑难杂症（超时、拒连）要抓包定位时，可以参考 tcpdump 实战笔记。
实战四：磁盘水位监控告警
场景：磁盘写满是最常见的故障，写个脚本每 5 分钟查一次水位，超阈值就推消息到群里，但同一个问题不能每 5 分钟轰炸一次。
#!/usr/bin/env python3
"""磁盘水位监控：超阈值 POST 到群机器人 webhook，带去抖"""
import json
import os
import shutil
import time
from pathlib import Path
import requests
WEBHOOK = os.environ.get("ROBOT_WEBHOOK", "")
THRESHOLD = 85 # 使用率百分比阈值
DEBOUNCE_SECONDS = 3600 # 同一挂载点告警最小间隔
STATE_FILE = Path("/tmp/disk-alert-state.json")
WATCH_PATHS = ["/", "/data"]
def load_state():
if STATE_FILE.exists():
return json.loads(STATE_FILE.read_text())
return {}
def save_state(state):
STATE_FILE.write_text(json.dumps(state))
def send_alert(mount, percent):
"""钉钉/企业微信 robot 通用的 text 消息格式"""
payload = {
"msgtype": "text",
"text": {"content": f"[磁盘告警] {mount} 使用率 {percent:.1f}%，超过阈值 {THRESHOLD}%"},
}
resp = requests.post(WEBHOOK, json=payload, timeout=10)
resp.raise_for_status()
def main():
if not WEBHOOK:
raise SystemExit("先 export ROBOT_WEBHOOK=<机器人 webhook 地址> 再运行")
state = load_state()
now = time.time()
for mount in WATCH_PATHS:
usage = shutil.disk_usage(mount)
percent = usage.used / usage.total * 100
if percent < THRESHOLD:
state.pop(mount, None) # 水位回落，清除记录，下次超阈值可重新告警
continue
last_sent = state.get(mount, 0)
if now - last_sent < DEBOUNCE_SECONDS:
print(f"{mount} 超阈值但在去抖期内，跳过")
continue
send_alert(mount, percent)
state[mount] = now
print(f"{mount} 使用率 {percent:.1f}%，已告警")
save_state(state)
if __name__ == "__main__":
main()
挂到 cron 里每 5 分钟跑一次（crontab -e）：
*/5 * * * * ROBOT_WEBHOOK="https://oapi.dingtalk.com/robot/send?access_token=xxx" /usr/bin/python3 /opt/scripts/disk_alert.py >> /var/log/disk-alert.log 2>&1
逐段要点：
shutil.disk_usage是标准库，零依赖拿到总量/已用/剩余；想顺带枚举挂载点就换psutil.disk_partitions()+psutil.disk_usage()。- 告警去抖是刚需。没有状态记录的话，磁盘 90% 挂着没人处理，群里就每 5 分钟收一条一模一样的告警，大家很快会把机器人屏蔽——等于没有告警。这里的办法：
/tmp下存一个 JSON，记录每个挂载点上次告警时间，一小时内不重复发；水位回落后清掉记录，恢复告警能力。 - webhook 的
msgtype: text载荷是钉钉和企业微信群机器人通用的格式，换平台只改 URL 不改代码；token 走环境变量，别硬编码进脚本——脚本进 git 是迟早的事，密钥泄露也是。 - 状态文件放
/tmp重启会清，清了顶多告警重发一次，可接受；介意的话换/var/lib/disk-alert/state.json，注意给脚本写权限。
扩展思路：加恢复通知（水位回落到阈值以下发一条「已恢复」）；阈值分级（85% 警告、95% 严重并 @ 值班人）；多指标合一（内存、load、inode 都用同一个去抖框架）。
工程化建议
脚本写多了，慢慢会攒出一套自己的规矩。几条见效最快的：
配置外置到 yaml。 主机列表、阈值、webhook 这类「会变的」东西别写在代码里，代码和配置分开改，读取用 yaml.safe_load（完整片段见附录《运维脚本模板库》）：
# config.yaml
threshold: 85
watch_paths:
- /
- /data
webhook: https://oapi.dingtalk.com/robot/send?access_token=xxx
logging 替代 print。 脚本进 cron 之后，print 的去向就成了玄学。用 logging 配「控制台 + 文件」双输出，级别分清 DEBUG/INFO/WARNING，附录《运维脚本模板库》里有现成的配置片段。
脚本打包分发。 单文件脚本想丢到没装依赖的机器上跑，两条路：python -m zipapp 把脚本和依赖打成一个 .pyz，目标机有 python3 就能跑；或者 pyinstaller -F tool.py 编成独立二进制，目标机连 Python 都不用装。
什么时候脚本该升级成 Ansible role。 一个 Python 脚本管几十台机器没问题，但出现这些信号就该换了：脚本开始维护「状态」而不是「执行动作」（比如要保证配置文件和某份模板一致，而不是把文件推过去就完事）；任务要按角色分组编排；团队里别人也要看懂和改动。Ansible 的幂等、handler、inventory 正好解决这些问题，系列文章见 Ansible 核心概念与架构。
小结
回看整个系列五篇的脉络：第 1 篇基础速通把语法和环境垫平，第 2 篇内置模块给了文件、时间、正则这些随身工具，第 3 篇网络与远程打通了 SSH 和 HTTP 两条对外的路，第 4 篇系统与进程把单机管理能力补全，本篇把四者组装成四个能直接用的实战脚本。附录的模板库是平时攒下的骨架片段，写新脚本时从那里起手比从零开始快得多。Python 不是运维自动化的终点——机器多了、状态复杂了，自然会走到 Ansible 那条路上去，但这些脚本能力在任何一条路上都用得上。