Dieser Artikel stellt Ihnen hauptsächlich die Methode zur Verwendung von Python zur Implementierung von Netzwerktests vor. Der Artikel bietet einen bestimmten Referenz- und Lernwert Freunde, lasst uns gemeinsam einen Blick darauf werfen.
Vorwort
Kürzlich bat mich ein Klassenkamerad, beim Schreiben eines Tools zum Testen des Netzwerks zu helfen. Aus beruflichen Gründen dauerte es zeitweise lange, bis eine relativ vollständige Version verfügbar war. Tatsächlich verwende ich Python relativ selten, daher schreibe ich grundsätzlich Programme, während ich Informationen überprüfe.
Die Hauptlogik des Programms ist wie folgt:
Lesen Sie die IP-Liste in einer Excel-Datei und rufen Sie dann mithilfe von Multithreading Ping auf, um die Netzwerkparameter zu zählen jeder IP und schließlich werden die Ergebnisse in eine Excel-Datei ausgegeben.
Der Code lautet wie folgt:
#! /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 p_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?')
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 = p_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.'%fileNach dem Login kopieren
Dieser Code bezieht sich auf die Implementierung anderer Personen, obwohl er nicht besonders kompliziert ist einmal eine einfache Erklärung.
Bisher ist mein Verständnis des
Zeichensatzes in Python nicht gut genug, daher ist der reguläre Ausdruck-Übereinstimmungscode nicht stark genug , aber im Moment kann es kaum funktionieren und es könnte notwendig sein, es in Zukunft zu ändern!
Zusammenfassung
Das obige ist der detaillierte Inhalt vonAusführliche Erläuterung der Skriptfreigabe zur Implementierung von Netzwerktests in Python. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!
