Shell 与 Python 自动化运维脚本开发：从手工操作到高效自动化

运维工作中存在大量重复性操作，如日志清理、备份、监控脚本等。将这些操作自动化不仅能提升效率，还能减少人为错误。Shell 脚本适合系统级任务和管道组合，Python 则在复杂逻辑和数据处理场景更加强大。本文分享自动化运维脚本开发的经验与最佳实践。

一、Shell 脚本编写的核心原则

Shell 脚本是 Linux 运维的基础技能，贯穿日常工作的方方面面。

错误处理机制是脚本可靠性的保障。使用 set -e 让脚本在任何命令失败时立即退出，使用 set -u 检测未定义的变量引用，使用 set -o pipefail 确保管道中任何一个命令失败都能被捕获。这些选项的组合能够大幅提升脚本的健壮性。

日志输出规范化便于问题排查。定义统一的日志函数，包含时间戳、日志级别、调用位置等信息。生产环境脚本应当同时输出到标准输出和日志文件。

参数校验与帮助信息是专业脚本的标配。使用 getopts 或 getopt 解析命令行参数，参数错误时给出清晰的使用说明并退出。

#!/bin/bash
# 规范化 Shell 脚本模板

# 严格模式
set -euo pipefail
IFS=$'\n\t'

# 脚本目录和日志配置
readonly SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
readonly LOG_FILE="/var/log/$(basename "${BASH_SOURCE[0]}").log"

# 日志函数
log() {
    local level="$1"
    shift
    local message="$*"
    local timestamp=$(date '+%Y-%m-%d %H:%M:%S')
    echo "[${timestamp}] [${level}] ${message}" | tee -a "${LOG_FILE}"
}

log_info() { log "INFO" "$@"; }
log_warn() { log "WARN" "$@"; }
log_error() { log "ERROR" "$@"; }

# 错误处理函数
on_error() {
    local exit_code=$?
    log_error "脚本执行失败，行号: ${1}, 命令: ${2}"
    exit "${exit_code}"
}

trap 'on_error ${LINENO} "${BASH_COMMAND}"' ERR

# 参数解析示例
usage() {
    cat << EOF
用法: $0 [选项]

选项:
    -h, --help              显示帮助信息
    -d, --directory DIR     指定工作目录
    -r, --retention DAYS    保留天数 (默认: 7)
    -v, --verbose           详细输出

示例:
    $0 -d /var/log -r 30 -v
EOF
    exit 1
}

# 默认值
work_dir="/var/log"
retention_days=7
verbose=false

# 解析参数
while [[ $# -gt 0 ]]; do
    case "$1" in
        -h|--help) usage ;;
        -d|--directory) work_dir="$2"; shift 2 ;;
        -r|--retention) retention_days="$2"; shift 2 ;;
        -v|--verbose) verbose=true; shift ;;
        *) log_error "未知参数: $1"; usage ;;
    esac
done

log_info "开始执行清理任务，工作目录: ${work_dir}，保留天数: ${retention_days}"

二、Python 脚本的工程化实践

当任务复杂度超出 Shell 的能力范围时，Python 是更好的选择。

项目结构规范化对于长期维护的脚本尤为重要。包含配置模块、日志模块、核心逻辑模块、入口脚本的标准结构，便于扩展和测试。

日志配置集中管理。使用 Python 标准库的 logging 模块，配置日志格式、级别、输出目标。生产环境应当同时输出到文件和控制台。

配置文件与代码分离。敏感信息（如数据库密码、API 密钥）不应硬编码在代码中，应当存储在配置文件或环境变量中。

#!/usr/bin/env python3
"""
日志清理脚本 - Python 版本
"""

import os
import sys
import argparse
import logging
import configparser
from pathlib import Path
from datetime import datetime, timedelta
from typing import List

