当前位置：首页 > news >正文

Xshell自动化脚本大赛实战案例及深度分析

news 2025/8/29 14:15:57

实战案例一：多服务器 Nginx 批量部署与验证脚本

一、需求背景

某企业新采购 20 台 CentOS 8 服务器，需快速完成 Nginx 安装、配置默认首页、启动服务并验证可用性，传统手动操作需逐台登录执行命令，耗时且易漏配，急需自动化脚本实现 “一键部署 + 结果校验”。

二、技术架构

采用 “Python 主导 + Bash 辅助” 模式：

Bash 脚本：负责执行 Nginx 安装、配置、启动的系统命令，输出执行状态日志；
Python 脚本：读取服务器列表、调用 Bash 脚本、校验服务可用性、生成部署报告。

三、脚本实现

Bash 执行脚本（nginx_install.sh）

\#!/bin/bash\# 接收参数：服务器IP、用户名、密码（实际场景建议用免密登录，此处为演示）IP=\$1USER=\$2PASS=\$3\# 远程执行Nginx安装与配置sshpass -p \$PASS ssh \$USER@\$IP "yum install epel-release -y &&yum install nginx -y &&echo 'Xshell Automation Test' > /usr/share/nginx/html/index.html &&systemctl start nginx &&systemctl enable nginx"\# 检查Nginx服务状态SERVICE\_STATUS=\$(sshpass -p \$PASS ssh \$USER@\$IP "systemctl is-active nginx")echo "\$IP - Nginx服务状态：\$SERVICE\_STATUS" >> deploy\_log.txt

Python 控制脚本（batch_nginx_deploy.py）

import subprocessimport pandas as pdimport requests\# 读取Excel服务器列表（IP、用户名、密码）server\_list = pd.read\_excel("server\_list.xlsx")success\_count = 0fail\_list = \[]\# 遍历服务器执行部署for index, row in server\_list.iterrows():ip = row\["IP"]user = row\["Username"]password = row\["Password"]\# 调用Bash脚本subprocess.run(\["./nginx\_install.sh", ip, user, password], check=False)\# 校验Nginx可用性（访问默认首页）try:response = requests.get(f"http://{ip}", timeout=5)if response.status\_code == 200 and "Xshell Automation Test" in response.text:print(f"{ip} - Nginx部署成功，页面验证通过")success\_count += 1else:print(f"{ip} - Nginx部署失败，页面访问异常")fail\_list.append(ip)except Exception as e:print(f"{ip} - Nginx部署失败，原因：{str(e)}")fail\_list.append(ip)\# 生成部署报告report = f"""Nginx批量部署报告\================总服务器数量：{len(server\_list)}成功数量：{success\_count}失败数量：{len(fail\_list)}失败服务器IP：{fail\_list if fail\_list else '无'}"""with open("nginx\_deploy\_report.txt", "w") as f:f.write(report)print(report)

四、案例优势分析

效率提升：手动部署 20 台服务器需 1.5 小时，脚本仅需 8 分钟，耗时缩短 91%；
结果可控：Python 自动校验服务状态与页面内容，避免 “部署成功但服务不可用” 的隐性问题；
可复用性：修改 Excel 服务器列表即可适配不同批次服务器，无需修改核心逻辑。

实战案例二：多服务器 Nginx 日志异常排查脚本

一、需求背景

某企业 10 台 Web 服务器（Nginx）频繁出现 502 错误，运维人员需逐台下载日志、搜索错误关键词，耗时且易遗漏，需脚本实现 “批量日志采集 + 异常分析 + 报告生成”。

二、技术架构

采用 “Bash 采集 + Python 分析” 模式：

Bash 脚本：批量下载目标服务器 Nginx 错误日志；
Python 脚本：解析日志、统计异常频次、定位异常服务器。

三、脚本实现

Bash 日志采集脚本（log_download.sh）

