用Python教你寫一個批量ping

[TOC]

前言

最近幾天，剛好需要配合防火墻替換的割接方案褪猛，需要去批量ping測試20+個C類網(wǎng)段网杆，約5000+個地址，我同事在網(wǎng)上找的工具也不能很好的一次性ping完所有網(wǎng)段的IP地址伊滋，心想碳却，我來幫你搞定，就花些時間劈里啪啦的調(diào)試下代碼笑旺，其中還是遇到一些疑難雜癥的昼浦，比如下所列：

• 使用的模塊，必須得有回顯代碼0或1讓你判斷是通還是不通筒主，另外ping結(jié)果也要保存关噪？
• 使用的模塊，ping回顯是通的乌妙，但實(shí)際用電腦ping是不通的使兔，你說妖不妖？
• 綜合測試了scrapy藤韵、pythonping和subprocess模塊虐沥，最終還是選擇subprocess內(nèi)置模塊符合我心意。

想要寫一個批量的ping腳本泽艘，首先需要厘清下編寫的思路欲险，要怎么去實(shí)現(xiàn)？

這是我整理的思路匹涮，我會一步一步教大家怎么去實(shí)現(xiàn)天试，咋們接著往下看...

實(shí)驗環(huán)境

• 系統(tǒng)：windows 10 和 Centos7.6
• python工具：Pycharm Pro
• 模塊：內(nèi)置模塊

備注：沒啥特別的要求。

代碼分解

如何讀取文本中的所有網(wǎng)段然低？

說明：通過上下文管理with讀取文本中的所有IP網(wǎng)段信息

假如你把收集的所有IP網(wǎng)段都集中放到了一個文本當(dāng)中喜每，那么我們在執(zhí)行腳本的時候，怎么把它提取出來呢脚翘？

# 存放ip網(wǎng)段的文本路徑(當(dāng)前目錄)
IP_INFO_PATH = 'IPnet.cfg'
# 以只讀權(quán)限打開文本
with open(IP_INFO_PATH, 'r') as f:
    # for循環(huán)遍歷每一行
    for line in f.readlines():
        #去掉前后空白字符
        line = line.strip()
        # 忽略空白行或注釋行
        if ( len(line) == 1 or line.startswith('#') ):
            continue
        print(line)

執(zhí)行結(jié)果如下所示：

192.168.100.0/24
192.168.101.0/24
192.168.102.0/24
192.168.103.0/24
192.168.104.0/24

代碼解釋：

• 先定義好存放ip網(wǎng)段的文本路徑：IP_INFO_PATH = 'IPnet.cfg'
• 打開文本通過上下文管理with灼卢，格式：with open('文本路徑'，'只讀方式') as '別名'
• readlines()方法可以讀取行信息来农，并通過for循環(huán)就遍歷所有行信息鞋真；
• len(line) == 1, 表示空白行長度為1，記住不是0沃于。

如何獲取網(wǎng)段的主機(jī)地址涩咖？

說明：該模塊我以前有詳細(xì)寫過海诲，請在公眾號搜索<用Python幫你實(shí)現(xiàn)IP子網(wǎng)計算>

• 如給你一個網(wǎng)段，你怎么去ping呢檩互？
  對于我們專業(yè)來說特幔，心算就可以得出來了，其他同學(xué)就拿起子網(wǎng)計算工具吧闸昨，那么蚯斯，我就通過ipaddress模塊給大家演示下：

  假如有一個網(wǎng)段：192.168.100.0/24, 快速計算所有的主機(jī)地址。

  # 導(dǎo)入內(nèi)置模塊
  import ipaddress
  
  IPnet = '192.168.100.0/24'
  # 創(chuàng)建一個空的列表饵较，用于存放主機(jī)地址信息
  ip_list = []
  # 通過ip_network方法拍嵌，賦值給temp，這是一個生成器
  # 已經(jīng)計算出所有主機(jī)了循诉，需要迭代出來
  temp = ipaddress.ip_network(IPnet, strict=False).hosts()
  # for循環(huán)出主機(jī)地址横辆，添加到ip_list這個列表
  for ip in temp:
      # print(type(ip), ip)
      ip_list.append(str(ip))
  print(ip_list)
  

  這一步，我們已經(jīng)掌握如何計算一個網(wǎng)段的所有主機(jī)地址了茄猫。

如何執(zhí)行ping程序狈蚤？

subprocess是一個內(nèi)置模塊，它可以讓你創(chuàng)建一個新的子程序來運(yùn)行划纽。通常使用如下3個方法：

• subprocess.run()
• subprocess.call()
• subprocess.Popen()

這里脆侮，我只介紹簡單好用的run()方法就夠滿足我們的需求了。

