第318篇：用 Python 开发自动化巡检脚本

关键词：Python、自动化巡检、设备健康检查、Netmiko、Paramiko、巡检报告、告警规则

一、引言

第317篇介绍了华为 MDA 框架的设备管理方式。在实际运维中，每日/每周对网络设备进行巡检是必不可少的工作：检查设备 CPU/内存利用率、接口错误计数、路由数量、协议状态、温度等指标，及时发现潜在故障。

传统的手工巡检（逐台登录设备执行 display 命令）耗时耗力、容易遗漏。用 Python 开发自动化巡检脚本，可以批量采集、自动分析、生成报告，大幅提升运维效率。

二、核心概念

2.1 巡检项目分类

2.2 巡检流程

2.3 巡检工具选型

三、技术细节

3.1 基于 Netmiko 的基础巡检

$ pip install netmiko jinja2 pyyaml

3.1.1 单设备巡检框架

# scripts/device_inspection.py
from netmiko import ConnectHandler
import re
import json
from datetime import datetime

class DeviceInspector:
    """设备巡检器"""

    # 巡检命令定义（华为设备）
    INSPECTION_COMMANDS = {
        "cpu_memory": "display device",
        "temperature": "display temperature",
        "interface_status": "display interface brief",
        "bgp_summary": "display bgp peer all",
        "ospf_summary": "display ospf peer brief",
        "isis_summary": "display isis peer brief",
        "route_count": "display ip routing-table statistics",
        "arp_count": "display arp statistics",
        "uptime": "display version",
        "ntp_status": "display ntp status",
        "clock": "display clock",
        "board_status": "display device",
        "power_status": "display power-supply",
        "fan_status": "display fan",
    }

    # 告警阈值
    THRESHOLDS = {
        "cpu_usage_warning": 70,
        "cpu_usage_critical": 90,
        "memory_usage_warning": 80,
        "memory_usage_critical": 95,
        "temperature_warning": 60,
        "temperature_critical": 75,
        "interface_error_warning": 100,
        "interface_down_critical": True,  # 接口 Down 即告警
    }

    def __init__(self, device):
        self.device = device
        self.results = {
            "device": device.get("hostname", device["host"]),
            "ip": device["host"],
            "timestamp": datetime.now().isoformat(),
            "checks": {},
            "alerts": [],
        }
        self.connection = None

    def connect(self):
        """建立 SSH 连接"""
        self.connection = ConnectHandler(**self.device)
        return self

    def disconnect(self):
        """断开连接"""
        if self.connection:
            self.connection.disconnect()

    def execute_command(self, command, description=""):
        """执行单条巡检命令"""
        try:
            output = self.connection.send_command(command, read_timeout=30)
            return {"output": output, "status": "success"}
        except Exception as e:
            return {"output": str(e), "status": "error", "error": str(e)}

    def run_all_checks(self):
        """执行全部巡检项"""
        for check_name, command in self.INSPECTION_COMMANDS.items():
            print(f"  执行: {check_name}...")
            result = self.execute_command(command)
            self.results["checks"][check_name] = result

            # 解析并检查阈值
            self._analyze_check(check_name, result)

        return self.results

    def _analyze_check(self, check_name, result):
        """分析巡检结果并生成告警"""
        if result["status"] != "success":
            self._add_alert(check_name, "critical", f"命令执行失败: {result.get('error')}")
            return

        output = result["output"]

        if check_name == "cpu_memory":
            self._check_cpu_memory(output)
        elif check_name == "temperature":
            self._check_temperature(output)
        elif check_name == "interface_status":
            self._check_interfaces(output)
        elif check_name == "bgp_summary":
            self._check_bgp_peers(output)
        elif check_name == "ospf_summary":
            self._check_ospf_peers(output)

    def _check_cpu_memory(self, output):
        """检查 CPU 和内存利用率"""
        # 解析 CPU 使用率
        cpu_match = re.search(r'CPU\s+utilization:\s+(\d+)%', output, re.IGNORECASE)
        if cpu_match:
            cpu = int(cpu_match.group(1))
            if cpu >= self.THRESHOLDS["cpu_usage_critical"]:
                self._add_alert("cpu_usage", "critical", f"CPU 利用率 {cpu}%")
            elif cpu >= self.THRESHOLDS["cpu_usage_warning"]:
                self._add_alert("cpu_usage", "warning", f"CPU 利用率 {cpu}%")

        # 解析内存使用率
        mem_match = re.search(r'Memory\s+utilization:\s+(\d+)%', output, re.IGNORECASE)
        if mem_match:
            mem = int(mem_match.group(1))
            if mem >= self.THRESHOLDS["memory_usage_critical"]:
                self._add_alert("memory_usage", "critical", f"内存利用率 {mem}%")
            elif mem >= self.THRESHOLDS["memory_usage_warning"]:
                self._add_alert("memory_usage", "warning", f"内存利用率 {mem}%")

    def _check_temperature(self, output):
        """检查温度"""
        temp_pattern = re.compile(r'(\d+)\s*°?\s*C', re.IGNORECASE)
        temps = temp_pattern.findall(output)

        for temp_str in temps:
            temp = int(temp_str)
            if temp >= self.THRESHOLDS["temperature_critical"]:
                self._add_alert("temperature", "critical", f"温度 {temp}°C")
            elif temp >= self.THRESHOLDS["temperature_warning"]:
                self._add_alert("temperature", "warning", f"温度 {temp}°C")

    def _check_interfaces(self, output):
        """检查接口状态"""
        lines = output.strip().split("\n")[2:]  # 跳过表头

        for line in lines:
            parts = line.split()
            if len(parts) < 4:
                continue

            iface = parts[0]
            phys_status = parts[1] if len(parts) > 1 else ""
            proto_status = parts[2] if len(parts) > 2 else ""

            # 跳过管理接口和 Loopback
            if iface.lower().startswith(("lo", "me", "inloop")):
                continue

            # 检查接口 Down
            if proto_status.lower() == "down":
                self._add_alert("interface_down", "critical",
                              f"接口 {iface} 协议状态 Down")

            # 检查错误计数
            if len(parts) > 6:
                try:
                    errors = int(parts[6]) if parts[6].isdigit() else 0
                    if errors > self.THRESHOLDS["interface_error_warning"]:
                        self._add_alert("interface_errors", "warning",
                                      f"接口 {iface} 错误计数 {errors}")
                except (ValueError, IndexError):
                    pass

    def _check_bgp_peers(self, output):
        """检查 BGP 邻居状态"""
        lines = output.strip().split("\n")[2:]

        for line in lines:
            parts = line.split()
            if len(parts) < 3:
                continue

            peer_ip = parts[0]
            state = parts[-1] if parts else ""

            if state.upper() != "Established" and state.upper() != "ESTABLISHED":
                self._add_alert("bgp_peer", "critical",
                              f"BGP 对等体 {peer_ip} 状态: {state}")

    def _check_ospf_peers(self, output):
        """检查 OSPF 邻居状态"""
        lines = output.strip().split("\n")[2:]

        for line in lines:
            parts = line.split()
            if len(parts) < 4:
                continue

            neighbor = parts[0]
            state = parts[-1]

            if state.upper() not in ("Full", "FULL", "2-Way", "2-WAY"):
                self._add_alert("ospf_peer", "warning",
                              f"OSPF 邻居 {neighbor} 状态: {state}")

    def _add_alert(self, check_name, severity, message):
        """添加告警"""
        alert = {
            "check": check_name,
            "severity": severity,
            "message": message,
            "timestamp": datetime.now().isoformat(),
        }
        self.results["alerts"].append(alert)

    def __enter__(self):
        self.connect()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.disconnect()

