简介：一套面向网络运维工程师与Python学习者的网络设备巡检自动化工具，基于telnetlib标准库实现远程登录与命令执行，可解决多厂商路由器、交换机日常状态检查中人工登录效率低、易遗漏的问题。压缩包内共七个文件，包含一个可直接运行的Python主脚本、五个INI格式配置文件（分别管理设备地址、登录口令、巡检命令及邮件接收人）和一份 Markdown 说明文档，整体大小仅6KB，结构清晰。脚本覆盖从认证登录、命令下发、输出解析到结果存储与邮件报告发送的完整流程，同时内置异常处理逻辑，并预留定时任务扩展思路，便于二次开发。已有6140人学习借鉴，适合具备Python基础、希望快速搭建内网设备巡检机制的网络管理员直接套用，也可作为学习telnetlib网络编程的实用范例，并提供了清晰的二次开发入口。

1. 用脚本替代手工巡检：下班前半小时跑完五十台设备

每天下午四点，网工或运维都要做同一件事：登录设备、敲 show 命令、看日志、查状态、记到表格里。五十台设备一台台 telnet 过去输指令，快的话一个半小时，赶上设备响应慢或者回显刷屏，两小时也打不住。这个项目的思路很直接——既然这些命令是固定的，就让 Python 脚本替你登录、执行、抓回显、整理成报告。

项目定位是一套纯 Python 的自动巡检框架：用 paramiko 做 SSH 连接，支持多台设备并发执行巡检命令，把结果输出成结构化报告。它适合两类人：一是每天被重复巡检折磨的运维工程师，想先把机械化操作交给脚本，留出时间处理真正的问题；二是刚开始做自动化运维的 Python 开发者，拿一套完整的巡检链路做参考，看连接管理、命令交互、并发调度、异常兜底是怎么组织在一起的。整体代码量不大，但覆盖了一条完整的自动化巡检链路：设备清单解析、SSH 会话建立、命令批量下发、结果收集、报告生成。

2. 技术选型与整体架构：paramiko 做连接层，自带调度器管并发

2.1 为什么选 paramiko 而不是 netmiko

Python 做网络设备自动化，绕不开两个库：paramiko 和 netmiko。这个项目选的是 paramiko，因为它的定位更接近底层协议实现，看代码能理解 SSH 交互的原始过程，二次开发时不会觉得被框架绑住。

paramiko 做的事情可以概括成三块：建立 SSH 连接、维护会话通道、收发数据。它不需要额外启动外部进程，纯 Python 实现，部署时只要 pip 装上就行。netmiko 则是基于 paramiko 封装的平台适配层，好处是命令交互细节被抹平了，但换来的是排障时要多翻一层封装逻辑。这个项目的设备种类不算多，常见的交换机、路由器、防火墙几类，paramiko 手写分发规则完全够用。

项目对 paramiko 的使用是包装成一个连接管理类，把连接建立、调用、断开封装成上下文管理器。核心思路是让上层调用方不关心 SSH 连接的底层细节，只传入设备信息和要执行的命令，内部维护认证和会话语义。

import paramiko class SSHClient: def __init__(self, host, port, username, password, timeout=10): self.host = host self.port = port self.username = username self.password = password self.timeout = timeout self.client = None def __enter__(self): self.connect() return self def __exit__(self, exc_type, exc_val, exc_tb): if self.client: self.client.close() return False def connect(self): self.client = paramiko.SSHClient() self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) self.client.connect( hostname=self.host, port=self.port, username=self.username, password=self.password, timeout=self.timeout, look_for_keys=False, allow_agent=False )

这段代码的关键在设计上：用__enter__和__exit__做上下文管理，确保设备连接用完后一定关闭，避免长时间巡检后文件描述符泄漏。set_missing_host_key_policy(AutoAddPolicy())的意思是首次连接时自动接受陌生主机指纹，这在设备 IP 经常轮换的内网场景里很实用，但也意味着有中间人攻击的风险——如果对安全要求高，改成WarningPolicy或手动维护 known_hosts 更稳妥。

look_for_keys=False和allow_agent=False这两个参数很多人会漏：前者禁掉自动读取本地私钥文件，后者禁掉 SSH agent 代理认证。显式关掉它们，可以保证认证时只走账密，避免本机私钥干扰导致认证行为不确定。

2.2 协议通道与命令执行：exec_command 和交互式 shell 怎么选

