Python自动化巡检华为交换机（含告警判断+钉钉推送）

本文基于Netmiko库通过SSH协议批量巡检华为VRP交换机，自动采集硬件、资源、接口、告警信息，内置异常规则判断，一旦检测到故障/资源超限，自动推送Markdown告警消息到钉钉群；每台设备输出完整巡检日志文件。

适用：华为VRP5 / VRP8 交换机、路由器

依赖库：netmiko、requests

实现能力：多线程并发巡检 + 自动告警识别 + 钉钉机器人推送 + 本地日志留存

华为交换机巡检项清单

辅助收集命令：

display diagnostic-information 一键采集全套诊断信息

display current-configuration 查看设备当前运行配置

前置准备

Python依赖安装

pip install netmiko requests

华为交换机侧最小SSH配置

sysname Switch
aaa
 local-user admin password cipher Admin@123
 local-user admin service-type ssh
 local-user admin privilege level 15
ssh user admin authentication-type password
stelnet server enable

说明：创建SSH账号，权限等级15，开启SSH服务

钉钉自定义机器人配置

1. 钉钉群 → 群设置 → 智能群助手 → 添加机器人 → 自定义机器人
2. 获取Webhook地址填入脚本
3. 安全设置选择加签，复制密钥；不开启加签则密钥留空
4. 自定义关键词：巡检、告警，防止消息被钉钉拦截

完整Python巡检脚本

from netmiko import ConnectHandler
from concurrent.futures import ThreadPoolExecutor
import re
import requests
import time
import hmac, hashlib, base64, urllib.parse

# ===================== 【配置区，按需修改】=====================
# 设备清单
DEVICE_LIST = [
    {
        "device_type": "huawei",
        "ip": "192.168.1.1",
        "username": "admin",
        "password": "Admin@123",
        "port": 22,
    }
]

# 巡检命令集合
CHECK_COMMANDS = [
    "display version",
    "display device",
    "display cpu-usage",
    "display memory-usage",
    "display environment",
    "display power",
    "display fan",
    "display interface brief",
    "display alarm active",
    "display logbuffer",
]

# 钉钉机器人配置
DING_WEBHOOK = "[https://oapi.dingtalk.com/robot/send?access_token=XXX](https://oapi.dingtalk.com/robot/send?access_token=XXX)"
DING_SECRET = "SECXXX"# 加签密钥，不加签留空
AT_ALL = False# 是否@所有人
# ==============================================================

# 正则匹配规则
ALARM_PATTERN = re.compile(r"(Critical|Major)", re.IGNORECASE)
CPU_PATTERN = re.compile(r"five seconds:\s*(\d+)%", re.IGNORECASE)
DOWN_PATTERN = re.compile(r"Down", re.IGNORECASE)


def ding_push(content: str):
    """钉钉Markdown消息推送，支持加签"""
    timestamp = str(round(time.time() * 1000))
    url = DING_WEBHOOK

    # 加签逻辑
    if DING_SECRET:
        sign_str = f"{timestamp}\n{DING_SECRET}"
        hmac_code = hmac.new(DING_SECRET.encode(), sign_str.encode(), hashlib.sha256).digest()
        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
        url = f"{DING_WEBHOOK}&timestamp={timestamp}&sign={sign}"

    payload = {
        "msgtype": "markdown",
        "markdown": {
            "title": "华为交换机巡检告警",
            "text": content
        },
        "at": {"isAtAll": AT_ALL}
    }
    try:
        res = requests.post(url, json=payload, timeout=10)
        print("钉钉推送结果：", res.json())
    except Exception as e:
        print("钉钉推送失败：", str(e))


def check_device(dev_info):
    ip = dev_info["ip"]
    alarm_flag = False
    alarm_msg = []
    report_text = f"\n===== 设备 {ip} 巡检报告 =====\n"

    try:
        with ConnectHandler(**dev_info) as conn:
            print(f"[{ip}] 连接成功，开始巡检")
            for cmd in CHECK_COMMANDS:
                output = conn.send_command(cmd)
                report_text += f"\n--- {cmd} ---\n{output}\n"

                # 识别活跃告警
                if cmd == "display alarm active":
                    if ALARM_PATTERN.search(output):
                        alarm_flag = True
                        alarm_msg.append(f"【{ip}】存在紧急/重要告警！")
                        alarm_msg.append(output.strip())

                # CPU利用率判断
                if cmd == "display cpu-usage":
                    cpu_match = CPU_PATTERN.search(output)
                    if cpu_match:
                        cpu_val = int(cpu_match.group(1))
                        if cpu_val > 80:
                            alarm_flag = True
                            alarm_msg.append(f"【{ip}】CPU使用率过高：{cpu_val}%")

                # 端口Down检测
                if cmd == "display interface brief":
                    if DOWN_PATTERN.search(output):
                        alarm_flag = True
                        alarm_msg.append(f"【{ip}】存在端口Down")

        # 保存巡检报告到本地txt
        with open(f"huawei_{ip}_report.txt", "w", encoding="utf-8") as f:
            f.write(report_text)

        # 存在异常推送钉钉告警
        if alarm_flag:
            ding_content = "### 🚨 网络设备巡检异常告警\n" + "\n".join(alarm_msg)
            ding_push(ding_content)
        else:
            print(f"[{ip}] 无异常")

    except Exception as e:
        err_info = f"【{ip}】设备连接失败：{str(e)}"
        print(err_info)
        ding_push(f"### ❌ 设备巡检失败\n{err_info}")


if __name__ == "__main__":
    # 多线程并发，最大并发5台设备，按需调整
    with ThreadPoolExecutor(max_workers=5) as pool:
        pool.map(check_device, DEVICE_LIST)

脚本功能说明

1. 多线程并发巡检：默认最大并发5台，避免SSH会话占满，提升批量巡检速度
2. 自动异常判断
• 识别display alarm active中Critical、Major级别告警
• CPU 5秒利用率超过80%触发告警
• 检测接口状态Down
3. 钉钉告警推送：使用Markdown格式推送告警；设备SSH连接失败也会推送报错信息
4. 本地日志留存：每台设备单独生成huawei_IP_report.txt，保存完整CLI回显用于事后排查

常见踩坑点 

1. 分页 ---- More ----：Netmiko内置华为驱动自动处理分页，不要使用raw_send_command
2. SSH算法兼容：老旧VRP设备SSH加密算法弱，ConnectHandler可增加参数ssh_file_transfer=False
3. VRP版本差异：VRP5与VRP8命令输出格式略有区别，正则规则需要微调适配
4. 并发数量：不要设置过大并发，建议5~10以内，防止设备SSH会话耗尽
5. 密码特殊字符：账号密码含特殊符号时，Python字符串注意转义

后续扩展方向

• 读取Excel设备清单，不用硬编码设备列表
• 生成带颜色标记的HTML巡检报告
• 巡检数据写入SQLite数据库，做CPU、内存历史趋势监控
• 增加风扇、电源、温度、端口错包正则判断
• 配置定时任务（Linux crontab / Windows任务计划程序），实现每日自动巡检