• 一個簡單的示例：

  import subprocess
  
  ip = '114.114.114.114'
  # windows上cmd的ping操作格式
  res = subprocess.run(['ping', '-n', '2', '-w', '500', ip])
  print('-'*50)
  print('類對象:', res)
  print('什么類型:', type(res))
  print('返回代碼值:', res.returncode)
  

  執(zhí)行結(jié)果如下所示：

  如果不想再終端上看到ping的執(zhí)行結(jié)果, 增加參數(shù)stdout=subprocess.PIPE：

  說明：簡單理解就是把執(zhí)行結(jié)果隱藏起來阿浓。

  # 修改代碼如下
  res = subprocess.run(['ping', '-n', '2', '-w', '500', ip], stdout=subprocess.PIPE)
  

• 如何把執(zhí)行的結(jié)果寫入到文本當(dāng)中他嚷？

  說明：既然隱藏了ping結(jié)果，那就要把結(jié)果賦值給一個變量芭毙，就可以拿到結(jié)果了.

  import subprocess
  
  ip_list  = ['8.8.8.8', '114.114.114.114', '114.114.114.115']
  for ip in ip_list:
      res = subprocess.run(['ping', '-n', '2', '-w', '500', ip], stdout=subprocess.PIPE)
      # 將輸出標(biāo)準(zhǔn)流解碼成中文
      output = res.stdout.decode('gbk')
      # ping通代碼為0
      if res.returncode == 0:
          print('%-20s%-20s' % (ip, 'success'))
          # ping通結(jié)果寫入文本
          with open('ping_success.txt', 'a+') as f:
              f.write('%-20s%-20s' % (ip, 'success'))
          # ping執(zhí)行結(jié)果寫入文本
          with open('ping_result.txt', 'a+') as f:
              f.write(output)
              f.write('-'*50)
      # ping不通代碼為1
      else:
          print('%-20s%-20s' % (ip, 'failure'))
          # ping不通結(jié)果寫入文本
          with open('ping_failure.txt', 'a+') as f:
              f.write('%-20s%-20s' % (ip, 'success'))
          # ping執(zhí)行結(jié)果寫入文本
          with open('ping_result.txt', 'a+') as f:
              f.write(output)
              f.write('-' * 50)
  

  執(zhí)行結(jié)果如下所示：

代碼執(zhí)行完成后自動創(chuàng)建了文件，這里就不把貼出來了卸耘，自行打開文本進(jìn)行驗證退敦。

• 如何并發(fā)執(zhí)行ping呢？

  使用多進(jìn)程multiprocessing的異步方式來看看并發(fā)的效果是怎么樣的(多線程threading就不用了)蚣抗。

  說明：multiprocessing侈百，簡單理解就是通過CPU多核進(jìn)行并發(fā)處理，本示例只演示其功能翰铡。

  from multiprocessing.pool import ThreadPool
  from datetime import datetime
  import subprocess
  
  THREADING_NUM = 10
  pool = ThreadPool(THREADING_NUM)
  
  def do_ping(ip):
      res = subprocess.run(['ping', '-n', '2', '-w', '500', ip], stdout=subprocess.PIPE)
      output = res.stdout.decode('gbk')
      if res.returncode == 0:
          print('%-20s%-20s' % (ip, 'success'))
          with open('ping_success.txt', 'a+') as f:
              f.write('%-20s%-20s' % (ip, 'success'))
  
          with open('ping_result.txt', 'a+') as f:
              f.write(output)
              f.write('-' * 50)
      else:
          print('%-20s%-20s' % (ip, 'failure'))
          with open('ping_failure.txt', 'a+') as f:
              f.write('%-20s%-20s' % (ip, 'success'))
  
          with open('ping_result.txt', 'a+') as f:
              f.write(output)
              f.write('-' * 50)
              
  if __name__ == '__main__':
      start_time = datetime.now()
      ip_list = ['1.1.1.1', '1.1.1.2','1.1.1.3', '8.8.8.8', '114.114.114.114', '114.114.114.115']
      for ip in ip_list:
          pool.apply_async(do_ping, args=(ip,))
      pool.close()
      pool.join()
  
      end_time = datetime.now()
      print('All done.總花費(fèi)時間{:0.2f}s.'.format((end_time - start_time).total_seconds()))
  

  執(zhí)行結(jié)果如下所示：

  說明：具體效果钝域，大家可通過大規(guī)模網(wǎng)段進(jìn)行測試，方可看出其效果明顯锭魔。

完整代碼

