运维自动化脚本分享：提升效率的Shell与Python实战

作者：ai | 发布时间：2026-05-08 13:00

运维自动化脚本分享：提升效率的Shell与Python实战

在日常的运维工作中，面对成百上千的服务器、繁杂的日常巡检以及突发的故障排查，纯手工操作不仅效率低下，而且极易因人为疏忽引发线上事故。自动化是运维转型的核心，而编写可复用的脚本则是实现自动化的第一步。

本文将分享四个在日常运维中极为实用的自动化脚本，涵盖日志清理、系统巡检、批量命令执行和数据库备份，帮助大家从繁琐的重复劳动中解放出来。

一、 Shell脚本实战：系统级运维利器

Shell脚本贴近系统底层，执行效率高，是处理文件、文本和系统命令的最佳选择。

1. 自动化日志归档与清理脚本

随着业务运行，Nginx、App等日志文件极易撑爆磁盘。此脚本可自动查找指定目录下超过指定天数的日志，进行Gzip压缩归档，并删除过期的归档文件。


#!/bin/bash
# 描述：日志自动归档与清理脚本
# 作者：运维自动化系列

LOG_DIR="/var/log/app"
ARCHIVE_DIR="/var/log/app/archive"
DAYS_TO_COMPRESS=3      # 超过3天的日志进行压缩
DAYS_TO_DELETE=30       # 超过30天的归档日志进行删除

# 确保归档目录存在
mkdir -p $ARCHIVE_DIR

echo "===== 开始日志归档清理: $(date) ====="

# 1. 压缩老旧日志（排除 .gz 文件和目录）
find $LOG_DIR -maxdepth 1 -type f ! -name "*.gz" -mtime +$DAYS_TO_COMPRESS | while read log_file; do
    file_name=$(basename "$log_file")
    echo "正在压缩: $log_file"
    gzip -c "$log_file" > "$ARCHIVE_DIR/${file_name}.gz"
    if [ $? -eq 0 ]; then
        > "$log_file"  # 压缩成功后清空原日志（避免直接删除导致句柄丢失）
        echo "已清空原日志: $log_file"
    fi
done

# 2. 删除过期的归档日志
find $ARCHIVE_DIR -type f -name "*.gz" -mtime +$DAYS_TO_DELETE | while read old_archive; do
    echo "正在删除过期归档: $old_archive"
    rm -f "$old_archive"
done

echo "===== 清理完成: $(date) ====="

实践要点：清理日志时直接rm可能导致进程依然向已删除的文件写数据（句柄未释放），造成隐性磁盘空间占用。这里采用> "$log_file"清空文件内容的方式，更为安全。

2. 系统健康巡检一键输出脚本

排查故障时，往往需要快速了解系统当前状态。此脚本一键抓取CPU、内存、磁盘及关键进程信息。


#!/bin/bash
# 描述：系统健康状态快速巡检脚本

GREEN='\033[0;32m'
RED='\033[0;31m'
NC='\033[0m' # No Color

echo -e "${GREEN}===== 系统健康巡检报告 =====${NC}"
echo "主机名: $(hostname)"
echo "系统版本: $(cat /etc/redhat-release 2>/dev/null || cat /etc/os-release | grep PRETTY_NAME)"
echo "内核版本: $(uname -r)"
echo "运行时间/负载: $(uptime | awk -F'load average:' '{print $2}')"
echo ""

echo -e "${GREEN}[内存使用情况]${NC}"
free -h | awk 'NR==1{printf "%-10s %-10s %-10s %-10s\n",$1,$2,$3,$4} NR==2{printf "%-10s %-10s %-10s %-10s\n",$1,$2,$3,$4}'
MEM_USAGE=$(free | awk '/Mem/{printf("%.1f"), $3/$2*100}')
if (( $(echo "$MEM_USAGE > 85" | bc -l) )); then
    echo -e "${RED}警告: 内存使用率已达 ${MEM_USAGE}%！${NC}"
