第324篇：自动化巡检与智能报告生成

关键词

自动化巡检、智能报告、网络健康检查、巡检脚本、HTML 报告、定时巡检、巡检模板

一、自动化巡检概述

1.1 为什么要自动化巡检

传统巡检 vs 自动化巡检：

  传统巡检：
  ┌──────────────────────────────────────────┐
  │  工程师逐台 SSH 登录设备                   │
  │  ├─ show version                          │
  │  ├─ display interface brief               │
  │  ├─ display bgp peer                      │
  │  ├─ display ospf peer                     │
  │  └─ ...（可能漏掉某个关键命令）            │
  │                                           │
  │  结果记录在 Excel 或 Word 中               │
  │  ├─ 数据靠人工复制粘贴                    │
  │  ├─ 容易出错和漏项                        │
  │  └─ 无法做趋势分析                        │
  │                                           │
  │  100 台设备 × 每天巡检 = 4-6 小时         │
  │  └─ 每天花半天时间做重复劳动              │
  └──────────────────────────────────────────┘

  自动化巡检：
  ┌──────────────────────────────────────────┐
  │  脚本自动登录所有设备                       │
  │  ├─ 标准化命令集                          │
  │  ├─ 并发执行（100 台设备 3-5 分钟）       │
  │  └─ 不遗漏任何命令                        │
  │                                           │
  │  自动生成 HTML/PDF 报告                    │
  │  ├─ 数据已格式化                          │
  │  ├─ 异常数据自动标记                      │
  │  ├─ 合规检查集成                          │
  │  └─ 趋势变化一目了然                      │
  │                                           │
  │  100 台设备 × 每天巡检 = 5 分钟           │
  │  └─ 工程师只需关注异常部分                │
  └──────────────────────────────────────────┘

1.2 巡检架构

自动化巡检系统架构：

二、巡检模板定义

2.1 分级巡检

巡检级别设计：

  日常巡检（每天）：
  ┌──────────────────────────────────────────┐
  │  目标：快速发现异常                        │
  │  耗时：5 分钟（100 台设备）                │
  │                                           │
  │  检查项：                                  │
  │  ┌─ 设备可达性（ping 设备管理 IP）        │
  │  ├─ 关键接口状态（上联/下联口）           │
  │  ├─ BGP/OSPF 邻居状态                    │
  │  ├─ CPU/内存利用率                       │
  │  └─ 环境温度（异常高温告警）              │
  └──────────────────────────────────────────┘

  深度巡检（每周）：
  ┌──────────────────────────────────────────┐
  │  目标：全面体检                            │
  │  耗时：15 分钟（100 台设备）               │
  │                                           │
  │  检查项（在日常巡检基础上增加）：            │
  │  ┌─ 所有接口状态和错误计数                │
  │  ├─ 路由表检查（数量变化/黑洞路由）       │
  │  ├─ 日志错误检查                         │
  │  ├─ NTP 同步状态                         │
  │  ├─ LLDP 邻居一致性检查                  │
  │  ├─ VRRP/堆叠状态                       │
  │  └─ QoS 队列丢弃检查                     │
  └──────────────────────────────────────────┘

  全量巡检（每月）：
  ┌──────────────────────────────────────────┐
  │  目标：配置基线审计                        │
  │  耗时：30 分钟（100 台设备）               │
  │                                           │
  │  检查项（在深度巡检基础上增加）：            │
  │  ┌─ 全量 running-config 采集             │
  │  ├─ 配置漂移检查（与基线对比）            │
  │  ├─ 合规检查（安全基线/最佳实践）         │
  │  ├─ 版本审计（推荐版本/已知漏洞）         │
  │  ├─ ACL 规则审计                         │
  │  └─ 设备证书有效期检查                   │
  └──────────────────────────────────────────┘

2.2 巡检命令模板

#!/usr/bin/env python3
# patrol_templates.py — 巡检命令模板

from dataclasses import dataclass
from typing import List, Dict, Optional