3.1.2 多设备批量巡检

# scripts/batch_inspection.py
from device_inspection import DeviceInspector
import yaml
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime

def run_batch_inspection(inventory_file="devices/inventory.yaml", max_workers=10):
    """批量巡检入口"""
    with open(inventory_file, "r", encoding="utf-8") as f:
        inventory = yaml.safe_load(f)

    devices = inventory.get("devices", [])
    all_results = []
    total_alerts = {"critical": 0, "warning": 0, "info": 0}

    print(f"开始巡检 {len(devices)} 台设备...")
    print("=" * 60)

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(inspect_single_device, dev): dev
            for dev in devices
        }

        for future in as_completed(futures):
            device = futures[future]
            try:
                result = future.result()
                all_results.append(result)

                # 统计告警
                for alert in result.get("alerts", []):
                    severity = alert.get("severity", "info")
                    total_alerts[severity] = total_alerts.get(severity, 0) + 1

                # 输出摘要
                alert_count = len(result.get("alerts", []))
                status = "✓" if alert_count == 0 else f"⚠ {alert_count}告警"
                print(f"{status} {result['device']}: {alert_count} 告警")

            except Exception as e:
                all_results.append({
                    "device": device.get("hostname", device["host"]),
                    "ip": device["host"],
                    "timestamp": datetime.now().isoformat(),
                    "checks": {},
                    "alerts": [{
                        "check": "connection",
                        "severity": "critical",
                        "message": f"连接失败: {e}",
                    }],
                })
                total_alerts["critical"] += 1
                print(f"✗ {device.get('hostname', device['host'])}: 连接失败")

    return all_results, total_alerts

