第314篇：Python 自动化脚本集

关键词

Python 脚本、配置备份、自动巡检、批量变更、合规检查、异常检测、运维工具集

一、脚本架构

1.1 项目结构

network_scripts/
├── config/
│   └── devices.yaml        # 设备清单
├── lib/
│   ├── __init__.py
│   ├── connector.py         # 连接管理
│   ├── logger.py            # 日志工具
│   └── utils.py             # 通用工具
├── scripts/
│   ├── backup_config.py     # 配置备份
│   ├── daily_inspection.py  # 每日巡检
│   ├── deploy_vlan.py       # VLAN 批量部署
│   ├── compliance_check.py  # 合规检查
│   ├── health_check.py      # 健康检查
│   └── anomaly_detect.py    # 异常检测
├── templates/               # Jinja2 模板
├── output/                  # 输出目录
├── logs/                    # 日志
└── requirements.txt         # 依赖

1.2 公共模块

#!/usr/bin/env python3
# lib/connector.py — 连接管理

"""
统一连接管理模块
支持 Netmiko / NAPALM / NCClient 三种后端
"""

from netmiko import ConnectHandler
from netmiko.exceptions import (
    NetmikoTimeoutException,
    NetmikoAuthenticationException,
)
import yaml
import os
from typing import Dict, Optional, List

class DeviceConnector:
    """设备连接管理器"""

    def __init__(self, config_path: str = "config/devices.yaml"):
        self.devices = self._load_config(config_path)

    def _load_config(self, path: str) -> Dict:
        """加载设备配置"""
        with open(path, "r", encoding="utf-8") as f:
            return yaml.safe_load(f)["devices"]

    def get_device(self, name: str) -> Optional[Dict]:
        """获取单台设备信息"""
        for dev in self.devices:
            if dev["name"] == name:
                return dev
        return None

    def get_devices_by_role(self, role: str) -> List[Dict]:
        """按角色获取设备"""
        return [d for d in self.devices if d.get("role") == role]

    def connect_netmiko(self, device: Dict) -> Optional[ConnectHandler]:
        """Netmiko 连接"""
        try:
            conn = ConnectHandler(
                device_type=device.get("device_type", "huawei"),
                host=device["mgmt_ip"],
                username=device.get("username", "admin"),
                password=device.get("password", ""),
                timeout=30,
            )
            return conn
        except NetmikoTimeoutException:
            print(f"  [超时] {device['name']}: 连接超时")
        except NetmikoAuthenticationException:
            print(f"  [认证失败] {device['name']}: 用户名/密码错误")
        return None

    def send_commands(
        self, conn: ConnectHandler, commands: List[str]
    ) -> str:
        """安全执行命令"""
        try:
            output = conn.send_config_set(commands)
            return output
        except Exception as e:
            return f"命令执行错误: {e}"

    def backup_config(self, conn: ConnectHandler) -> str:
        """获取运行配置"""
        return conn.send_command("display current-configuration")

    def close(self, conn: ConnectHandler):
        """关闭连接"""
        if conn:
            conn.disconnect()

#!/usr/bin/env python3
# lib/logger.py — 日志工具

import logging
import os
from datetime import datetime

