Python 自动化运维：让你从 “搬砖” 中解放出来的 5 个必备工具

pip install paramiko  # 推荐安装 3.4.0 最新稳定版

import paramiko
import time
from typing import List

def ssh_execute_command(hosts: List[dict], command: str) -> dict:
    """
    批量登录服务器执行指定命令
    :param hosts: 服务器列表，格式 [{"host": "IP", "port": 22, "user": "用户名", "password": "密码"}, ...]
    :param command: 要执行的命令
    :return: 执行结果，key 为服务器IP，value 为命令输出
    """
    results = {}
    # 创建 SSH 客户端配置
    ssh_config = paramiko.SSHClient()
    # 自动添加未知主机密钥（生产环境建议手动配置 known_hosts）
    ssh_config.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    
    for host in hosts:
        try:
            # 建立 SSH 连接
            ssh_config.connect(
                hostname=host["host"],
                port=host["port"],
                username=host["user"],
                password=host["password"],
                timeout=10  # 超时时间 10 秒
            )
            # 执行命令
            stdin, stdout, stderr = ssh_config.exec_command(command)
            # 获取输出结果（注意编码）
            output = stdout.read().decode("utf-8").strip()
            error = stderr.read().decode("utf-8").strip()
            
            if error:
                results[host["host"]] = f"执行失败: {error}"
            else:
                results[host["host"]] = output
            
            # 关闭连接
            ssh_config.close()
            time.sleep(0.5)  # 避免连接过快被防火墙拦截
        except Exception as e:
            results[host["host"]] = f"连接失败: {str(e)}"
    
    return results

# 示例：批量执行 df -h
if __name__ == "__main__":
    # 配置要操作的服务器列表（生产环境建议从配置文件读取，不要硬编码密码）
    servers = [
        {"host": "192.168.1.10", "port": 22, "user": "root", "password": "your_password"},
        {"host": "192.168.1.11", "port": 22, "user": "root", "password": "your_password"},
    ]
    # 执行磁盘查看命令
    disk_results = ssh_execute_command(servers, "df -h")
    # 打印结果
    for ip, result in disk_results.items():
        print(f"=== 服务器 {ip} 磁盘信息 ===")
        print(result)
        print("-" * 50)

pip install fabric  # 推荐安装 3.2.2 最新稳定版

# 文件名：fabfile.py（Fabric 约定的默认文件名）
from fabric import Connection, task
from invoke import Responder

# 定义服务器列表（生产环境建议用环境变量/配置文件）
SERVERS = ["192.168.1.10", "192.168.1.11"]
USER = "root"
PASSWORD = "your_password"

# 定义 sudo 密码响应器（避免手动输入密码）
sudo_responder = Responder(
    pattern=r"\[sudo\] password for .*:",
    response=f"{PASSWORD}\n"
)

@task
def restart_nginx(c):
    """
    批量重启 Nginx 服务的 Fabric 任务
    :param c: Fabric 自动传入的 Connection 对象
    """
    # 循环连接每台服务器
    for host in SERVERS:
        print(f"=== 操作服务器 {host} ===")
        # 建立连接
        conn = Connection(
            host=host,
            user=USER,
            connect_kwargs={"password": PASSWORD}
        )
        # 执行重启命令（需要 sudo 权限）
        result = conn.sudo(
            "systemctl restart nginx",
            watchers=[sudo_responder],  # 自动输入 sudo 密码
            warn=True  # 命令失败时不中断后续操作
        )
        # 输出结果
        if result.ok:
            print(f"{host} Nginx 重启成功")
        else:
            print(f"{host} Nginx 重启失败: {result.stderr}")
        # 关闭连接
        conn.close()

# 执行方式：在终端运行
# fab restart_nginx

pip install ansible  # 推荐安装 9.1.0 最新稳定版

from ansible import context
from ansible.cli import CLI
from ansible.executor.playbook_executor import PlaybookExecutor
from ansible.inventory.manager import InventoryManager
from ansible.parsing.dataloader import DataLoader
from ansible.vars.manager import VariableManager
import os

def run_ansible_playbook(playbook_path: str, inventory_path: str):
    """
    用 Python API 执行 Ansible Playbook
    :param playbook_path: Playbook 文件路径
    :param inventory_path: 主机清单文件路径
    """
    # 初始化 Ansible 上下文
    context.CLIARGS = CLI.setup_cli_args({
        "connection": "ssh",
        "module_path": "",
        "forks": 10,  # 并行执行的进程数
        "become": True,
        "become_method": "sudo",
        "become_user": "root",
        "check": False,
        "diff": False,
        "verbosity": 1
    })
    
    # 加载数据（如 Playbook、清单）
    loader = DataLoader()
    # 加载主机清单
    inventory = InventoryManager(loader=loader, sources=[inventory_path])
    # 变量管理
    variable_manager = VariableManager(loader=loader, inventory=inventory)
    
    # 初始化 Playbook 执行器
    pbex = PlaybookExecutor(
        playbooks=[playbook_path],
        inventory=inventory,
        variable_manager=variable_manager,
        loader=loader,
        passwords={"conn_pass": "your_password", "become_pass": "your_password"}
    )
    
    # 执行 Playbook
    result = pbex.run()
    # 输出执行结果
    if result == 0:
        print("Playbook 执行成功！")
    else:
        print(f"Playbook 执行失败，返回码: {result}")