@dataclass
class PatrolCommand:
    """巡检命令定义"""
    name: str                     # 命令名称（如 bgp_peer）
    cli_command: str              # 实际 CLI 命令
    description: str              # 检查目的
    critical: bool = False        # 是否关键检查
    parser: str = "text"          # 解析方式

    def __post_init__(self):
        self.cli_command = self.cli_command.strip()


# =============================================
# 华为 VRP 设备巡检命令模板
# =============================================

HUAWEI_VRP_DAILY = [
    PatrolCommand(
        name="device_uptime",
        cli_command="display version",
        description="设备运行时间",
        critical=False,
        parser="uptime",
    ),
    PatrolCommand(
        name="cpu_usage",
        cli_command="display cpu-usage",
        description="CPU 利用率",
        critical=True,
        parser="cpu",
    ),
    PatrolCommand(
        name="memory_usage",
        cli_command="display memory-usage",
        description="内存利用率",
        critical=True,
        parser="memory",
    ),
    PatrolCommand(
        name="interface_brief",
        cli_command="display interface brief",
        description="接口状态概览",
        critical=True,
        parser="interface_brief",
    ),
    PatrolCommand(
        name="bgp_peer",
        cli_command="display bgp peer",
        description="BGP 邻居状态",
        critical=True,
        parser="bgp_peer",
    ),
    PatrolCommand(
        name="ospf_peer",
        cli_command="display ospf peer brief",
        description="OSPF 邻居状态",
        critical=True,
        parser="ospf_peer",
    ),
    PatrolCommand(
        name="temperature",
        cli_command="display device temperature",
        description="设备温度",
        critical=False,
        parser="temperature",
    ),
]

HUAWEI_VRP_WEEKLY = HUAWEI_VRP_DAILY + [
    PatrolCommand(
        name="interface_errors",
        cli_command="display interface",
        description="接口错误计数",
        critical=False,
        parser="interface_errors",
    ),
    PatrolCommand(
        name="routing_table",
        cli_command="display ip routing-table statistics",
        description="路由表统计",
        critical=False,
        parser="routing_stats",
    ),
    PatrolCommand(
        name="log_buffer",
        cli_command="display logbuffer",
        description="日志错误检查",
        critical=False,
        parser="log",
    ),
    PatrolCommand(
        name="ntp_status",
        cli_command="display ntp status",
        description="NTP 同步状态",
        critical=False,
        parser="ntp",
    ),
    PatrolCommand(
        name="lldp_neighbor",
        cli_command="display lldp neighbor brief",
        description="LLDP 邻居",
        critical=False,
        parser="lldp",
    ),
]

HUAWEI_VRP_MONTHLY = HUAWEI_VRP_WEEKLY + [
    PatrolCommand(
        name="running_config",
        cli_command="display current-configuration",
        description="当前配置",
        critical=False,
        parser="config",
    ),
    PatrolCommand(
        name="acl_config",
        cli_command="display acl all",
        description="ACL 规则",
        critical=False,
        parser="acl",
    ),
    PatrolCommand(
        name="certificate",
        cli_command="display pki certificate",
        description="证书状态",
        critical=False,
        parser="cert",
    ),
]


# =============================================
# 思科 IOS 设备巡检命令模板
# =============================================

CISCO_IOS_DAILY = [
    PatrolCommand(
        name="device_uptime",
        cli_command="show version | include uptime",
        description="设备运行时间",
        critical=False,
    ),
    PatrolCommand(
        name="cpu_usage",
        cli_command="show processes cpu | include CPU",
        description="CPU 利用率",
        critical=True,
    ),
    PatrolCommand(
        name="memory_usage",
        cli_command="show memory statistics",
        description="内存利用率",
        critical=True,
    ),
    PatrolCommand(
        name="interface_brief",
        cli_command="show ip interface brief",
        description="接口状态概览",
        critical=True,
    ),
    PatrolCommand(
        name="bgp_peer",
        cli_command="show bgp summary",
        description="BGP 邻居状态",
        critical=True,
    ),
    PatrolCommand(
        name="ospf_peer",
        cli_command="show ip ospf neighbor",
        description="OSPF 邻居状态",
        critical=True,
    ),
]