def setup_logger(name: str, log_dir: str = "logs") -> logging.Logger:
    """配置日志器"""
    os.makedirs(log_dir, exist_ok=True)

    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)

    # 文件日志
    today = datetime.now().strftime("%Y%m%d")
    fh = logging.FileHandler(
        os.path.join(log_dir, f"{name}_{today}.log"),
        encoding="utf-8",
    )
    fh.setLevel(logging.DEBUG)

    # 控制台日志
    ch = logging.StreamHandler()
    ch.setLevel(logging.INFO)

    # 格式
    formatter = logging.Formatter(
        "%(asctime)s [%(levelname)s] %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )
    fh.setFormatter(formatter)
    ch.setFormatter(formatter)

    logger.addHandler(fh)
    logger.addHandler(ch)

    return logger

#!/usr/bin/env python3
# lib/utils.py — 通用工具

from datetime import datetime
import re

def parse_uptime(uptime_str: str) -> int:
    """解析运行时间字符串为秒数"""
    total_seconds = 0
    patterns = [
        (r"(\d+) year", 365 * 24 * 3600),
        (r"(\d+) week", 7 * 24 * 3600),
        (r"(\d+) day", 24 * 3600),
        (r"(\d+) hour", 3600),
        (r"(\d+) minute", 60),
        (r"(\d+) second", 1),
    ]
    for pattern, multiplier in patterns:
        match = re.search(pattern, uptime_str)
        if match:
            total_seconds += int(match.group(1)) * multiplier
    return total_seconds

def ip_in_subnet(ip: str, subnet: str) -> bool:
    """检查 IP 是否在子网内"""
    from ipaddress import ip_address, ip_network
    try:
        return ip_address(ip) in ip_network(subnet, strict=False)
    except ValueError:
        return False

def timestamp() -> str:
    """获取当前时间戳"""
    return datetime.now().strftime("%Y%m%d_%H%M%S")

def human_size(bytes_val: int) -> str:
    """字节转可读格式"""
    for unit in ["B", "KB", "MB", "GB", "TB"]:
        if bytes_val < 1024:
            return f"{bytes_val:.1f}{unit}"
        bytes_val /= 1024
    return f"{bytes_val:.1f}PB"

class ConfigDiff:
    """配置差异比较"""

    @staticmethod
    def diff(old_config: str, new_config: str) -> str:
        """比较两个配置的差异"""
        old_lines = old_config.splitlines()
        new_lines = new_config.splitlines()

        old_set = set(line.strip() for line in old_lines if line.strip())
        new_set = set(line.strip() for line in new_lines if line.strip())

        added = new_set - old_set
        removed = old_set - new_set

        lines = []
        if added:
            lines.append("=== 新增配置 ===")
            for line in sorted(added):
                lines.append(f"+ {line}")
        if removed:
            lines.append("\n=== 删除配置 ===")
            for line in sorted(removed):
                lines.append(f"- {line}")

        return "\n".join(lines) if lines else "无差异"

二、配置备份脚本

#!/usr/bin/env python3
# scripts/backup_config.py — 配置备份脚本

"""
网络设备配置备份脚本
功能：备份所有设备的 running-config，保存到日期目录
"""

import sys
import os

# 添加项目根目录到路径
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))

from lib.connector import DeviceConnector
from lib.logger import setup_logger
from lib.utils import timestamp
from concurrent.futures import ThreadPoolExecutor, as_completed
import os

# 日志
log = setup_logger("backup")

def backup_device(device):
    """备份单台设备"""
    connector = DeviceConnector()
    conn = connector.connect_netmiko(device)
    if not conn:
        return device["name"], False, "连接失败"

    try:
        # 获取配置
        config = connector.backup_config(conn)
        if not config:
            return device["name"], False, "配置为空"

        # 保存文件
        backup_dir = f"output/backup/{timestamp()}"
        os.makedirs(backup_dir, exist_ok=True)

        filename = f"{backup_dir}/{device['name']}_running.cfg"
        with open(filename, "w", encoding="utf-8") as f:
            f.write(config)

        log.info(f"✓ {device['name']}: 配置已备份 ({len(config)} 字符)")
        return device["name"], True, filename

    except Exception as e:
        log.error(f"✗ {device['name']}: {e}")
        return device["name"], False, str(e)
    finally:
        connector.close(conn)

def main():
    """主函数"""
    log.info("=== 全网配置备份开始 ===")

    connector = DeviceConnector()
    devices = connector.devices
    log.info(f"设备总数: {len(devices)}")

    success = 0
    failed = 0

    with ThreadPoolExecutor(max_workers=5) as ex:
        futures = {ex.submit(backup_device, dev): dev for dev in devices}
        for future in as_completed(futures):
            name, ok, msg = future.result()
            if ok:
                success += 1
            else:
                failed += 1

    log.info(f"=== 备份完成: 成功 {success}, 失败 {failed} ===")

    if failed > 0:
        sys.exit(1)

