Python小程序网络耗时监控

近日接到人行通知，要求对金融行业IPv6规模部署情况进行专项排查，其中一项内容就是针对门户网站连接稳定性进行监控，具体要求就是：

金融服务机构提供正式报告说明稳定性测试数据。（不少于15天时间，每隔1小时发起门户网站连接，记录连接失败率），应网络同事要求，结合自己最近学习的Python，写了如下程序，将执行结果保存到文件中，源码如下，仅供Python爱好者一起分析学习：

************************************************

#coding=utf-8

import requests

import csv

import datetime

import time

from requests.exceptions import RequestException

from threading import Timer

#本程序运行每间隔指定时间，产生一条数据，数据文件为csv格式，每行文件内容为：

#当前时间,联通IPv6耗时,联通IPv4耗时，联通耗时结果,电信IPv6耗时,电信IPv4耗时，电信耗时结果

#是否满足要求公式：IPv6耗时-IPv4耗时<75ms，通过sucess;IPv6耗时-IPv4耗时>75ms，不通过fail

#定义获取时间方法

def get_time():

url1 = 'http://[xx:xx:xxxx::x]'#门户网站联通6，需要调整为自己的IP

url2 = 'http://xx.xx.xx.xx' #门户网站联通4，需要调整为自己的IP

url3 = 'http://[xx:xx:xx::x]' #门户网站电信6，需要调整为自己的IP

url4 = 'http://xx.x.xx.xx' #门户网站电信4，需要调整为自己的IP

destList = [url1,url2,url3,url4]

#print(destList)

now_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')

timeList = []

dataList = []

#当前时间

dataList.Append(str(now_time))

timeOut = 'timeOut'

timePass = 'sucess'

timeFail = 'fail'

other = 'none'

#IPv6-IPv4耗时最低合格时间75ms

timeLimit = 75

#获取每个请求的时间ms，超时数组则存放timeOut

for url in destList:

#print(url)

try:

reqDest = requests.get(url)

timeList.append(reqDest.elapsed.total_seconds()*1000)

except (RequestException):

timeList.append(timeOut)

#准备数据行

#联通线路请求耗时数据

if(timeOut != timeList[0]):

dataList.append(str(round(timeList[0],3))+'ms')

else:

dataList.append(str(timeList[0]))

if(timeOut != timeList[1]):

dataList.append(str(round(timeList[1],3))+'ms')

else:

dataList.append(str(timeList[1]))

if(timeOut != timeList[0] and timeOut != timeList[1]):

ltTimeLast= timeList[0]-timeList[1]

if(ltTimeLast < timeLimit):

dataList.append(timePass)

else:

dataList.append(timeFail)

else:

dataList.append(other)

#电信请求耗时数据

if(timeOut != timeList[2]):

dataList.append(str(round(timeList[2],3))+'ms')

else:

dataList.append(str(timeList[2]))

if(timeOut != timeList[3]):

dataList.append(str(round(timeList[3],3))+'ms')

else:

dataList.append(str(timeList[3]))

if(timeOut != timeList[2] and timeOut != timeList[3]):

dxTimeLast= timeList[2]-timeList[3]

if(dxTimeLast < timeLimit):

dataList.append(timePass)

else:

dataList.append(timeFail)

else:

dataList.append(other)

return dataList

#定义写文件title方法

def writeFileTitle():

with open('IPv6-IPv4TimeResult.csv', "a+") as file:

titleList = ['当前时间','联通IPv6耗时','联通IPv4耗时','联通耗时结果','电信IPv6耗时','电信IPv4耗时','电信耗时结果']

csv_file = csv.writer(file)

csv_file.writerow(titleList)

file.close()

#定义增量写一行数据方法

def writeReq_time():

with open('IPv6-IPv4TimeResult.csv', "a+") as file:

begin_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')

print(begin_time + ' begin to request')

csv_file = csv.writer(file)

list = []

list.append(get_time())

datas = list

csv_file.writerows(datas)

end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')

print(end_time + ' end to write')

file.close()

#主程序入口

#调用文件title

writeFileTitle()

#线程延迟时间1s执行

timer_interval=1

#线程间隔3600s执行

timer_sleep = 3600

t=Timer(timer_interval,writeReq_time)

t.start()

while True:

time.sleep(timer_sleep)

writeReq_time()

*******************************

程序执行界面及结果如下：

程序运行界面