fi
echo ""

echo -e "${GREEN}[磁盘使用情况 (使用率>80%标红)]${NC}"
df -h | awk 'NR==1 || /^\/dev/' | while read line; do
    usage=$(echo $line | awk '{print $5}' | tr -d '%')
    if [ "$usage" -gt 80 ] 2>/dev/null; then
        echo -e "${RED}$line${NC}"
    else
        echo "$line"
    fi
done
echo ""

echo -e "${GREEN}[僵尸进程检查]${NC}"
ZOMBIE=$(ps aux | awk '{if($8=="Z") print}' | wc -l)
if [ "$ZOMBIE" -gt 0 ]; then
    echo -e "${RED}发现 $ZOMBIE 个僵尸进程！${NC}"
    ps aux | awk '{if($8=="Z") print}'
else
    echo "未发现僵尸进程。"
fi

二、 Python脚本实战：复杂逻辑与跨平台交互

当涉及复杂的数据结构处理、网络请求或跨平台操作时，Python的生态和可读性远胜于Shell。

1. 基于Paramiko的批量主机命令执行脚本

在没有Ansible或SaltStack的轻量级环境中，运维常需在多台机器上执行同一命令。此脚本利用paramiko库实现并发执行。


#!/usr/bin/env python3
# 描述：批量主机命令并发执行脚本
# 依赖：pip install paramiko

import paramiko
import concurrent.futures

# 服务器列表 (实际应用中可从CMDB或文件读取)
SERVERS = [
    {"host": "192.168.1.10", "username": "root", "password": "your_password"},
    {"host": "192.168.1.11", "username": "root", "password": "your_password"},
]
COMMAND = "uptime"  # 需要执行的命令

def execute_command(server_info, command):
    host = server_info['host']
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        ssh.connect(hostname=host, 
                    username=server_info['username'], 
                    password=server_info['password'], 
                    timeout=5)
        stdin, stdout, stderr = ssh.exec_command(command)
        result = stdout.read().decode('utf-8').strip()
        error = stderr.read().decode('utf-8').strip()
        if error:
            return f"[{host}] 执行失败: {error}"
        return f"[{host}] 执行成功:\n{result}"
    except Exception as e:
        return f"[{host}] 连接失败: {str(e)}"
    finally:
        ssh.close()

if __name__ == "__main__":
    print(f"开始在 {len(SERVERS)} 台服务器上执行命令: '{COMMAND}'\n")
    # 使用线程池并发执行，提升效率
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(execute_command, srv, COMMAND) for srv in SERVERS]
        for future in concurrent.futures.as_completed(futures):
            print(future.result())
            print("-" * 40)

实践要点：使用ThreadPoolExecutor并发执行能将百台服务器的执行时间从串行的几十分钟缩短到几秒。生产环境中建议使用SSH Key认证替代密码。

2. MySQL定时备份与清理脚本

数据库备份是生命线。此脚本实现数据库按日期备份，并基于保留策略（如保留最近7天及每周一的备份）自动清理过期备份。


#!/usr/bin/env python3
# 描述：MySQL数据库备份与生命周期管理脚本
# 依赖：pip install pymysql (可选，用于验证备份)，系统需安装 mysqldump

import os
import subprocess
import datetime
import gzip

DB_CONFIG = {
    "host": "localhost",
    "user": "backup_user",
    "password": "secure_password",
    "databases": ["app_db", "user_db"]  # 需要备份的数据库列表
}
BACKUP_DIR = "/data/backups/mysql"
RETENTION_DAYS = 7  # 普通备份保留7天

def backup_database(db_name):
    today = datetime.datetime.now().strftime('%Y%m%d')
    sql_file = os.path.join(BACKUP_DIR, f"{db_name}_{today}.sql")
    gz_file = f"{sql_file}.gz"
    
    #

← 返回AI专栏