导航：首页 > 网络安全 >

python中如何通过脚本进行双线路切换

发表于：2026-10-31 作者：千家信息网编辑

千家信息网最后更新 2026年10月31日，这篇文章主要为大家展示了"python中如何通过脚本进行双线路切换"，内容简而易懂，条理清晰，希望能够帮助大家解决疑惑，下面让小编带领大家一起研究并学习一下"python中如何通过脚本进行双线路切换"

千家信息网最后更新 2026年10月31日python中如何通过脚本进行双线路切换

这篇文章主要为大家展示了"python中如何通过脚本进行双线路切换"，内容简而易懂，条理清晰，希望能够帮助大家解决疑惑，下面让小编带领大家一起研究并学习一下"python中如何通过脚本进行双线路切换"这篇文章吧。

这两天写了一个脚本，用于线路切换，目前还有些小bug，还没有想好逻辑结构如何改，反正想着想着头发就没有了。

要求：1.所有用户连接ISP是通过Core01的G0/1出局，Core02的G0/1是Shutdown状态

2. 当Core01的G0/1光线线路出问题时，通过脚本去把Core02的G0/1　UP

***更新代码，增加微信提醒功能

拓扑如下图：

脚本如下：

#!/usr/bin/env python3# -*- coding: utf-8 -*-import timeimport reimport sysimport subprocess from sendwxInfo import * from netmiko import ConnectHandlerfrom netmiko.ssh_exception import NetMikoTimeoutException,NetMikoAuthenticationExceptiondef auth(Conn):        def wrapper(ip,username,password):        device = {                 'device_type': 'cisco_ios',                 'ip': ip,                 'username': username,                 'password': password,                  }        try:                       connect = ConnectHandler(**device)            connect.enable()        except (EOFError, NetMikoTimeoutException):            print(u" 网络设备%s:  无法连接!请确认该设备IPAddress是否可达！"%ip)              return                     except (EOFError, NetMikoAuthenticationException):            print(u" 网络设备%s:  用户名与密码错误!请确认账号与密码！" %ip)            return        res = Conn(ip,username,password,connect)        return res     return wrapper@authdef upInterface(ip,username,password,connect):    cmd = ['interface g0/1','no shutdown']    connect.send_config_set(cmd)    res = connect.send_command('show ip int brief')    connect.disconnect()@authdef downInterface(ip,username,password,connect):    cmd = ['interface g0/1','shutdown']    connect.send_config_set(cmd)    connect.disconnect()def sendInfo():    weixinInfo = WeChat('https://qyapi.weixin.qq.com/cgi-bin')    return weixinInfo def masterLogs():    filename = '/var/log/syslog-ng/172.16.200.21/messages'    file = subprocess.Popen('tail -n 1 ' +filename , shell=True, stdout=subprocess.PIPE,stderr=subprocess.PIPE)    masterData = str(file.stdout.readline(),encoding='utf-8')    if re.match('.*sla 1 state Up -> Down.*',masterData):        res01 ="设备Core01连接互联网的端口G0/１已经断开，请管理员确认业务。"        downInterface('172.16.200.21','admin','Password.123')        upInterface('172.16.200.22','admin','Password.123')        res02 = 'Intetnet线路已经从Core01的G0/1切换至Core02的G0/1端口，请管理员确认业务。'        res03 = '重要信息!'        message = "{0}\n{1}\n\n{2}".format(res03,res01,res02)               info = sendInfo()        info.sendMessage(message)def slaveLogs():     filename = '/var/log/syslog-ng/172.16.200.22/messages'     file = subprocess.Popen('tail -n 1 ' +filename , shell=True, stdout=subprocess.PIPE,stderr=subprocess.PIPE)     slaveData  = str(file.stdout.readline(),encoding='utf-8')     if re.match('.*sla 1 state Up -> Down.*',slaveData):         res01= "设备Core02连接互联网的端口G0/１已经断开，请管理员确认业务。"         downInterface('172.16.200.22','admin','Password.123')         upInterface('172.16.200.21','admin','Password.123')         res02='Internet线路已经从Core02的G0/1切换至Core01的G0/1端口，请管理员确认业务。'         res03 = '重要信息!'         message = "{0}\n{1}\n\n{2}".format(res03,res01,res02)         info = sendInfo()         info.sendMessage(message)        if __name__ == "__main__":    while True:            masterLogs()        time.sleep(5)            slaveLogs()            time.sleep(5)

微信提醒功能代码

#!/usr/bin/env python3 # -*- coding: utf-8 -*-import urllib,jsonimport urllib.parseimport urllib.requestimport sysclass WeChat(object):        __token_id = ''         def __init__(self,url):                self.__url = url.rstrip('/')                #自行更改                self.__corpid = 'XXXXXXXXXX'                self.__secret = 'XXXXXXXXXXXXXX'                def authID(self):                params = {'corpid':self.__corpid, 'corpsecret':self.__secret}                data = urllib.parse.urlencode(params)                content = self.getToken(data)                try:                        self.__token_id = content['access_token']                except KeyError:                        raise KeyError        def getToken(self,data,url_prefix='/'):                url = self.__url + url_prefix + 'gettoken?'                try:                    response = urllib.request.Request(url + data)                except KeyError:                        raise KeyError                result = urllib.request.urlopen(response)                content = json.loads(result.read())                return content        def postData(self,data,url_prefix='/'):                url = self.__url + url_prefix + 'message/send?access_token=%s' % self.__token_id                request = urllib.request.Request(url,data)                try:                        result = urllib.request.urlopen(request)                except urllib.request.HTTPError as e:                        if hasattr(e,'reason'):                                print ('reason',e.reason)                        elif hasattr(e,'code'):                                print ('code',e.code)                        return 0                else:                        content = json.loads(result.read())                        result.close()                return content        def sendMessage(self,message):                self.authID()                data = json.dumps({                                'touser':"",                                #自行更改                                'toparty':"XXX",                                'msgtype':"text",                                #自行更改                                'agentid':"XXXXXX",                                'text':{                                                'content':message                                },                                'safe':"0"                },ensure_ascii=False)                response = self.postData(data.encode('utf-8'))

主备设备的目前状态如下：所有流量从Core01的G0/1出局

备用路由器Core02的G0/1口是关闭状态，（注意一点: Core01与Core02各自的G0/1配置相同）

从互联网ISP去Ping 企业的IP:　169.254.100.1状态是OK的。

以下是测试抓取日志以及线路切换后互联网ISP去ping 企业的IP: 169.100.1状态

Core01连接互联网的端口已经断开