Pythonでネットワークテストを実装するためのスクリプト共有について詳しく解説

この記事では、主に Python を使用してネットワーク テストを実装する方法を紹介します。この記事は、参考と学習のために詳細なサンプル コードを提供します。必要な友人が一緒に読むことができます。見てください。

はじめに

最近、クラスメートから、ネットワークをテストするツールの作成を手伝ってほしいと頼まれました。仕事の都合上、比較的完成度の高いバージョンを断続的に提供するのに長い時間がかかりました。実際、私は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 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 = &#39;empty&#39;
 ms = &#39;=[0-9]+ms&#39;.decode("utf8")
 print "%s" %ms
 print "%s" %xpin
 mstime=re.search(ms,xpin)
 if not mstime:
  MS=&#39;timeout&#39;
  return MS
 else:
  MS=mstime.group().split(&#39;=&#39;)[1]
  return MS.strip(&#39;ms&#39;)
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 &#39;pin output = %s&#39; %MS
  if MS == &#39;timeout&#39;:
   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 &#39;nrows %s&#39; %nrows
 ips = []
 for i in range(nrows):
  ips.append(table.cell_value(i, 0))
  print table.cell_value(i, 0)
 return ips
 

if name == &#39;main&#39;:
 file = &#39;ping_result.txt&#39;
 times = 10
 network = &#39;&#39;
 thread_num = 10
 args = sys.argv[1:]
 try:
  (opts, getopts) = getopt.getopt(args, &#39;n:t:c:h?&#39;)
 except:
  print "\nInvalid command line option detected."
  usage()
  sys.exit(1)
 for opt, arg in opts:
  if opt in (&#39;-n&#39;):
   network = arg
  if opt in (&#39;-h&#39;, &#39;-?&#39;):
   usage()
   sys.exit(0)
  if opt in (&#39;-t&#39;):
   times = int(arg)
  if opt in (&#39;-c&#39;):
   thread_num = int(arg)
 f = open(file, &#39;w&#39;)
 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 [&#39;TEST&#39;] 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 == &#39;TEST&#39;:
   ips = TEST
  else:
   ips = readIpsInFile(network)
  print &#39;Starting...&#39;
  threads = []
  nest_list = p_list(ips, thread_num)
  loops = range(len(nest_list))
  print &#39;Total %s Threads is working...&#39; %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(&#39;%.2f&#39;%value[1]))
   sheet1.write(it, 2, str(&#39;%.2f&#39;%value[2]))
   sheet1.write(it, 3, str(&#39;%.2f&#39;%value[3]))
   it+=1
   f.write(line + &#39;\t&#39;+ str(&#39;%.2f&#39;%value[1]) + &#39;\t&#39;+ str(&#39;%.2f&#39;%value[2]) + &#39;\t&#39;+ str(&#39;%.2f&#39;%value[3]) + &#39;\n&#39;)
  f.close()
  workbook.save(&#39;ping_result.xls&#39;)
  print &#39;Work Done. please check result %s and ping_result.xls.&#39;%file

このコードは他の人の実装を参考にしていますが、特に複雑ではないので簡単に説明します。

• Excel の読み取りと書き込みは xlrd と xlwt を使用し、基本的にいくつかの単純な api を使用します。

• スレッドを使用してマルチスレッドの同時実行性を実現します。これは、POSIX 標準の インターフェース によく似ています。 thread_func は、 関数 を処理するスレッドです。その入力には IP のリストが含まれるため、各 IP は関数内の ループ を通じて処理されます。

• さらに、Python コマンドは Windows では互換性がないため、subprocess モジュールが使用されます。

これまでのところ、Python の 文字セット の理解が十分ではないため、 正規表現 マッチング コードは十分強力ではありませんが、現時点ではかろうじて機能します。将来的には変えてください！

概要

以上がPythonでネットワークテストを実装するためのスクリプト共有について詳しく解説の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。