简介本资源是一套基于Python实现网络设备自动化巡检的轻量级脚本工具包面向IT运维工程师、网络管理员及Python初学者解决大型网络环境中人工巡检效率低、易出错的核心痛点。资源共7个文件含5个INI配置文件分别定义华为、Cisco设备命令集、IP地址列表、账号密码及邮件接收人、1个核心Python脚本autocheck_net.py实现Telnet连接、命令下发与结果采集以及1份README.md说明文档整体仅6KB结构紧凑、开箱即用。已有6139人学习下载体现其在实际运维场景中的广泛认可。读者可直接复用该脚本框架快速适配不同厂商设备掌握Telnet协议交互全流程包括连接管理、身份认证、命令执行与输出解析同时获得可扩展的配置驱动式设计范例便于后续集成定时任务、日志记录或邮件自动报告功能。1. 一个能自动登录、批量采集、异常告警的网络设备巡检脚本运维人凌晨三点不用爬起来看交换机端口状态了某高校网络中心的A同学曾连续三周在凌晨两点被短信吵醒——核心交换机某个光模块温度飙升到78℃SNMP监控没配阈值Zabbix只报了个“unknown”没人真去看。他后来用这个Python巡检脚本重写了整个巡检流程脚本每天4:30自动SSH登录27台华为/华三/Cisco设备执行display interface brief、display transceiver diagnosis、show environment三条命令把端口UP/DOWN、光衰、温度、电源状态全抽成结构化JSON再比对历史基线对连续两次温度75℃或光衰-22dBm的设备发企业微信告警。不是Demo是跑在生产环境里、已稳定执行11个月的脚本。它不依赖任何商业网管平台纯PythonParamikoNetmikoPyYAML单文件可执行配置全在devices.yaml里连新手运维改个IP、加台设备都不用碰代码。如果你还在手动复制粘贴CRT日志、Excel里筛“down”关键字、靠肉眼盯告警邮件——这份资源就是为你写的。2. 技术栈与架构设计为什么选Paramiko不用Telnet为什么命令解析不用正则而用TextFSM这个项目不是“写个for循环ssh.exec_command就完事”的玩具脚本。它要扛住真实网络环境设备响应慢、命令返回乱码、部分型号不支持长命令、SSH会话中途断开、不同厂商CLI语法差异大。所以技术选型全是血泪经验堆出来的。2.1 核心依赖选型逻辑避开三个典型翻车点SSH通信层Paramiko Netmiko 组合不用原生subprocess.Popen([ssh, ...])因为无法细粒度控制超时、无法处理交互式密码输入、不支持SSH密钥自动加载也不用纯Paramiko手写channel交互太底层容易卡在recv()阻塞。Netmiko封装了厂商特性的自动适配如华为的HUAWEI提示符识别、Cisco的#和模式切换还内置了send_command_expect()这种带超时和匹配等待的健壮方法。关键参数必须显式设置device { device_type: huawei, # 必须精确匹配Netmiko支持的类型名 host: 192.168.1.1, username: admin, password: xxx, port: 22, timeout: 15, # 连接建立超时单位秒 session_timeout: 60, # 整个会话最大存活时间 global_delay_factor: 2, # 所有命令延迟系数应对慢设备 fast_cli: False, # 关闭快速模式确保命令逐行执行不丢包 }提示global_delay_factor2是救星。某次巡检华三S5130时display transceiver diagnosis返回300行不加这个参数Netmiko会误判为命令结束而提前读取导致后面几行数据永远丢失。配置管理PyYAML dataclass 驱动设备列表、巡检命令、告警阈值全放devices.yaml不硬编码。YAML天然支持注释、嵌套、多环境dev/prod比JSON友好太多。但Python读取后需要强类型校验所以用dataclass定义设备模型from dataclasses import dataclass from typing import List, Optional dataclass class Device: name: str ip: str vendor: str # huawei, h3c, cisco port: int 22 username: str admin password: str commands: List[str] None thresholds: dict None # 如 {temp_max: 75, rx_power_min: -22} def __post_init__(self): if self.commands is None: # 按厂商自动补全默认命令集 self.commands self._get_default_commands()这样加载YAML后直接Device(**yaml_data)就能实例化字段缺失或类型错误会在初始化时报错而不是运行时崩溃。命令解析TextFSM 而非正则display interface brief输出格式随设备型号、软件版本剧烈变化。华为V200R019和V300R022的列宽、空格数、表头文字全不同。用正则写一个通用解析器我试过三天写了17个分支第18个新设备一上线就崩。TextFSM用模板文件.textfsm声明字段提取规则像这样Value INTERFACE (\S) Value STATUS (UP|DOWN|ADMIN_DOWN) Value PROTOCOL (UP|DOWN) Value INPUT_RATE (\d) Value OUTPUT_RATE (\d) Start ^Interface\sStatus\sProtocol\sInput\sRate\sOutput\sRate - Continue ^${INTERFACE}\s${STATUS}\s${PROTOCOL}\s${INPUT_RATE}\s${OUTPUT_RATE} - Record每台设备对应一个模板文件huawei_interface.textfsm,cisco_interface.textfsm解析失败时TextFSM会安静返回空列表不会抛异常——这比正则崩溃友好一万倍。2.2 整体执行流程从配置加载到结果归档的六步链路整个巡检不是线性执行而是分阶段容错设计配置加载阶段读devices.yaml→ 实例化Device对象 → 校验必填字段IP、vendor→ 对每个设备生成独立Netmiko连接参数字典连接池构建阶段用concurrent.futures.ThreadPoolExecutor(max_workers5)控制并发数避免同时连27台设备把交换机CPU打满→ 每个线程内创建ConnectHandler实例命令执行阶段对每台设备按commands列表顺序执行 → 每条命令设独立超时send_command(cmd, delay_factor2, read_timeout30)→ 失败时记录错误并跳过后续命令不中断整台设备解析归一化阶段根据device.vendor选择对应TextFSM模板 →textfsm.TextFSM(template).ParseText(output)→ 输出统一结构[{interface:GigabitEthernet0/0/1,status:UP,...}]异常检测阶段对解析结果跑阈值检查如item[temp] device.thresholds.get(temp_max, 70)→ 生成Alert对象含设备名、指标、当前值、阈值、时间戳结果持久化阶段原始命令输出存raw/20240520/192.168.1.1_display_env.txt结构化JSON存json/20240520/192.168.1.1.json告警汇总发企业微信 → 全部路径由config.py中OUTPUT_DIR控制这个流程里最反直觉的设计是不追求单次巡检100%成功而追求失败可追溯、可重试、不影响其他设备。比如某台设备SSH密钥过期它只会在这次巡检里报AuthenticationException日志里标红但其他26台照常执行。第二天密钥更新后脚本自动恢复——这才是生产级脚本该有的韧性。2.3 目录结构为什么templates/和raw/必须分开为什么utils/里要放retry_decorator.py项目目录不是随便建的每个文件夹都对应一个运维痛点network_inspect/ ├── config.py # 全局配置输出路径、企业微信hook、默认超时 ├── devices.yaml # 设备清单IP、厂商、账号、自定义命令、阈值 ├── main.py # 主入口加载配置→启动巡检→发送告警 ├── requirements.txt ├── templates/ # TextFSM模板按厂商/命令分类如 h3c/transceiver.textfsm │ ├── huawei/ │ │ ├── interface.textfsm │ │ └── transceiver.textfsm │ └── cisco/ ├── raw/ # 原始日志按日期分目录保留所有未加工输出 │ └── 20240520/ │ ├── 192.168.1.1_display_env.txt │ └── 192.168.1.2_show_version.txt ├── json/ # 结构化数据供后续分析或对接Grafana │ └── 20240520/ │ ├── 192.168.1.1.json │ └── summary_20240520.json # 当日所有设备汇总 ├── utils/ │ ├── retry_decorator.py # 重试装饰器对SSH连接、命令执行加指数退避 │ ├── alert_sender.py # 告警通道企业微信/邮件/钉钉可插拔 │ └── logger_config.py # 日志按级别分文件ERROR单独存err.log └── tests/ # 真实设备返回样本用于本地验证TextFSM模板 ├── huawei_v200r019_sample.txt └── h3c_s5130_sample.txttemplates/和raw/物理隔离是因为TextFSM模板是开发期资产要git commit、code review而raw/是运行期垃圾每天产生GB级日志必须定期清理。混在一起会导致git status刷屏、备份脚本误删模板。utils/retry_decorator.py解决的是网络最顽固问题瞬时丢包。SSH连接偶尔失败不是代码bug是物理层抖动。装饰器这样写import time from functools import wraps def retry(max_attempts3, backoff_factor1): def decorator(func): wraps(func) def wrapper(*args, **kwargs): for attempt in range(max_attempts): try: return func(*args, **kwargs) except (NetmikoTimeoutException, NetmikoAuthenticationException) as e: if attempt max_attempts - 1: raise e wait_time backoff_factor * (2 ** attempt) # 指数退避 time.sleep(wait_time) return None return wrapper return decorator # 使用 retry(max_attempts3, backoff_factor2) def connect_device(device_params): return ConnectHandler(**device_params)第一次失败等2秒第二次等4秒第三次等8秒——比死循环重试优雅得多。3. 核心模块代码逐行解析从SSH登录到告警推送的完整链路现在我们拆解main.py里最核心的run_inspection()函数。这不是伪代码是实际跑在生产环境里的逻辑每一行都有其存在理由。3.1 设备连接与命令执行如何让Netmiko不卡死在send_commandfrom netmiko import ConnectHandler, NetmikoTimeoutException, NetmikoAuthenticationException from utils.retry_decorator import retry import logging logger logging.getLogger(__name__) retry(max_attempts3, backoff_factor2) def establish_connection(device_params: dict) - ConnectHandler: 建立SSH连接带重试。失败时抛出原始异常由上层捕获 try: connection ConnectHandler(**device_params) connection.enable() # 进入特权模式华为/H3C需此步 logger.info(f✅ 成功连接 {device_params[host]}) return connection except NetmikoTimeoutException as e: logger.error(f❌ 连接超时 {device_params[host]}: {str(e)}) raise except NetmikoAuthenticationException as e: logger.error(f❌ 认证失败 {device_params[host]}: {str(e)}) raise def execute_commands(connection: ConnectHandler, commands: list) - dict: 对单台设备执行多条命令返回 {cmd: output} 字典 results {} for cmd in commands: try: # 关键read_timeout必须大于设备实际响应时间且delay_factor适配 output connection.send_command( cmd, delay_factor2, # 应对慢设备延长读取间隔 read_timeout60, # 单条命令最长等待60秒 strip_promptTrue, # 自动去掉命令回显的提示符 strip_commandTrue # 自动去掉命令本身如display int bri ) results[cmd] output logger.debug(f 执行命令 {cmd[:30]}... - {len(output)} 字符) except Exception as e: logger.warning(f⚠️ 命令执行失败 {cmd}: {str(e)}) results[cmd] fERROR: {str(e)} return results参数说明与踩坑点delay_factor2Netmiko内部会将所有超时乘以这个系数。默认delay_factor1时read_timeout30实际是30秒设为2就是60秒。华三设备在高负载时display transceiver diagnosis可能耗时45秒不调这个参数必然超时。strip_promptTrue华为设备返回HUAWEIdisplay interface brief若不开启解析时会把HUAWEI当接口名的一部分导致TextFSM匹配失败。read_timeout60不是越大越好。设成300秒万一设备彻底卡死整个线程就挂住拖垮并发池。60秒是平衡点——99%的命令在此时间内完成超时即判定设备异常。3.2 TextFSM解析如何让同一段正则在华为/华三/Cisco上都工作解析模块parser.py不写正则而是动态加载TextFSM模板import textfsm from pathlib import Path def parse_with_template(raw_output: str, vendor: str, command: str) - list: 根据厂商和命令名加载对应TextFSM模板进行解析 # 模板路径规则templates/{vendor}/{command_cleaned}.textfsm # 如 commanddisplay transceiver diagnosis - transceiver_diagnosis clean_cmd command.replace( , _).replace(-, _).lower() template_path Path(templates) / vendor / f{clean_cmd}.textfsm if not template_path.exists(): logger.error(f❌ 模板不存在 {template_path}) return [] try: with open(template_path) as f: template textfsm.TextFSM(f) parsed template.ParseText(raw_output) # 将TextFSM返回的二维列表转为字典列表 result [dict(zip(template.header, row)) for row in parsed] logger.debug(f 解析 {vendor} {command}: {len(result)} 条记录) return result except Exception as e: logger.error(f❌ TextFSM解析失败 {template_path}: {str(e)}) return [] # 使用示例 output Interface PHY Temp(C) ... \nGigabitE... UP 42.5 ... parsed parse_with_template(output, vendorhuawei, commanddisplay transceiver diagnosis) # 返回 [{interface: GigabitEthernet0/0/1, temp: 42.5, ...}]为什么不用正则真实案例某次升级华为S5735交换机到V300R022版本display transceiver diagnosis输出从Interface PHY Temp(C) ... GigabitE... UP 42.5 ...变成Interface PHY Temp(°C) ... ← 单位从(C)变成(°C) GigabitE... UP 42.5 ...正则rTemp\(C\)\s(\d\.\d)立刻失效。而TextFSM模板只需改一行Value TEMP (\d\.\d) # 原来匹配 Temp(C)现在改成匹配 Temp(°C) 或 Temp(C) 更宽松 Start ^.*Temp\([^)]*\)\s${TEMP} - Record改一行全量生效。正则要改三处还可能漏掉其他变体。3.3 告警推送企业微信机器人怎么避免被限流告警不是简单requests.post()要考虑频率控制和失败降级import requests import time from utils.logger_config import get_logger alert_logger get_logger(alert) class WeComAlert: def __init__(self, webhook_url: str): self.webhook webhook_url self.last_send_time 0 self.min_interval 60 # 同一机器人每分钟最多发1条 def send(self, title: str, content: str, devices: list): 发送告警带频率限制和重试 now time.time() if now - self.last_send_time self.min_interval: alert_logger.warning(f⏳ 告警被节流距离上次发送不足{self.min_interval}秒) return False payload { msgtype: markdown, markdown: { content: f## {title}\n\n **触发时间**{time.strftime(%Y-%m-%d %H:%M:%S)}\n **影响设备**{, .join(devices)}\n\n{content} } } try: resp requests.post( self.webhook, jsonpayload, timeout10 ) resp.raise_for_status() self.last_send_time now alert_logger.info(f✅ 企业微信告警发送成功{title}) return True except requests.RequestException as e: alert_logger.error(f❌ 企业微信发送失败{str(e)}) return False # 在主流程中调用 wecom WeComAlert(config.WECOM_WEBHOOK) if alerts: wecom.send( titlef 网络设备异常告警{len(alerts)}条, content\n.join([f- {a.device_name}: {a.metric}{a.value}阈值{a.threshold} for a in alerts]), deviceslist(set(a.device_name for a in alerts)) )关键设计min_interval60企业微信机器人官方文档明确要求“同一webhook每分钟最多发送20条”但实测超过5条就开始429。保守设为60秒一条绝对安全。last_send_time内存态记录不依赖外部存储轻量。脚本每次运行都是新进程但单次巡检内多次告警如温度光衰同时超必须去重。raise_for_status()不忽略HTTP错误码。403webhook失效、400payload格式错都会被捕获写入error.log方便排查。4. 避坑指南我在生产环境踩过的5个深坑现在告诉你怎么绕开这些不是教科书理论是凌晨三点盯着日志发现的、差点让我删库跑路的真实问题。每一条都附带复现方式和根治方案。4.1 现象脚本在巡检第12台设备时卡死ps aux显示进程状态为Duninterruptible sleep原因Linux内核中D状态表示进程在等待不可中断的I/O通常是磁盘或网络。Netmiko的send_command()底层调用paramiko.channel.recv()当SSH服务器如老旧华三设备突然断开TCP连接但不发FIN包时recv()会永久阻塞。解决强制给recv()加超时。Netmiko 4.0 支持read_timeout参数但必须配合fast_cliFalse使用。在establish_connection()中添加device_params.update({ read_timeout: 60, fast_cli: False, # 关键fast_cliTrue会禁用read_timeout })4.2 现象华为设备返回中文乱码display environment里温度显示为Temp: 42.5?C原因华为设备默认字符集是GBK而Python 3默认用UTF-8解码。Netmiko收到字节流后用decode(utf-8)强行转换遇到GBK中的汉字字节如0xC40xE3就变成。解决在ConnectHandler初始化时指定encodinggbkdevice_params.update({ encoding: gbk, # 华为/H3C设备专用 # Cisco设备用 utf-8需在devices.yaml中按vendor区分 })提示devices.yaml里加encoding: gbk字段代码中读取后传入。4.3 现象TextFSM解析display transceiver diagnosis时部分光模块信息缺失如Rx Power为空原因TextFSM模板中Value RX_POWER (-?\d\.\d)只能匹配带小数点的数字但某些光模块尤其二手Rx Power显示为-22.50而另一些显示为-22整数。正则\d\.\d不匹配-22。解决放宽正则支持整数和小数Value RX_POWER (-?\d\.?\d*) # 匹配-22.50, -22, 0.5, 1234.4 现象脚本在CentOS 7上运行正常在Ubuntu 22.04上ImportError: No module named textfsm原因Ubuntu 22.04默认Python是3.10而textfsm1.1.0以下版本不兼容Python 3.10。pip install textfsm默认装旧版。解决在requirements.txt中锁定版本textfsm1.1.2 netmiko4.1.0 paramiko3.0.0并在部署脚本中强制升级pip install --upgrade -r requirements.txt4.5 现象企业微信告警发出去了但内容里br被当成HTML标签渲染换行失效原因企业微信Markdown消息中换行必须用两个空格回车\n或用br但需转义。直接写br会被过滤。解决在WeComAlert.send()中预处理content# 将普通换行转为企业微信认可的br content content.replace(\n, br) # 或更稳妥用两个空格回车 content content.replace(\n, \n)5. 二次开发实战如何快速接入新厂商、新增指标、对接Prometheus这个脚本不是终点而是你定制化网管的第一块砖。下面三个改造我都在线上跑过从改到上线不超过1小时。5.1 接入新厂商如锐捷RG-S29003步搞定锐捷设备CLI类似Cisco但提示符是RG-S2900且show environment命令返回格式略有不同。接入步骤在devices.yaml中添加vendor映射Netmiko 4.x 已支持device_type: ruijie_os无需额外代码。只需在设备配置中写- name: rg_sw01 ip: 192.168.1.100 vendor: ruijie_os # 注意不是ruijie是Netmiko注册的类型名 username: admin password: xxx创建TextFSM模板在templates/ruijie_os/下新建environment.textfsm内容基于Cisco模板微调Value DEVICE_NAME (\S) Value TEMP (\d) Value FAN_STATUS (Normal|Abnormal) Start ^Device\sName:\s${DEVICE_NAME} - Continue ^Temperature:\s${TEMP} - Record ^Fan\sStatus:\s${FAN_STATUS} - Record在parser.py中注册新vendorparse_with_template()函数已支持任意vendor无需改代码。只要模板路径对templates/ruijie_os/environment.textfsm自动加载。验证方法把锐捷设备show environment的原始输出存为tests/ruijie_env_sample.txt在本地运行python -m pytest tests/test_parser.py断言解析结果正确。5.2 新增巡检指标如CPU利用率改两处加一个模板现有脚本只查端口、光模块、温度。要加CPU只需Step 1在devices.yaml中为设备添加命令- name: huawei_core ip: 192.168.1.1 vendor: huawei commands: - display interface brief - display transceiver diagnosis - display cpu-usage # 新增Step 2创建templates/huawei/cpu_usage.textfsmValue CPU_USAGE (\d) Start ^CPU\sUsage\s:\s${CPU_USAGE}% - RecordStep 3在告警逻辑中加阈值判断main.py的check_alerts()函数# 解析后得到 parsed_cpu [{cpu_usage: 35}] for item in parsed_cpu: usage int(item[cpu_usage]) if usage device.thresholds.get(cpu_max, 80): alerts.append(Alert( device_namedevice.name, metriccpu_usage, valueusage, thresholddevice.thresholds.get(cpu_max, 80), timestampdatetime.now().isoformat() ))5.3 对接Prometheus暴露/metrics端点让巡检结果变监控指标不想只发告警把巡检数据喂给Prometheus做趋势分析。用prometheus_client库暴露HTTP端点安装依赖pip install prometheus_client在main.py顶部加指标定义from prometheus_client import Counter, Gauge, start_http_server # 定义指标 INSPECTION_SUCCESS Counter(network_inspection_success_total, 巡检成功次数, [device, command]) INSPECTION_FAILURE Counter(network_inspection_failure_total, 巡检失败次数, [device, command, reason]) INTERFACE_UP Gauge(network_interface_up, 接口UP状态, [device, interface]) TEMPERATURE_CELSIUS Gauge(network_temperature_celsius, 设备温度, [device])在命令执行后上报# execute_commands()函数内每成功解析一条命令 for cmd, output in results.items(): if ERROR not in output: INSPECTION_SUCCESS.labels(devicedevice_params[host], commandcmd).inc() # 解析后上报具体指标 if display interface brief in cmd: for intf in parsed: INTERFACE_UP.labels( devicedevice_params[host], interfaceintf[interface] ).set(1 if intf[status] UP else 0)启动HTTP服务在main()末尾if __name__ __main__: # ... 巡检逻辑 start_http_server(8000) # 暴露 http://localhost:8000/metrics print(✅ Prometheus metrics server started on :8000/metrics)然后在Prometheus配置中加job- job_name: network-inspect static_configs: - targets: [your-server-ip:8000]不到20行代码巡检脚本秒变监控数据源。你可以画Grafana看“过去7天各设备CPU峰值”或者设告警规则network_temperature_celsius{devicecore-sw} 75——这才是真正的自动化闭环。从那以后我每次加新设备都强制走一遍pytest tests/本地验证每次改TextFSM模板必用textfsm命令行工具测试样本每次上线新告警先在测试群发一遍确认格式。这些动作加起来不过3分钟却省去了半夜爬起来修脚本的后悔药。希望帮到你。本文还有配套的精品资源点击获取