def inspect_single_device(device):
    """巡检单台设备"""
    with DeviceInspector(device) as inspector:
        return inspector.run_all_checks()

# 执行批量巡检
results, alerts = run_batch_inspection()

# 保存结果
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
with open(f"inspection_reports/inspection_{timestamp}.json", "w",
          encoding="utf-8") as f:
    json.dump({
        "timestamp": datetime.now().isoformat(),
        "summary": alerts,
        "results": results,
    }, f, indent=2, ensure_ascii=False)

print(f"\n巡检完成!")
print(f"  严重告警: {alerts['critical']}")
print(f"  警告: {alerts['warning']}")

3.2 巡检报告生成（HTML 格式）

# scripts/inspection_report.py
from jinja2 import Template
import json

HTML_REPORT_TEMPLATE = Template("""
<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <title>网络巡检报告 - {{ timestamp }}</title>
    <style>
        body { font-family: 'Microsoft YaHei', Arial, sans-serif; margin: 20px; }
        .header { background: #2c3e50; color: white; padding: 20px; border-radius: 8px; }
        .summary { display: flex; gap: 20px; margin: 20px 0; }
        .summary-card {
            flex: 1; padding: 15px; border-radius: 8px;
            text-align: center; color: white;
        }
        .critical { background: #e74c3c; }
        .warning { background: #f39c12; }
        .normal { background: #27ae60; }
        .device-card {
            border: 1px solid #ddd; border-radius: 8px;
            margin: 10px 0; padding: 15px;
        }
        .device-card.has-alert { border-left: 4px solid #e74c3c; }
        .device-card.ok { border-left: 4px solid #27ae60; }
        .alert-item {
            padding: 8px; margin: 5px 0; border-radius: 4px;
            background: #ffeaea;
        }
        .alert-item.warning { background: #fff3cd; }
        table { width: 100%; border-collapse: collapse; margin: 10px 0; }
        th, td { padding: 8px; border: 1px solid #ddd; text-align: left; }
        th { background: #f5f5f5; }
    </style>
</head>
<body>
    <div class="header">
        <h1>🔍 网络巡检报告</h1>
        <p>巡检时间: {{ timestamp }}</p>
        <p>设备总数: {{ results|length }} 台</p>
    </div>

    <div class="summary">
        <div class="summary-card critical">
            <h2>{{ summary.critical }}</h2>
            <p>严重告警</p>
        </div>
        <div class="summary-card warning">
            <h2>{{ summary.warning }}</h2>
            <p>警告</p>
        </div>
        <div class="summary-card normal">
            <h2>{{ normal_count }}</h2>
            <p>正常设备</p>
        </div>
    </div>

    {% for result in results %}
    <div class="device-card {% if result.alerts %}has-alert{% else %}ok{% endif %}">
        <h3>{{ result.device }} ({{ result.ip }})</h3>
        {% if result.alerts %}
            <p><strong>{{ result.alerts|length }} 条告警</strong></p>
            {% for alert in result.alerts %}
            <div class="alert-item {% if alert.severity == 'warning' %}warning{% endif %}">
                <strong>[{{ alert.severity|upper }}]</strong>
                {{ alert.check }}: {{ alert.message }}
            </div>
            {% endfor %}
        {% else %}
            <p>✅ 所有检查项正常</p>
        {% endif %}
    </div>
    {% endfor %}

    <div style="margin-top: 20px; color: #999; text-align: center;">
        <p>报告生成时间: {{ generation_time }}</p>
    </div>
</body>
</html>
""")