\#!/bin/bash\# 服务器列表文件（每行格式：IP 用户名 密码）SERVER\_FILE=\$1LOG\_DIR="./nginx\_error\_logs"\# 创建本地日志存储目录mkdir -p \$LOG\_DIR\# 批量下载日志while read -r ip user pass; doecho "正在下载\$ip的Nginx错误日志..."sshpass -p \$pass scp \$user@\$ip:/var/log/nginx/error.log \$LOG\_DIR/\${ip}\_error.logdone < \$SERVER\_FILE

Python 日志分析脚本（log_analysis.py）

import reimport osfrom collections import defaultdict\# 日志存储目录LOG\_DIR = "./nginx\_error\_logs"\# 502错误正则表达式ERROR\_PATTERN = r"\\\[(\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2})\\] .\* 502 Bad Gateway"\# 统计各服务器502错误频次error\_stats = defaultdict(list)\# 遍历日志文件分析for filename in os.listdir(LOG\_DIR):if filename.endswith("\_error.log"):ip = filename.split("\_")\[0]log\_path = os.path.join(LOG\_DIR, filename)with open(log\_path, "r") as f:for line in f:match = re.search(ERROR\_PATTERN, line)if match:error\_time = match.group(1)error\_stats\[ip].append(error\_time)\# 生成分析报告report = "Nginx 502错误分析报告\n"report += "====================\n"report += f"分析服务器数量：{len(error\_stats)}\n"report += "各服务器错误统计：\n"for ip, times in error\_stats.items():report += f"- {ip}：共{len(times)}次错误，发生时间：{times\[:5]}...（仅显示前5条）\n"\# 定位错误最频繁的服务器if error\_stats:most\_error\_ip = max(error\_stats, key=lambda x: len(error\_stats\[x]))report += f"\n错误最频繁服务器：{most\_error\_ip}（{len(error\_stats\[most\_error\_ip])}次）\n"report += "建议优先排查该服务器的后端服务（如PHP-FPM、Tomcat）状态\n"with open("nginx\_error\_report.txt", "w") as f:f.write(report)print(report)

四、案例优势分析

排查效率：手动分析 10 台服务器日志需 40 分钟，脚本仅需 5 分钟，效率提升 87.5%；
精准定位：自动统计错误频次，快速锁定问题最严重的服务器，减少排查范围；
指导性强：报告中给出排查建议，降低运维人员技术门槛。

实战案例三：MySQL 数据库自动备份与异常恢复脚本

一、需求背景

某企业 3 台 MySQL 服务器需每日凌晨 2 点备份数据库，若备份失败或数据库服务宕机，需立即通知运维人员并尝试恢复，传统手动备份易遗忘，故障响应滞后。

二、技术架构

采用 “Python 监控 + Bash 执行” 模式：

Bash 脚本：执行 MySQL 备份、恢复命令；
Python 脚本：定时触发备份、监控服务状态、发送告警通知、调用恢复脚本。

三、脚本实现

Bash 备份与恢复脚本（mysql_backup_restore.sh）

\#!/bin/bash\# 接收参数：操作类型（backup/restore）、IP、用户名、密码、数据库名ACTION=\$1IP=\$2USER=\$3PASS=\$4DB\_NAME=\$5BACKUP\_DIR="./mysql\_backups"\# 创建备份目录mkdir -p \$BACKUP\_DIRif \[ "\$ACTION" = "backup" ]; then\# 执行备份BACKUP\_FILE="\${BACKUP\_DIR}/\${IP}\_\${DB\_NAME}\_\$(date +%Y%m%d).sql"mysqldump -h \$IP -u \$USER -p\$PASS \$DB\_NAME > \$BACKUP\_FILEif \[ \$? -eq 0 ]; thenecho "备份成功：\$BACKUP\_FILE"exit 0elseecho "备份失败"exit 1fiif \[ "\$ACTION" = "restore" ]; then\# 恢复最新备份LATEST\_BACKUP=\$(ls -t \${BACKUP\_DIR}/\${IP}\_\${DB\_NAME}\_\*.sql | head -1)if \[ -z "\$LATEST\_BACKUP" ]; thenecho "无备份文件可恢复"exit 1fimysql -h \$IP -u \$USER -p\$PASS \$DB\_NAME < \$LATEST\_BACKUPif \[ \$? -eq 0 ]; thenecho "恢复成功：\$LATEST\_BACKUP"exit 0elseecho "恢复失败"exit 1fifi

