python 判断网络通断同时检测网络的状态

思路：通过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}}

结果存放

巴特西

python 判断网络通断同时检测网络的状态

最新文章

热门文章