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 None

2. 执行远程命令

连接成功后，执行命令并获取输出：

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), -1

3. 异常处理：脚本的“安全带”

永远不要裸奔！ 任何网络操作、文件操作、命令执行，都必须包裹在 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 = f"echo '{new_name}' > /etc/hostname"out, 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 = f"sed -i 's/{old_name}/{new_name}/g' /etc/hosts"out, err, code = execute_command(client, cmd2)if code != 0:print(f"[FAIL] {host}: 修改 /etc/hosts 失败 - {err}")return False# 3. 立即生效 (注意：某些系统可能需要 reboot，这里用 hostnamectl 尝试)cmd3 = f"hostnamectl 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 + ".gz"target_path = os.path.join(BACKUP_DIR, compressed_name)if not os.path.exists(target_path):# 执行 gzip 命令# 注意：这里使用 subprocess 调用系统命令，比 Python 原生压缩更快cmd = f"gzip -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 paramiko`
`AuthenticationException: Authentication failed`	密码错误，或服务器禁用了密码登录	检查密码；确认 `/etc/ssh/sshd_config` 中 `PasswordAuthentication yes`
`Connection refused`	服务器端口未开放，或防火墙拦截	检查 `firewalld` 或 `iptables` 规则；确认 SSH 端口是 22 还是其他
`TimeoutError`	网络不通，或服务器负载过高无响应	增加超时时间参数；检查网络连通性 `ping`
`Permission denied`	当前用户无权限读取/写入目标文件	使用 `sudo` 运行；检查文件所有者 `ls -l`

特别提示：如果连接 SSH 时卡在“Waiting for host key confirmation”，是因为脚本没有处理指纹确认。务必在 connect_ssh 函数中加入 client.set_missing_host_key_policy(paramiko.AutoAddPolicy())。