用Python教你寫一個批量ping

[TOC]

前言

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

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

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

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

實驗環(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)段的主機地址？

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

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

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

# 導(dǎo)入內(nèi)置模塊
import ipaddress

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

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

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

subprocess是一個內(nèi)置模塊，它可以讓你創(chuàng)建一個新的子程序來運行。通常使用如下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.總花費時間{: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):
    """
    單機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()

    # 單主機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.總花費時間{:0.2f}s.'.format((end_time - start_time).total_seconds()))

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

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

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

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av

【干貨】用Python教你寫一個批量ping

【干貨】用Python教你寫一個批量ping

前言

實驗環(huán)境

代碼分解

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

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

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

完整代碼

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九 欧美,1769亚洲,黄色成人av

【干貨】用Python教你寫一個批量ping

前言

實驗環(huán)境

代碼分解

如何讀取文本中的所有網(wǎng)段？

如何獲取網(wǎng)段的主機地址？

如何執(zhí)行ping程序？

完整代碼

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av

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

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

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