paramiko 执行远程命令有两种方式：exec_command和invoke_shell。前者每条命令独立建一个远程进程，一次性返回输出，适合执行完就走的批处理命令；后者开一个持久化 shell，像人坐终端前一样逐条敲命令，适合需要切换 enable 模式或分页交互的场景。

这个项目对两种方式做了一个约定：普通巡检命令用exec_command，需要进入特权模式或带交互流程的设备单独走invoke_shell。原因是常见的厂商设备执行 show 命令不需要常驻会话，独立通道反而好控制超时和结果拼接。但类似华为设备的display命令如果开启分页，输出会被截断，这时候就得靠交互式 shell 发送screen-length 0 temporary关掉分页。

def exec_commands(self, commands, wait=2): results = {} for cmd in commands: stdin, stdout, stderr = self.client.exec_command(cmd, timeout=self.timeout) exit_code = stdout.channel.recv_exit_status() output = stdout.read().decode('utf-8', errors='replace') if exit_code != 0: output = f"exit_code={exit_code}\n" + output results[cmd] = output time.sleep(wait) return results

注意这段里的两个设计细节：recv_exit_status()会阻塞直到远程命令结束，拿到退出码后结合 stdout 内容判断这条命令是否执行成功；errors='replace'处理编码问题时不会让脚本崩掉——设备回显里常有非 UTF-8 字符，直接 decode 会抛异常，替换成占位符更稳。命令之间加了 sleep，是为了避免连续下发时设备 CPU 过高或回显粘连。

2.3 并发调度与任务拆分：线程池按设备拆分巡检单元

单台设备依次巡检，五十台设备平均每台 30 秒，跑完约二十五分钟。如果每个设备的巡检之间没有依赖关系，完全可以上并发。项目用concurrent.futures.ThreadPoolExecutor做并发，把每台设备的一次巡检定义成一个独立任务单元。

from concurrent.futures import ThreadPoolExecutor, as_completed def run_inspection(device, commands): result = {} try: with SSHClient(**device) as ssh: result['device'] = device['host'] result['status'] = 'success' result['data'] = ssh.exec_commands(commands) except Exception as e: result['device'] = device['host'] result['status'] = 'failed' result['error'] = str(e) return result def batch_inspect(devices, commands, max_workers=10): results = [] with ThreadPoolExecutor(max_workers=max_workers) as pool: future_map = {pool.submit(run_inspection, dev, commands): dev for dev in devices} for future in as_completed(future_map): results.append(future.result()) return results

max_workers是并发上限。它不代表随便调大就能更快，设备 CPU 和带宽有限，并发太高会导致部分设备连接超时或报错。常见做法是控制在 5 到 15 之间，先按 10 跑一轮，观察失败率再调。as_completed的好处是结果回来一个处理一个，不需要等最慢的设备完成，报告可以增量生成。

这个并发模型有一个值得注意的点：每个设备任务里先捕获异常，保证一台设备挂了不会拖垮整批。

3. 核心模块拆解：命令匹配、设备清单解析和报告输出

3.1 设备清单解析

设备的 IP、用户名、密码、端口、设备型号这些信息，项目里用一个 YAML 文件维护。选 YAML 而不是写死在代码里，原因是设备信息经常变，开关机器、改密码都是常态，让运维直接改外部配置比改代码门槛低得多。

devices: - host: 192.168.10.1 port: 22 username: admin password: securepass device_type: huawei enable_password: admin@123 - host: 192.168.10.2 port: 22 username: admin password: anotherpass device_type: h3c

解析这块用 PyYAML 加载，然后做一层字段校验，缺 host 或 username 的设备直接跳过并给出日志提示。device_type这个字段是给命令映射用的，不同厂商的命令词不一样：思科的show version，华为对应display version；思科接口状态用show interface status，华为是display interface brief。设备型号不匹配会得到一堆命令错误提示，所以命令表要和设备类型严格对应。

3.2 命令分组与按设备类型下发

把命令和设备类型绑定，是这个项目比较实用的设计。每类设备维护一份命令清单，巡检时根据device_type选取对应的命令列表执行。

COMMAND_MAP = { "huawei": [ "display version", "display device", "display interface brief", "display cpu-usage", "display memory-usage", "display alarm active", ], "h3c": [ "display version", "display device manuinfo", "display interface brief", "display cpu-usage", "display memory", "display alarm", ], "cisco": [ "show version", "show inventory", "show ip interface brief", "show processes cpu", "show memory statistics", "show logging summary", ], }