# 示例：执行安装 Nginx 的 Playbook
if __name__ == "__main__":
    # 1. 先创建 Playbook 文件（nginx_install.yml）
    playbook_content = """
- hosts: all
  tasks:
    - name: 安装 Nginx
      apt:
        name: nginx=1.24.0
        state: present
        update_cache: yes
      when: ansible_os_family == "Debian"
    
    - name: 启动并启用 Nginx
      service:
        name: nginx
        state: started
        enabled: yes
    """
    with open("nginx_install.yml", "w") as f:
        f.write(playbook_content)
    
    # 2. 创建主机清单文件（inventory.ini）
    inventory_content = """
[web_servers]
192.168.1.10
192.168.1.11
    """
    with open("inventory.ini", "w") as f:
        f.write(inventory_content)
    
    # 3. 执行 Playbook
    run_ansible_playbook("nginx_install.yml", "inventory.ini")
    
    # 4. 清理临时文件（可选）
    os.remove("nginx_install.yml")
    os.remove("inventory.ini")

pip install psutil  # 推荐安装 5.9.6 最新稳定版

import psutil
import time
import requests

def get_cpu_usage(interval: int = 5) -> float:
    """
    获取 CPU 平均使用率（按指定间隔采样）
    :param interval: 采样间隔（秒）
    :return: CPU 使用率（百分比）
    """
    return psutil.cpu_percent(interval=interval)

def send_wechat_alert(content: str, webhook_url: str):
    """
    发送企业微信告警
    :param content: 告警内容
    :param webhook_url: 企业微信机器人 webhook 地址
    """
    headers = {"Content-Type": "application/json"}
    data = {
        "msgtype": "text",
        "text": {"content": content}
    }
    try:
        response = requests.post(webhook_url, json=data, headers=headers, timeout=10)
        if response.json()["errcode"] == 0:
            print("告警发送成功")
        else:
            print(f"告警发送失败: {response.text}")
    except Exception as e:
        print(f"告警发送异常: {str(e)}")

# 监控主逻辑
if __name__ == "__main__":
    # 企业微信机器人 webhook（替换成你的）
    WEBHOOK_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your_key"
    # CPU 告警阈值
    CPU_THRESHOLD = 80.0
    
    print("开始监控 CPU 使用率...")
    while True:
        cpu_usage = get_cpu_usage()
        print(f"当前 CPU 使用率: {cpu_usage}%")
        
        # 超过阈值发送告警
        if cpu_usage > CPU_THRESHOLD:
            alert_content = f"【服务器告警】CPU 使用率超过阈值！\n当前使用率: {cpu_usage}%\n阈值: {CPU_THRESHOLD}%\n时间: {time.strftime('%Y-%m-%d %H:%M:%S')}"
            send_wechat_alert(alert_content, WEBHOOK_URL)
        
        # 每 60 秒检查一次
        time.sleep(60)

pip install schedule  # 推荐安装 1.2.0 最新稳定版

import schedule
import time
import logging
import os
from datetime import datetime

# 配置日志（聚合脚本执行日志）
def setup_logging():
    """配置日志输出：同时输出到文件和控制台"""
    # 日志文件路径
    log_file = f"ops_automation_{datetime.now().strftime('%Y%m%d')}.log"
    # 日志格式
    log_format = "%(asctime)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s"
    
    # 配置根日志器
    logging.basicConfig(
        level=logging.INFO,
        format=log_format,
        handlers=[
            # 输出到文件
            logging.FileHandler(log_file, encoding="utf-8"),
            # 输出到控制台
            logging.StreamHandler()
        ]
    )

def clean_log_files(log_dir: str, keep_days: int = 7):
    """
    清理指定目录下的旧日志文件
    :param log_dir: 日志目录
    :param keep_days: 保留最近几天的日志
    """
    logger = logging.getLogger(__name__)
    try:
        # 检查目录是否存在
        if not os.path.exists(log_dir):
            logger.warning(f"日志目录不存在: {log_dir}")
            return
        
        # 计算过期时间（秒）
        expire_time = time.time() - (keep_days * 24 * 3600)
        
        # 遍历目录下的文件
        for filename in os.listdir(log_dir):
            file_path = os.path.join(log_dir, filename)
            # 只处理文件（跳过目录）
            if os.path.isfile(file_path):
                # 获取文件最后修改时间
                file_mtime = os.path.getmtime(file_path)
                # 判断是否过期
                if file_mtime < expire_time:
                    # 删除文件
                    os.remove(file_path)
                    logger.info(f"已删除过期日志文件: {file_path}")
        
        logger.info("日志清理任务执行完成")
    except Exception as e:
        logger.error(f"日志清理任务执行失败: {str(e)}", exc_info=True)

# 定时任务配置
if __name__ == "__main__":
    # 初始化日志
    setup_logging()
    logger = logging.getLogger(__name__)
    
    # 配置定时任务：每天凌晨 2 点执行日志清理
    schedule.every().day.at("02:00").do(
        clean_log_files,
        log_dir="/var/log/nginx",  # Nginx 日志目录
        keep_days=7  # 保留 7 天日志
    )
    
    logger.info("定时任务已启动，每天凌晨 2 点清理日志文件...")
    
    # 循环执行定时任务
    while True:
        schedule.run_pending()
        time.sleep(60)  # 每分钟检查一次任务