# =============================================
# 巡检模板管理器
# =============================================

PATROL_TEMPLATES = {
    "huawei_vrp": {
        "daily": HUAWEI_VRP_DAILY,
        "weekly": HUAWEI_VRP_WEEKLY,
        "monthly": HUAWEI_VRP_MONTHLY,
    },
    "cisco_ios": {
        "daily": CISCO_IOS_DAILY,
        "weekly": CISCO_IOS_DAILY,  # 简化
        "monthly": CISCO_IOS_DAILY,
    },
}


def get_patrol_commands(
    device_type: str,
    patrol_level: str = "daily",
) -> List[PatrolCommand]:
    """获取指定级别和厂商的巡检命令"""
    template = PATROL_TEMPLATES.get(device_type, {})
    return template.get(patrol_level, [])

三、巡检执行引擎

3.1 核心引擎

#!/usr/bin/env python3
# patrol_engine.py — 巡检执行引擎

from netmiko import ConnectHandler
from concurrent.futures import (
    ThreadPoolExecutor, as_completed
)
from typing import Dict, List, Any, Optional
from datetime import datetime
import time
import json
import logging

from patrol_templates import (
    PatrolCommand, get_patrol_commands
)

logger = logging.getLogger(__name__)


class PatrolResult:
    """单台设备巡检结果"""

    def __init__(self, hostname: str, device_type: str):
        self.hostname = hostname
        self.device_type = device_type
        self.timestamp = datetime.now().isoformat()
        self.commands: Dict[str, Any] = {}
        self.errors: List[str] = []
        self.alerts: List[Dict] = []
        self.alive = False

    def add_command_result(
        self, cmd: PatrolCommand, output: str
    ):
        """添加命令执行结果"""
        self.commands[cmd.name] = {
            "command": cmd.cli_command,
            "output": output,
            "description": cmd.description,
        }

    def add_alert(
        self, level: str, message: str, detail: str = ""
    ):
        """添加告警"""
        self.alerts.append({
            "level": level,
            "message": message,
            "detail": detail,
        })

    def to_dict(self) -> Dict:
        return {
            "hostname": self.hostname,
            "device_type": self.device_type,
            "timestamp": self.timestamp,
            "alive": self.alive,
            "commands": self.commands,
            "errors": self.errors,
            "alerts": self.alerts,
            "alert_count": len(self.alerts),
        }