命令清单的设计考虑是：每条命令抓的信息类别要能对应到巡检项的某个维度——版本信息用于资产核对，接口状态用于判断链路抖动和端口 down，CPU 和内存用于容量风险预警，告警信息是故障排查入口。后续想加新设备类型，只需要在映射表里加一个 key，填充对应的命令，不用改动执行逻辑。

3.3 报告生成：把原始回显整理成能直接用的巡检记录

设备执行完命令返回的是原始文本，直接看也可以，但五十台设备的回显堆在一起，找问题要翻很久。项目把结果整理成 HTML 报告，按设备分组，每台设备下面列出每条命令的返回，关键指标抽出来放在设备头部摘要区。

报告生成的核心代码大致是这个思路：

def generate_report(results, output_path="report.html"): html_parts = [] html_parts.append("<html><head><meta charset='utf-8'></head><body>") for device_result in results: device = device_result['device'] if device_result['status'] != 'success': html_parts.append(f"<h3>{device} - 巡检失败</h3>") html_parts.append(f"<pre>{device_result.get('error', '')}</pre>") continue html_parts.append(f"<h3>{device} - 正常</h3>") for cmd, output in device_result['data'].items(): html_parts.append(f"<h4>{cmd}</h4>") html_parts.append(f"<pre>{output}</pre>") html_parts.append("</body></html>") with open(output_path, "w", encoding="utf-8") as f: f.write("\n".join(html_parts))

这个模块有几个可以扩展的点。报告目前是静态 HTML，如果想接到现有监控平台，只需要把数据结构改成 JSON 或 CSV 再输出。巡检结果里如果存在连续多条命令返回exit_code != 0的设备，可以单独标记为异常设备，报告开头加一段汇总——这些都可以在generate_report内部加逻辑，不需要动执行链。

4. 部署与运行：从克隆到产出第一份报告的完整流程

4.1 环境准备和依赖安装

项目对 Python 版本没有太高要求，3.8 以上就够。依赖集中在一个文件里，安装用 pip 批量处理。

pip install paramiko pyyaml

paramiko 是 SSH 通道的核心，pyyaml 负责解析设备清单。如果安装 paramiko 时遇到cryptography编译报错，大概率是缺 OpenSSL 开发头文件，在 Debian 系环境下先装libssl-dev再重装就能过。Windows 环境建议直接装对应 Python 版本的 wheel，避免本地编译翻车。

4.2 配置设备清单并试跑

先用两个设备做试点，不要一上来就灌五十台。把设备清单文件按上面 YAML 格式填好，然后跑单机模式。

python inspect.py --config devices.yaml --device 192.168.10.1

--device参数用来指定只巡检某一台设备，这是排查问题的关键开关。第一台设备能正常出报告，再放开全量并发巡检。

python inspect.py --config devices.yaml --output report.html

跑的过程中如果看到Connection refused或Authentication failed，先确认这台设备的 SSH 服务是否开启、账密是否一致，再排查代码逻辑——这两个错误绝大多数情况下和设备侧配置有关，不是脚本 bug。

4.3 定时跑巡检：cron 和计划任务配置

巡检是典型的定时任务场景，脚本本身不需要常驻，用系统定时器调度最合适。Linux 下配置 crontab，在工作日每天下午固定时间执行，输出文件名带日期，保留最近一个月的报告。

0 16 * * 1-5 cd /opt/inspection && python inspect.py --config devices.yaml --output reports/report_$(date +\%Y\%m\%d).html >> logs/inspect.log 2>&1

注意%在 crontab 里有特殊含义，date命令里的格式化符号要加反斜杠转义，否则定时任务会静默失败。Windows 环境下用任务计划程序创建计划任务，操作类似，但要注意 Python 解释器的绝对路径。

4.4 输出校验

拿到报告后不要只盯着成功跑完的日志，打开 HTML 抽查几台设备。重点看接口状态里有没有 down 的端口、CPU 利用率有没有超阈值、设备版本是否和资产台账一致。报告内容对不上预期，先回查COMMAND_MAP里命令适不适合当前设备型号。设备型号识别依赖device_type字段，这个字段写错的话，即使连接成功，命令也会全部执行失败。

5. 踩坑与排查：SSH 会话中断、编码乱码和并发失败

5.1 回显不完整或截断

一条display interface brief在华为交换机上可能输出几十行，如果设备默认开启了分页显示，SSH 通道只会收到第一屏内容，后面的被卡在分页状态。

