思路：通过http判断网络通断，通过ping获取网络的状态
注意：不同平台下，调用的系统命令返回格式可能不同，跨平台使用的时候，注意调整字符串截取的值
主程序：network_testing_v0.3.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2019-04-19 09:50
# @Author : yuyongxr
# @Site :
# @File : network_testing_v0.1.py
# @Software: PyCharm
# from icmp_test import *
from http_test import *
from config import *
import threading
import time
import json http_status = True def start_check_http(url,title):
global http_status
while True:
if check_http(url,title):
http_status = True
# print("http_status :", http_status)
else:
http_status = False
print("http_status :", http_status)
file = open("http_logs.txt", "a")
file.write(time.asctime(time.localtime(time.time()))+ " : http status is " +str(http_status) + "\n")
time.sleep(1) def check_continuous():
url,title,ip_address,times,domain,dnsserver = analyze_config()
t_http = threading.Thread(target=start_check_http,args=(url,title))
t_http.start() def icmp_continuous(): global http_status
global avg_list
global lost_list avg_list = []
lost_list = []
url,title,ip_address,times,domain,dnsserver = analyze_config() while True:
if http_status:
min, avg, max, lost = "","","",""
try:
min, avg, max, lost = shend_icmp_packet(ip_address, times) avg_flag = 0
avg_list.append(float(avg)) for a in avg_list:
avg_flag = avg_flag + a
avg_flag = avg_flag/len(avg_list) lost_flag = 0
lost_list.append(float(lost))
for a in lost_list:
lost_flag = lost_flag + a
lost_flag = lost_flag / len(lost_list) file = open("result.json")
content = file.read()
file.close()
content = json.loads(content)
content["ping"]["min"] =float(content["ping"]["min"]) if float(content["ping"]["min"]) <= float(min) else min
content["ping"]["avg"] = avg_flag
content["ping"]["max"] = float(content["ping"]["max"]) if float(content["ping"]["max"]) >= float(max) else max
content["ping"]["lost"] = lost_flag
content = json.dumps(content) file = open("result.json","w")
file.write(content)
file.close() print(min, avg, max, lost)
except:
pass
else:
time.sleep(1)
continue
time.sleep(5) def main():
c_check = threading.Thread(target=check_continuous)
i_icmp = threading.Thread(target=icmp_continuous) c_check.start()
i_icmp.start() if __name__ == '__main__':
main()
主程序
获取网络状态：icmp_test.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2019-04-19 18:08
# @Author : yuyongxr
# @Site :
# @File : icmp_test.py
# @Software: PyCharm
import os def shend_icmp_packet(ip_address,times):
try:
response = os.popen('ping -c ' + times + ' '+ ip_address).read()
# 取出丢包率
lost = response[response.index("%")-4:response.index("%")]
#取出指定的延时字符串
res = list(response)
index = 0
count = 0
for r in res:
count += 1
if r == "=" :
index = count
response = response[index + 1:-4] # 取出执行的延迟
i = 0
j = []
res1 = list(response)
for r in res1:
i += 1
if r == "/" :
j.append(i) min = response[:j[0]-1]
avg = response[j[0]:j[1]-1]
max = response[j[1]:j[2]-1]
return min,avg,max,lost
except :
print("ping exec error")
file = open("icmp_logs.txt","a")
file.write(time.asctime(time.localtime(time.time())) +" ping exec error \n")
获取网络状态
判断网络通断：http_test.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2019-04-19 18:09
# @Author : yuyongxr
# @Site :
# @File : http_test.py
# @Software: PyCharm import requests,bs4,time
from requests.exceptions import ReadTimeout,ConnectTimeout,HTTPError,ConnectionError def send_http_packet(url):
requests.packages.urllib3.disable_warnings()
user_agent = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/73.0.3683.86 Safari/537.36"
headers = {'User-Agent': user_agent}
url = "http://" + url
response_html= ""
file = open("http_logs.txt","a")
try:
response = requests.get(url, headers)
response_html = response.content.decode()
return response_html
except ReadTimeout:
print('Read Timeout')
file.write(time.asctime(time.localtime(time.time())) + " Read Timeout \n")
file.close()
return False
except ConnectTimeout:
print('Connect Timeout')
file.write(time.asctime(time.localtime(time.time())) + " Connect Timeout \n")
file.close()
return False
except HTTPError:
print('HTTP Error')
file.write(time.asctime(time.localtime(time.time())) + " HTTP Error \n")
file.close()
return False
except ConnectionError:
print('Connection Error')
file.write(time.asctime(time.localtime(time.time())) + " Connection Error \n")
file.close()
return False def check_http(url,title):
html = send_http_packet(url)
file = open("http_logs.txt","a")
if html != False and title != False :
soup = bs4.BeautifulSoup(html, 'lxml')
html_title = ""
html_title = soup.title.text
if title in html_title:
return True
else:
return False
else:
print('html or title is None')
file.write(time.asctime(time.localtime(time.time())) + " html or title is None \n")
file.close()
return False
判断网络通断
配置文件：config.json
{
"dns":{"domain":"www.baidu.com","dnsserver":"114.114.114.114"},
"icmp":{"ip_address":"www.baidu.com","times":""},
"http":{"url":"www.baidu.com","title":"百度"}
}
配置文件
解析配置文件：config.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2019-04-19 18:28
# @Author : yuyongxr
# @Site :
# @File : config.py
# @Software: PyCharm import json def read_config():
file = open("config.json")
content = file.read()
file.close()
return content def analyze_config():
config = read_config()
config = json.loads(config) url = config["http"]["url"]
title = config["http"]["title"] ip_address = config["icmp"]["ip_address"]
times = config["icmp"]["times"] domain = config["dns"]["domain"]
dnsserver = config["dns"]["dnsserver"] return url,title,ip_address,times,domain,dnsserver
解析配置文件
结果存储文件：result.json
{"ping": {"min": 46.828, "avg": 137.0165, "max": 150.587, "lost": 0.7446808510638298}}
结果存放