转载自:https://mp.weixin.qq.com/s/P5hkCBmJmGxMjM3FT5NbwA

前段时间《Python 的交换机批量的巡检脚本》是基于Paramiko开发,经过多位大佬的指点,我这次升级采用了Netmiko框架,可实现多厂商设备兼容。

核心功能

针对企业网络中大量交换机的运维需求,开发此自动化巡检工具。通过标准化检查流程,提升网络设备维护效率,降低人工操作失误风险。
✅ 多厂商设备支持(Cisco IOS/Huawei/H3C)
✅ 关键配置自动采集(版本信息/接口状态/运行配置)
✅ 智能日志记录与错误追踪
✅ 批量设备巡检支持

技术亮点

模块选型对比

  • Paramiko(旧项目方案):基础SSH协议库,需自行处理设备交互逻辑(需手动维护会话状态/适配不同厂商指令/实现错误重试机制),适合深度定制场景

  • Netmiko(当前方案):基于Paramiko二次开发的网络设备自动化框架,主要增强功能包括:

  • • 多厂商协议适配(内置Cisco、Huawei、H3C等更多设备)
    • 会话生命周期管理(自动处理SSH连接超时与重连机制)
    • 智能交互优化(自动处理分页显示、命令提示符匹配、异常恢复)
    • 标准化输出处理(自动清理ANSI转义字符与多余空行)
    • 批处理执行引擎(支持命令队列的原子化执行与结果收集)

代码架构解析

核心模块说明

  1. 设备连接管理
# 使用ConnectHandler统一连接入口
with netmiko.ConnectHandler(
    device_type=switch['device_type'],
    ip=switch['ip'],
    # ... 认证参数
) as net_connect:
  1. 设备类型适配
# 通过device_type自动匹配厂商协议
if switch['device_type'] in ['cisco_ios']:
    net_connect.enable()  # Cisco专属特权模式
  1. 批处理执行引擎
# 标准化命令执行流程
for command in commands:
    output = net_connect.send_command(command)
    logging.info(f'### Output for {command}...')
  • • 采用Netmiko实现多厂商设备兼容
  • • 异常处理机制覆盖常见网络故障场景
  • • 可视化实时日志输出
  • • 配置信息加密存储

使用说明

环境要求

pip install netmiko==4.2.0

配置文件示例(switches.txt)

# 格式:IP,用户名,登录密码,enable密码,设备类型
192.168.1.1,admin,password123,enablepass,cisco_ios
10.0.0.1,huawei_admin,Huawei@123,,huawei

运行命令

python switch_inspection.py

典型应用场景

  1. 日常网络设备健康检查
  2. 变更配置前基准状态备份
  3. 故障排查时快速获取设备信息
  4. 合规性审计数据采集

设备类型支持

厂商 设备类型标识 适用场景
Cisco cisco_ios IOS系统交换机(Catalyst系列)
Cisco cisco_nxos NX-OS系统交换机(Nexus系列)
Huawei huawei VRP系统交换机(S系列)
H3C h3c Comware V7系统交换机
Juniper juniper_junos Junos系统交换机(EX系列)

注:具体设备类型请参考Netmiko官方文档

注意事项

⚠️ 建议在隔离环境测试后再正式使用
⚠️ 配置文件需设置访问权限
⚠️ 确保执行主机与设备网络可达

完整代码

import netmiko
import logging
import re

logging.basicConfig(filename='switch_inspection.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 定义巡检交换机的函数
definspect_switch(switch):
    try:
        # 建立与交换机的连接
        with netmiko.ConnectHandler(
            device_type=switch['device_type'],
            ip=switch['ip'],
            username=switch['username'],
            password=switch['password'],
            secret=switch['enable_password']
        ) as net_connect:
            # 记录成功连接的日志
            logging.info(f'成功连接到 {switch["ip"]}')

            # 如果是 Cisco IOS 设备,进入 enable 模式
            if switch['device_type'] in ['cisco_ios']:
                net_connect.enable()
                # 记录进入 enable 模式的日志
                logging.info(f'已进入 {switch["ip"]} 的 enable 模式')
            else:
                # 记录非 Cisco IOS 设备跳过进入 enable 模式的日志
                logging.info(f'{switch["device_type"]} 设备不需要进入 enable 模式,跳过此步骤。')

            # 定义要执行的命令列表
            commands = [
                'show version',
                'show interfaces status',
                'show running-config'
            ]

            # 遍历命令列表并执行
            for command in commands:
                output = net_connect.send_command(command)
                # 记录命令输出的日志
                logging.info(f'### Output for {command} on {switch["ip"]} ###\n{output}')
                # 打印命令输出
                print(f'### Output for {command} on {switch["ip"]} ###')
                print(output)

            # 记录断开连接的日志
            logging.info(f'已断开与 {switch["ip"]} 的连接')
    except netmiko.NetMikoTimeoutException as e:
        # 记录连接超时的错误日志
        logging.error(f'连接到 {switch["ip"]} 超时: {e}')
        # 打印连接超时的错误信息
        print(f'Error connecting to {switch["ip"]}: 连接超时')
    except netmiko.NetMikoAuthenticationException as e:
        # 记录认证失败的错误日志
        logging.error(f'连接到 {switch["ip"]} 认证失败: {e}')
        # 打印认证失败的错误信息
        print(f'Error connecting to {switch["ip"]}: 认证失败')
    except Exception as e:
        # 记录未知错误的日志
        logging.error(f'连接到 {switch["ip"]} 发生未知错误: {e}')
        # 打印未知错误的信息
        print(f'Error connecting to {switch["ip"]}: {e}')

defread_switches_from_txt(file_path):
    switches = []
    withopen(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            if line.strip() andnot line.startswith('#'):
                ip, username, password, enable_password, device_type = line.strip().split(',')
                switches.append({
                    'ip': ip,
                    'username': username,
                    'password': password,
                    'enable_password': enable_password,
                    'device_type': device_type
                })
    return switches

# 主程序入口
if __name__ == "__main__":
    # 定义交换机信息文件的路径
    file_path = "switches.txt"
    # 从文件中读取交换机信息
    switches = read_switches_from_txt(file_path)
    # 遍历交换机列表并进行巡检
    for switch in switches:
        inspect_switch(switch)