Python网络设备批量配置脚本解析
目录
-
脚本概述
-
代码解析
-
导入模块
-
日志配置
-
核心函数config_device
-
主程序逻辑
-
-
使用说明
-
脚本优化建议
-
完整代码
脚本概述
这是一个使用Python编写的网络设备批量配置脚本,主要功能是通过SSH协议批量登录多台网络设备(如路由器、交换机等),并执行预定义的配置命令。脚本采用了多线程技术,可以同时管理多台设备,大大提高了批量操作的效率。
代码解析
导入模块
import paramiko
import getpass
import logging
import time
from concurrent.futures import ThreadPoolExecutor
-
paramiko: 用于实现SSH协议的Python库,提供SSH客户端功能
-
getpass: 安全获取用户输入的密码(虽然本脚本中未直接使用)
-
logging: Python标准日志模块,用于记录操作日志
-
time: 提供时间相关功能,这里主要用于命令执行间隔
-
ThreadPoolExecutor: 来自concurrent.futures模块,实现线程池功能
日志配置
logging.basicConfig(filename='network_ops.log',level=logging.INFO,format='%(asctime)s-%(levelname)s-%(message)s'
)
配置日志系统:
-
日志输出到文件
network_ops.log
-
日志级别为INFO(记录一般操作信息)
-
日志格式包含时间戳、日志级别和消息内容
核心函数config_device
def config_device(ip,username,password,commands):try:client=paramiko.SSHClient()client.set_missing_host_key_policy(paramiko.AutoAddPolicy())client.connect(hostname=ip,username=username,password=password,timeout=10)logging.info(f'Successfully connected to {ip}')shell=client.invoke_shell()for cmd in commands:shell.send(cmd+'\n')time.sleep(1)output=shell.recv(65535).decode('utf-8')logging.info(f'{ip} output:\n{output}')except Exception as e:logging.error(f'{ip} failed:{str(e)}')finally:client.close()
函数功能解析:
-
创建SSHClient对象并设置自动添加主机密钥策略
-
连接目标设备(IP、用户名、密码),设置10秒超时
-
获取交互式shell会话
-
循环发送每条命令,每条命令间隔1秒
-
接收命令输出并记录到日志
-
异常处理:记录连接或执行失败的原因
-
最终确保SSH连接被关闭
主程序逻辑
if __name__=='__main__':username=input('username:')password=getpass.getpass('password:')with open(r'C:\Users\23608\Desktop\ipadd_python.txt','r')as f:ips=[line.strip() for line in f.readlines()]with open(r'C:\Users\23608\Desktop\cmd_file.txt','r')as f:commands=[line.strip() for line in f.readlines()]with ThreadPoolExecutor(max_workers=5) as excutor:for ip in ips:excutor.submit(config_device,ip,username,password,commands)
执行流程:
-
获取用户名和密码
-
从两个文本文件读取:
-
ipadd_python.txt
: 存储设备IP地址列表 -
cmd_file.txt
: 存储要执行的命令列表
-
-
创建最大5个线程的线程池
-
为每个IP提交一个配置任务到线程池
使用说明
-
准备两个文本文件:
-
ipadd_python.txt
: 每行一个设备IP地址 -
cmd_file.txt
: 每行一条要执行的命令
-
-
运行脚本,输入设备的用户名和密码
-
查看
network_ops.log
文件获取执行结果
完整代码
import paramiko
import getpass
import logging
import time
from concurrent.futures import ThreadPoolExecutorlogging.basicConfig(filename='network_ops.log',level=logging.INFO,format='%(asctime)s-%(levelname)s-%(message)s'
)def config_device(ip,username,password,commands):try:client=paramiko.SSHClient()client.set_missing_host_key_policy(paramiko.AutoAddPolicy())client.connect(hostname=ip,username=username,password=password,timeout=10)logging.info(f'Successfully connected to {ip}')shell=client.invoke_shell()for cmd in commands:shell.send(cmd+'\n')time.sleep(1)output=shell.recv(65535).decode('utf-8')logging.info(f'{ip} output:\n{output}')except Exception as e:logging.error(f'{ip} failed:{str(e)}')finally:client.close()if __name__=='__main__':username=input('username:')password=getpass.getpass('password:')with open(r'C:\Users\23608\Desktop\ipadd_python.txt','r')as f:ips=[line.strip() for line in f.readlines()]with open(r'C:\Users\23608\Desktop\cmd_file.txt','r')as f:commands=[line.strip() for line in f.readlines()]with ThreadPoolExecutor(max_workers=5) as excutor:for ip in ips:excutor.submit(config_device,ip,username,password,commands)