Heim > Artikel > Backend-Entwicklung > Ausführliche Erläuterung der Skriptfreigabe zur Implementierung von Netzwerktests in Python
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.'%file
Dieser Code bezieht sich auf die Implementierung anderer Personen, obwohl er nicht besonders kompliziert ist einmal eine einfache Erklärung.
Excel liest und schreibt mit xlrd und xlwt, im Grunde mit einer einfachen API.
verwendet Threading, um Multi-Thread-Parallelität zu erreichen, was der POSIX-Standard-Schnittstelle sehr ähnlich ist. thread_func ist die Thread-Verarbeitungsfunktion . Ihre Eingabe enthält eine Liste von IPs, sodass jede IP innerhalb der Funktion durch eine -Schleife verarbeitet wird.
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!