page contents

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

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

attachments-2026-07-nscKh8Ay6a4473e8ad1b4.png如果你是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实现自动化——你会发现,省下来的时间,可以用来学习更多技术,而不是加班。

更多相关技术内容咨询欢迎前往并持续关注好学星城论坛了解详情。

想高效系统的学习Python编程语言,推荐大家关注一个微信公众号:Python编程学习圈。每天分享行业资讯、技术干货供大家阅读,关注即可免费领取整套Python入门到进阶的学习资料以及教程,感兴趣的小伙伴赶紧行动起来吧。

attachments-2022-05-rLS4AIF8628ee5f3b7e12.jpg

 

你可能感兴趣的文章

相关问题

0 条评论

请先 登录 后评论
Pack
Pack

2179 篇文章

作家榜 »

  1. 轩辕小不懂 2403 文章
  2. 小柒 2228 文章
  3. Pack 2179 文章
  4. Nen 576 文章
  5. 王昭君 209 文章
  6. 文双 71 文章
  7. 小威 64 文章
  8. Cara 36 文章