使用类和函数优化Python代码示例
# *utf-* python3
# 导入所需模块
from paramiko import SSHClient, AutoAddPolicy # SSH连接库
from time import sleep, localtime, strftime, time # 时间相关库
from ncclient import manager # NETCONF连接库
class HCIE():
def __init__(self, ip: str, usr: str, pwd: str):
self.hostname = ip # 设备IP地址
self.username = usr # SSH用户名
self.password = pwd # SSH密码
self.client = self._get_connect() # SSH连接对象
self.cli = self.client.invoke_shell() # SSH交互式终端对象
self.cli.send('n\n') #操作步骤
self.cli.send('screen-length 0 temporary\n') # 临时关闭分页显示
self.cli.recv(65535)
# 连接设备
def _get_connect(self):
ssh_con = SSHClient() # 创建SSH连接对象
ssh_con.set_missing_host_key_policy(AutoAddPolicy) # 未知主机密钥的策略
ssh_con.connect(hostname=self.hostname, username=self.username, password=self.password) # 连接设备
return ssh_con
# 运行命令
def cmd_running(self):
ssh_con = self._get_connect()
f = open(Command, 'r') # 打开命令列表文件
cmd_list = f.readlines() # 读取命令列表
for cmd in cmd_list:
self.cli.send(cmd) # 发送命令
sleep(1)
dis_this = self.cli.recv(65535).decode() # 接收命令输出
if cmd == 'display fan\n': # 如果是显示风扇命令
state = dis_this.find('Normal') # 判断是否存在'Normal'
if state < 0:
print('All fans are faulty') # 如果不存在，输出所有风扇异常
print(dis_this) # 输出命令输出内容
f.close() # 关闭文件
ssh_con.close() # 关闭SSH连接
# 保存配置
def save_running(self):
ssh_con = self._get_connect()
cmd = 'save ' + name_time # 构造保存配置命令
self.cli.send('{}\n'.format(cmd)) # 发送保存配置命令
self.cli.send('y\n') # 确认保存
self.cli.send('y\n')
sleep(1)
dis_this = self.cli.recv(65535).decode() # 接收命令输出
print(dis_this) # 输出命令输出内容
ssh_con.close() # 关闭SSH连接
# 下载配置文件
def download_file(self):
ssh_con = self._get_connect()
sftp_con = ssh_con.open_sftp() # 创建SFTP连接对象
l_path = name_time # 本地保存路径
r_path = '/' + l_path # 远程保存路径
sftp_con.get(localpath=l_path, remotepath=r_path) # 下载远程文件到本地
sleep(1)
ssh_con.close() # 关闭SSH连接
# 进入netconf模式
def netconf(self):
ssh_con = self._get_connect()
self.cli.send('sys\n') # 进入系统视图
self.cli.send('netconf\n') # 进入NETCONF模式
self.cli.send('y\n') # 确认进入NETCONF模式
self.cli.send('source ip 10.1.0.6 port 830\n') # 设置NETCONF连接参数
sleep(10)
dis_this = self.cli.recv(65535).decode() # 接收命令输出
print(dis_this) # 输出命令输出内容
ssh_con.close() # 关闭SSH连接
# 关闭连接
def close(self):
self.client.close() # 关闭SSH连接
class HCIEOptimize():
def __init__(self, dev_ip: str, ssh_usr: str, ssh_pwd: str, nc_usr: str, nc_pwd: str, command: str, name_time: str, nc_conf_syslog: str):
self.dev_ip = dev_ip
self.ssh_usr = ssh_usr
self.ssh_pwd = ssh_pwd
self.nc_usr = nc_usr
self.nc_pwd = nc_pwd
self.Command = command
self.name_time = name_time
self.nc_conf_syslog = nc_conf_syslog
# 编辑syslog服务器地址
def edit_loghost(self):
with manager.connect_ssh(
host=self.dev_ip,
username=self.nc_usr,
password=self.nc_pwd,
hostkey_verify=False,
device_params={'name': 'huaweiyang'}
) as m:
m.edit_config(config=self.nc_conf_syslog, target='running') # 编辑配置
sleep(1)
# 数据循环处理
def datacom_loop(self):
now_time = time() # 记录当前时间
while True:
datacom = HCIE(self.dev_ip, self.ssh_usr, self.ssh_pwd)
datacom.cmd_running() # 运行命令
last_time = time() # 记录当前时间
if last_time - now_time >= 86400: # 如果时间间隔大于等于1天
datacom.save_running() # 保存配置
datacom.download_file() # 下载配置
now_time = time() # 记录当前时间
elif last_time - now_time < 300: # 如果时间间隔小于5分钟
datacom.save_running() # 保存配置
datacom.download_file() # 下载配置
datacom.close() # 关闭SSH连接
sleep(300) # 休眠300秒
# 进入netconf模式
def netconf(self):
netconf = HCIE(self.dev_ip, self.ssh_usr, self.ssh_pwd)
netconf.netconf() # 进入NETCONF模式
netconf.close() # 关闭SSH连接
def run(self):
try:
self.netconf() # 进入NETCONF模式
self.edit_loghost() # 编辑syslog服务器地址
self.datacom_loop() # 数据循环处理
except KeyboardInterrupt:
print('Monitoring has stopped!') # 监控停止提示
if __name__ == '__main__':
dev_ip = '10.1.0.6' # 设备IP地址
ssh_usr = 'sshuser' # SSH用户名
ssh_pwd = 'Huawei@123' # SSH密码
nc_usr = 'netconf' # NETCONF用户名
nc_pwd = 'Huawei@123' # NETCONF密码
Command= 'command.txt'
name_time = strftime('%Y_%m_%d', localtime(time())) + '_X_T1_AGG1_vrpcfg.zip'
nc_conf_syslog = '''
<config>
<syslog:syslog xmlns:syslog='urn:ietf:params:xml:ns:yang:ietf-syslog'>
<syslog:log-actions>
<syslog:remote>
<syslog:destination>
<syslog:name>log-center server</syslog:name>
<syslog:udp>
<syslog:address xmlns:xc='urn:ietf:params:xml:ns:netconf:base:1.0' xc:operation='merge'>10.1.60.2</syslog:address>
</syslog:udp>
</syslog:destination>
</syslog:remote>
</syslog:log-actions>
</syslog:syslog>
</config>
'''
hcie_opt = HCIEOptimize(dev_ip, ssh_usr, ssh_pwd, nc_usr, nc_pwd, Command, name_time, nc_conf_syslog)
hcie_opt.run()
代码分析
-
HCIE类：
- 定义了与网络设备交互的类，包含连接、执行命令、保存配置、下载配置、进入NETCONF模式等功能。
- 通过
__init__方法初始化连接信息，使用_get_connect方法创建SSH连接对象。 - 使用
cmd_running方法执行命令列表，并根据命令类型进行结果判断和输出。 - 使用
save_running方法保存设备配置，使用download_file方法下载配置到本地。 - 使用
netconf方法进入NETCONF模式，使用close方法关闭连接。
-
HCIEOptimize类：
- 定义了优化后的类，包含了设备信息、命令文件、时间戳、NETCONF配置等参数。
- 使用
edit_loghost方法编辑syslog服务器地址，并使用datacom_loop方法进行数据循环处理。 - 在
datacom_loop方法中，每隔5分钟或每天自动保存配置并下载配置。 - 使用
netconf方法进入NETCONF模式，并使用run方法执行整个流程。
-
代码优化：
- 将代码封装成类，提高代码可读性和复用性。
- 将多个功能函数拆分，提高代码模块化程度。
- 使用参数化和函数调用，减少代码重复性。
- 添加注释，提高代码可读性和理解性。
总结
通过使用类和函数对Python代码进行优化，可以提高代码的可读性、复用性和维护性，使其更易于理解和修改。
原文地址: https://www.cveoy.top/t/topic/obak 著作权归作者所有。请勿转载和采集!