13

使用Python实现批量ping操作

 3 years ago
source link: http://www.cnblogs.com/yemuchangan-007/p/12832770.html
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.

在日常的工作中，我们通常会有去探测目标主机是否存活的应用场景，单个的服务器主机可以通过计算机自带的DOS命令来执行，但是业务的存在往往不是单个存在的，通常都是需要去探测C段的主机（同一个网段下的存活主机），这样使用DOS来进行操作是不可取，探测的速度太慢了，不满足实际需要。一般批量的操作需要使用脚本进行一键部署执行，本文主要通过使用Python语言来实现批量ping的操作（使用多线程实现Python批量处理）

Python版本 ：Python3

使用的第三方库：subprocess, logging, threading, queue

日志导出模块功能：

def set_logging_format():
    logging.basicConfig(level=logging.INFO,
                        format='%(message)s',
                        filename='ping_host.log',
                        filemode='w'
                        )
    console = logging.StreamHandler()
    console.setLevel(logging.INFO)
    formatter = logging.Formatter('%(message)s')
    console.setFormatter(formatter)
    logging.getLogger('').addHandler(console)
多线程实现批量操作：

threads = []
    THREAD_NUM = 20
    user_iput = input('please input modren: ')
    if user_iput == 'addr':
        IP_L = IP_LIST
    if user_iput == 'file':
        IP_L = IP_QUEUE
    for i in range (THREAD_NUM):
        t = threading.Thread(target = ping_IP,args = (IP_L,))
        threads.append(t)
    for i in range (THREAD_NUM):
        threads[i].start()
    for i in range (THREAD_NUM):
        threads[i].join()
完整代码部分：

import subprocess
import logging
import datetime
import time
import threading
from queue import Queue
import sys

# 实现日志导出
def set_logging_format():
    logging.basicConfig(level=logging.INFO,
                        format='%(message)s',
                        filename='ping_host.log',
                        filemode='w'
                        )
    console = logging.StreamHandler()
    console.setLevel(logging.INFO)
    formatter = logging.Formatter('%(message)s')
    console.setFormatter(formatter)
    logging.getLogger('').addHandler(console)

# 将需要ping 连接的IP加入队列

def insert_ip_queue(ip_list_path):
    IP_QUEUE = Queue()
    with open (ip_list_path,'r') as f:
        for ip in f.readlines():
            IP_QUEUE.put(ip)
        f.close()
    return IP_QUEUE

def IP_list ():
    ip_list = Queue()
    for i in range (1,255):
        ip = '192.168.1.' + str(i)
        ip_list.put(ip)
    return ip_list
# print (IP_list())

#print (IP_list())

#定义 ping 函数

def ping_IP (IP_QUEUE):
    while not IP_QUEUE.empty():
        ip = IP_QUEUE.get().strip('\n')
        #print (ip)
        res = subprocess.call('ping -w 1000 -n 1 %s' % ip , stdout=subprocess.PIPE,shell=True)
        #print (res)
        if res == 0:
            h =subprocess.getoutput('ping' + ' ' + ip)
        #print (h)





            if 'TTL=' in h:
                res = ('网络可以正常连通平均延时 = %s' % h.split('平均 = ')[1])
        else:
            
            res = '网络连接失败!'
        today = datetime.datetime.now().strftime("%Y - %m - %d %H : %M : %S")
        logging.info("%s IP =  %s %s" % (today,ip,res))

def main ():
    set_logging_format()
    ip_list_path = './hostip.txt'
    IP_QUEUE = insert_ip_queue(ip_list_path)
    IP_LIST = IP_list()

    threads = []
    THREAD_NUM = 20
    user_iput = input('please input modren: ')
    if user_iput == 'addr':
        IP_L = IP_LIST
    if user_iput == 'file':
        IP_L = IP_QUEUE
    for i in range (THREAD_NUM):
        t = threading.Thread(target = ping_IP,args = (IP_L,))
        threads.append(t)
    for i in range (THREAD_NUM):
        threads[i].start()
    for i in range (THREAD_NUM):
        threads[i].join()

if __name__ == '__main__':
    main()

report

Recommend

About Joyk


Aggregate valuable and interesting links.
Joyk means Joy of geeK