class LogCleaner:
    """日志清理器"""
    
    def __init__(self, config_path: str = "/etc/logcleaner.conf"):
        self.config = self._load_config(config_path)
        self.logger = self._setup_logging()
    
    def _load_config(self, config_path: str) -> configparser.ConfigParser:
        """加载配置文件"""
        config = configparser.ConfigParser()
        if os.path.exists(config_path):
            config.read(config_path)
        return config
    
    def _setup_logging(self) -> logging.Logger:
        """配置日志"""
        logger = logging.getLogger("logcleaner")
        log_level = self.config.get("logging", "level", fallback="INFO")
        logger.setLevel(getattr(logging, log_level.upper(), logging.INFO))
        
        formatter = logging.Formatter(
            "[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s",
            datefmt="%Y-%m-%d %H:%M:%S"
        )
        
        # 文件 handler
        log_file = self.config.get("logging", "file", fallback="/var/log/logcleaner.log")
        os.makedirs(os.path.dirname(log_file), exist_ok=True)
        file_handler = logging.FileHandler(log_file)
        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)
        
        # 控制台 handler
        console_handler = logging.StreamHandler()
        console_handler.setFormatter(formatter)
        logger.addHandler(console_handler)
        
        return logger
    
    def find_old_logs(self, directory: str, days: int) -> List[Path]:
        """查找超过指定天数的日志文件"""
        cutoff_date = datetime.now() - timedelta(days=days)
        old_logs = []
        
        for root, _, files in os.walk(directory):
            for filename in files:
                filepath = Path(root) / filename
                # 检查修改时间
                mtime = datetime.fromtimestamp(filepath.stat().st_mtime)
                if mtime < cutoff_date:
                    old_logs.append(filepath)
        
        self.logger.info(f"在 {directory} 中找到 {len(old_logs)} 个过期日志文件")
        return old_logs
    
    def cleanup(self, directory: str, retention_days: int, dry_run: bool = False) -> int:
        """执行清理任务"""
        old_logs = self.find_old_logs(directory, retention_days)
        
        if not old_logs:
            self.logger.info("没有需要清理的日志文件")
            return 0
        
        total_size = sum(log.stat().st_size for log in old_logs)
        self.logger.info(f"将清理 {len(old_logs)} 个文件，总大小: {total_size / 1024 / 1024:.2f} MB")
        
        deleted_count = 0
        for log in old_logs:
            try:
                if dry_run:
                    self.logger.info(f"[DRY RUN] 将删除: {log}")
                else:
                    log.unlink()
                    self.logger.info(f"已删除: {log}")
                deleted_count += 1
            except Exception as e:
                self.logger.error(f"删除失败 {log}: {e}")
        
        return deleted_count


def main():
    parser = argparse.ArgumentParser(description="日志清理工具")
    parser.add_argument("-d", "--directory", default="/var/log", help="日志目录")
    parser.add_argument("-r", "--retention", type=int, default=7, help="保留天数")
    parser.add_argument("--dry-run", action="store_true", help="仅模拟运行")
    parser.add_argument("-c", "--config", default="/etc/logcleaner.conf", help="配置文件")
    
    args = parser.parse_args()
    
    cleaner = LogCleaner(config_path=args.config)
    cleaner.logger.info(f"开始清理任务，目录: {args.directory}，保留天数: {args.retention}")
    
    deleted = cleaner.cleanup(args.directory, args.retention, args.dry_run)
    
    cleaner.logger.info(f"清理完成，共删除 {deleted} 个文件")
    return 0 if deleted >= 0 else 1


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

三、常用自动化场景与脚本模板

日志轮转与清理是最常见的自动化任务。生产服务器的日志如果不做清理，可能在短期内填满磁盘。脚本应当按照文件修改时间而非创建时间判断过期，考虑保留正在被写入的日志文件。

服务健康检查与告警是保障服务可用性的重要手段。健康检查脚本应当覆盖多个维度：进程存活、端口监听、HTTP 端点响应、数据库连接等。发现问题后通过钉钉、企业微信等渠道及时通知。

备份脚本需要考虑完整性、可靠性和恢复便捷性。备份完成后应当验证备份文件完整性，定期进行恢复演练。

#!/bin/bash
# 服务健康检查与告警脚本

set -euo pipefail

# 配置
readonly SERVICES=("nginx" "mysql" "redis" "app")
readonly WEB_PORT=8080
readonly WEB_PATH="/healthz"
readonly DINGTALK_WEBHOOK="https://oapi.dingtalk.com/robot/send?access_token=xxx"
readonly LOG_FILE="/var/log/health_check.log"

log() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $*" >> "${LOG_FILE}"
}

send_alert() {
    local message="$1"
    local curl_data=$(cat <<EOF
{
    "msgtype": "text",
    "text": {
        "content": "[健康检查告警] ${message}"
    }
}
EOF
)
    curl -s -X POST "${DINGTALK_WEBHOOK}" \
        -H "Content-Type: application/json" \
        -d "${curl_data}" > /dev/null
}

check_process() {
    local service="$1"
    if pgrep -x "${service}" > /dev/null; then
        log "${service}: 进程运行中"
        return 0
    else
        log "${service}: 进程未运行"
        return 1
    fi
}

check_port() {
    local port="$1"
    if nc -z localhost "${port}" 2>/dev/null; then
        log "端口 ${port}: 监听正常"
        return 0
    else
        log "端口 ${port}: 无法连接"
        return 1
    fi
}

check_web() {
    local port="$1"
    local path="$2"
    local http_code=$(curl -s -o /dev/null -w "%{http_code}" "http://localhost:${port}${path}")
    if [[ "${http_code}" == "200" ]]; then
        log "Web 健康检查: 正常 (${http_code})"
        return 0
    else
        log "Web 健康检查: 异常 (${http_code})"
        return 1
    fi
}

# 主检查逻辑
failed=0

for service in "${SERVICES[@]}"; do
    check_process "${service}" || ((failed++))
done

check_port "${WEB_PORT}" || ((failed++))

check_web "${WEB_PORT}" "${WEB_PATH}" || ((failed++))

# 发送告警
if [[ ${failed} -gt 0 ]]; then
    message="$(hostname) 健康检查失败，${failed} 项检查未通过"
    log "${message}"
    send_alert "${message}"
    exit 1
fi

log "健康检查全部通过"
exit 0

四、脚本的部署与调度

脚本编写完成后，需要配合调度机制实现自动化执行。

crontab 调度是 Linux 自带的定时任务工具。通过 /etc/crontab 或 crontab -e 命令配置定时执行规则。cron 表达式从左到右表示：分钟、小时、日期、月份、星期。

systemd timer 是更现代的调度方式，相比 crontab 提供了更好的日志集成、服务依赖管理和权限控制。

ansible 或 puppet 批量部署脚本到多台服务器，配合配置管理工具统一管理脚本版本和调度配置。

# crontab 配置示例
# 每天凌晨 3 点执行日志清理
0 3 * * * /usr/local/bin/log_cleaner.sh -d /var/log -r 30 >> /var/log/log_cleaner.log 2>&1

# 每 5 分钟执行健康检查
*/5 * * * * /usr/local/bin/health_check.sh >> /var/log/health_check.log 2>&1

# 每周日凌晨 4 点执行数据库备份
0 4 * * 0 /usr/local/bin/backup_db.sh >> /var/log/backup.log 2>&1

# systemd timer 配置示例
# /etc/systemd/system/logcleaner.timer
[Unit]
Description=日志清理定时任务

[Timer]
OnCalendar=daily
Persistent=true
Unit=logcleaner.service

[Install]
WantedBy=timers.target

五、总结

Shell 和 Python 是运维自动化的两把利器。

Shell 脚本适合系统级任务和管道组合，优势在于与系统命令的无缝集成。编写 Shell 脚本应当遵循严格模式、规范化日志、完整参数校验的原则。错误处理是脚本可靠性的关键，set -euo pipefail 组合应当作为标准配置。

Python 脚本在复杂逻辑、数据处理、API 调用等场景更加灵活。工程化的项目结构、配置与代码分离、完善的异常处理是专业 Python 脚本的标志。

自动化脚本的价值在于减少人工操作、提升效率、降低错误率。但自动化也可能放大错误——一个 buggy 的自动化脚本可能造成比人工操作更大的损失。因此，自动化脚本在投入生产前需要充分测试，关键操作应当保留人工审批环节。建议建立自动化脚本的评审机制，确保脚本质量和安全性。