Python + Netmiko实现华为交换机批量巡检自动化脚本实战
网络运维中，面对成百上千台华为交换机，人工SSH登录逐台巡检 display version、display interface brief 等命令效率低下且容易出错。本文提供一套基于 Python 3.11 与 Netmiko 的批量自动化巡检方案，通过配置文件管理设备清单、并发执行命令、解析输出并生成 Excel 报告，同时集成重试、超时、阈值告警等机制。所有代码均可在实际环境中直接部署。一、环境准备与依赖安装
推荐使用 Python 3.8+（本文用 3.11）。核心依赖为 Netmiko（SSH 连接与命令执行）、Pandas（数据处理）、openpyxl（Excel 写入）、Paramiko（SSH 底层库）。可选 ntc-templates 和 textfsm 用于结构化解析输出。安装命令：
pip install netmiko pandas openpyxl paramiko
设备清单存放在 devices.csv 中，格式如下（严禁硬编码密码）：
ip,hostname,device_type,username,password,secret
192.168.1.1,SW-Core-01,huawei,admin,YourPass,enable_pass
192.168.1.2,SW-Access-01,huawei,admin,YourPass,enable_pass
交换机需开启 SSH 并配置只读权限巡检账号，脚本服务器需保证与目标网络连通。并发建议控制在 50~100 以内，避免设备负载过高。
二、项目结构与核心模块
项目目录：
huawei_inspection/
├── devices.csv # 设备清单
├── config.py # 全局配置（命令列表、超时、重试、阈值）
├── inspection.py # 主程序入口
├── parsers.py # 命令输出解析函数
├── reporter.py # 报告生成模块
└── logs/ # 日志目录
三、全局配置文件 config.py
定义巡检命令列表、并发线程数、超时参数、重试策略以及阈值告警：
# config.py
INSPECTION_COMMANDS = [
"display version",
"display device",
"display fan",
"display power",
"display cpu-usage",
"display memory",
"display interface brief",
"display interface description",
"display arp",
"display mac-address",
"display logbuffer",
"display temperature",
]
MAX_WORKERS = 50
CONNECT_TIMEOUT = 30
READ_TIMEOUT = 30
MAX_RETRIES = 3
RETRY_DELAY = 2
REPORT_DIR = "./reports"
LOG_DIR = "./logs"
CPU_THRESHOLD = 80
MEM_THRESHOLD = 80
TEMP_THRESHOLD = 60
四、连接与命令执行模块 connector.py
通过 Netmiko 连接设备，支持 enable 密码、自动关闭分页、使用指数退避重试。核心函数：
# connector.py（核心代码片段）
from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationException
import config
def connect_huawei_device(device_info):
device = {
'device_type': 'huawei',
'host': device_info['ip'],
'username': device_info['username'],
'password': device_info['password'],
'port': 22,
'timeout': config.CONNECT_TIMEOUT,
'read_timeout': config.READ_TIMEOUT,
}
if 'secret' in device_info and device_info['secret']:
device['secret'] = device_info['secret']
try:
conn = ConnectHandler(**device)
if device_info.get('secret'):
conn.enable()
return conn
except (NetmikoAuthenticationException, NetmikoTimeoutException) as e:
return None
def inspect_single_device(device_info, commands=None):
# 含重试机制，最多尝试3次，间隔递增
for attempt in range(1, config.MAX_RETRIES + 1):
conn = connect_huawei_device(device_info)
if conn is None:
if attempt < config.MAX_RETRIES:
time.sleep(config.RETRY_DELAY * attempt)
continue
try:
outputs = run_commands(conn, commands)
conn.disconnect()
return {'status':'success', 'outputs':outputs, ...}
except:
conn.disconnect()
if attempt == config.MAX_RETRIES:
return {'status':'failure', ...}
五、输出解析模块 parsers.py
使用正则表达式从命令输出中提取结构化数据，并判断是否超过阈值。支持解析 version、cpu-usage、memory、interface brief、power、fan、temperature、logbuffer 等。例如 parse_cpu 函数：
# parsers.py（片段）
import re
import config
def parse_cpu(output):
result = {'cpu_usage': 'N/A', 'alarm': False}
match = re.search(r'CPU Usage\s*:\s*(\d+)%', output, re.I)
if not match:
match = re.search(r'CPU utilization\s*[\s:]+(\d+)%', output, re.I)
if match:
cpu_val = int(match.group(1))
result['cpu_usage'] = cpu_val
result['alarm'] = cpu_val > config.CPU_THRESHOLD
return result
解析后的数据会汇总到 parsed 字典中，供报告生成使用。
六、报告生成模块 reporter.py
利用 Pandas 将巡检结果写入 Excel，包含汇总 sheet 和明细 sheet（可选）。自动在 REPORT_DIR 下生成带时间戳的文件：
# reporter.py（简化）
import pandas as pd
from datetime import datetime
def generate_excel_report(results, output_path=None):
if not output_path:
output_path = f"./reports/巡检报告_{datetime.now():%Y%m%d_%H%M%S}.xlsx"
summary_data = []
for res in results:
row = {
'IP地址': res['ip'],
'主机名': res['hostname'],
'巡检状态': '成功' if res['status']=='success' else '失败',
'CPU使用率': res.get('parsed',{}).get('cpu',{}).get('cpu_usage', 'N/A'),
# ... 其它字段
}
summary_data.append(row)
df = pd.DataFrame(summary_data)
df.to_excel(output_path, index=False)
return output_path
七、主程序与定时执行
主程序 inspection.py 读取 devices.csv，使用 ThreadPoolExecutor 并发调用 inspect_single_device，然后解析所有输出并生成报告。如需定时执行，可用系统的 cron 或 Windows 任务计划程序调用该脚本。
八、常见问题与最佳实践
- 连接超时/失败：检查 SSH 服务、防火墙、账号权限。增加重试与指数退避。
- 命令输出分页：Netmiko 的 send_command 已自动处理分页，但也可在命令后加 | no-more。
- 权限不足：巡检账号使用只读视图或 privilege level。
- 性能瓶颈：分批巡检（每批 50 台），避免同时打开过多 SSH 连接。
- 密码安全：使用 CSV 存储时注意权限，或对接 vault 服务。
- 结果验证：定期人工抽检脚本记录与设备实际状态是否一致。
九、总结
本方案通过 Python + Netmiko 实现了华为交换机的批量巡检自动化，将原本数小时的人工操作缩短到分钟级，并输出标准化报告便于存档分析。代码可直接套用，只需调整设备清单和阈值参数即可适配不同规模网络。对于上千台交换机环境，建议结合异步框架（如 asyncio）进一步优化并发性能。
Re: Python + Netmiko实现华为交换机批量巡检自动化脚本实战
非常实用的方案！我之前也尝试过类似思路，但并发控制这块没处理好，经常有设备超时。你这里的重试加指数退避和阈值预警机制很到位，尤其CPU、内存、温度三个阈值正好是我们日常运维最关注的。配置文件的模块化设计也很清晰，后续扩展命令或者增加厂商支持应该很方便。 有一个小建议：设备清单里密码明文保存还是有点风险，能否考虑用环境变量或者加密配置文件（比如使用python的keyring库）来存储？另外日志目录和报告目录自动创建的逻辑，代码里是否也包含了？期待后续完整代码分享！Re: Python + Netmiko实现华为交换机批量巡检自动化脚本实战
很实用的方案，结构清晰，代码模块化做得很好。配置文件独立管理命令和阈值，方便维护和扩展。想问一下，在解析`display version`和`display interface brief`这类输出时，你们是用正则还是TextFSM模板？如果不同系列（比如CE和S系列）输出格式有差异，脚本里是怎么兼容处理的？另外，并发50个设备时，有没有遇到SSH连接数限制或者设备CPU冲高的现象？Re: Python + Netmiko实现华为交换机批量巡检自动化脚本实战
感谢分享，这套方案非常完整实用！尤其配置文件单独抽取、重试和阈值告警都考虑到了，直接复用门槛很低。想请教一下：并发50的时候，网络抖动导致部分设备连接失败，日志里会不会有清晰的记录或者单独汇总到报告里？另外，`display logbuffer`输出通常很长，解析成Excel时你是按条拆分还是直接存原文？期待看到更多生产环境调优的细节。
页:
[1]