前言
最近同学让我帮忙写一个测试网络的工具。由于工作上的事情,断断续续地拖了很久才给出一个相对完整的版本。其实,我Python用的比较少,所以基本都是边查资料边写程序。
程序的主要逻辑如下:
读取一个excel文件中的ip列表,然后使用多线程调用ping统计每个ip的网络参数,最后把结果输出到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 测试次数;默认为1000;" print "\t-c concurrent number 并行线程数目:默认为10" print "\t-h|-" print "\t 输出为当前目录文件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,次数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"\nInvalid command line option detected." usage() sys.exit(1) for opt, arg in opts: if opt in ('-n'): network = arg if opt in ('-h', '-"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
这段代码参照了别人的实现,虽然不是特别复杂,这里还是简单解释一下。
- excel读写使用了xlrd和xlwt,基本都是使用了一些简单的api。
- 使用了threading实现多线程并发,和POSIX标准接口非常相似。thread_func是线程的处理函数,它的输入包含了一个ip的List,所以在函数内部通过循环处理各个ip。
- 此外,Python的commands在Windows下并不兼容,所以使用了subprocess模块。
到目前为止,我对Python里面字符集的理解还不到位,所以正则表达式匹配的代码并不够强壮,不过目前勉强可以工作,以后有必要再改咯!
总结
以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作能带来一定的帮助,如果有疑问大家可以留言交流,谢谢大家对的支持。
广告合作:本站广告合作请联系QQ:858582 申请时备注:广告合作(否则不回)
免责声明:本站资源来自互联网收集,仅供用于学习和交流,请遵循相关法律法规,本站一切资源不代表本站立场,如有侵权、后门、不妥请联系本站删除!
免责声明:本站资源来自互联网收集,仅供用于学习和交流,请遵循相关法律法规,本站一切资源不代表本站立场,如有侵权、后门、不妥请联系本站删除!
暂无评论...
更新日志
2024年11月25日
2024年11月25日
- 凤飞飞《我们的主题曲》飞跃制作[正版原抓WAV+CUE]
- 刘嘉亮《亮情歌2》[WAV+CUE][1G]
- 红馆40·谭咏麟《歌者恋歌浓情30年演唱会》3CD[低速原抓WAV+CUE][1.8G]
- 刘纬武《睡眠宝宝竖琴童谣 吉卜力工作室 白噪音安抚》[320K/MP3][193.25MB]
- 【轻音乐】曼托凡尼乐团《精选辑》2CD.1998[FLAC+CUE整轨]
- 邝美云《心中有爱》1989年香港DMIJP版1MTO东芝首版[WAV+CUE]
- 群星《情叹-发烧女声DSD》天籁女声发烧碟[WAV+CUE]
- 刘纬武《睡眠宝宝竖琴童谣 吉卜力工作室 白噪音安抚》[FLAC/分轨][748.03MB]
- 理想混蛋《Origin Sessions》[320K/MP3][37.47MB]
- 公馆青少年《我其实一点都不酷》[320K/MP3][78.78MB]
- 群星《情叹-发烧男声DSD》最值得珍藏的完美男声[WAV+CUE]
- 群星《国韵飘香·贵妃醉酒HQCD黑胶王》2CD[WAV]
- 卫兰《DAUGHTER》【低速原抓WAV+CUE】
- 公馆青少年《我其实一点都不酷》[FLAC/分轨][398.22MB]
- ZWEI《迟暮的花 (Explicit)》[320K/MP3][57.16MB]