利用Python實(shí)現(xiàn)網(wǎng)絡(luò)測試的腳本分享
前言
最近同學(xué)讓我?guī)兔懸粋€測試網(wǎng)絡(luò)的工具。由于工作上的事情,斷斷續(xù)續(xù)地拖了很久才給出一個相對完整的版本。其實(shí),我Python用的比較少,所以基本都是邊查資料邊寫程序。
程序的主要邏輯如下:
讀取一個excel文件中的ip列表,然后使用多線程調(diào)用ping統(tǒng)計每個ip的網(wǎng)絡(luò)參數(shù),最后把結(jié)果輸出到excel文件中。
代碼如下所示:
#! /usr/bin/env python # -*- coding: UTF-8 -*- # File: pingtest_test.py # Date: 2008-09-28 # Author: Michael Field # Modified By:intheworld # Date: 2017-4-17 import sys import os import getopt import commands import subprocess import re import time import threading import xlrd import xlwt TEST = [ '220.181.57.217', '166.111.8.28', '202.114.0.242', '202.117.0.20', '202.112.26.34', '202.203.128.33', '202.115.64.33', '202.201.48.2', '202.114.0.242', '202.116.160.33', '202.202.128.33', ] RESULT={} def usage(): print "USEAGE:" print "\t%s -n TEST|excel name [-t times of ping] [-c concurrent number(thread nums)]" %sys.argv[0] print "\t TEST為簡單測試的IP列表" print "\t-t times 測試次數(shù);默認(rèn)為1000;" print "\t-c concurrent number 并行線程數(shù)目:默認(rèn)為10" print "\t-h|-?, 幫助信息" print "\t 輸出為當(dāng)前目錄文件ping_result.txt 和 ping_result.xls" print "for example:" print "\t./ping_test.py -n TEST -t 1 -c 10" def div_list(ls,n): if not isinstance(ls,list) or not isinstance(n,int): return [] ls_len = len(ls) print 'ls length = %s' %ls_len if n<=0 or 0==ls_len: return [] if n > ls_len: return [] elif n == ls_len: return [[i] for i in ls] else: j = ls_len/n k = ls_len%n ### j,j,j,...(前面有n-1個j),j+k #步長j,次數(shù)n-1 ls_return = [] for i in xrange(0,(n-1)*j,j): ls_return.append(ls[i:i+j]) #算上末尾的j+k ls_return.append(ls[(n-1)*j:]) return ls_return def pin(IP): try: xpin=subprocess.check_output("ping -n 1 -w 100 %s" %IP, shell=True) except Exception: xpin = 'empty' ms = '=[0-9]+ms'.decode("utf8") print "%s" %ms print "%s" %xpin mstime=re.search(ms,xpin) if not mstime: MS='timeout' return MS else: MS=mstime.group().split('=')[1] return MS.strip('ms') def count(total_count,I): global RESULT nowsecond = int(time.time()) nums = 0 oknums = 0 timeout = 0 lostpacket = 0.0 total_ms = 0.0 avgms = 0.0 maxms = -1 while nums < total_count: nums += 1 MS = pin(I) print 'pin output = %s' %MS if MS == 'timeout': timeout += 1 lostpacket = timeout*100.0 / nums else: oknums += 1 total_ms = total_ms + float(MS) if oknums == 0: oknums = 1 maxms = float(MS) avgms = total_ms / oknums else: avgms = total_ms / oknums maxms = max(maxms, float(MS)) RESULT[I] = (I, avgms, maxms, lostpacket) def thread_func(t, ipList): if not isinstance(ipList,list): return else: for ip in ipList: count(t, ip) def readIpsInFile(excelName): data = xlrd.open_workbook(excelName) table = data.sheets()[0] nrows = table.nrows print 'nrows %s' %nrows ips = [] for i in range(nrows): ips.append(table.cell_value(i, 0)) print table.cell_value(i, 0) return ips if __name__ == '__main__': file = 'ping_result.txt' times = 10 network = '' thread_num = 10 args = sys.argv[1:] try: (opts, getopts) = getopt.getopt(args, 'n:t:c:h?') except: print "\nInvalid command line option detected." usage() sys.exit(1) for opt, arg in opts: if opt in ('-n'): network = arg if opt in ('-h', '-?'): usage() sys.exit(0) if opt in ('-t'): times = int(arg) if opt in ('-c'): thread_num = int(arg) f = open(file, 'w') workbook = xlwt.Workbook() sheet1 = workbook.add_sheet("sheet1", cell_overwrite_ok=True) if not isinstance(times,int): usage() sys.exit(0) if network not in ['TEST'] and not os.path.exists(os.path.join(os.path.dirname(__file__), network)): print "The network is wrong or excel file does not exist. please check it." usage() sys.exit(0) else: if network == 'TEST': ips = TEST else: ips = readIpsInFile(network) print 'Starting...' threads = [] nest_list = div_list(ips, thread_num) loops = range(len(nest_list)) print 'Total %s Threads is working...' %len(nest_list) for ipList in nest_list: t = threading.Thread(target=thread_func,args=(times,ipList)) threads.append(t) for i in loops: threads[i].start() for i in loops: threads[i].join() it = 0 for line in RESULT: value = RESULT[line] sheet1.write(it, 0, line) sheet1.write(it, 1, str('%.2f'%value[1])) sheet1.write(it, 2, str('%.2f'%value[2])) sheet1.write(it, 3, str('%.2f'%value[3])) it+=1 f.write(line + '\t'+ str('%.2f'%value[1]) + '\t'+ str('%.2f'%value[2]) + '\t'+ str('%.2f'%value[3]) + '\n') f.close() workbook.save('ping_result.xls') print 'Work Done. please check result %s and ping_result.xls.'%file
這段代碼參照了別人的實(shí)現(xiàn),雖然不是特別復(fù)雜,這里還是簡單解釋一下。
- excel讀寫使用了xlrd和xlwt,基本都是使用了一些簡單的api。
- 使用了threading實(shí)現(xiàn)多線程并發(fā),和POSIX標(biāo)準(zhǔn)接口非常相似。thread_func是線程的處理函數(shù),它的輸入包含了一個ip的List,所以在函數(shù)內(nèi)部通過循環(huán)處理各個ip。
- 此外,Python的commands在Windows下并不兼容,所以使用了subprocess模塊。
到目前為止,我對Python里面字符集的理解還不到位,所以正則表達(dá)式匹配的代碼并不夠強(qiáng)壯,不過目前勉強(qiáng)可以工作,以后有必要再改咯!
總結(jié)
以上就是這篇文章的全部內(nèi)容了,希望本文的內(nèi)容對大家的學(xué)習(xí)或者工作能帶來一定的幫助,如果有疑問大家可以留言交流,謝謝大家對腳本之家的支持。
相關(guān)文章
python3實(shí)現(xiàn)ftp服務(wù)功能(服務(wù)端 For Linux)
這篇文章主要介紹了python3實(shí)現(xiàn)ftp服務(wù)功能,服務(wù)端 For Linux,具有一定的參考價值,感興趣的小伙伴們可以參考一下2017-03-032018年P(guān)ython值得關(guān)注的開源庫、工具和開發(fā)者(總結(jié)篇)
本文給大家總結(jié)了2018年P(guān)ython值得關(guān)注的開源庫、工具和開發(fā)者,需要的朋友可以參考下2018-01-01django生產(chǎn)環(huán)境搭建(uWSGI+django+nginx+python+MySQL)
本文主要介紹了django生產(chǎn)環(huán)境搭建,主要包括uWSGI+django+nginx+python+MySQL,具有一定的參考價值,感興趣的小伙伴們可以參考一下2021-08-08以一個投票程序的實(shí)例來講解Python的Django框架使用
這篇文章主要介紹了以一個投票程序的實(shí)例來講解Python的Django框架使用,Django是Python世界中人氣最高的MVC框架,需要的朋友可以參考下2016-02-02