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 用于结构化解析输出。安装命令：

2. pip install netmiko pandas openpyxl paramiko

复制代码

设备清单存放在 devices.csv 中，格式如下（严禁硬编码密码）：

2. ip,hostname,device_type,username,password,secret
3. 192.168.1.1,SW-Core-01,huawei,admin,YourPass,enable_pass
4. 192.168.1.2,SW-Access-01,huawei,admin,YourPass,enable_pass

复制代码

交换机需开启 SSH 并配置只读权限巡检账号，脚本服务器需保证与目标网络连通。并发建议控制在 50~100 以内，避免设备负载过高。

---

二、项目结构与核心模块

项目目录：

2. huawei_inspection/
3. ├── devices.csv          # 设备清单
4. ├── config.py            # 全局配置（命令列表、超时、重试、阈值）
5. ├── inspection.py        # 主程序入口
6. ├── parsers.py           # 命令输出解析函数
7. ├── reporter.py          # 报告生成模块
8. └── logs/                # 日志目录

复制代码

---

三、全局配置文件 config.py

定义巡检命令列表、并发线程数、超时参数、重试策略以及阈值告警：

2. # config.py
3. INSPECTION_COMMANDS = [
4.     "display version",
5.     "display device",
6.     "display fan",
7.     "display power",
8.     "display cpu-usage",
9.     "display memory",
10.     "display interface brief",
11.     "display interface description",
12.     "display arp",
13.     "display mac-address",
14.     "display logbuffer",
15.     "display temperature",
16. ]
17. MAX_WORKERS = 50
18. CONNECT_TIMEOUT = 30
19. READ_TIMEOUT = 30
20. MAX_RETRIES = 3
21. RETRY_DELAY = 2
22. REPORT_DIR = "./reports"
23. LOG_DIR = "./logs"
24. CPU_THRESHOLD = 80
25. MEM_THRESHOLD = 80
26. TEMP_THRESHOLD = 60

复制代码

---

四、连接与命令执行模块 connector.py

通过 Netmiko 连接设备，支持 enable 密码、自动关闭分页、使用指数退避重试。核心函数：

2. # connector.py（核心代码片段）
3. from netmiko import ConnectHandler
4. from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationException
5. import config
7. def connect_huawei_device(device_info):
8.     device = {
9.         'device_type': 'huawei',
10.         'host': device_info['ip'],
11.         'username': device_info['username'],
12.         'password': device_info['password'],
13.         'port': 22,
14.         'timeout': config.CONNECT_TIMEOUT,
15.         'read_timeout': config.READ_TIMEOUT,
16.     }
17.     if 'secret' in device_info and device_info['secret']:
18.         device['secret'] = device_info['secret']
19.     try:
20.         conn = ConnectHandler(**device)
21.         if device_info.get('secret'):
22.             conn.enable()
23.         return conn
24.     except (NetmikoAuthenticationException, NetmikoTimeoutException) as e:
25.         return None
27. def inspect_single_device(device_info, commands=None):
28.     # 含重试机制，最多尝试3次，间隔递增
29.     for attempt in range(1, config.MAX_RETRIES + 1):
30.         conn = connect_huawei_device(device_info)
31.         if conn is None:
32.             if attempt < config.MAX_RETRIES:
33.                 time.sleep(config.RETRY_DELAY * attempt)
34.             continue
35.         try:
36.             outputs = run_commands(conn, commands)
37.             conn.disconnect()
38.             return {'status':'success', 'outputs':outputs, ...}
39.         except:
40.             conn.disconnect()
41.             if attempt == config.MAX_RETRIES:
42.                 return {'status':'failure', ...}

复制代码

---

五、输出解析模块 parsers.py

使用正则表达式从命令输出中提取结构化数据，并判断是否超过阈值。支持解析 version、cpu-usage、memory、interface brief、power、fan、temperature、logbuffer 等。例如 parse_cpu 函数：

2. # parsers.py（片段）
3. import re
4. import config
6. def parse_cpu(output):
7.     result = {'cpu_usage': 'N/A', 'alarm': False}
8.     match = re.search(r'CPU Usage\s*:\s*(\d+)%', output, re.I)
9.     if not match:
10.         match = re.search(r'CPU utilization\s*[\s:]+(\d+)%', output, re.I)
11.     if match:
12.         cpu_val = int(match.group(1))
13.         result['cpu_usage'] = cpu_val
14.         result['alarm'] = cpu_val > config.CPU_THRESHOLD
15.     return result

复制代码

解析后的数据会汇总到 parsed 字典中，供报告生成使用。

---

六、报告生成模块 reporter.py

利用 Pandas 将巡检结果写入 Excel，包含汇总 sheet 和明细 sheet（可选）。自动在 REPORT_DIR 下生成带时间戳的文件：

2. # reporter.py（简化）
3. import pandas as pd
4. from datetime import datetime
6. def generate_excel_report(results, output_path=None):
7.     if not output_path:
8.         output_path = f"./reports/巡检报告_{datetime.now():%Y%m%d_%H%M%S}.xlsx"
9.     summary_data = []
10.     for res in results:
11.         row = {
12.             'IP地址': res['ip'],
13.             '主机名': res['hostname'],
14.             '巡检状态': '成功' if res['status']=='success' else '失败',
15.             'CPU使用率': res.get('parsed',{}).get('cpu',{}).get('cpu_usage', 'N/A'),
16.             # ... 其它字段
17.         }
18.         summary_data.append(row)
19.     df = pd.DataFrame(summary_data)
20.     df.to_excel(output_path, index=False)
21.     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）进一步优化并发性能。

热心网友7

非常实用的方案！我之前也尝试过类似思路，但并发控制这块没处理好，经常有设备超时。你这里的重试加指数退避和阈值预警机制很到位，尤其CPU、内存、温度三个阈值正好是我们日常运维最关注的。配置文件的模块化设计也很清晰，后续扩展命令或者增加厂商支持应该很方便。 有一个小建议：设备清单里密码明文保存还是有点风险，能否考虑用环境变量或者加密配置文件（比如使用python的keyring库）来存储？另外日志目录和报告目录自动创建的逻辑，代码里是否也包含了？期待后续完整代码分享！