def generate_html_report(results, summary, output_file="inspection_report.html"):
    """生成 HTML 巡检报告"""
    normal_count = sum(1 for r in results if not r.get("alerts"))

    html = HTML_REPORT_TEMPLATE.render(
        timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        results=results,
        summary=summary,
        normal_count=normal_count,
        generation_time=datetime.now().isoformat(),
    )

    with open(output_file, "w", encoding="utf-8") as f:
        f.write(html)

    print(f"HTML 报告已生成: {output_file}")

3.3 巡检告警通知

# scripts/inspection_alerts.py
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import json

def send_alert_email(alerts, recipients=["admin@example.com"],
                     sender="inspection@example.com"):
    """发送告警邮件"""
    if not alerts:
        return

    critical_alerts = [a for a in alerts if a["severity"] == "critical"]
    warning_alerts = [a for a in alerts if a["severity"] == "warning"]

    if not critical_alerts and not warning_alerts:
        return

    # 构建邮件正文
    body = "<h2>网络巡检告警通知</h2>"
    body += f"<p>巡检时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>"

    if critical_alerts:
        body += "<h3>🔴 严重告警</h3><ul>"
        for alert in critical_alerts:
            body += f"<li>{alert['device']}: {alert['message']}</li>"
        body += "</ul>"

    if warning_alerts:
        body += "<h3>🟡 警告</h3><ul>"
        for alert in warning_alerts:
            body += f"<li>{alert['device']}: {alert['message']}</li>"
        body += "</ul>"

    msg = MIMEMultipart("alternative")
    msg["Subject"] = f"网络巡检告警 - {len(critical_alerts)} 严重 / {len(warning_alerts)} 警告"
    msg["From"] = sender
    msg["To"] = ", ".join(recipients)
    msg.attach(MIMEText(body, "html", "utf-8"))

    try:
        server = smtplib.SMTP("smtp.example.com", 587)
        server.starttls()
        server.login(sender, "password")
        server.sendmail(sender, recipients, msg.as_string())
        server.quit()
        print(f"告警邮件已发送至: {recipients}")
    except Exception as e:
        print(f"邮件发送失败: {e}")

def send_wechat_alert(alerts, webhook_url):
    """发送企业微信告警"""
    critical_alerts = [a for a in alerts if a["severity"] == "critical"]

    if not critical_alerts:
        return

    content = "🔴 网络巡检严重告警\n\n"
    for alert in critical_alerts:
        content += f"• {alert['device']}: {alert['message']}\n"

    payload = {
        "msgtype": "text",
        "text": {"content": content},
    }

    import requests
    try:
        requests.post(webhook_url, json=payload, timeout=10)
    except Exception as e:
        print(f"企微告警发送失败: {e}")

3.4 定时巡检调度

# scripts/scheduled_inspection.py
import schedule
import time
from batch_inspection import run_batch_inspection
from inspection_report import generate_html_report
from inspection_alerts import send_alert_email

def daily_inspection_job():
    """每日巡检任务"""
    print(f"[{datetime.now()}] 开始每日巡检...")

    results, summary = run_batch_inspection()

    # 生成报告
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    report_file = f"inspection_reports/daily_{timestamp}.html"
    generate_html_report(results, summary, report_file)

    # 收集所有告警
    all_alerts = []
    for r in results:
        for alert in r.get("alerts", []):
            alert["device"] = r["device"]
            all_alerts.append(alert)

    # 发送告警
    if summary["critical"] > 0:
        send_alert_email(all_alerts)

    print(f"巡检完成: {summary}")

# 调度配置
schedule.every().day.at("09:00").do(daily_inspection_job)
schedule.every().monday.at("08:00").do(lambda: print("周报提醒"))

print("巡检调度已启动...")
while True:
    schedule.run_pending()
    time.sleep(60)

四、实战配置/案例分析

4.1 案例一：综合巡检——接口 + 路由 + 协议

# scripts/comprehensive_inspection.py
from netmiko import ConnectHandler
import re