Python 监控脚本（mysql_monitor.py）

import subprocessimport scheduleimport timeimport smtplibfrom email.mime.text import MIMEText\# 邮件配置（企业邮箱为例）SMTP\_SERVER = "smtp.xxx.com"SMTP\_PORT = 465EMAIL\_USER = "ops@xxx.com"EMAIL\_PASS = "xxx"ALERT\_RECEIVERS = \["admin@xxx.com"]\# 服务器配置SERVER\_CONFIGS = \[{"ip": "192.168.1.101", "user": "root", "pass": "xxx", "db": "app\_db"},{"ip": "192.168.1.102", "user": "root", "pass": "xxx", "db": "user\_db"},{"ip": "192.168.1.103", "user": "root", "pass": "xxx", "db": "log\_db"}]\# 发送告警邮件def send\_alert(title, content):msg = MIMEText(content, "plain", "utf-8")msg\["From"] = EMAIL\_USERmsg\["To"] = ",".join(ALERT\_RECEIVERS)msg\["Subject"] = titlewith smtplib.SMTP\_SSL(SMTP\_SERVER, SMTP\_PORT) as server:server.login(EMAIL\_USER, EMAIL\_PASS)server.sendmail(EMAIL\_USER, ALERT\_RECEIVERS, msg.as\_string())\# 备份任务def backup\_mysql():for config in SERVER\_CONFIGS:ip = config\["ip"]user = config\["user"]password = config\["pass"]db = config\["db"]result = subprocess.run(\["./mysql\_backup\_restore.sh", "backup", ip, user, password, db],capture\_output=True, text=True)if result.returncode != 0:alert\_title = f"MySQL备份失败 - {ip}"alert\_content = f"服务器{ip}的{db}数据库备份失败，原因：{result.stderr}"send\_alert(alert\_title, alert\_content)else:print(f"服务器{ip}的{db}数据库备份成功")\# 检查MySQL服务状态并尝试恢复def check\_mysql\_status():for config in SERVER\_CONFIGS:ip = config\["ip"]user = config\["user"]password = config\["pass"]db = config\["db"]\# 检查服务是否存活status\_result = subprocess.run(\[f"sshpass -p {password} ssh {user}@{ip} 'systemctl is-active mysqld'"],shell=True, capture\_output=True, text=True)if status\_result.stdout.strip() != "active":\# 尝试重启服务restart\_result = subprocess.run(\[f"sshpass -p {password} ssh {user}@{ip} 'systemctl restart mysqld'"],shell=True, capture\_output=True, text=True)time.sleep(10)\# 再次检查状态recheck\_result = subprocess.run(\[f"sshpass -p {password} ssh {user}@{ip} 'systemctl is-active mysqld'"],shell=True, capture\_output=True, text=True)if recheck\_result.stdout.strip() == "active":alert\_title = f"MySQL服务恢复 - {ip}"alert\_content = f"服务器{ip}的MySQL服务已重启恢复"send\_alert(alert\_title, alert\_content)else:\# 重启失败，尝试恢复备份restore\_result = subprocess.run(\["./mysql\_backup\_restore.sh", "restore", ip, user, password, db],capture\_output=True, text=True)if restore\_result.returncode == 0:alert\_title = f"MySQL恢复成功 - {ip}"alert\_content = f"服务器{ip}的MySQL服务重启失败，已通过备份恢复"send\_alert(alert\_title, alert\_content)else:alert\_title = f"MySQL故障紧急 - {ip}"alert\_content = f"服务器{ip}的MySQL服务宕机，重启与恢复均失败，需人工处理"send\_alert(alert\_title, alert\_content)\# 定时任务：每日2点备份，每5分钟检查服务状态schedule.every().day.at("02:00").do(backup\_mysql)schedule.every(5).minutes.do(check\_mysql\_status)\# 启动定时任务print("MySQL监控与备份脚本已启动...")while True:schedule.run\_pending()time.sleep(60)