if __name__ == "__main__":
    main()

三、每日巡检脚本

#!/usr/bin/env python3
# scripts/daily_inspection.py — 每日巡检脚本

"""
网络设备每日自动巡检
收集：设备健康状态、接口状态、CPU/内存、日志摘要
输出：HTML 巡检报告
"""

import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))

from lib.connector import DeviceConnector
from lib.logger import setup_logger
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
import json

log = setup_logger("inspection")

def inspect_device(device):
    """巡检单台设备"""
    connector = DeviceConnector()
    conn = connector.connect_netmiko(device)
    if not conn:
        return device["name"], {"status": "failed", "error": "连接失败"}

    result = {
        "name": device["name"],
        "mgmt_ip": device["mgmt_ip"],
        "role": device.get("role", "unknown"),
        "status": "success",
        "timestamp": datetime.now().isoformat(),
    }

    try:
        # 1. 系统信息
        version = conn.send_command("display version")
        result["version"] = version.split("\n")[0] if version else "N/A"

        # 2. 设备名称
        hostname = conn.send_command(
            "display current-configuration | include sysname"
        )
        result["hostname"] = hostname.replace("sysname ", "").strip()

        # 3. CPU 使用率
        cpu = conn.send_command("display cpu-usage")
        # 解析 CPU 行
        for line in cpu.split("\n"):
            if "CPU Usage" in line:
                result["cpu"] = line.strip()
                break
            elif "cpu-usage" in line.lower():
                result["cpu"] = line.strip()

        # 4. 内存使用率
        mem = conn.send_command("display memory-usage")
        for line in mem.split("\n"):
            if "Memory Usage" in line or "Memory:" in line:
                result["memory"] = line.strip()
                break

        # 5. 接口状态摘要
        interfaces = conn.send_command(
            "display interface brief"
        )
        up_count = interfaces.count("up")
        down_count = interfaces.count("down") - interfaces.count("undo shutdown")
        result["interfaces"] = {
            "total": up_count + down_count,
            "up": up_count,
            "down": down_count,
        }

        # 6. 日志告警
        logs = conn.send_command(
            "display logbuffer | include error|down|failed|critical"
        )
        result["log_alerts"] = len([l for l in logs.split("\n") if l.strip()])

        # 7. 环境状态
        env = conn.send_command("display environment")
        result["environment"] = {}
        for line in env.split("\n"):
            if "temperature" in line.lower():
                result["environment"]["temperature"] = line.strip()
            elif "fan" in line.lower():
                result["environment"]["fan"] = line.strip()
            elif "power" in line.lower():
                result["environment"]["power"] = line.strip()

        log.info(f"✓ {device['name']}: CPU={result.get('cpu', 'N/A')}, "
                 f"接口 UP={result.get('interfaces', {}).get('up', 0)}, "
                 f"告警={result.get('log_alerts', 0)}")

    except Exception as e:
        result["status"] = "error"
        result["error"] = str(e)
        log.error(f"✗ {device['name']}: {e}")
    finally:
        connector.close(conn)

    return device["name"], result

