简介:面向网络运维工程师与Python网络编程学习者,这份资源提供了一套基于telnetlib库的设备自动巡检脚本,可批量连接交换机、路由器等网络设备,执行状态检查并汇总输出,解决人工登录多台设备效率低、易遗漏的问题。包内共7个文件,以核心Python脚本为主,配合5个ini配置文件和1个Markdown说明文档,压缩包整体仅6KB;ini按功能拆分,分别存放华为与Cisco设备的巡检命令、登录密码、网络地址清单及邮件收件人列表,便于按实际环境修改使用。目前已有6140人学习下载,适合希望快速落地设备巡检自动化、减少重复操作的运维人员参考。通过学习脚本整体结构,可以掌握telnet连接、命令下发、输出解析及结果邮件发送的完整实现思路,并进一步扩展至配置备份、故障告警等运维场景。1. Python 脚本自动巡检网络设备:把深夜手工敲命令变成早上看报告某次值班,一台接入交换机 CPU 飙到 90%,同事凌晨爬起来逐条敲命令、复制、粘贴、存成文本,敲到天亮才发现漏了两台设备。那之后我把巡检改成了 Python 脚本自动巡检网络设备:脚本按一份设备清单登录设备、逐条下发命令、把回显落盘并抽出关键指标,最后生成一份汇总报告。适用范围是维护几十台到几百台网络设备的运维场景,适合被重复登录、逐条敲命令、复制粘贴折磨过的人。它解决的不是“配置自动下发”这类变更操作,而是“定期看一眼设备是否还健康”这个低频但必须做的活。2. 先搭巡检框架:SSH 会话管理与命令采集清单设计2.1 巡检对象先变成一张清单,而不是一堆散 IP写脚本之前我习惯先把“要巡检什么”落成一份结构化清单。脚本入口读这份清单,而不是在代码里写死 IP。清单里每个设备需要包含连接地址、登录名、设备角色,以及要执行的命令组。命令组单独定义,方便不同角色设备复用不同的巡检内容。下面这份 JSON 就是我最常用的一种组织方式:{ "devices": [ { "name": "core-sw-01", "host": "192.0.2.11", "username": "ops", "password": "encrypted_placeholder", "device_type": "vendor_a", "command_group": "switch_base" }, { "name": "access-sw-12", "host": "192.0.2.24", "username": "ops", "password": "encrypted_placeholder", "device_type": "vendor_b", "command_group": "switch_base" } ], "command_groups": { "switch_base": [ "display version", "display cpu", "display memory", "display interface" ] } }用 JSON 而不是直接写在 Python 里,好处是加设备不用改代码。凡是新增一台设备,只要在devices数组里加一条记录,改command_group就能让脚本执行不同的命令集合。device_type字段决定 SSH 库用哪种交互协议去协商,不同厂商设备的命令提示符和分页方式不一样,这个字段不能省。密码字段在正式环境里不要让脚本以明文保存,常见做法是读环境变量,或者在清单文件里加密后由脚本解密,这里只作占位。命令分组我一般按三个维度设计:基础信息(版本、序列号)、健康状态(CPU、内存、温度)、链路状态(接口 up/down、错包计数)。版本信息平时用不上,但出故障追溯变更时要靠它,所以每次都采集,成本低、价值大。2.2 会话层:一次连接、逐条命令、完整读回显连接设备这一步是整套脚本的地基。我直接使用网络自动化专用的 SSH 交互库,它比手写 socket 会话省心得多,对分页、命令回显结束符做了封装,底层实际上还是 Paramiko 那套 SSH 实现,但对网络设备的交互习惯做了适配。核心连接和命令执行的骨架是这样:from netmiko import ConnectHandler device = { "device_type": "vendor_a", "host": "192.0.2.11", "username": "ops", "password": "encrypted_placeholder", "timeout": 30, # 单条命令等待回显的超时秒数 "conn_timeout": 15, # TCP 连接建立的超时秒数 "fast_cli": False, # 关闭快速模式,保障每条命令都读到完整回显 } conn = ConnectHandler(**device) for cmd in ["display version", "display cpu", "display interface"]: output = conn.send_command(cmd, read_timeout=30) print(f"===== {cmd} =====") print(output) conn.disconnect()这段代码的逻辑很直白:建立一条 SSH 会话,按顺序逐条发送命令,每次send_command会一直读到命令提示符再次出现为止,然后打印回显。注意循环里是“逐条发送”,不是一次性把所有命令灌进去,因为设备交互终端不支持并发输入。timeout和conn_timeout是两个不同的超时,前者管单条命令的回显,后者管建连阶段,网络差的设备建议把conn_timeout放到 15 秒以上,否则拨号链路偶发延迟会导致误判设备不可达。fast_cli默认开启时,库会做回显缓存来提速,但代价是有可能拿到上一轮的残影输出,巡检场景牺牲一点速度保准确,所以我显式关掉。连接建立后我先发一条设备相关命令关闭分页输出,这个细节放到第 4 章专门讲,它直接关系到最后一条命令会不会卡死。2.3 多设备并发:同一设备必须串行,不同设备才并行几十台设备的巡检如果一台台串行跑,一台快则几秒、慢则几十秒,全部跑完可能要半小时。我的做法是“设备级并行、设备内串行”:用线程池同时登录多台设备,但每台设备内部的命令仍然按顺序执行。这样既缩短整体时间,又不会在同一台设备上抢占会话。线程池的代码不复杂:from concurrent.futures import ThreadPoolExecutor, as_completed def inspect_one(device): try: conn = ConnectHandler(**device) results = {} for cmd in device["command_group"]: results[cmd] = conn.send_command(cmd, read_timeout=30) conn.disconnect() return device["name"], results, None except Exception as exc: return device["name"], None, str(exc) with ThreadPoolExecutor(max_workers=6) as pool: futures = [pool.submit(inspect_one, dev) for dev in devices] for future in as_completed(futures): name, results, err = future.result() if err: print(f"[FAIL] {name}: {err}") else: print(f"[OK] {name}: {len(results)} commands")max_workers=6不是拍脑袋定的,常见设备的远程登录并发上限在 5 到 10 之间,超过后新连接会被直接拒绝,所以保守取 6。如果设备数量少或者设备性能差,降到 3 更稳。每个任务里最关键的约束是:inspect_one函数内部是串行执行命令的,即使多个任务并发跑,同一台设备不会出现两条命令同时发送的情况。凡是看到“同一设备开两个会话分别查 CPU 和查接口”的写法,我建议直接改掉,设备侧的登录模块会因此频繁报错,看起来像密码错了,实际上是被并发挤掉的。3. 采集与解析:从设备回显里抽出可用状态3.1 回显就三种形态:键值、表格、长文本,解析姿势各不同命令回显虽然看起来五花八门,剥开看就三类。第一类是键值对,比如CPU usage is 23%、Memory free: 123456 KB,用正则抽数字和单位就行。第二类是表格,比如接口摘要表,表头一行、数据多行,按列拆分最直接。第三类是长文本,比如完整配置文件,这种不适合做结构化解析,直接落盘留作审计查询。CPU、内存、温度这些健康指标,我通常放在同一个函数里解析:import re def parse_health(output): """从健康类命令回显中抽取 CPU、内存、温度。""" cpu_match = re.search(r"CPU\s+usage\s+is\s+(\d+)%", output, re.IGNORECASE) mem_match = re.search(r"Memory\s+free:\s+(\d+)\s+KB", output, re.IGNORECASE) temp_match = re.search(r"Temperature\s*:\s*(\d+)\s*C", output, re.IGNORECASE) return { "cpu_usage": int(cpu_match.group(1)) if cpu_match else None, "memory_free_kb": int(mem_match.group(1)) if mem_match else None, "temperature_c": int(temp_match.group(1)) if temp_match else None, }这段代码的核心是三个正则。第一个CPU\s+usage\s+is\s+(\d+)%要求“CPU usage is”和数字之间可以有多处空白,数字后面必须跟百分号,这样不会把23%和0.23这种不同单位的输出混淆。第二个匹配Memory free: 123456 KB,注意我显式保留了单位KB,因为有的厂商输出是MB,解析出来要除以 1024 才对得上。第三个温度正则容错差一些,因为不同设备输出格式差异最大,看到解析不出来回None就好,不要抛异常。解析原则是“解析失败留给报告层判断”,不要让单台设备解析异常拖垮整批巡检。3.2 接口状态表:按行拆、按列取、按关键字归类接口摘要类的回显经常长这样,表头固定、每行一个接口,中间可能有空行:Interface Status InErrors OutErrors GigabitEthernet0/1 up 0 12 GigabitEthernet0/2 down 5 0 GigabitEthernet0/3 up 0 0对这种表格输出,我的解析步骤是:先按行分割,跳过空行和表头;然后按空白拆分列;再根据Status列的值归类。代码长一点但结构清晰:def parse_interfaces(output): interfaces = [] for line in output.splitlines(): line = line.strip() if not line or line.startswith("Interface"): continue parts = line.split() if len(parts) < 4: continue name, status, in_errors, out_errors = parts[:4] interfaces.append({ "name": name, "status": status.lower(), "in_errors": int(in_errors), "out_errors": int(out_errors), }) return interfacessplit()默认按任意连续空白切分,这在处理设备表格时比手动指定分隔符可靠,因为厂商回显有时用两个空格、有时用制表符对齐。line.startswith("Interface")用于跳过表头,实际场景里表头文字可能不同,更稳妥的判断是第一列无法转换成接口名格式。这里我为了可读性用了简写,生产脚本里一般会加一层“尝试识别接口名开头”的逻辑。另外要注意一点:InErrors和OutErrors这类列并不是每台设备都输出,有的设备只给up/down状态。字段缺失时字典少两个键就好,不要硬补None,报告阶段会区分“没数据”和“数据为 0”。解析接口状态最大的价值不在 up 接口数,而在错包数。接口 down 是显性故障,错包持续增长是隐性故障,可能影响业务但链路状态还是 up。所以上面代码里把in_errors和out_errors单独解析成数字,后续报告里看到错包持续增长就能提前介入。3.3 解析后的记录统一打平成字典,为报告层铺路巡检脚本最终要生成报告,所以解析结果需要统一收拢成一个扁平的记录结构。我的习惯是每条设备产出一条字典,把健康指标和接口统计放在同一层:def build_device_record(name, health, interfaces): up_count = sum(1 for i in interfaces if i["status"] == "up") down_count = sum(1 for i in interfaces if i["status"] == "down") record = { "device": name, "timestamp": datetime.now().isoformat(timespec="seconds"), "cpu_usage": health["cpu_usage"], "memory_free_kb": health["memory_free_kb"], "temperature_c": health["temperature_c"], "interface_total": len(interfaces), "interface_up": up_count, "interface_down": down_count, "down_interface_names": [i["name"] for i in interfaces if i["status"] == "down"], } return record这个函数把前面解析的两类数据合并成一条设备记录,down_interface_names单独列出来是因为报告里要能直接看到“谁 down 了”,而不是只看到一个数字。interface_total和interface_up、interface_down之间不一定满足严格的等式,因为有些接口状态可能是admin down或deleted,所以不要用up_count + down_count去反推总数。这里顺带提醒一句:解析时保留原始状态值,不要只留一个up/down布尔值,后续判断接口抖动或者状态变更时原始值更有用。4. 巡检避坑:连接超时、分页卡死、编码乱码与表项缺失的排查记录4.1 命令发出后脚本像卡住一样,是分页符把回显截断了现象:脚本执行某条命令时长时间无返回,等了很久才超时,而且超时前拿到的回显只有半截,最后停在--More--那一行。原因:设备默认开启分页输出,命令结果超过一屏时会停下来等待用户按空格继续。SSH 库发送命令后一直等命令提示符重新出现,但提示符被分页符挡住了,等于双方都在等对方。解决:建立会话后第一时间发送关闭分页的命令,然后再发巡检命令。关闭分页的命令在不同设备上写法不同,常见的是terminal length 0或screen-length disable。我把这一步放在连接对象建立之后、命令循环之前:conn = ConnectHandler(**device) conn.send_command("terminal length 0", read_timeout=10) for cmd in command_list: output = conn.send_command(cmd, read_timeout=30)这段代码把关闭分页命令当作“会话初始化”的一部分,和巡检命令分开。要注意read_timeout=10不能设太短,有的设备首次会话还要做协议协商,回显慢一点也正常。从那以后我每写一个适配新设备的脚本,第一件事就是先手工把这条命令敲一遍确认设备接受它。4.2 同一台设备反复登录失败,但手工敲密码能登上去现象:并发巡检时,某个设备连续报认证失败,把并发数降到 1 之后又能正常登录。原因:设备对远程登录连接数有限制,并发一高,新连接占用全部登录槽位,后续连接被拒绝,表现却是“认证失败”而不是“连接被拒绝”。设备侧日志里能看到大量“登录模块资源不足”的报错。解决:把同一台设备的所有巡检命令放进一个会话里执行,而不是拆成多个任务并发登录。线程池并发只用在不同的设备之间。同时把max_workers收敛到 6 以下,如果设备是低端款,甚至要降到 3。巡检是低频操作,跑慢一点没有关系,反复重试反而把设备登录模块拖垮。4.3 中文回显全是乱码,解析正则一个都匹配不上现象:设备名称、描述信息里的中文全部变成??或�,正则匹配不到预期的字面量。原因:设备回显的编码和终端默认编码不一致。常见设备回显是 GBK 编码,而 SSH 库默认按 UTF-8 解码,导致中文字节被错误翻译。解决:连接时显式指定编码,或者在拿到原始字节后按 GBK 解码。SSH 库层面通常有encoding参数:device = { "device_type": "vendor_a", "host": "192.0.2.11", "username": "ops", "password": "encrypted_placeholder", "encoding": "gbk", }指定encoding="gbk"之后,回显里的中文能正常显示,正则匹配中文描述也才能生效。排查这个问题最直接的办法是打印回显的repr看十六进制字节,再判断编码类型。实测经验是:新设备默认 UTF-8,老设备默认 GBK,多厂商环境里干脆统一用一个检测函数去嗅探编码,避免每台设备单独配置。4.4 CPU 值显示 90%,巡检报告却判定正常现象:设备 CPU 已经 90%,但报告输出cpu_usage是None,告警逻辑直接跳过了。原因:两台设备的回显格式不同,一台是CPU usage is 90%,另一台是CPU usage is 90,或者数值和百分号之间有空格。正则要求必须匹配%,后者匹配失败返回None。解决:放宽单位匹配,允许百分号存在或缺失,同时对数值范围做二次校验:cpu_match = re.search(r"CPU\s+usage\s+is\s+(\d+)\s*%?", output, re.IGNORECASE)%?表示百分号可选,\s*容忍数字和百分号之间的空白。改成这样后两种格式都能解析。但要注意另一种翻车情况:有的设备回显是CPU busy: 0.5这种小数格式,(\d+)匹配不到。遇到这种我会把正则升级为(\d+(?:\.\d+)?)并保留原始字符串,在报告层再做类型判断。解析这种事没有一劳永逸的写法,只能靠“采集样本回显,逐条正则对照”来兜底。4.5 接口表少了十几行,但设备上确实有那么多接口现象:用display interface获取接口状态,解析出的接口列表数量比实际少,表格后半段被截断。原因:设备关闭分页不彻底。有的设备terminal length 0只对当前会话有效,但接口数量特别多的时候,命令回显还是被设备侧按输出块断开,SSH 库读到的回显不到完整内容就以为命令结束了。解决:读取结束后校验回显完整性。我的习惯是采集接口命令后,统计回显行数,同时把接口列表最后几行是否以“空白行+结束标记”作为判断依据。更通用的兜底做法是解析完成后做一次数量校验:if len(interfaces) < expected_count: print(f"[WARN] {name}: interfaces parsed {len(interfaces)}, retry once")代码里没有魔法,就是“发现问题、重试一次”。实测里重跑一次绝大多数情况下能拿到完整回显。如果重试两次仍然缺失,就把这条设备标成“回显不完整”,宁可缺失也不要伪造一份错误报告让值班的人白跑一趟。5. 生成巡检报告:文本落盘、CSV 汇总与告警判断5.1 原始回显按设备按日期归档,解析结果丢了还能翻原稿解析结果只保存结构化字段,但回显原文必须完整保留。故障复盘时,只有解析数据没有原文,很多判断做不了。我按“日期/设备名/命令名”的目录结构归档:inspection/ 2026-01-15/ core-sw-01/ display-version.log display-cpu.log display-interface.log落盘逻辑用一个函数统一处理:import datetime from pathlib import Path def save_raw_output(base_dir, device_name, command, output): date_str = datetime.date.today().isoformat() safe_name = command.replace(" ", "-") + ".log" path = Path(base_dir) / date_str / device_name / safe_name path.parent.mkdir(parents=True, exist_ok=True) path.write_text(output, encoding="utf-8") return pathbase_dir是巡检根目录,device_name直接用了清单里的设备名而不是 IP,这样看目录就知道是哪台设备。文件名由命令名转换而来,空格替换成连字符,避免文件名里出现奇怪字符。path.parent.mkdir(parents=True, exist_ok=True)一次性建好两级目录,不需要提前手工创建。这套归档结构的好处是:报告里说不清的问题,打开对应日期目录就能看到原始回显,不再需要重新登录设备。5.2 CSV 汇总表:一屏看完所有设备的状态归档是给回溯用的,日常值班更需要的是一张能扫一眼看完的表。CSV 格式最适合发早报和导入表格工具。汇总表每行一台设备,字段就是第 3 章的build_device_record生成的字典。字段含义示例device设备名core-sw-01timestamp巡检时间2026-01-15T06:00:00cpu_usageCPU 使用率百分比45memory_free_kb空闲内存 KB 数123456temperature_c温度摄氏度42interface_total接口总数48interface_upup 接口数45interface_downdown 接口数3写 CSV 的代码:import csv def write_summary_csv(summary_path, records): fieldnames = [ "device", "timestamp", "cpu_usage", "memory_free_kb", "temperature_c", "interface_total", "interface_up", "interface_down", ] with open(summary_path, "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for record in records: writer.writerow({k: record.get(k) for k in fieldnames})utf-8-sig是刻意选的,Excel 打开 UTF-8 文件乱码,加 BOM 后直接双击就能正确显示中文。csv.DictWriter按fieldnames顺序写列,不用关心字典键的顺序。这里只总结了核心字段,down_interface_names字段因为可能包含多个接口名,直接用逗号拼接写成字符串塞进 CSV 的一列也可以,但更推荐单独输出到文本里,避免 Excel 把列拆乱。5.3 告警判断:规则写在代码里,标注在报告上有了结构化记录,告警判断就是纯规则匹配。我的规则很简单,不搞机器学习,巡检脚本要的是可解释性,规则写在后面这张表里:告警类型触发条件CPU 过载cpu_usage 大于等于 80内存紧张memory_free_kb 小于某个设备阈值高温temperature_c 大于等于 60接口异常interface_down 大于等于 2告警代码:def check_alerts(record): alerts = [] if record["cpu_usage"] is not None and record["cpu_usage"] >= 80: alerts.append("cpu-high") if record["temperature_c"] is not None and record["temperature_c"] >= 60: alerts.append("temp-high") if record["interface_down"] and record["interface_down"] >= 2: alerts.append("multi-interface-down") return alerts for record in records: alerts = check_alerts(record) if alerts: print(f"[ALERT] {record['device']}: {', '.join(alerts)}")告警逻辑里先判None再比大小,避免某台设备没采到温度数据时直接报高温。interface_down >= 2这个阈值我是按机房冗余链路设计的,单条链路 down 可能是业务方操作,两条及以上同时 down 基本是设备侧问题。阈值不要硬编码在函数里,建议放到清单文件的alert_config段,不同网络规模阈值不同,稍后调整也方便。这段代码执行完后,终端上会列出所有告警设备,配合 CSV 汇总表,早报内容就够了。6. 进阶:把巡检脚本接进值班流6.1 用命令行参数指定设备范围,不要改代码巡检脚本刚写完时还很“个人工具”,参数全在代码里。我后来加了命令行入口,指定设备清单文件和输出目录:import argparse parser = argparse.ArgumentParser(description="network inspection") parser.add_argument("--inventory", required=True, help="path to device JSON") parser.add_argument("--outdir", default="./inspection", help="output base dir") args = parser.parse_args()这样调度系统或者手动执行时只要改参数,不用动脚本。清单文件还可以按区域拆分,核心设备和接入设备分开巡检,频率不同。6.2 状态增量对比:接口 down 不重要,多了哪些 down 才重要定时巡检跑起来之后,报告看多了会发现一个现象:某条接口长期 down,每天早报都有它,看多了就麻木了。真正需要关注的是“这次巡检相比上次,哪些接口状态变了”。我加了一个函数对比两次巡检的接口状态:def diff_interfaces(previous, current): changed = [] for name in set(previous) | set(current): if previous.get(name) != current.get(name): changed.append((name, previous.get(name), current.get(name))) return changedset(previous) | set(current)取两个字典的键合并,不存在的接口用None表示。对比结果会单独输出一段“状态变化”清单,列在报告最前面。这个函数的价值在于把“巡检快照”变成了“变更记录”,对值班早报来说,变化才是需要行动的信号。把脚本接进值班流之后,我养成了一个习惯:每次上线新设备类型,先手工登录这台设备敲一遍清单里的命令,把回显存下来和脚本结果逐行对照。解析正则对不对、分页有没有关闭、编码对不对,这个环节全都能暴露出来。从那以后我每次写新适配,都强制走一遍对照流程,不再凭感觉调参。这套脚本陪伴我度过了很多个不用半夜爬起来敲命令的晚上,希望这篇笔记里的流程和坑位对你也有用。本文还有配套的精品资源,点击获取
简介:面向网络运维工程师与Python网络编程学习者,这份资源提供了一套基于telnetlib库的设备自动巡检脚本,可批量连接交换机、路由器等网络设备,执行状态检查并汇总输出,解决人工登录多台设备效率低、易遗漏的问题。包内共7个文件,以核心Python脚本为主,配合5个ini配置文件和1个Markdown说明文档,压缩包整体仅6KB;ini按功能拆分,分别存放华为与Cisco设备的巡检命令、登录密码、网络地址清单及邮件收件人列表,便于按实际环境修改使用。目前已有6140人学习下载,适合希望快速落地设备巡检自动化、减少重复操作的运维人员参考。通过学习脚本整体结构,可以掌握telnet连接、命令下发、输出解析及结果邮件发送的完整实现思路,并进一步扩展至配置备份、故障告警等运维场景。1. Python 脚本自动巡检网络设备:把深夜手工敲命令变成早上看报告某次值班,一台接入交换机 CPU 飙到 90%,同事凌晨爬起来逐条敲命令、复制、粘贴、存成文本,敲到天亮才发现漏了两台设备。那之后我把巡检改成了 Python 脚本自动巡检网络设备:脚本按一份设备清单登录设备、逐条下发命令、把回显落盘并抽出关键指标,最后生成一份汇总报告。适用范围是维护几十台到几百台网络设备的运维场景,适合被重复登录、逐条敲命令、复制粘贴折磨过的人。它解决的不是“配置自动下发”这类变更操作,而是“定期看一眼设备是否还健康”这个低频但必须做的活。2. 先搭巡检框架:SSH 会话管理与命令采集清单设计2.1 巡检对象先变成一张清单,而不是一堆散 IP写脚本之前我习惯先把“要巡检什么”落成一份结构化清单。脚本入口读这份清单,而不是在代码里写死 IP。清单里每个设备需要包含连接地址、登录名、设备角色,以及要执行的命令组。命令组单独定义,方便不同角色设备复用不同的巡检内容。下面这份 JSON 就是我最常用的一种组织方式:{ "devices": [ { "name": "core-sw-01", "host": "192.0.2.11", "username": "ops", "password": "encrypted_placeholder", "device_type": "vendor_a", "command_group": "switch_base" }, { "name": "access-sw-12", "host": "192.0.2.24", "username": "ops", "password": "encrypted_placeholder", "device_type": "vendor_b", "command_group": "switch_base" } ], "command_groups": { "switch_base": [ "display version", "display cpu", "display memory", "display interface" ] } }用 JSON 而不是直接写在 Python 里,好处是加设备不用改代码。凡是新增一台设备,只要在devices数组里加一条记录,改command_group就能让脚本执行不同的命令集合。device_type字段决定 SSH 库用哪种交互协议去协商,不同厂商设备的命令提示符和分页方式不一样,这个字段不能省。密码字段在正式环境里不要让脚本以明文保存,常见做法是读环境变量,或者在清单文件里加密后由脚本解密,这里只作占位。命令分组我一般按三个维度设计:基础信息(版本、序列号)、健康状态(CPU、内存、温度)、链路状态(接口 up/down、错包计数)。版本信息平时用不上,但出故障追溯变更时要靠它,所以每次都采集,成本低、价值大。2.2 会话层:一次连接、逐条命令、完整读回显连接设备这一步是整套脚本的地基。我直接使用网络自动化专用的 SSH 交互库,它比手写 socket 会话省心得多,对分页、命令回显结束符做了封装,底层实际上还是 Paramiko 那套 SSH 实现,但对网络设备的交互习惯做了适配。核心连接和命令执行的骨架是这样:from netmiko import ConnectHandler device = { "device_type": "vendor_a", "host": "192.0.2.11", "username": "ops", "password": "encrypted_placeholder", "timeout": 30, # 单条命令等待回显的超时秒数 "conn_timeout": 15, # TCP 连接建立的超时秒数 "fast_cli": False, # 关闭快速模式,保障每条命令都读到完整回显 } conn = ConnectHandler(**device) for cmd in ["display version", "display cpu", "display interface"]: output = conn.send_command(cmd, read_timeout=30) print(f"===== {cmd} =====") print(output) conn.disconnect()这段代码的逻辑很直白:建立一条 SSH 会话,按顺序逐条发送命令,每次send_command会一直读到命令提示符再次出现为止,然后打印回显。注意循环里是“逐条发送”,不是一次性把所有命令灌进去,因为设备交互终端不支持并发输入。timeout和conn_timeout是两个不同的超时,前者管单条命令的回显,后者管建连阶段,网络差的设备建议把conn_timeout放到 15 秒以上,否则拨号链路偶发延迟会导致误判设备不可达。fast_cli默认开启时,库会做回显缓存来提速,但代价是有可能拿到上一轮的残影输出,巡检场景牺牲一点速度保准确,所以我显式关掉。连接建立后我先发一条设备相关命令关闭分页输出,这个细节放到第 4 章专门讲,它直接关系到最后一条命令会不会卡死。2.3 多设备并发:同一设备必须串行,不同设备才并行几十台设备的巡检如果一台台串行跑,一台快则几秒、慢则几十秒,全部跑完可能要半小时。我的做法是“设备级并行、设备内串行”:用线程池同时登录多台设备,但每台设备内部的命令仍然按顺序执行。这样既缩短整体时间,又不会在同一台设备上抢占会话。线程池的代码不复杂:from concurrent.futures import ThreadPoolExecutor, as_completed def inspect_one(device): try: conn = ConnectHandler(**device) results = {} for cmd in device["command_group"]: results[cmd] = conn.send_command(cmd, read_timeout=30) conn.disconnect() return device["name"], results, None except Exception as exc: return device["name"], None, str(exc) with ThreadPoolExecutor(max_workers=6) as pool: futures = [pool.submit(inspect_one, dev) for dev in devices] for future in as_completed(futures): name, results, err = future.result() if err: print(f"[FAIL] {name}: {err}") else: print(f"[OK] {name}: {len(results)} commands")max_workers=6不是拍脑袋定的,常见设备的远程登录并发上限在 5 到 10 之间,超过后新连接会被直接拒绝,所以保守取 6。如果设备数量少或者设备性能差,降到 3 更稳。每个任务里最关键的约束是:inspect_one函数内部是串行执行命令的,即使多个任务并发跑,同一台设备不会出现两条命令同时发送的情况。凡是看到“同一设备开两个会话分别查 CPU 和查接口”的写法,我建议直接改掉,设备侧的登录模块会因此频繁报错,看起来像密码错了,实际上是被并发挤掉的。3. 采集与解析:从设备回显里抽出可用状态3.1 回显就三种形态:键值、表格、长文本,解析姿势各不同命令回显虽然看起来五花八门,剥开看就三类。第一类是键值对,比如CPU usage is 23%、Memory free: 123456 KB,用正则抽数字和单位就行。第二类是表格,比如接口摘要表,表头一行、数据多行,按列拆分最直接。第三类是长文本,比如完整配置文件,这种不适合做结构化解析,直接落盘留作审计查询。CPU、内存、温度这些健康指标,我通常放在同一个函数里解析:import re def parse_health(output): """从健康类命令回显中抽取 CPU、内存、温度。""" cpu_match = re.search(r"CPU\s+usage\s+is\s+(\d+)%", output, re.IGNORECASE) mem_match = re.search(r"Memory\s+free:\s+(\d+)\s+KB", output, re.IGNORECASE) temp_match = re.search(r"Temperature\s*:\s*(\d+)\s*C", output, re.IGNORECASE) return { "cpu_usage": int(cpu_match.group(1)) if cpu_match else None, "memory_free_kb": int(mem_match.group(1)) if mem_match else None, "temperature_c": int(temp_match.group(1)) if temp_match else None, }这段代码的核心是三个正则。第一个CPU\s+usage\s+is\s+(\d+)%要求“CPU usage is”和数字之间可以有多处空白,数字后面必须跟百分号,这样不会把23%和0.23这种不同单位的输出混淆。第二个匹配Memory free: 123456 KB,注意我显式保留了单位KB,因为有的厂商输出是MB,解析出来要除以 1024 才对得上。第三个温度正则容错差一些,因为不同设备输出格式差异最大,看到解析不出来回None就好,不要抛异常。解析原则是“解析失败留给报告层判断”,不要让单台设备解析异常拖垮整批巡检。3.2 接口状态表:按行拆、按列取、按关键字归类接口摘要类的回显经常长这样,表头固定、每行一个接口,中间可能有空行:Interface Status InErrors OutErrors GigabitEthernet0/1 up 0 12 GigabitEthernet0/2 down 5 0 GigabitEthernet0/3 up 0 0对这种表格输出,我的解析步骤是:先按行分割,跳过空行和表头;然后按空白拆分列;再根据Status列的值归类。代码长一点但结构清晰:def parse_interfaces(output): interfaces = [] for line in output.splitlines(): line = line.strip() if not line or line.startswith("Interface"): continue parts = line.split() if len(parts) < 4: continue name, status, in_errors, out_errors = parts[:4] interfaces.append({ "name": name, "status": status.lower(), "in_errors": int(in_errors), "out_errors": int(out_errors), }) return interfacessplit()默认按任意连续空白切分,这在处理设备表格时比手动指定分隔符可靠,因为厂商回显有时用两个空格、有时用制表符对齐。line.startswith("Interface")用于跳过表头,实际场景里表头文字可能不同,更稳妥的判断是第一列无法转换成接口名格式。这里我为了可读性用了简写,生产脚本里一般会加一层“尝试识别接口名开头”的逻辑。另外要注意一点:InErrors和OutErrors这类列并不是每台设备都输出,有的设备只给up/down状态。字段缺失时字典少两个键就好,不要硬补None,报告阶段会区分“没数据”和“数据为 0”。解析接口状态最大的价值不在 up 接口数,而在错包数。接口 down 是显性故障,错包持续增长是隐性故障,可能影响业务但链路状态还是 up。所以上面代码里把in_errors和out_errors单独解析成数字,后续报告里看到错包持续增长就能提前介入。3.3 解析后的记录统一打平成字典,为报告层铺路巡检脚本最终要生成报告,所以解析结果需要统一收拢成一个扁平的记录结构。我的习惯是每条设备产出一条字典,把健康指标和接口统计放在同一层:def build_device_record(name, health, interfaces): up_count = sum(1 for i in interfaces if i["status"] == "up") down_count = sum(1 for i in interfaces if i["status"] == "down") record = { "device": name, "timestamp": datetime.now().isoformat(timespec="seconds"), "cpu_usage": health["cpu_usage"], "memory_free_kb": health["memory_free_kb"], "temperature_c": health["temperature_c"], "interface_total": len(interfaces), "interface_up": up_count, "interface_down": down_count, "down_interface_names": [i["name"] for i in interfaces if i["status"] == "down"], } return record这个函数把前面解析的两类数据合并成一条设备记录,down_interface_names单独列出来是因为报告里要能直接看到“谁 down 了”,而不是只看到一个数字。interface_total和interface_up、interface_down之间不一定满足严格的等式,因为有些接口状态可能是admin down或deleted,所以不要用up_count + down_count去反推总数。这里顺带提醒一句:解析时保留原始状态值,不要只留一个up/down布尔值,后续判断接口抖动或者状态变更时原始值更有用。4. 巡检避坑:连接超时、分页卡死、编码乱码与表项缺失的排查记录4.1 命令发出后脚本像卡住一样,是分页符把回显截断了现象:脚本执行某条命令时长时间无返回,等了很久才超时,而且超时前拿到的回显只有半截,最后停在--More--那一行。原因:设备默认开启分页输出,命令结果超过一屏时会停下来等待用户按空格继续。SSH 库发送命令后一直等命令提示符重新出现,但提示符被分页符挡住了,等于双方都在等对方。解决:建立会话后第一时间发送关闭分页的命令,然后再发巡检命令。关闭分页的命令在不同设备上写法不同,常见的是terminal length 0或screen-length disable。我把这一步放在连接对象建立之后、命令循环之前:conn = ConnectHandler(**device) conn.send_command("terminal length 0", read_timeout=10) for cmd in command_list: output = conn.send_command(cmd, read_timeout=30)这段代码把关闭分页命令当作“会话初始化”的一部分,和巡检命令分开。要注意read_timeout=10不能设太短,有的设备首次会话还要做协议协商,回显慢一点也正常。从那以后我每写一个适配新设备的脚本,第一件事就是先手工把这条命令敲一遍确认设备接受它。4.2 同一台设备反复登录失败,但手工敲密码能登上去现象:并发巡检时,某个设备连续报认证失败,把并发数降到 1 之后又能正常登录。原因:设备对远程登录连接数有限制,并发一高,新连接占用全部登录槽位,后续连接被拒绝,表现却是“认证失败”而不是“连接被拒绝”。设备侧日志里能看到大量“登录模块资源不足”的报错。解决:把同一台设备的所有巡检命令放进一个会话里执行,而不是拆成多个任务并发登录。线程池并发只用在不同的设备之间。同时把max_workers收敛到 6 以下,如果设备是低端款,甚至要降到 3。巡检是低频操作,跑慢一点没有关系,反复重试反而把设备登录模块拖垮。4.3 中文回显全是乱码,解析正则一个都匹配不上现象:设备名称、描述信息里的中文全部变成??或�,正则匹配不到预期的字面量。原因:设备回显的编码和终端默认编码不一致。常见设备回显是 GBK 编码,而 SSH 库默认按 UTF-8 解码,导致中文字节被错误翻译。解决:连接时显式指定编码,或者在拿到原始字节后按 GBK 解码。SSH 库层面通常有encoding参数:device = { "device_type": "vendor_a", "host": "192.0.2.11", "username": "ops", "password": "encrypted_placeholder", "encoding": "gbk", }指定encoding="gbk"之后,回显里的中文能正常显示,正则匹配中文描述也才能生效。排查这个问题最直接的办法是打印回显的repr看十六进制字节,再判断编码类型。实测经验是:新设备默认 UTF-8,老设备默认 GBK,多厂商环境里干脆统一用一个检测函数去嗅探编码,避免每台设备单独配置。4.4 CPU 值显示 90%,巡检报告却判定正常现象:设备 CPU 已经 90%,但报告输出cpu_usage是None,告警逻辑直接跳过了。原因:两台设备的回显格式不同,一台是CPU usage is 90%,另一台是CPU usage is 90,或者数值和百分号之间有空格。正则要求必须匹配%,后者匹配失败返回None。解决:放宽单位匹配,允许百分号存在或缺失,同时对数值范围做二次校验:cpu_match = re.search(r"CPU\s+usage\s+is\s+(\d+)\s*%?", output, re.IGNORECASE)%?表示百分号可选,\s*容忍数字和百分号之间的空白。改成这样后两种格式都能解析。但要注意另一种翻车情况:有的设备回显是CPU busy: 0.5这种小数格式,(\d+)匹配不到。遇到这种我会把正则升级为(\d+(?:\.\d+)?)并保留原始字符串,在报告层再做类型判断。解析这种事没有一劳永逸的写法,只能靠“采集样本回显,逐条正则对照”来兜底。4.5 接口表少了十几行,但设备上确实有那么多接口现象:用display interface获取接口状态,解析出的接口列表数量比实际少,表格后半段被截断。原因:设备关闭分页不彻底。有的设备terminal length 0只对当前会话有效,但接口数量特别多的时候,命令回显还是被设备侧按输出块断开,SSH 库读到的回显不到完整内容就以为命令结束了。解决:读取结束后校验回显完整性。我的习惯是采集接口命令后,统计回显行数,同时把接口列表最后几行是否以“空白行+结束标记”作为判断依据。更通用的兜底做法是解析完成后做一次数量校验:if len(interfaces) < expected_count: print(f"[WARN] {name}: interfaces parsed {len(interfaces)}, retry once")代码里没有魔法,就是“发现问题、重试一次”。实测里重跑一次绝大多数情况下能拿到完整回显。如果重试两次仍然缺失,就把这条设备标成“回显不完整”,宁可缺失也不要伪造一份错误报告让值班的人白跑一趟。5. 生成巡检报告:文本落盘、CSV 汇总与告警判断5.1 原始回显按设备按日期归档,解析结果丢了还能翻原稿解析结果只保存结构化字段,但回显原文必须完整保留。故障复盘时,只有解析数据没有原文,很多判断做不了。我按“日期/设备名/命令名”的目录结构归档:inspection/ 2026-01-15/ core-sw-01/ display-version.log display-cpu.log display-interface.log落盘逻辑用一个函数统一处理:import datetime from pathlib import Path def save_raw_output(base_dir, device_name, command, output): date_str = datetime.date.today().isoformat() safe_name = command.replace(" ", "-") + ".log" path = Path(base_dir) / date_str / device_name / safe_name path.parent.mkdir(parents=True, exist_ok=True) path.write_text(output, encoding="utf-8") return pathbase_dir是巡检根目录,device_name直接用了清单里的设备名而不是 IP,这样看目录就知道是哪台设备。文件名由命令名转换而来,空格替换成连字符,避免文件名里出现奇怪字符。path.parent.mkdir(parents=True, exist_ok=True)一次性建好两级目录,不需要提前手工创建。这套归档结构的好处是:报告里说不清的问题,打开对应日期目录就能看到原始回显,不再需要重新登录设备。5.2 CSV 汇总表:一屏看完所有设备的状态归档是给回溯用的,日常值班更需要的是一张能扫一眼看完的表。CSV 格式最适合发早报和导入表格工具。汇总表每行一台设备,字段就是第 3 章的build_device_record生成的字典。字段含义示例device设备名core-sw-01timestamp巡检时间2026-01-15T06:00:00cpu_usageCPU 使用率百分比45memory_free_kb空闲内存 KB 数123456temperature_c温度摄氏度42interface_total接口总数48interface_upup 接口数45interface_downdown 接口数3写 CSV 的代码:import csv def write_summary_csv(summary_path, records): fieldnames = [ "device", "timestamp", "cpu_usage", "memory_free_kb", "temperature_c", "interface_total", "interface_up", "interface_down", ] with open(summary_path, "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for record in records: writer.writerow({k: record.get(k) for k in fieldnames})utf-8-sig是刻意选的,Excel 打开 UTF-8 文件乱码,加 BOM 后直接双击就能正确显示中文。csv.DictWriter按fieldnames顺序写列,不用关心字典键的顺序。这里只总结了核心字段,down_interface_names字段因为可能包含多个接口名,直接用逗号拼接写成字符串塞进 CSV 的一列也可以,但更推荐单独输出到文本里,避免 Excel 把列拆乱。5.3 告警判断:规则写在代码里,标注在报告上有了结构化记录,告警判断就是纯规则匹配。我的规则很简单,不搞机器学习,巡检脚本要的是可解释性,规则写在后面这张表里:告警类型触发条件CPU 过载cpu_usage 大于等于 80内存紧张memory_free_kb 小于某个设备阈值高温temperature_c 大于等于 60接口异常interface_down 大于等于 2告警代码:def check_alerts(record): alerts = [] if record["cpu_usage"] is not None and record["cpu_usage"] >= 80: alerts.append("cpu-high") if record["temperature_c"] is not None and record["temperature_c"] >= 60: alerts.append("temp-high") if record["interface_down"] and record["interface_down"] >= 2: alerts.append("multi-interface-down") return alerts for record in records: alerts = check_alerts(record) if alerts: print(f"[ALERT] {record['device']}: {', '.join(alerts)}")告警逻辑里先判None再比大小,避免某台设备没采到温度数据时直接报高温。interface_down >= 2这个阈值我是按机房冗余链路设计的,单条链路 down 可能是业务方操作,两条及以上同时 down 基本是设备侧问题。阈值不要硬编码在函数里,建议放到清单文件的alert_config段,不同网络规模阈值不同,稍后调整也方便。这段代码执行完后,终端上会列出所有告警设备,配合 CSV 汇总表,早报内容就够了。6. 进阶:把巡检脚本接进值班流6.1 用命令行参数指定设备范围,不要改代码巡检脚本刚写完时还很“个人工具”,参数全在代码里。我后来加了命令行入口,指定设备清单文件和输出目录:import argparse parser = argparse.ArgumentParser(description="network inspection") parser.add_argument("--inventory", required=True, help="path to device JSON") parser.add_argument("--outdir", default="./inspection", help="output base dir") args = parser.parse_args()这样调度系统或者手动执行时只要改参数,不用动脚本。清单文件还可以按区域拆分,核心设备和接入设备分开巡检,频率不同。6.2 状态增量对比:接口 down 不重要,多了哪些 down 才重要定时巡检跑起来之后,报告看多了会发现一个现象:某条接口长期 down,每天早报都有它,看多了就麻木了。真正需要关注的是“这次巡检相比上次,哪些接口状态变了”。我加了一个函数对比两次巡检的接口状态:def diff_interfaces(previous, current): changed = [] for name in set(previous) | set(current): if previous.get(name) != current.get(name): changed.append((name, previous.get(name), current.get(name))) return changedset(previous) | set(current)取两个字典的键合并,不存在的接口用None表示。对比结果会单独输出一段“状态变化”清单,列在报告最前面。这个函数的价值在于把“巡检快照”变成了“变更记录”,对值班早报来说,变化才是需要行动的信号。把脚本接进值班流之后,我养成了一个习惯:每次上线新设备类型,先手工登录这台设备敲一遍清单里的命令,把回显存下来和脚本结果逐行对照。解析正则对不对、分页有没有关闭、编码对不对,这个环节全都能暴露出来。从那以后我每次写新适配,都强制走一遍对照流程,不再凭感觉调参。这套脚本陪伴我度过了很多个不用半夜爬起来敲命令的晚上,希望这篇笔记里的流程和坑位对你也有用。本文还有配套的精品资源,点击获取