class PatrolEngine:
    """巡检执行引擎"""

    def __init__(
        self,
        max_workers: int = 10,
        command_timeout: int = 30,
    ):
        self.max_workers = max_workers
        self.command_timeout = command_timeout

    def patrol_device(
        self,
        device_info: Dict,
        commands: List[PatrolCommand],
    ) -> PatrolResult:
        """巡检单台设备"""
        result = PatrolResult(
            hostname=device_info.get("hostname", ""),
            device_type=device_info.get("device_type", ""),
        )

        try:
            conn = ConnectHandler(**device_info)
            conn.enable()
            result.alive = True

            # 获取设备 hostname
            prompt = conn.find_prompt()
            result.hostname = prompt.rstrip("#>")

            # 执行每条巡检命令
            for cmd in commands:
                try:
                    output = conn.send_command(
                        cmd.cli_command,
                        read_timeout=self.command_timeout,
                    )
                    result.add_command_result(cmd, output)

                    # 异常检测
                    self._detect_anomalies(result, cmd, output)

                except Exception as e:
                    error_msg = (
                        f"命令 {cmd.name} 执行失败: {e}"
                    )
                    result.errors.append(error_msg)
                    logger.warning(
                        f"{result.hostname}: {error_msg}"
                    )

            conn.disconnect()

        except Exception as e:
            result.alive = False
            result.errors.append(f"设备连接失败: {e}")
            logger.error(
                f"{device_info.get('host')}: {e}"
            )

        return result

    def _detect_anomalies(
        self,
        result: PatrolResult,
        cmd: PatrolCommand,
        output: str,
    ):
        """检测异常"""
        # CPU 利用率过高
        if cmd.name == "cpu_usage":
            import re
            match = re.search(r"(\d+)%", output)
            if match:
                cpu = int(match.group(1))
                if cpu > 80:
                    result.add_alert(
                        "WARNING",
                        f"CPU 利用率过高: {cpu}%",
                        cmd.cli_command,
                    )
                elif cpu > 90:
                    result.add_alert(
                        "CRITICAL",
                        f"CPU 利用率极高: {cpu}%",
                        cmd.cli_command,
                    )

        # 内存利用率过高
        if cmd.name == "memory_usage":
            import re
            match = re.search(r"(\d+)%", output)
            if match:
                mem = int(match.group(1))
                if mem > 80:
                    result.add_alert(
                        "WARNING",
                        f"内存利用率过高: {mem}%",
                        cmd.cli_command,
                    )

        # 接口 down
        if cmd.name == "interface_brief":
            for line in output.splitlines():
                if "down" in line.lower() and "GigabitEthernet" in line:
                    result.add_alert(
                        "WARNING",
                        f"接口 DOWN: {line.strip()}",
                        cmd.cli_command,
                    )

        # BGP 邻居非 Established
        if cmd.name == "bgp_peer":
            if "Established" not in output:
                result.add_alert(
                    "CRITICAL",
                    "BGP 邻居异常 - 无 Established 状态的 peer",
                    cmd.cli_command,
                )

    def patrol_all(
        self,
        devices: List[Dict],
        patrol_level: str = "daily",
    ) -> Dict[str, PatrolResult]:
        """并发巡检所有设备"""
        results = {}

        with ThreadPoolExecutor(
            max_workers=self.max_workers
        ) as executor:
            futures = {}
            for dev in devices:
                device_type = dev.get("device_type", "huawei_vrp")
                commands = get_patrol_commands(
                    device_type, patrol_level
                )
                future = executor.submit(
                    self.patrol_device, dev, commands
                )
                futures[future] = dev

            for future in as_completed(futures):
                result = future.result()
                results[result.hostname] = result

        return results

3.2 异常检测器

class AnomalyDetector:
    """巡检异常检测器"""

    # 默认阈值
    THRESHOLDS = {
        "cpu_usage": {"warning": 70, "critical": 90},
        "memory_usage": {"warning": 70, "critical": 90},
        "temperature": {"warning": 45, "critical": 55},
        "interface_errors": {"warning": 100, "critical": 1000},
    }

    def __init__(
        self,
        thresholds: Optional[Dict] = None,
        baseline_data: Optional[Dict] = None,
    ):
        self.thresholds = thresholds or self.THRESHOLDS
        self.baseline = baseline_data or {}

    def check_cpu(self, cpu_value: float) -> Optional[Dict]:
        """检查 CPU 利用率"""
        threshold = self.thresholds.get("cpu_usage", {})
        if cpu_value >= threshold.get("critical", 90):
            return {
                "level": "CRITICAL",
                "message": f"CPU {cpu_value}% (阈值: {threshold.get('critical')}%)",
            }
        elif cpu_value >= threshold.get("warning", 70):
            return {
                "level": "WARNING",
                "message": f"CPU {cpu_value}% (阈值: {threshold.get('warning')}%)",
            }
        return None

    def check_memory(self, mem_value: float) -> Optional[Dict]:
        """检查内存利用率"""
        threshold = self.thresholds.get("memory_usage", {})
        if mem_value >= threshold.get("critical", 90):
            return {
                "level": "CRITICAL",
                "message": f"内存 {mem_value}% (阈值: {threshold.get('critical')}%)",
            }
        elif mem_value >= threshold.get("warning", 70):
            return {
                "level": "WARNING",
                "message": f"内存 {mem_value}% (阈值: {threshold.get('warning')}%)",
            }
        return None

    def check_interface_errors(
        self, error_count: int
    ) -> Optional[Dict]:
        """检查接口错误计数"""
        threshold = self.thresholds.get("interface_errors", {})
        if error_count >= threshold.get("critical", 1000):
            return {
                "level": "CRITICAL",
                "message": f"接口错误 {error_count} (阈值: {threshold.get('critical')})",
            }
        elif error_count >= threshold.get("warning", 100):
            return {
                "level": "WARNING",
                "message": f"接口错误 {error_count} (阈值: {threshold.get('warning')})",
            }
        return None

    def check_trend(
        self,
        metric: str,
        current_value: float,
        device: str,
    ) -> Optional[Dict]:
        """基于基线的趋势异常检测"""
        device_baseline = self.baseline.get(device, {})
        baseline_value = device_baseline.get(metric)

        if baseline_value is None:
            return None

        change_ratio = abs(
            current_value - baseline_value
        ) / baseline_value if baseline_value else 0

        if change_ratio > 0.3:  # 变化超过 30%
            return {
                "level": "WARNING",
                "message": (
                    f"{metric} 趋势异常: "
                    f"当前 {current_value} vs "
                    f"基线 {baseline_value} "
                    f"(变化 {change_ratio*100:.0f}%)"
                ),
            }
        return None