class ComprehensiveInspector:
    """综合巡检器"""

    def __init__(self, device):
        self.device = device
        self.connection = None
        self.results = {}

    def connect(self):
        self.connection = ConnectHandler(**self.device)

    def inspect_all(self):
        """执行综合巡检"""
        checks = {
            "interface_errors": self._check_interface_errors,
            "interface_flapping": self._check_interface_flapping,
            "bgp_peer_count": self._check_bgp_peer_count,
            "route_count": self._check_route_count,
            "arp_table": self._check_arp_table,
            "disk_usage": self._check_disk_usage,
        }

        for name, check_func in checks.items():
            try:
                self.results[name] = check_func()
            except Exception as e:
                self.results[name] = {"status": "error", "error": str(e)}

        return self.results

    def _check_interface_errors(self):
        """检查接口错误计数"""
        output = self.connection.send_command("display interface")
        errors = {}

        for line in output.split("\n"):
            # 匹配接口名
            iface_match = re.match(r'^(\w+/\d+/\d+)', line)
            current_iface = iface_match.group(1) if iface_match else None

            # 匹配错误计数
            if "Input errors:" in line:
                count_match = re.search(r'Input errors:\s+(\d+)', line)
                if count_match and current_iface:
                    count = int(count_match.group(1))
                    if count > 0:
                        errors[current_iface] = count

        return {
            "interfaces_with_errors": errors,
            "total_affected": len(errors),
            "status": "warning" if errors else "normal",
        }

    def _check_bgp_peer_count(self):
        """检查 BGP 对等体数量和状态"""
        output = self.connection.send_command("display bgp peer all")
        peers = {"established": 0, "other": 0, "details": []}

        for line in output.split("\n")[2:]:
            parts = line.split()
            if len(parts) >= 3:
                peer_ip = parts[0]
                state = parts[-1]
                if state.upper() == "ESTABLISHED":
                    peers["established"] += 1
                else:
                    peers["other"] += 1
                    peers["details"].append(
                        f"{peer_ip}: {state}"
                    )

        return {
            **peers,
            "status": "critical" if peers["other"] > 0 else "normal",
        }

    def _check_route_count(self):
        """检查路由表规模"""
        output = self.connection.send_command("display ip routing-table statistics")
        total_match = re.search(r'Total number:\s+(\d+)', output)

        if total_match:
            total = int(total_match.group(1))
            status = "warning" if total > 100000 else "normal"
            return {"total_routes": total, "status": status}

        return {"status": "error", "error": "无法解析路由数量"}

    def _check_disk_usage(self):
        """检查磁盘使用率"""
        output = self.connection.send_command("display filesystem")
        usage_match = re.search(r'Used:\s+(\d+)%', output)

        if usage_match:
            usage = int(usage_match.group(1))
            status = "critical" if usage > 90 else ("warning" if usage > 80 else "normal")
            return {"usage_percent": usage, "status": status}

        return {"status": "error", "error": "无法解析磁盘使用率"}

# 使用
inspector = ComprehensiveInspector({
    "device_type": "huawei",
    "host": "10.1.1.1",
    "username": "admin",
    "password": "HuaWei@123",
})
inspector.connect()
results = inspector.inspect_all()
print(json.dumps(results, indent=2, ensure_ascii=False))
inspector.connection.disconnect()

4.2 案例二：故障案例——巡检脚本导致设备过载

故障现象：每日巡检时，50 台设备同时收到 SSH 连接请求，导致多台设备 CPU 飙升至 90% 以上，影响业务。

根因分析：

1. 批量巡检时并发数过高（默认 max_workers=10），设备同时处理多个 SSH 会话
2. 每条巡检命令都是独立的 SSH 交互，I/O 开销大
3. 巡检时间窗口集中在同一时刻

解决方案：

# scripts/optimized_inspection.py
import time
from concurrent.futures import ThreadPoolExecutor

def optimized_batch_inspection(devices, max_workers=5, delay_between=2):
    """优化版批量巡检"""
    results = []

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = []
        for device in devices:
            # 限制并发，逐批提交
            futures.append(
                executor.submit(inspect_single_device, device)
            )
            time.sleep(delay_between)  # 设备间延迟，避免同时连接

        for future in as_completed(futures):
            results.append(future.result())

    return results

# 优化参数
# - max_workers: 5（原 10）
# - delay_between: 2 秒（新增）
# - 巡检时间：避开业务高峰期（凌晨 2:00）

预防措施：

五、总结

思考题

1. 使用 Netmiko 和 NETCONF 进行巡检，各有什么优势和局限？什么场景下必须用 Netmiko？
2. 如果设备返回的 CLI 输出格式因版本不同而变化，如何保证巡检脚本的兼容性？
3. 大规模网络（1000+ 设备）巡检时，串行、并行和分时巡检各有什么优缺点？

下篇预告：第319篇：配置备份与合规检查自动化工具