def generate_html_report(results):
    """生成 HTML 巡检报告"""
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    healthy = sum(1 for r in results.values() if r["status"] == "success")
    total = len(results)

    # 只保留成功的结果用于 HTML
    success_results = {
        k: v for k, v in results.items() if v["status"] == "success"
    }

    html = f"""<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8">
    <title>网络设备每日巡检报告</title>
    <style>
        body {{ font-family: Arial, sans-serif; margin: 20px; }}
        h1 {{ color: #333; }}
        .summary {{ background: #f0f8ff; padding: 15px; border-radius: 5px; }}
        .device {{ border: 1px solid #ddd; margin: 10px 0; padding: 10px; border-radius: 5px; }}
        .success {{ border-left: 4px solid #4CAF50; }}
        .failed {{ border-left: 4px solid #f44336; }}
        .error {{ border-left: 4px solid #ff9800; }}
        .ok {{ color: green; }}
        .warn {{ color: orange; }}
        .critical {{ color: red; }}
        table {{ border-collapse: collapse; width: 100%; }}
        th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
        th {{ background: #4CAF50; color: white; }}
    </style>
</head>
<body>
    <h1>📋 网络设备每日巡检报告</h1>
    <p>生成时间: {now}</p>
    <div class="summary">
        <p>设备总数: {total} | 健康: {healthy} | 异常: {total - healthy}</p>
    </div>
    <table>
        <tr>
            <th>设备</th><th>角色</th><th>CPU</th><th>内存</th>
            <th>接口(UP/总)</th><th>告警</th><th>状态</th>
        </tr>"""

    for name, r in success_results.items():
        cpu_str = r.get("cpu", "N/A")
        mem_str = r.get("memory", "N/A")
        intf = r.get("interfaces", {})
        intf_str = f"{intf.get('up', 0)}/{intf.get('total', 0)}"
        alerts = r.get("log_alerts", 0)
        status_class = "ok" if alerts == 0 else ("warn" if alerts < 5 else "critical")
        status_str = "正常" if alerts == 0 else f"{alerts} 条告警"

        html += f"""
        <tr>
            <td>{name}</td>
            <td>{r.get('role', '-')}</td>
            <td>{cpu_str}</td>
            <td>{mem_str}</td>
            <td>{intf_str}</td>
            <td class="{status_class}">{status_str}</td>
            <td>{r.get('status')}</td>
        </tr>"""

    # 异常设备
    failed_results = {
        k: v for k, v in results.items() if v["status"] != "success"
    }
    if failed_results:
        html += """</table>
        <h2>❌ 异常设备</h2>
        <table><tr><th>设备</th><th>错误</th></tr>"""
        for name, r in failed_results.items():
            html += f"<tr><td>{name}</td><td>{r.get('error', '未知')}</td></tr>"
        html += "</table>"
    else:
        html += "</table>"

    html += """
    <hr>
    <p><em>自动生成，如有问题请联系网络团队</em></p>
</body>
</html>"""

    output_dir = "output/inspection"
    os.makedirs(output_dir, exist_ok=True)
    filename = f"{output_dir}/inspection_{datetime.now().strftime('%Y%m%d')}.html"
    with open(filename, "w", encoding="utf-8") as f:
        f.write(html)
    log.info(f"巡检报告已生成: {filename}")
    return filename

def main():
    log.info("=== 每日巡检开始 ===")
    connector = DeviceConnector()
    results = {}

    with ThreadPoolExecutor(max_workers=5) as ex:
        futures = {ex.submit(inspect_device, dev): dev for dev in connector.devices}
        for future in as_completed(futures):
            name, result = future.result()
            results[name] = result

    # 生成报告
    report_file = generate_html_report(results)
    print(f"巡检完成，报告: {report_file}")

if __name__ == "__main__":
    main()

四、VLAN 批量部署

#!/usr/bin/env python3
# scripts/deploy_vlan.py — VLAN 批量部署脚本

"""
VLAN 批量部署脚本
从 YAML 配置读取 VLAN 规划，批量下发到指定设备组
"""

import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))

from lib.connector import DeviceConnector
from lib.logger import setup_logger
import yaml

log = setup_logger("vlan_deploy")

def load_vlan_plan(plan_file: str) -> dict:
    """加载 VLAN 规划"""
    with open(plan_file, "r", encoding="utf-8") as f:
        return yaml.safe_load(f)

def deploy_vlans(device, vlans):
    """在设备上部署 VLAN"""
    connector = DeviceConnector()
    conn = connector.connect_netmiko(device)
    if not conn:
        return device["name"], False, "连接失败"

    try:
        commands = []
        for vlan in vlans:
            commands.append(f"vlan {vlan['id']}")
            if "name" in vlan:
                commands.append(f"name {vlan['name']}")
            commands.append("quit")

        log.info(f"  {device['name']}: 正在创建 {len(vlans)} 个 VLAN...")
        output = conn.send_config_set(commands)

        # 验证
        verify = conn.send_command("display vlan summary")
        conn.save()

        log.info(f"✓ {device['name']}: VLAN 部署成功")
        return device["name"], True, output
    except Exception as e:
        log.error(f"✗ {device['name']}: {e}")
        return device["name"], False, str(e)
    finally:
        connector.close(conn)