四、智能报告生成

4.1 HTML 报告生成

#!/usr/bin/env python3
# patrol_report.py — 巡检报告生成

from typing import Dict, List
from datetime import datetime
import json


class PatrolReportGenerator:
    """巡检报告生成器"""

    @staticmethod
    def generate_html(
        results: Dict[str, PatrolResult],
        output_file: str = "patrol_report.html",
    ):
        """生成 HTML 巡检报告"""

        # 统计
        total = len(results)
        alive = sum(1 for r in results.values() if r.alive)
        dead = total - alive
        total_alerts = sum(
            len(r.alerts) for r in results.values()
        )
        critical_alerts = sum(
            len([a for a in r.alerts
                 if a["level"] == "CRITICAL"])
            for r in results.values()
        )
        warning_alerts = sum(
            len([a for a in r.alerts
                 if a["level"] == "WARNING"])
            for r in results.values()
        )

        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        # 告警设备列表
        alerted_devices = [
            r for r in results.values()
            if r.alerts
        ]

        # 生成告警表格行
        alert_rows = ""
        for r in alerted_devices:
            for alert in r.alerts:
                level_class = (
                    "critical"
                    if alert["level"] == "CRITICAL"
                    else "warning"
                )
                alert_rows += f"""
            <tr class="{level_class}">
                <td>{r.hostname}</td>
                <td><span class="badge badge-{level_class}">{alert['level']}</span></td>
                <td>{alert['message']}</td>
            </tr>"""

        # 设备状态列表
        device_rows = ""
        for hostname, result in sorted(results.items()):
            status_icon = "✅" if result.alive else "❌"
            alert_count = len(result.alerts)
            alert_badge = (
                f'<span class="badge badge-danger">{alert_count}</span>'
                if alert_count > 0 else '<span class="badge badge-success">0</span>'
            )
            device_rows += f"""
            <tr>
                <td>{hostname}</td>
                <td>{result.device_type}</td>
                <td>{status_icon} {'在线' if result.alive else '离线'}</td>
                <td>{alert_badge}</td>
            </tr>"""

        # 完整 HTML
        html = f"""
<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="utf-8">
    <title>网络巡检报告</title>
    <style>
        * {{ margin: 0; padding: 0; box-sizing: border-box; }}
        body {{ font-family: 'Microsoft YaHei', Arial, sans-serif;
                background: #f5f5f5; padding: 20px; }}
        .header {{ background: linear-gradient(135deg, #1a237e, #283593);
                   color: white; padding: 30px; border-radius: 10px;
                   margin-bottom: 20px; }}
        .header h1 {{ font-size: 28px; margin-bottom: 10px; }}
        .summary {{ display: grid;
                    grid-template-columns: repeat(auto-fit, minmax(180px, 1fr));
                    gap: 15px; margin-bottom: 20px; }}
        .summary-card {{ background: white; padding: 20px; border-radius: 10px;
                         text-align: center; box-shadow: 0 2px 4px rgba(0,0,0,0.1); }}
        .summary-card .number {{ font-size: 36px; font-weight: bold; }}
        .summary-card .label {{ font-size: 14px; color: #666; margin-top: 5px; }}
        .green {{ color: #4CAF50; }} .red {{ color: #f44336; }}
        .orange {{ color: #ff9800; }}
        table {{ width: 100%; border-collapse: collapse;
                 background: white; border-radius: 10px;
                 overflow: hidden; box-shadow: 0 2px 4px rgba(0,0,0,0.1);
                 margin-bottom: 20px; }}
        th, td {{ padding: 12px 15px; text-align: left;
                  border-bottom: 1px solid #f0f0f0; }}
        th {{ background: #283593; color: white; font-weight: 500; }}
        tr:hover {{ background: #f8f9fa; }}
        .badge {{ padding: 4px 8px; border-radius: 4px; font-size: 12px;
                  font-weight: bold; }}
        .badge-danger {{ background: #ffebee; color: #c62828; }}
        .badge-warning {{ background: #fff3e0; color: #ef6c00; }}
        .badge-success {{ background: #e8f5e9; color: #2e7d32; }}
        .critical {{ background: #fff5f5; }}
        .warning {{ background: #fffbf0; }}
        .section-title {{ font-size: 20px; margin: 20px 0 10px 0;
                          color: #1a237e; }}
    </style>
</head>
<body>
    <div class="header">
        <h1>📡 网络巡检报告</h1>
        <p>巡检时间: {now}</p>
        <p>巡检级别: 日常巡检</p>
    </div>

    <div class="summary">
        <div class="summary-card">
            <div class="number">{total}</div>
            <div class="label">总设备数</div>
        </div>
        <div class="summary-card">
            <div class="number green">{alive}</div>
            <div class="label">在线设备</div>
        </div>
        <div class="summary-card">
            <div class="number red">{dead}</div>
            <div class="label">离线设备</div>
        </div>
        <div class="summary-card">
            <div class="number red">{critical_alerts}</div>
            <div class="label">严重告警</div>
        </div>
        <div class="summary-card">
            <div class="number orange">{warning_alerts}</div>
            <div class="label">警告告警</div>
        </div>
    </div>

    <h2 class="section-title">告警列表</h2>
    <table>
        <tr>
            <th>设备</th>
            <th>级别</th>
            <th>告警内容</th>
        </tr>
        {alert_rows if alert_rows else '<tr><td colspan="3" style="text-align:center; color:#666;">🎉 无告警</td></tr>'}
    </table>

    <h2 class="section-title">设备状态</h2>
    <table>
        <tr>
            <th>设备名称</th>
            <th>设备类型</th>
            <th>状态</th>
            <th>告警</th>
        </tr>
        {device_rows}
    </table>
</body>
</html>
"""
        with open(output_file, "w", encoding="utf-8") as f:
            f.write(html)

        print(f"巡检报告已生成: {output_file}")
        return output_file

