当前位置:首页>python>用Python脚本把日常运维“自动化”起来,你也能成为效率达人

用Python脚本把日常运维“自动化”起来,你也能成为效率达人

  • 2026-10-11 05:38:14
用Python脚本把日常运维“自动化”起来,你也能成为效率达人

🐍 用Python脚本把日常运维“自动化”起来,你也能成为效率达人

如果你是IT运维或开发人员,一定对每天重复的“手工操作”深有体会:登录服务器查日志、批量改配置、定时备份、监控报警处理……这些工作看似简单,却极其消耗时间与精力。真正的高手,不是手速快,而是会用脚本替自己干活。

今天,我们就来聊聊如何用 Python 和 Shell 脚本,把日常运维中那些“高频、低技术含量”的操作自动化。以下内容均来自实战经验,代码可直接复用。


一、为什么选择Python做运维脚本?

很多老运维习惯用纯Shell,但面对复杂的文本解析、API调用、多线程任务时,Shell的局限性就暴露了。Python的优势在于:

• 易读易写:语法接近自然语言,团队协作成本低

• 生态强大:`os`, `subprocess`, `paramiko`, `requests`, `crontab` 等模块直接解决运维痛点

• 跨平台:Windows/Linux/macOS 一套代码通吃

但注意: 纯Shell适合“快速执行一条命令”,Python适合“多步骤、带逻辑判断、需异常处理”的场景。两者结合才是王道。


二、实战场景一:批量检查服务器磁盘使用率(SSH + 告警)

日常运维中,最怕服务器磁盘写满导致服务挂掉。手动登录每台机器执行 df -h 显然不现实。下面用 paramiko 实现批量SSH执行命令并解析结果。

import paramiko

import re

from datetime import datetime

# 服务器列表(生产环境建议从CMDB或配置文件读取)

servers = [

  {"host": "192.168.1.10", "user": "root", "pwd": "your_pass"},

  {"host": "192.168.1.11", "user": "root", "pwd": "your_pass"},

]

# 告警阈值(百分比)

ALARM_THRESHOLD = 85

def check_disk(host, user, pwd):

  try:

    ssh = paramiko.SSHClient()

    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    ssh.connect(host, username=user, password=pwd, timeout=10)

    stdin, stdout, stderr = ssh.exec_command("df -h | grep -v Filesystem | awk '{print $5, $6}'")

    output = stdout.read().decode().strip().split('\n')

    ssh.close()

    return output

  except Exception as e:

    return [f"ERROR: {str(e)}"]

def parse_and_alarm(lines):

  alarms = []

  for line in lines:

    if "ERROR" in line:

      alarms.append(line)

      continue

    # 匹配 "85% /data" 格式

    match = re.match(r'(\d+)%\s+(.+)', line)

    if match:

      usage = int(match.group(1))

      mount = match.group(2)

      if usage >= ALARM_THRESHOLD:

        alarms.append(f"[{mount}] 使用率 {usage}% 超过阈值!")

  return alarms

# 主流程

if __name__ == "__main__":

  print(f"=== 磁盘检查开始: {datetime.now()} ===")

  for srv in servers:

    host = srv["host"]

    print(f"正在检查 {host}...")

    result = check_disk(host, srv["user"], srv["pwd"])

    alarms = parse_and_alarm(result)

    if alarms:

      for alarm in alarms:

        print(f"  ⚠️ {host}: {alarm}")

    else:

      print(f"  ✅ {host}: 正常")

  print(f"=== 检查结束 ===")

要点说明:

• 使用 `paramiko` 实现SSH连接,注意设置超时避免卡死

• 正则提取百分比和挂载点,**阈值可配置**

• 实际生产可对接钉钉/企业微信机器人发送告警


三、实战场景二:定时清理过期日志(Shell + Python混合)

很多应用日志按天滚动,但不会自动删除。我们写一个 Python脚本 + crontab 的组合方案。

1. 核心Python脚本 `log_cleaner.py`

#!/usr/bin/env python3

import os

import time

import glob

# 配置区域

LOG_DIRS = [

  "/var/log/myapp",

  "/data/logs/nginx"

]

RETENTION_DAYS = 30  # 保留最近30天

DRY_RUN = False      # 设为True则只打印不删除

def clean_logs(base_dir, days):

  cutoff = time.time() - days * 86400

  # 匹配所有 .log 和 .log.gz 文件

  pattern = os.path.join(base_dir, "*.log*")

  files = glob.glob(pattern)

  deleted_count = 0

  for fpath in files:

    mtime = os.path.getmtime(fpath)

    if mtime < cutoff:

      if DRY_RUN:

        print(f"[模拟] 将删除: {fpath}")

      else:

        os.remove(fpath)

        print(f"[删除] {fpath}")

      deleted_count += 1

  return deleted_count

if __name__ == "__main__":

  total = 0

  for d in LOG_DIRS:

    if os.path.isdir(d):

      cnt = clean_logs(d, RETENTION_DAYS)

      total += cnt

      print(f"目录 {d}: 清理 {cnt} 个文件")

    else:

      print(f"警告: 目录 {d} 不存在")

  print(f"总计清理: {total} 个日志文件")

2. 配置crontab定时执行

# 每天凌晨2点执行

0 2 * * * /usr/bin/python3 /opt/scripts/log_cleaner.py >> /var/log/cleaner.log 2>&1

关键点:

• 使用 `os.path.getmtime` 获取文件修改时间,比文件名解析更可靠

• 设置 `DRY_RUN` 模式,初次运行先模拟验证

• 日志重定向到文件,方便排查问题


四、进阶技巧:用Python替代复杂Shell管道

有些运维任务用Shell管道也能做,但维护困难。Python的 subprocess 模块可以优雅地执行系统命令并捕获输出。

例如,需要统计Nginx日志中状态码为500的IP列表:

import subprocess

from collections import Counter

# 执行命令并实时读取输出

cmd = "grep ' 500 ' /var/log/nginx/access.log | awk '{print $1}'"

proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

ip_counter = Counter()

for line in proc.stdout:

  ip = line.decode().strip()

  if ip:

    ip_counter[ip] += 1

# 输出Top 5

print("500错误最多的IP:")

for ip, count in ip_counter.most_common(5):

  print(f"  {ip}: {count}次")

优势: 不用写复杂的awk/sed正则,直接用Python的Counter排序,代码更易读。


五、避坑指南:编写运维脚本的3个原则

1.幂等性:脚本无论执行多少次,结果一致。例如清理日志时先判断文件是否存在。

2.可观测性:每一步都打印日志,**建议使用 `logging` 模块**,而非简单的print。

3.异常处理:网络超时、文件不存在、权限不足都要捕获,避免脚本崩溃。

import logging

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

logging.info("开始执行备份任务...")

try:

  # 业务逻辑

  pass

except Exception as e:

  logging.error(f"任务失败: {e}", exc_info=True)


总结

从批量检查磁盘到定时清理日志,Python脚本能帮你把运维工作从“重复劳动”变成“一次编写、永久生效”。 建议从今天开始,把每周至少做3次的手动操作列出来,挑一个最简单的用Python实现自动化——你会发现,省下来的时间,可以用来学习更多技术,而不是加班。

关注我,为您分享更多的IT知识,*如果你有想自动化的运维场景,欢迎留言讨论。*

最新文章

随机文章