def main():
    """主函数"""
    import argparse
    parser = argparse.ArgumentParser(description="VLAN 批量部署")
    parser.add_argument("--plan", required=True, help="VLAN 规划 YAML 文件")
    parser.add_argument("--group", help="目标设备组（如 access, core）")
    parser.add_argument("--device", help="目标设备名（单个）")
    args = parser.parse_args()

    # 加载规划
    plan = load_vlan_plan(args.plan)
    vlans = plan.get("vlans", [])
    target_device_names = plan.get("target_devices", [])

    if not vlans:
        log.error("VLAN 规划为空")
        return

    log.info(f"=== VLAN 批量部署 ===")
    log.info(f"待创建 VLAN: {[v['id'] for v in vlans]}")

    # 获取目标设备
    connector = DeviceConnector()
    if args.device:
        dev = connector.get_device(args.device)
        devices = [dev] if dev else []
    elif args.group:
        devices = connector.get_devices_by_role(args.group)
    else:
        devices = [d for d in connector.devices
                   if d["name"] in target_device_names]

    if not devices:
        log.error("未找到目标设备")
        return

    log.info(f"目标设备: {[d['name'] for d in devices]}")

    # 部署
    success = 0
    for dev in devices:
        name, ok, msg = deploy_vlans(dev, vlans)
        if ok:
            success += 1

    log.info(f"=== 完成: {success}/{len(devices)} ===")

if __name__ == "__main__":
    main()

五、合规检查脚本

#!/usr/bin/env python3
# scripts/compliance_check.py — 合规检查脚本

import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))

from lib.connector import DeviceConnector
from lib.logger import setup_logger
import yaml
from concurrent.futures import ThreadPoolExecutor, as_completed

log = setup_logger("compliance")

# 合规检查规则
COMPLIANCE_RULES = {
    "SSH 版本": {"command": "display ssh server status",
                  "check": lambda o: "SSH version : 2" in o},
    "NTP 配置": {"command": "display ntp-service status",
                  "check": lambda o: "clock " in o or "synchronized" in o},
    "SNMP 配置": {"command": "display snmp-agent sys-info version",
                   "check": lambda o: "v3" in o or "SNMP" in o},
    "口令加密": {"command": "display current-configuration | include password",
                 "check": lambda o: "cipher" in o or "hash" in o},
    "日志服务器": {"command": "display logbuffer | include info-center loghost",
                   "check": lambda o: "loghost" in o or "192.168" in o},
}

def check_device(device):
    """检查单台设备合规性"""
    connector = DeviceConnector()
    conn = connector.connect_netmiko(device)
    if not conn:
        return device["name"], {"status": "failed", "error": "连接失败"}

    result = {
        "name": device["name"],
        "status": "checked",
        "checks": {},
    }

    try:
        for rule_name, rule in COMPLIANCE_RULES.items():
            output = conn.send_command(rule["command"])
            passed = rule["check"](output)
            result["checks"][rule_name] = {
                "passed": passed,
                "output": output[:100] + "..." if len(output) > 100 else output,
            }
    except Exception as e:
        result["status"] = "error"
        result["error"] = str(e)
    finally:
        connector.close(conn)

    # 汇总
    total = len(result["checks"])
    passed = sum(1 for c in result["checks"].values() if c["passed"])
    result["summary"] = f"{passed}/{total}"
    result["compliant"] = passed == total

    log.info(f"{'✓' if result['compliant'] else '✗'} {device['name']}: "
             f"{result['summary']} 合规")
    return device["name"], result