• 以下為本章更新的完整代碼例证，大家可在此基礎(chǔ)上進(jìn)行拓展和優(yōu)化：

  #!/usr/bin/env python3
  # -*- coding:UTF-8 -*-
  
  from multiprocessing import freeze_support
  from multiprocessing.pool import ThreadPool
  from datetime import datetime
  
  import subprocess
  import ipaddress
  import threading
  import logging
  import sys
  
  # ip網(wǎng)段文本路徑(當(dāng)前目錄下)
  IP_INFO_PATH = 'ip_list.txt'
  
  # 線程數(shù)()
  THREADING_NUM = 10
  # 進(jìn)程池
  pool = ThreadPool(THREADING_NUM)
  # 線程鎖
  queueLock = threading.Lock()
  
  # 打印消息
  def show_info(msg):
      queueLock.acquire()
      print(msg)
      queueLock.release()
  
  # 讀取文本，獲取ip網(wǎng)段信息
  def get_ips_info():
      try:
          with open(IP_INFO_PATH, 'r') as f:
              for line in f.readlines():
                  # 去掉前后空白
                  line = line.strip()
                  # 忽略空格行迷捧，len=1
                  if (
                          len(line) == 1 or
                          line.startswith('#')
                  ):
                      continue
  
                  yield line
  
      except FileNotFoundError as e:
          show_info('Can not find "{}"'.format(IP_INFO_PATH))
      except IndexError as e:
          show_info('"{}" format error'.format(IP_INFO_PATH))
  
  def do_one_ping(target_ip):
      """
      單機(jī)ping測試
      """
      res = ''
      if sys.platform == 'linux':
          res = subprocess.run(['ping', '-c', '2', '-W', '1000', target_ip], stdout=subprocess.PIPE)
          res_out = str(res.stdout.decode('gbk'))
      if sys.platform == 'win32':
          res = subprocess.call(['ping', '-n', '2', '-w', '1000', target_ip ], stdout = subprocess.PIPE)
      else:
          show_info('不支持該平臺系統(tǒng)织咧，非常抱歉!')
      # print(f'res狀態(tài)是: {res}')
      if res.returncode == 0:
          show_info('%-20s%-20s' % (target_ip, 'success'))
  
      else:
          show_info('%-20s%-20s' % (target_ip, 'failure'))
  
  def do_ping(target_ip):
      """
      批量ping測試
      """
      try:
          res , res_out  = '', ''
          # 判斷系統(tǒng)平臺胀葱，執(zhí)行對應(yīng)命令
          if sys.platform == 'linux':
              res = subprocess.run(['ping', '-c', '2', '-W', '1000', target_ip], stdout=subprocess.PIPE)
              res_out = str(res.stdout.decode('gbk'))
          # 判斷系統(tǒng)平臺，執(zhí)行對應(yīng)命令
          if sys.platform == 'win32':
              res = subprocess.run(['ping', '-n', '2', '-w', '1000', target_ip ], stdout = subprocess.PIPE)
              res_out = str(res.stdout.decode('gbk'))
          else:
              show_info('不支持該平臺系統(tǒng)笙蒙，非常抱歉!')
  
          # print(f'res狀態(tài)是: {res.returncode}')
          if res.returncode == 0:
              show_info('%-20s%-20s' % (target_ip, 'success'))
              # ping成功
              with open('LOG' + '/' + 'success_ping_result_' + LogTime + '.txt', 'a+') as f:
                  f.write('%-20s%-20s' % (target_ip, 'success') + '\n')
              # ping成功結(jié)果記錄
              with open('LOG' + '/' + 'record_ping_' + LogTime + '.txt', 'a+') as f:
                  f.write(res_out)
                  f.write('-' * 50)
          else:
              show_info('%-20s%-20s' % (target_ip, 'failure'))
              # ping失敗
              with open('LOG' + '/' + 'failure_ping_result_' + LogTime + '.txt', 'a+') as f:
                  f.write('%-20s%-20s' % (target_ip, 'failure') + '\n')
              # ping失敗結(jié)果記錄
              with open('LOG' + '/' + 'record_ping_' + LogTime + '.txt', 'a+') as f:
                  f.write(res_out)
                  f.write('-' * 50)
  
      except Exception as e:
          show_info(e)
  
  def get_ip_list(ip):
      """
      獲取ip列表
      """
      temp = ipaddress.ip_network(ip, False).hosts()
      ip_list = []
      for item in temp:
          ip_list.append(str(item))
      # print(ip_list)
      return ip_list
  
  
  if __name__ == '__main__':
      freeze_support()
      LogTime = datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
      start_time = datetime.now()
  
      # 單主機(jī)ping
      # do_one_ping('114.114.114.114')
  
      # 批量執(zhí)行ping
      ips_list = get_ips_info()
      for ips in ips_list:
          ip_list = get_ip_list(ips)
          for ip in ip_list:
              pool.apply_async(do_ping, args=(ip,))
      pool.close()
      pool.join()
  
      end_time = datetime.now()
      print('All done.總花費(fèi)時間{:0.2f}s.'.format((end_time - start_time).total_seconds()))
  

okay, 本章就先介紹這么多了抵屿，相信大家有了一個清晰的思路了，再把各個模塊ipaddress捅位、subprocess轧葛、multiprocessing都熟悉一邊，實(shí)操干起來艇搀，我相信大家能夠勝任朝群，并在此基礎(chǔ)上更新迭代，寫出更好中符、更優(yōu)雅的代碼姜胖，咋們下次見。

碼字不易淀散，如覺得本文章有用右莱，請動動手指給個愛心、分享給有需要的朋友拉档插，多謝各位慢蜓，晚安。

如果喜歡的我的文章郭膛，歡迎關(guān)注我的公號：點(diǎn)滴技術(shù)晨抡，掃碼關(guān)注，不定期分享