原因很明确：设备执行命令时检测到终端是交互式的，输出分页后等待用户按空格或回车继续。解决方法是执行巡检命令前，先发一条关闭分页的命令，华为是screen-length 0 temporary，思科是terminal length 0，H3C 是screen-length disable。把这条命令放在COMMAND_MAP对应设备组的开头，确保后续命令回显完整。

5.2 命令超时但设备仍在处理

高并发模式下，设备 CPU 过载会导致命令响应变慢，SSH 通道长时间没有数据返回，触发客户端的timeout。但此时设备还在执行命令，只是回显被堵住了。

解决办法是两层配合：连接层超时设短一点（比如 10 秒），保证连接失败能快速暴露；命令执行层超时设长一点（比如 30 秒），给设备足够的处理时间。两个超时分开设置，不要用同一个值，否则要么连接阶段容易误判，要么命令阶段长时间卡死。

5.3 并发过高导致连接失败率上升

把max_workers调到 30 去跑百台设备，结果发现失败率反而升高。设备的 SSH 服务通常有最大连接数限制，过多的并发连接会直接拒绝超出的部分。

原因就是同时建立的 SSH 会话数超过了设备允许的上限。解决方法是先把并发调小，观察设备稳定性和巡检耗时，再逐步加大。常见做法是控制在 5 到 10 之间，如果需要提高吞吐量，可以在两台设备之间加一个小的延时，避免所有连接在同一秒内发起。

5.4 密码含特殊字符解析失败

设备密码里包含@、:这类字符时，YAML 解析会出现问题。password: admin@123可能被解析成字符串没问题，但如果密码开头是特殊符号或包含#，YAML 会误判成注释。

解决方法是给密码字段加引号包裹：password: "abc#123"。如果字符串里包含双引号，用单引号包裹再加转义。在 YAML 里凡是疑似包含特殊字符的字符串，统一加引号是最省心的习惯。

5.5 设备回显里出现乱码

不同厂商设备默认字符编码不一样，多数是 UTF-8，但个别老设备是 GBK 或 Latin-1。直接按 UTF-8 decode 会得到一堆乱码，接口状态和告警信息没法读。

常见做法是先统一用 UTF-8 解码，加了errors='replace'后脚本不会崩。如果乱码比例高，在设备清单里加一个encoding字段，按设备指定编码解码。这样不用全局改代码，只在解析设备信息的地方加一行逻辑。

6. 二次开发方向：从巡检到配置对比和异常告警

6.1 把巡检结果接入配置备份

巡检本身已经拿到了多台设备的配置输出，但目前的报告是纯展示性的。可以加一个配置快照模块，定期把配置文件存到指定目录，按设备 IP 和日期命名。保存配置这个动作在很多监控场景下都顺手有用——回滚、变更对比、审计追溯都需要历史配置。

import os import hashlib def save_config_snapshot(device_host, config_text, base_dir="configs"): snapshot_dir = os.path.join(base_dir, device_host) os.makedirs(snapshot_dir, exist_ok=True) digest = hashlib.md5(config_text.encode("utf-8")).hexdigest() file_path = os.path.join(snapshot_dir, f"{digest}.cfg") with open(file_path, "w", encoding="utf-8") as f: f.write(config_text) return digest

这段代码用 MD5 对配置内容做摘要，文件内容有变化就会生成新的文件名，相同的配置不会重复落盘。迭代多次后同目录下会积累多个版本，可以拿来做配置漂移分析。

6.2 增加异常告警文本匹配

巡检的价值在于发现异常，而异常识别靠的是对回显做规则过滤。一个常见实现是给设备类型维护一组异常特征规则，比如接口状态里出现了down，或 CPU 使用率超过 90%，就在报告中打上异常标记并单独汇总。

正则匹配是这块的主流做法，阈值判断适合数值型指标。设备命令回显格式不完全一致，规则要按设备类型分组维护，保证误报率可控。告警命中后再接飞书或邮件通知，就是一套完整的异常巡检链路。

6.3 验证方式：比对自动巡检和手工执行的结果

改动代码后，拿三台不同厂商的设备做前后验证。对照方法是手工登录设备执行同一条命令，和脚本抓取的回显对比，确认内容一致。这样能发现两类问题：一是命令本身发错了，二是抓取时机太早导致回显不完整。

从那以后我每次改巡检逻辑都强制走一遍这个流程：先单台试跑、再做小规模并发、最后上全量定时任务。基本能过滤掉八成以上的低级问题。这套流程也可以复制到其他自动化脚本里——小范围验证永远比全量翻车再回滚省心。希望帮到你。

本文还有配套的精品资源，点击获取