def main():
    log.info("=== 合规检查开始 ===")
    connector = DeviceConnector()
    all_results = {}

    with ThreadPoolExecutor(max_workers=5) as ex:
        futures = {ex.submit(check_device, dev): dev for dev in connector.devices}
        for future in as_completed(futures):
            name, result = future.result()
            all_results[name] = result

    # 输出汇总
    print("\n=== 合规检查汇总 ===")
    compliant = sum(1 for r in all_results.values()
                    if r.get("compliant"))
    print(f"总设备: {len(all_results)}, 合规: {compliant}, "
          f"不合规: {len(all_results) - compliant}")
    print()

    for name, result in all_results.items():
        if not result.get("compliant", False):
            print(f"✗ {name}: {result.get('summary', 'N/A')}")
            for rule, check in result.get("checks", {}).items():
                if not check["passed"]:
                    print(f"   - {rule}: 不合规")
            print()

if __name__ == "__main__":
    main()

六、健康检查与告警

#!/usr/bin/env python3
# scripts/health_check.py — 健康检查脚本

"""
网络设备健康检查
检查项：CPU > 80%, 内存 > 80%, 接口 down, 温度异常
输出：告警清单
"""

import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))

from lib.connector import DeviceConnector
from lib.logger import setup_logger
import re
from datetime import datetime

log = setup_logger("health")

ALARM_THRESHOLDS = {
    "cpu_max": 80,
    "memory_max": 80,
    "interface_down_max": 5,
    "temperature_max": 50,
}

def extract_cpu_percent(cpu_str):
    """从 CPU 字符串提取百分比"""
    match = re.search(r"(\d+)%", cpu_str or "")
    return int(match.group(1)) if match else 0

def extract_memory_percent(mem_str):
    """从内存字符串提取百分比"""
    match = re.search(r"(\d+)%", mem_str or "")
    return int(match.group(1)) if match else 0

def health_check_device(device):
    """健康检查单台设备"""
    connector = DeviceConnector()
    conn = connector.connect_netmiko(device)
    if not conn:
        return device["name"], {"status": "failed", "error": "连接失败"}

    alarms = []
    info = {"name": device["name"], "status": "healthy"}

    try:
        # CPU 检查
        cpu_output = conn.send_command("display cpu-usage")
        cpu_pct = extract_cpu_percent(cpu_output)
        if cpu_pct > ALARM_THRESHOLDS["cpu_max"]:
            alarms.append(f"CPU 使用率 {cpu_pct}% 超过阈值 {ALARM_THRESHOLDS['cpu_max']}%")
        info["cpu"] = cpu_pct

        # 内存检查
        mem_output = conn.send_command("display memory-usage")
        mem_pct = extract_memory_percent(mem_output)
        if mem_pct > ALARM_THRESHOLDS["memory_max"]:
            alarms.append(f"内存使用率 {mem_pct}% 超过阈值 {ALARM_THRESHOLDS['memory_max']}%")
        info["memory"] = mem_pct

        # 接口检查
        intf_output = conn.send_command("display interface brief")
        down_ints = []
        for line in intf_output.split("\n"):
            if "*down" in line or "down" in line:
                # 排除管理口
                if "GigabitEthernet" in line or "Eth-Trunk" in line:
                    down_ints.append(line.split()[0] if line.split() else "unknown")
        if len(down_ints) > ALARM_THRESHOLDS["interface_down_max"]:
            alarms.append(f"接口 DOWN 数 {len(down_ints)} 超过阈值 {ALARM_THRESHOLDS['interface_down_max']}")
        info["interfaces_down"] = len(down_ints)
        info["down_interfaces"] = down_ints

        # 温度检查
        env_output = conn.send_command("display temperature all")
        for line in env_output.split("\n"):
            match = re.search(r"(\d+)", line)
            if match and "temperature" in line.lower() or "Current" in line:
                temp = int(match.group(1))
                if temp > ALARM_THRESHOLDS["temperature_max"]:
                    alarms.append(f"温度异常: {line.strip()}")
        info["temperature"] = env_output[:200]

    except Exception as e:
        info["status"] = "error"
        info["error"] = str(e)
    finally:
        connector.close(conn)

    if alarms:
        info["status"] = "alarm"
        info["alarms"] = alarms
        for alarm in alarms:
            log.warning(f"⚠ {device['name']}: {alarm}")
    else:
        log.info(f"✓ {device['name']}: 健康")

    return device["name"], info

