3个实操案例教你用Python自动化运维，迈克陈博客避坑指南
3个实操案例教你用Python自动化运维，迈克陈博客避坑指南
看了一堆教程还是不会写项目？别慌，这是大多数应届生的通病。
很多刚毕业的同学，对着屏幕发呆，感觉知识都懂，手一抖就报错。其实问题不在你笨，而在缺少一个能落地的避坑指南。
在迈克陈博客整理的这份实战手册里，我们直接跳过枯燥理论，用 Python 解决运维中最头疼的三个场景：批量改主机名、自动清理日志、健康检查告警。
这套方法我自己在 CSDN 上分享过，很多读者反馈说：“终于能把书本知识变成生产力了。”今天就把这 3 个核心脚本拆解给你看，确保你看完就能跑，跑了就不报错。
概念速懂：为什么运维需要 Python？
运维工程师的核心工作，本质上是“重复性劳动的自动化”。
以前我们写 Shell 脚本，处理文本很灵活，但面对复杂逻辑、API 调用、多线程时，Shell 就显得力不从心。Python 的优势在于：语法简洁：像写伪代码一样，减少语法错误。
生态丰富：requests 发 HTTP 请求，paramiko 连 SSH，psutil 查资源，应有尽有。
跨平台：Windows、Linux、Mac 通吃，方便本地调试。对于应届生来说，不需要精通所有库，只需掌握 标准库 + 几个常用第三方库，就能覆盖 80% 的日常运维场景。
记住一个原则：脚本不是用来炫技的，是用来稳定运行的。 一个 50 行但能稳定跑的脚本，远胜一个 500 行但动不动崩溃的“艺术品”。
环境准备：别在第一步就翻车
很多新手卡在环境配置上，花了三天装 Python，结果第二天发现路径没配好，全白干。
1. Python 版本选择
推荐直接安装 Python 3.9 或 3.10。为什么不用 3.12？ 部分老旧运维库（如某些老版本的 paramiko）对新版本兼容性还没跟上。
为什么不用 2.7？ 已经停止维护，新项目必须用 3.x。2. 虚拟环境隔离
严禁直接 pip install 到全局环境。这是运维大忌，会导致不同项目依赖冲突，最后你都不知道哪个包是谁装的。
推荐使用 venv（Python 3.3+ 自带）：
# 创建虚拟环境
python3 -m venv my_ops_env# 激活环境 (Linux/Mac)
source my_ops_env/bin/activate# 激活环境 (Windows)
my_ops_env\Scripts\activate激活后，命令行前面会多一个 (my_ops_env) 前缀，说明你进入了隔离环境。
3. 必备库安装
创建 requirements.txt 文件，内容如下：
requests==2.31.0
paramiko==3.4.0
psutil==5.9.8
colorama==0.4.6然后执行：
pip install -r requirements.txt避坑点：如果 pip install 速度慢，记得换国内源：
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple核心语法：运维脚本的“三板斧”
写运维脚本，90% 的时间在处理三件事：连服务器、执行命令、处理异常。
1. 连接 SSH 服务器
使用 paramiko 库，这是 Python 连接 Linux 服务器的标准方案。
import paramikodef connect_ssh(host, user, password, port=22):建立 SSH 连接:param host: 服务器 IP:param user: 用户名:param password: 密码 (生产环境建议用密钥，这里为了演示用密码):param port: 端口:return: SSH 客户端对象try:client = paramiko.SSHClient()# 关键：自动接受新主机的指纹，否则第一次连接会卡在确认提示client.set_missing_host_key_policy(paramiko.AutoAddPolicy())client.connect(hostname=host,port=port,username=user,password=password)print(f[SUCCESS] 已连接到 {host})return clientexcept paramiko.AuthenticationException:print(f[ERROR] 认证失败，检查用户名和密码)return Noneexcept paramiko.SSHException as e:print(f[ERROR] SSH 连接异常: {e})return None2. 执行远程命令
连接成功后，执行命令并获取输出：
def execute_command(client, command):执行远程命令:param client: SSH 客户端对象:param command: 要执行的命令字符串:return: (stdout, stderr, exit_code)try:stdin, stdout, stderr = client.exec_command(command)# 关键：必须读取输出，否则缓冲区满会导致连接挂起out = stdout.read().decode('utf-8')err = stderr.read().decode('utf-8')# 获取退出码，0 表示成功，非 0 表示失败exit_code = stdout.channel.recv_exit_status()return out, err, exit_codeexcept Exception as e:print(f[ERROR] 执行命令失败: {e})return , str(e), -13. 异常处理：脚本的“安全带”
永远不要裸奔！ 任何网络操作、文件操作、命令执行，都必须包裹在 try...except 中。
try:# 你的业务逻辑result = do_something()
except FileNotFoundError:print(文件不存在，请检查路径)
except Exception as e:# 捕获所有未预见的错误，记录日志import logginglogging.error(f发生未知错误: {e})raise # 重新抛出，让上层调用者知道出错了完整代码示例：三个实战场景
下面给出两个完整的、可运行的脚本。请复制保存为 .py 文件运行。
场景一：批量修改服务器主机名
痛点：新上架 20 台机器，默认主机名都是 node-01 这种，需要改成 web-01、db-01 等规范名称。手动改太累，sed 脚本又怕改错。
解决方案：读取 CSV 配置文件，自动登录修改 /etc/hostname 和 /etc/hosts。
import paramiko
import csv
import time# 服务器列表配置 (实际项目中建议从数据库或 CMDB 获取)
servers = [{host: 192.168.1.101, new_name: web-prod-01},{host: 192.168.1.102, new_name: web-prod-02},{host: 192.168.1.103, new_name: db-prod-01}
]SSH_USER = root
SSH_PASS = YourStrongPassword123! # 生产环境请替换为密钥认证def change_hostname(host, new_name):修改单台服务器主机名client = connect_ssh(host, SSH_USER, SSH_PASS)if not client:return Falsetry:# 1. 修改 /etc/hostnamecmd1 = fecho '{new_name}' /etc/hostnameout, err, code = execute_command(client, cmd1)if code != 0:print(f[FAIL] {host}: 修改 /etc/hostname 失败 - {err})return False# 2. 修改 /etc/hosts (替换旧主机名)# 先获取旧主机名out_old, _, _ = execute_command(client, hostname)old_name = out_old.strip()# 用 sed 替换 /etc/hosts 中的旧主机名cmd2 = fsed -i 's/{old_name}/{new_name}/g' /etc/hostsout, err, code = execute_command(client, cmd2)if code != 0:print(f[FAIL] {host}: 修改 /etc/hosts 失败 - {err})return False# 3. 立即生效 (注意：某些系统可能需要 reboot，这里用 hostnamectl 尝试)cmd3 = fhostnamectl set-hostname {new_name}out, err, code = execute_command(client, cmd3)if code != 0:# 如果 hostnamectl 失败，尝试直接重启网络服务或提示重启print(f[WARN] {host}: hostnamectl 失败，可能需要重启服务)print(f[SUCCESS] {host} - {new_name})return Trueexcept Exception as e:print(f[ERROR] {host}: {e})return Falsefinally:client.close()if __name__ == __main__:success_count = 0fail_count = 0for server in servers:print(f\n--- 开始处理 {server['host']} ---)if change_hostname(server['host'], server['new_name']):success_count += 1else:fail_count += 1# 避免请求过快，稍微停顿time.sleep(1)print(\n + =*30)print(f执行完毕: 成功 {success_count}, 失败 {fail_count})逐行讲解关键点：sed -i：-i 参数表示直接修改文件，不生成备份。生产环境建议加上 .bak 备份：sed -i.bak 's/old/new/g' file。
hostnamectl：这是 Systemd 系统修改主机名的标准命令，比直接改文件更规范。
finally：无论成功失败，都关闭 SSH 连接，防止连接池耗尽。场景二：日志自动清理与压缩
痛点：Nginx 日志每天几个 G，磁盘快满了。人工清理容易误删，find -mtime 命令记不住。
解决方案：扫描 /var/log/nginx，将 7 天前的 .log 文件压缩并移动到 /backup/logs，删除 30 天前的压缩文件。
import os
import subprocess
import time
from datetime import datetime, timedelta
from pathlib import PathLOG_DIR = /var/log/nginx
BACKUP_DIR = /backup/logs
COMPRESS_DAYS = 7 # 7 天前开始压缩
DELETE_DAYS = 30 # 30 天前直接删除def get_file_age_days(file_path):计算文件最后修改时间距今的天数mtime = os.path.getmtime(file_path)age_days = (time.time() - mtime) / (24 * 3600)return age_daysdef clean_logs():主清理逻辑# 确保备份目录存在Path(BACKUP_DIR).mkdir(parents=True, exist_ok=True)log_files = []# 遍历日志目录for filename in os.listdir(LOG_DIR):if filename.endswith('.log'):file_path = os.path.join(LOG_DIR, filename)if os.path.isfile(file_path):log_files.append(file_path)if not log_files:print(没有找到日志文件)returnfor file_path in log_files:age = get_file_age_days(file_path)file_name = os.path.basename(file_path)try:if age DELETE_DAYS:# 超过 30 天，直接删除os.remove(file_path)print(f[DELETED] {file_name} (Age: {age:.1f} days))elif age COMPRESS_DAYS:# 超过 7 天，压缩并移动# 检查是否已存在压缩文件compressed_name = file_name + .gztarget_path = os.path.join(BACKUP_DIR, compressed_name)if not os.path.exists(target_path):# 执行 gzip 命令# 注意：这里使用 subprocess 调用系统命令，比 Python 原生压缩更快cmd = fgzip -c {file_path} {target_path}result = subprocess.run(cmd, shell=True, capture_output=True, text=True)if result.returncode == 0:# 压缩成功，删除原文件os.remove(file_path)print(f[COMPRESSED] {file_name} - {compressed_name} (Age: {age:.1f} days))else:print(f[ERROR] 压缩失败 {file_name}: {result.stderr})else:# 如果备份目录已有同名文件，跳过或覆盖 (根据策略决定)print(f[SKIP] {file_name} 已存在于备份目录)except Exception as e:print(f[ERROR] 处理 {file_name} 出错: {e})if __name__ == __main__:print(f开始清理日志目录: {LOG_DIR})print(f规则: {COMPRESS_DAYS}天压缩, {DELETE_DAYS}天删除)clean_logs()print(清理任务完成)避坑点：不要直接 rm：先用 gzip 压缩，保留数据，防止误删重要日志。
subprocess.run：比 os.system 更安全，可以捕获输出和退出码。
权限问题：运行脚本的用户必须对 /var/log/nginx 有读权限，对 /backup/logs 有写权限。建议使用 sudo python3 clean_logs.py。常见报错：新手必踩的 5 个坑
在实际部署中，以下错误出现的频率超过 80%：报错信息
原因分析
解决方案ModuleNotFoundError: No module named 'paramiko'
虚拟环境未激活，或库未安装
检查是否在 venv 中运行；执行 pip install paramikoAuthenticationException: Authentication failed
密码错误，或服务器禁用了密码登录
检查密码；确认 /etc/ssh/sshd_config 中 PasswordAuthentication yesConnection refused
服务器端口未开放，或防火墙拦截
检查 firewalld 或 iptables 规则；确认 SSH 端口是 22 还是其他TimeoutError
网络不通，或服务器负载过高无响应
增加超时时间参数；检查网络连通性 pingPermission denied
当前用户无权限读取/写入目标文件
使用 sudo 运行；检查文件所有者 ls -l特别提示：如果连接 SSH 时卡在“Waiting for host key confirmation”，是因为脚本没有处理指纹确认。务必在 connect_ssh 函数中加入 client.set_missing_host_key_policy(paramiko.AutoAddPolicy())。
小结：从脚本到工程化
写运维脚本，能跑起来只是及格线。
真正成熟的运维脚本，还需要具备：日志记录：使用 logging 模块，将操作记录到 /var/log/ops/xxx.log，方便追溯。
配置分离：将 IP、密码、路径等硬编码内容提取到 config.yaml 或环境变量中。
幂等性：脚本运行多次，结果应该是一样的。比如改主机名，如果已经改过，不应该报错，而是提示“已是目标状态”。
监控告警：脚本失败时，发送钉钉/企微/邮件通知，而不是静默失败。在 CSDN 上，很多资深运维工程师分享的经验是：“脚本的价值不在于写了多少行，而在于它默默运行了多少天没有出错。”
对于应届生，建议你从最简单的“单机脚本”开始，逐步过渡到“批量脚本”，最后尝试接入 Ansible 或 SaltStack 等自动化平台。
互动话题：
你公司项目里是怎么处理自动化脚本的？是纯 Python，还是混合 Shell？有没有遇到过“脚本在测试环境正常，上生产就炸”的情况？欢迎在评论区分享你的踩坑经历，大家一起交流。