4.2 JSON 结构化报告

class PatrolJSONReport:
    """JSON 格式巡检报告（便于程序消费）"""

    def __init__(self, results: Dict[str, PatrolResult]):
        self.results = results

    def generate(self) -> Dict:
        """生成结构化报告"""
        return {
            "report_time": datetime.now().isoformat(),
            "summary": {
                "total_devices": len(self.results),
                "alive": sum(
                    1 for r in self.results.values()
                    if r.alive
                ),
                "dead": sum(
                    1 for r in self.results.values()
                    if not r.alive
                ),
                "total_alerts": sum(
                    len(r.alerts)
                    for r in self.results.values()
                ),
                "critical_alerts": sum(
                    len([a for a in r.alerts
                         if a["level"] == "CRITICAL"])
                    for r in self.results.values()
                ),
                "warning_alerts": sum(
                    len([a for a in r.alerts
                         if a["level"] == "WARNING"])
                    for r in self.results.values()
                ),
            },
            "devices": {
                hostname: result.to_dict()
                for hostname, result in self.results.items()
            },
        }

    def save(self, output_file: str):
        """保存为 JSON 文件"""
        data = self.generate()
        with open(output_file, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
        print(f"结构化报告已生成: {output_file}")

五、定时调度与通知

5.1 集成定时任务

#!/usr/bin/env python3
# schedule_patrol.py — 定时巡检调度

import schedule
import time
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders
import requests
import json


class PatrolScheduler:
    """巡检任务调度器"""

    def __init__(self, patrol_engine, report_generator):
        self.engine = patrol_engine
        self.report_generator = report_generator
        self.notifiers = []

    def add_notifier(self, notifier):
        """添加通知器"""
        self.notifiers.append(notifier)

    def run_daily_patrol(self):
        """执行每日巡检"""
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 开始每日巡检")

        # 加载设备清单
        with open("inventory.json", "r") as f:
            devices = json.load(f)

        # 执行巡检
        results = self.engine.patrol_all(
            devices, patrol_level="daily"
        )

        # 生成报告
        report_file = self.report_generator.generate_html(
            results,
            f"patrol_report_{time.strftime('%Y%m%d')}.html",
        )

        # 发送通知
        for notifier in self.notifiers:
            notifier.notify(results, report_file)

        print(f"每日巡检完成")

    def schedule_jobs(self):
        """配置定时任务"""

        # 每天早上 8 点和晚上 8 点执行日常巡检
        schedule.every().day.at("08:00").do(
            self.run_daily_patrol
        )
        schedule.every().day.at("20:00").do(
            self.run_daily_patrol
        )

        while True:
            schedule.run_pending()
            time.sleep(60)

5.2 企业微信通知

class WeChatNotifier:
    """企业微信通知"""

    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url

    def notify(self, results, report_file=None):
        """发送巡检结果通知"""
        critical_count = sum(
            len([a for a in r.alerts
                 if a["level"] == "CRITICAL"])
            for r in results.values()
        )
        warning_count = sum(
            len([a for a in r.alerts
                 if a["level"] == "WARNING"])
            for r in results.values()
        )

        if critical_count == 0 and warning_count == 0:
            message = (
                f"✅ 巡检完成\n"
                f"时间: {time.strftime('%Y-%m-%d %H:%M')}\n"
                f"设备: {len(results)} 台\n"
                f"状态: 全部正常，无告警"
            )
        else:
            message = (
                f"⚠️ 巡检完成（发现异常）\n"
                f"时间: {time.strftime('%Y-%m-%d %H:%M')}\n"
                f"设备: {len(results)} 台\n"
                f"严重告警: {critical_count} 个\n"
                f"警告: {warning_count} 个\n"
                f"请查看详细报告"
            )

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

        try:
            requests.post(
                self.webhook_url, json=payload, timeout=10
            )
        except Exception as e:
            print(f"企业微信通知失败: {e}")

六、最佳实践总结

自动化巡检最佳实践：

  1. 分级巡检
  ┌─ 日常 5 分钟覆盖关键指标
  ├─ 每周深入检查所有设备
  ├─ 每月全量合规审计
  └─ 变更后立即触发深度巡检

  2. 智能告警
  ┌─ 阈值告警（CPU > 80% 告警）
  ├─ 趋势告警（相比基线变化 > 30%）
  ├─ 关联告警（BGP down + 接口 down 一起报）
  └─ 告警聚合（同一设备多条告警合并）

  3. 报告可视化
  ┌─ 概览卡：一目了然的全网状态
  ├─ 告警列表：突出问题
  ├─ 设备列表：逐台状态
  └─ 趋势图：关键指标随时间变化

  4. 多通道通知
  ┌─ 严重告警：短信/电话
  ├─ 警告告警：IM（企业微信/钉钉）
  ├─ 信息提示：邮件月报
  └─ 报告归档：Web 门户

  5. 持续完善
  ┌─ 每次故障复盘后补充巡检项
  ├─ 定期调整告警阈值
  ├─ 收集工程师反馈优化报告
  └─ 与 CMDB 联动，新设备自动纳入巡检

下篇预告：第325篇 — 网络变更自动化与回滚，深入网络变更自动化的流程设计、执行与一键回滚机制。

下篇预告：第325篇 — 网络变更自动化与回滚，深入网络变更自动化的流程设计、执行与一键回滚机制。