def main():
    log.info("=== 健康检查开始 ===")
    connector = DeviceConnector()
    all_alarms = []

    for dev in connector.devices:
        name, info = health_check_device(dev)
        if info.get("status") == "alarm":
            all_alarms.append(info)

    # 告警汇总
    if all_alarms:
        print(f"\n⚠ 共发现 {len(all_alarms)} 台设备异常：")
        for alarm in all_alarms:
            print(f"\n{alarm['name']}:")
            for a in alarm.get("alarms", []):
                print(f"  • {a}")
    else:
        print("\n✓ 所有设备健康检查通过")

if __name__ == "__main__":
    main()

七、定时任务集成

#!/usr/bin/env python3
# schedule_tasks.py — 定时任务调度

"""
使用 schedule 库定时执行运维脚本
"""

import schedule
import time
import subprocess
import sys
import os
from lib.logger import setup_logger

log = setup_logger("scheduler")

def run_script(script_name: str):
    """运行脚本"""
    script_path = os.path.join("scripts", script_name)
    log.info(f"启动定时任务: {script_name}")
    try:
        result = subprocess.run(
            [sys.executable, script_path],
            capture_output=True,
            text=True,
            timeout=600,
        )
        if result.returncode == 0:
            log.info(f"✓ {script_name} 执行成功")
            for line in result.stdout.split("\n")[-5:]:
                if line.strip():
                    log.info(f"  {line}")
        else:
            log.error(f"✗ {script_name} 执行失败: {result.stderr[:200]}")
    except subprocess.TimeoutExpired:
        log.error(f"✗ {script_name} 执行超时")
    except Exception as e:
        log.error(f"✗ {script_name}: {e}")

def main():
    """配置定时任务"""
    # 每天凌晨 2:00 配置备份
    schedule.every().day.at("02:00").do(
        run_script, "backup_config.py"
    )

    # 每天上午 8:00 每日巡检
    schedule.every().day.at("08:00").do(
        run_script, "daily_inspection.py"
    )

    # 每周一 9:00 合规检查
    schedule.every().monday.at("09:00").do(
        run_script, "compliance_check.py"
    )

    # 每 30 分钟健康检查
    schedule.every(30).minutes.do(
        run_script, "health_check.py"
    )

    log.info("=== 定时任务调度器启动 ===")
    log.info("已配置任务:")
    log.info("  02:00 - 配置备份")
    log.info("  08:00 - 每日巡检")
    log.info("  周一 09:00 - 合规检查")
    log.info("  每 30 分钟 - 健康检查")

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

if __name__ == "__main__":
    main()

八、最佳实践总结

Python 自动化脚本开发规范：

  1. 代码组织
  ┌─ 公共模块复用（connector/logger/utils）
  ├─ 配置与代码分离（YAML 设备清单）
  ├─ 单一职责（每个脚本只做一件事）
  └─ 并行执行（ThreadPoolExecutor）

  2. 错误处理
  ┌─ 所有异常必须捕获
  ├─ 有意义的错误消息
  ├─ 失败设备不影响其他设备
  └─ 关键操作有重试机制

  3. 日志与审计
  ┌─ 记录所有操作
  ├─ 时间戳 + 设备名 + 操作内容
  ├─ 日志分级（DEBUG/INFO/WARNING/ERROR）
  └─ 变更操作记录到审计日志

  4. 执行安全
  ┌─ 先备份后变更
  ├─ 变更后验证
  ├─ 设置执行超时
  └─ 保留回滚能力

下篇预告：第315篇 — IPAM 地址管理，将介绍 IP 地址管理（IPAM）的核心概念、工具选型和自动化集成方案。