简介:Python连接交换机自动执行命令是一份面向网络运维与自动化初学者的实用代码包。资源聚焦通过SSH远程登录交换机并批量下发配置命令的典型场景,帮助读者摆脱逐台设备手动操作的繁琐流程,适用于日常巡检与批量配置变更。压缩包共3个文件,包含1个基于paramiko的Python脚本和2个txt辅助文件,整体仅2KB:脚本负责建立SSH连接、逐条执行命令并打印输出,两个txt分别存放IP地址和待执行命令清单,只需修改IP和命令即可快速复用。目前已有169人学习下载。通过学习其中脚本,读者可掌握读取本地文件、循环遍历设备、自动执行命令并捕获响应的完整思路,并能了解异常处理、日志记录等进阶做法。适合刚接触网络自动化、希望用Python提升设备管理效率的工程师参考学习。
1. 连接交换机自动执行命令:先想清楚脚本要替你干多少活
某次凌晨割接,我对着 24 台交换机一台台敲命令,轮到第 19 台时漏了一条命令,第二天业务报障,我花了两个通宵才定位到是那条漏掉的配置没下发。那之后我给自己定了一条规矩:凡是“登录设备、敲命令、看回显、存日志”这串重复动作,全部交给 Python。这篇文章要讲的 Python 连接交换机自动执行命令,就是把一批交换机的 IP 清单和一组命令清单交给脚本,由脚本按顺序自动登录、执行命令、抓回显、落盘。它能让每天登几十台设备的网络运维省下一半时间,也是配置备份、批量巡检、割接变更这类自动化动作的共同地基。适合网络运维、数据中心值班、做自动化平台的工程师,会 Python 基础语法加一点 SSH 概念就能跟上后面的步骤。
2. 选型先行:Paramiko 与 Netmiko 怎么选,连通性怎么自查
2.1 Paramiko 是 SSH 封装,Netmiko 是网络设备封装
先理清这两个库的关系,选错会走弯路。Paramiko 是 SSH 协议的 Python 实现,它做的是“建一条 SSH 会话、发字符串、收字符串”这件事。Netmiko 是在 Paramiko 之上针对网络设备做的封装,它把登录后等待提示符、命令分页、设备类型差异这些脏活都包了。这点非常关键:直接用 Paramiko 连交换机,你要自己处理“命令打出去了但设备还在翻页”、要在代码里写死提示符匹配;Netmiko 则把这些行为按设备类型整理好了。
我的选择逻辑是:日常批量执行命令、做巡检采集,直接上 Netmiko;只有要做 SSH 隧道、端口转发、或者想完全自己控制会话细节时,才退回 Paramiko。两者不是替代关系,Netmiko 底层本来就走 Paramiko。下面用一张表看清楚差异。
| 对比点 | Paramiko | Netmiko |
|---|---|---|
| 协议层级 | SSH 协议层 | 网络设备行为层 |
| 提示符等待 | 自己写循环判断 | 内置,按设备类型匹配 |
| 命令分页 | 自己发命令关分页 | 自动处理大部分分页 |
| 设备类型差异 | 完全不感知 | 通过 device_type 适配 |
| 适用场景 | 隧道、定制 SSH 操作 | 批量执行、配置备份、巡检 |
从项目周期角度看,用 Netmiko 能把“连接交换机自动执行命令”这件事的代码量压到三分之一,而且踩坑更少。后面章节的核心脚本我都以 Netmiko 为主,Paramiko 只保留一版最小实现用来对照。
2.2 环境准备:安装依赖并确认 22 端口可达
动手前先装好依赖。在虚拟环境里执行下面命令,安装时不指定版本,直接用当前能装到的新版本即可,新版本对老设备兼容性没有明显回退。
pip install paramiko netmiko装完顺手确认一下你本机能连到交换机的 SSH 端口。很多“脚本连不上”的现场,根因不是代码,是网络层面根本没放通。先用系统自带的 ssh 客户端做一次探测:
ssh -o ConnectTimeout=5 -o StrictHostKeyChecking=no ops@172.16.10.1参数含义:ConnectTimeout=5表示 5 秒连不上就放弃,避免挂死;StrictHostKeyChecking=no跳过首次连接的指纹确认。如果这条命令能进入密码输入界面,说明 SSH 服务层面是通的;如果直接卡住或报 timeout,下面的脚本也不用急着调试,先查网络。为了脚本里能判断端口通不通,还可以用 Python 做一次纯 TCP 探测:
import socket s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(5) try: s.connect(("172.16.10.1", 22)) print("22 端口可达") except socket.error as e: print("端口不通:", e) finally: s.close()这里只做连通性判断,不碰认证。端口可达不代表账号密码能通过,但端口不通一定可以排除脚本问题。这一步花一分钟,能省掉后面一小时的排错时间。我每次写新设备的自动化脚本前都会先把这段跑一遍,确认网络基线是好的,再往脚本上找原因。
注意:
StrictHostKeyChecking=no只适合开发和测试环境,生产环境建议把指纹加入 known_hosts,否则有会话劫持风险。后面章节会给出更稳妥的主机密钥处理思路。
3. 最小直连脚本:从 SSH 认证到拿到第一条命令回显
3.1 用 Paramiko 直连:exec_command 适合非交互命令
先用 Paramiko 写一个最小版本,目的是理解 SSH 会话里“连接、认证、执行、读取”的基本流程。这里的命令是假设设备开启了对应用户的 SSH 权限,账号具备执行权限。
import paramiko client = paramiko.SSHClient() # 自动接受首次连接时陌生的主机指纹 client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) client.connect( hostname="172.16.10.1", port=22, username="ops", password="YourPassword", # 生产环境改用 getpass 或密钥文件 timeout=10, # 建立 TCP 连接的超时 banner_timeout=15, # 等待 SSH banner 的时间 auth_timeout=15, # 等待认证完成的时间 look_for_keys=False, # 不尝试本机 SSH 私钥文件 allow_agent=False, # 不尝试 SSH agent ) # exec_command 适合执行一条命令后等待返回,不进入交互式 shell stdin, stdout, stderr = client.exec_command("display version", timeout=20) output = stdout.read().decode("utf-8", errors="ignore") print(output) client.close()重点说一下为什么用exec_command而不是invoke_shell。在 Paramiko 里exec_command是“执行单条命令并收取输出”,会话结束后命令就退出;invoke_shell则是在远端开一个交互式 shell,你需要自己发命令、自己判断提示符、自己处理输出。自动执行交换机的多条命令,听起来交互式更灵活,但实现麻烦:提示符会变、分页会卡、输出没有明确的结束标志。exec_command一次一条,逻辑清晰,适合脚本化的场景。
连接参数里,timeout管的是 TCP 连接阶段,auth_timeout管的是用户名密码认证阶段,两者独立。很多人在账号密码错误时看到脚本卡很久才报错,就是因为只设了timeout,没设auth_timeout。上面这组参数基本是我调试设备的固定起始值,确认网络稳定后可以把banner_timeout和auth_timeout再调小。
提示:
AutoAddPolicy在真实环境里会掩盖中间人攻击风险。如果你在公司内部网络且有严格安全要求,首次手工登录一次设备,把指纹固化到 known_hosts,再去掉set_missing_host_key_policy。
3.2 用 Netmiko 跑同一条命令:device_type 是关键一行
看完底层实现,再看常规做法。Netmiko 把认证、提示符等待、分页处理都收进ConnectHandler和send_command里,脚本短得多。
from netmiko import ConnectHandler device = { "device_type": "cisco_ios", # 按设备厂商与命令行风格填写 "host": "172.16.10.1", "username": "ops", "password": "YourPassword", "port": 22, "timeout": 60, # 连接建立超时 "global_delay_factor": 2, # 全局命令发送/读取的延时倍率 } conn = ConnectHandler(**device) # expect_string 是命令回显后要匹配的提示符,read_timeout 是单条命令最长等待 output = conn.send_command( "display version", expect_string=r"#", read_timeout=60, ) print(output) conn.disconnect()device_type是 Netmiko 里最需要较真的一行。它决定库按什么规则匹配提示符、怎么关分页、回显怎么清洗。填错了最常见的症状是连接成功但send_command卡死,因为你把某厂商的命令行风格套到另一个厂商上,提示符匹配不上。怎么查当前 Netmiko 支持哪些值:
python -c "import netmiko; print(netmiko.platforms)"这段会把所有设备类型标识打印出来,按你手头设备的命令行风格对照着选。判断命令行风格的办法很简单:登录设备后看一眼提示符和命令体系,比如某类设备是${hostname}#配display系列命令,另一类是>配show系列命令,device_type就按这个风格选,而不是按品牌名记。选错类型时,库通常会提示“无法识别提示符”,而不是直接报连接失败,这本身就是一条线索。
| 参数 | 作用 | 我的常用值 |
|---|---|---|
| device_type | 设备命令行风格 | 文档列表里对应项 |
| global_delay_factor | 全局发送/读取延时倍率 | 老设备设 2~4 |
| read_timeout | 单条命令最长等待秒数 | 60 |
| expect_string | 当前提示符正则 | 按设备实际提示符 |
global_delay_factor是排查“命令发出去了但读取超时”的常用旋钮。老设备 CPU 高、console 响应慢时,保持默认值 1 容易在回显未结束时就被send_command判定超时,调到 2 或 3 通常能缓解。它不是越快越好,延时倍率过大会拖慢整体批量速度,所以数值以“稳定不超时”为准,能用 2 就不要调到 3。
4. 批量执行落地:设备清单、命令清单与结果落盘
4.1 把设备与命令清单抽成 YAML
直连脚本跑通之后,下一步是处理多台设备。我的习惯是不把设备信息写在脚本里,而是抽到一个独立的 YAML 文件。这样改设备、改命令都不用动代码,运维同事拿到文件填 IP 就能用。
devices: - name: core-a host: 172.16.10.1 username: ops password: YourPassword device_type: cisco_ios - name: access-b host: 172.16.10.2 username: ops password: YourPassword device_type: ruijie_os commands: - display version - display current-configuration字段说明:name是设备别名,用于生成日志文件名;host是管理 IP;device_type沿用前面说的设备类型标识;commands是统一要执行的命令清单。密码写在明文 YAML 里有风险,只在内部测试网段可用。真实环境我一般只写用户名,密码留空,脚本运行时用getpass交互输入,或者用后面会提到的密钥方式。
读取 YAML 的脚本片段:
import yaml with open("inventory.yaml", "r", encoding="utf-8") as f: cfg = yaml.safe_load(f) print(cfg["devices"]) print(cfg["commands"])yaml.safe_load是这里唯一正确的读取方式,不要用yaml.load,前者不允许任意对象反序列化,后者在加载不可信 YAML 时有代码执行风险。配置文件编码固定用 UTF-8,避免设备名或命令里出现中文时在 Windows 下读取报错。
4.2 循环执行并把回显落盘
有了清单,批量逻辑就是一层循环套一层循环:外层遍历设备,内层遍历命令,每条命令回显写入单独文件。下面是可抄作业的核心脚本,我在实际项目中使用的就是这个结构的精简版。
import datetime import yaml from netmiko import ConnectHandler with open("inventory.yaml", "r", encoding="utf-8") as f: cfg = yaml.safe_load(f) # 给这批任务生成一个时间戳前缀,方便按批次归档 stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") failed = [] for dev in cfg["devices"]: log_name = f"{dev['name']}_{stamp}.txt" with open(log_name, "w", encoding="utf-8") as log: log.write(f"# device: {dev['name']} ({dev['host']})\n") try: conn = ConnectHandler( device_type=dev["device_type"], host=dev["host"], username=dev["username"], password=dev["password"], timeout=60, ) # 先关闭分页,避免长回显卡在读一半的位置 conn.send_command_timing("screen-length 0 temporary") conn.send_command_timing("terminal length 0") for cmd in cfg["commands"]: out = conn.send_command( cmd, expect_string=r"#", read_timeout=60, ) log.write(f"### CMD: {cmd}\n{out}\n") conn.disconnect() print(f"[OK] {dev['name']}") except Exception as e: log.write(f"ERROR: {e}\n") failed.append(dev["name"]) print(f"[FAIL] {dev['name']}: {e}") if failed: print("失败设备:", failed)逻辑说明:日志文件名用“设备名_时间戳”组合,同一个批次扫出来的文件不会互相覆盖,而且按时间能看出是哪个周期的采集。每台设备一个输出文件,比所有设备挤在一个大文件里好用,排查问题直接看单设备文件即可。内层命令循环里,每条命令前写一行### CMD:分隔,后面跟完整回显,方便人工核对和脚本二次解析。
send_command_timing和send_command的区别在这里体现得很清楚:send_command会等待匹配expect_string,适合执行后等待提示符返回;send_command_timing只按固定延时发送和读取,适合“发一条命令立刻继续”的场景。关分页这类命令,用send_command_timing更稳,因为它不依赖提示符匹配,设备返回快慢都按延时走。上面先发screen-length 0 temporary再发terminal length 0,是为了同时兼容两类命令行风格的设备,某类设备认其中一个命令就生效。
提示:关分页的命令在不同设备上可能报“未知命令”。如果某台老设备不识别这两个关分页命令,日志里它会以错误回显出现,但不影响后续命令执行。真正要担心的是“关分页没生效还继续跑”,那才会导致长命令被分页符卡住。
4.3 连接超时重试与失败名单处理
批量执行时最让人烦躁的是某台设备刚好在重启或网络抖动,脚本报一个超时,整台设备后面的命令全没跑。我的处理方式是:对“连接阶段”做有限次重试,对“命令执行阶段”不做重试。原因很简单——采集类命令重跑安全,但变更类命令(比如保存配置)盲目重跑可能造成重复操作。重试只放在建立连接这一步,是对连接瞬断的容错。
import time MAX_RETRY = 2 RETRY_WAIT = 5 for dev in cfg["devices"]: for attempt in range(MAX_RETRY + 1): try: conn = ConnectHandler( device_type=dev["device_type"], host=dev["host"], username=dev["username"], password=dev["password"], timeout=60, ) break # 连接成功,跳出重试循环 except Exception as e: print(f"{dev['name']} 第 {attempt + 1} 次连接失败: {e}") if attempt == MAX_RETRY: raise time.sleep(RETRY_WAIT) # 后续命令执行代码放在连接成功之后重试参数建议:MAX_RETRY=2表示最多连 3 次,RETRY_WAIT=5是两次尝试之间等 5 秒。太激进的连续重试没有意义,设备如果正在重启,5 秒到 10 秒的间隔更可能等到它起来;如果设备彻底挂了,重试再多次也是浪费。失败名单由 4.2 里的failed列表承担,批量跑完后打印出来,我再根据名单单独处理问题设备,而不是整批重跑。
这里还有个容易被忽略的点:打印日志和文件日志都要记录“哪一步失败”。脚本出错时,单独看[FAIL]不够,还要知道失败发生在连接阶段还是命令阶段。我在日志里会把ConnectHandler之外的异常也带上设备名写进文件,这样翻日志时能直接定位是这台设备密码变了,还是某条命令回显超时。
5. 避坑排查:连接失败、分页截断与中文乱码的 5 个真实现场
5.1 连接超时:先查链路,再查设备 SSH 配置
现象:脚本在ConnectHandler处抛TimeoutError,报Connection timed out,偶尔伴随Unable to connect to remote host。
原因:绝大多数不是代码问题,而是三层网络根本没通,或设备 SSH 服务没起来,或管理网段设有访问控制。我见过对着代码调了半天才发现,设备管理接口的管理网段和本机不在同一路由表里。
解决:先回到 2.2 的端口探测脚本和ssh命令,确认 TCP 22 端口通不通。如果端口通但脚本失败,再检查账号是否被锁定、设备 SSH 认证方式是否只允许密钥。如果端口不通,登录设备带外检查 SSH 服务状态以及管理接口的访问控制列表。这一条是最基础的排错顺序:网络层 → 服务层 → 认证层,不要一上来就怀疑 Python 代码。
5.2 回显停在分页符:先关掉设备分页
现象:脚本执行一条长命令后卡住,日志里回显最后一行是--- More ---或者类似分页标记,read_timeout超时后脚本报错退出。
原因:设备默认开启分页输出,长命令的完整回显被切成多屏,而脚本读输出时没有继续翻页的动作,就只能停在分页提示符处干等。
解决:连接成功后第一时间发送关闭分页命令,不同命令行风格的设备命令不同。某类设备用terminal length 0,另一类用screen-length 0 temporary。我在 4.2 的脚本里两条都发,就是为了覆盖不同设备。注意是temporary,只对当前会话有效,不会改设备的全局配置,会话断开后自动恢复默认分页,这点很安全。还有一条经验:关分页命令自身也会产生回显,要用send_command_timing而非send_command发送,否则可能因为匹配不到分页提示符又卡一次。
5.3 中文回显乱码:把编码显式定死
现象:回显里的中文变成æ¥è®¯这类乱码,或者读出来的字符串里中文全是问号。这在设备描述、接口备注带中文时特别常见。
原因:设备输出的是字节流,脚本解码用的字符集和设备的输出字符集不一致。很多老设备默认输出 GBK 或 GB2312,而脚本里默认decode("utf-8")自然对不上。
解决:先确认设备输出编码,再在读取时显式指定。把 3.1 里的读取行改掉:
output = stdout.read().decode("gbk", errors="replace")如果设备输出 UTF-8,则用decode("utf-8", errors="ignore")。我的判断习惯是:国产设备、年代稍久的老设备优先试 GBK;较新的设备优先试 UTF-8。errors="replace"能保证最坏情况下也是可见的替换符而不是抛异常中断脚本。文件写入端固定用 UTF-8,不要跟着设备编码走,否则后面用不同工具打开日志又是一轮乱码。
5.4 一台设备挂了,整批直接停摆
现象:批量脚本从第一台开始跑,第一台连不上后抛异常,脚本直接退出,后面 23 台一台都没跑。初版脚本最容易犯这个错。
原因:批量循环里没有做异常隔离,任何一台设备的失败都会中断整个 for 循环。这是“单点失败拖垮全批”的典型。
解决:每台设备一个独立的try-except,如 4.2 所示。异常只记录到该设备的日志文件和failed列表,然后继续下一台。批量结束后统一打印失败设备清单。这里有个心理预期要调整:脚本跑完不代表全部成功,如果失败清单里有设备,一定去对应日志文件里看异常原因。不要因为[OK]刷屏就忽略那一小串[FAIL]。
5.5 命令执行成功但没生效:缺少回显校验
现象:脚本正常执行完命令,返回[OK],但人工登录设备检查时发现配置根本没有变化,或者命令被设备语法检查拒绝了,回显里其实有报错,只是脚本没看。
原因:send_command的返回是命令的回显文本,脚本如果不主动去读这段文本内容,就不会发现“命令被拒绝”这类软错误。脚本判断的是“命令发送成功”,而不是“命令执行成功”。
解决:关键命令执行后,对回显做关键字校验。比如执行配置下发命令后,再抓一次配置回显,搜索目标关键字是否存在。代码留到第 6 章展开。这里先说排查习惯:脚本里看到的[OK]只是连接和发送层面的成功,真正的成功要以回显内容为准。如果日志里回显本身带了语法错误提示,问题的根因就在命令格式或设备型号的差异上。
6. 从脚本到工具:结果校验、定时执行与割接前的最后一关
6.1 给命令回显加关键字断言
批量脚本能跑只是第一步,值得花时间做的是让脚本“会检查自己”。我写了一个最小的断言函数,挂在每条写命令后面:
def assert_in_output(cmd, expect, output): if expect not in output: raise RuntimeError(f"{cmd} 的回显中没有期望关键字 {expect}")用法是执行完命令后,再抓一次相关配置回显,搜索目标行。比如给接口下发描述后,抓接口配置,判断回显里是否有那段描述文本。断言失败时抛异常,由批量脚本的异常处理记录到该设备的失败清单。这样“命令敲进去”和“配置确实生效”就被区分开了。
6.2 定时执行:采集可以无人值守,变更必须有人在场
脚本稳定后,把它放到定时任务里跑采集类命令是安全且高效的。备份配置文件、定期抓取设备状态、统计接口流量,这类只读操作即使半夜执行也不会产生副作用。但变更类命令——保存配置、下发配置、重启接口——我不建议无人值守。实在要跑,至少满足两个条件:变更前先执行一次配置备份,变更后逐条断言回显。我见过有人把配置下发挂进定时任务,第二天发现命令被设备拒绝而没有备份可回滚,只能人工救场。定时执行的是“采集”,变更永远要有人盯。
6.3 割接前我把最后一道关设在“对比”
现在我做割接,脚本的最后一个动作不是执行,而是对比。先在测试环境或同型号备机上跑一遍完整命令集,把回显存成基线,再在目标设备上执行,最后做一次文本级 diff。回显里的版本号、接口状态、配置片段,任何意外差异都会在提交前暴露出来。
有一次我全量下发接口描述命令,脚本显示全部成功,结果某台设备的回显里根本没有那行描述。后来排查是那台设备的命令行风格不同,命令被语法检查拦下,而旧版脚本没看回显。之后我养成了习惯:每条命令的回显都要存、都要对比、都要用关键字断言确认。这也是我想提醒你的:Python 替你把命令敲进去只是第一步,把“敲进去”变成“确实生效”,才是这套脚本真正值钱的地方。希望帮到你。
本文还有配套的精品资源,点击获取