Python中的多个ping脚本

22

我找不到任何关于Python和网络的好易学文档。在这种情况下,我只是试图编写一个简单的脚本,以便可以对多台远程机器进行ping测试。

for ping in range(1,10):
   ip="127.0.0."+str(ping)
   os.system("ping -c 3 %s" % ip)

像这样的简单脚本可以很好地ping通机器,但我希望脚本返回“活动”“无响应”。这让我想起也许我需要查找时间模块,我认为time.sleep(5)之后,应该有一个break语句。这让我想到for循环里面应该有一个while循环。我不确定100%,完全可能我正在走错方向 :/ 如果有人能提供帮助或指向一些文档,那就太好了。


我不确定为什么您认为需要时间模块?我建议研究如何解析subprocess的STDOUT(这是您应该使用而不是os.system())。 - ernie
2
尝试使用 Scapy - kichik
这是一个例子 https://github.com/lbaby/javalearn/blob/master/shell/ppe.py - lbaby
不要使用Scapy,因为对于任何需要吞吐量或可靠性的应用程序来说,Scapy都是糟糕的选择...这两者都是监控应用程序所必需的。 - Mike Pennington
9个回答

31

尝试使用subprocess.call。它会保存所使用程序的返回值。

根据我的ping手册,它在成功时返回0,在发送ping但未收到回复时返回2,任何其他值表示错误。

# typo error in import
import subprocess

for ping in range(1,10):
    address = "127.0.0." + str(ping)
    res = subprocess.call(['ping', '-c', '3', address])
    if res == 0:
        print "ping to", address, "OK"
    elif res == 2:
        print "no response from", address
    else:
        print "ping to", address, "failed!"

3
不用谢。请记得将对你有帮助的答案标记为“已采纳”(点击问题投票计数下方的绿色勾号),这可以向其他访客传达答案有帮助的信息。 :-) - Roland Smith
1
subprocess.call() 一次只能ping一个ip。如果要同时ping多个ip,则可以使用 subprocess.Popen()。参考链接:https://dev59.com/umct5IYBdhLWcg3wa83p#12102040 - jfs

14

这个脚本:

import subprocess
import os
with open(os.devnull, "wb") as limbo:
        for n in xrange(1, 10):
                ip="192.168.0.{0}".format(n)
                result=subprocess.Popen(["ping", "-c", "1", "-n", "-W", "2", ip],
                        stdout=limbo, stderr=limbo).wait()
                if result:
                        print ip, "inactive"
                else:
                        print ip, "active"

将会输出类似这样的结果:

192.168.0.1 active
192.168.0.2 active
192.168.0.3 inactive
192.168.0.4 inactive
192.168.0.5 inactive
192.168.0.6 inactive
192.168.0.7 active
192.168.0.8 inactive
192.168.0.9 inactive

如果你将limbo替换为subprocess.PIPE并在Popen对象上使用communicate(),就可以捕获输出:

p=Popen( ... )
output=p.communicate()
result=p.wait()

这种方法可以获取命令的返回值并捕获文本。根据手册,如果需要灵活操作子进程,则通常建议使用 Popen 类处理子进程的创建和管理,它提供了很多灵活性,使开发人员能够处理不常见的情况而不是使用方便函数。

该模块中的底层进程创建和管理由 Popen 类处理。它提供了很大的灵活性,使开发人员能够处理不常见的情况而不是使用方便函数。


请在评论中留下您投反对票的解释,以便答案可以得到改进。 - hochl

6
非常感谢您的帮助。我已经将其修改为适用于Windows平台。我还设置了一个较短的超时时间,这样没有返回值的IP地址就不会等待5秒钟。这是来自hochl源代码的内容。
import subprocess
import os
with open(os.devnull, "wb") as limbo:
        for n in xrange(200, 240):
                ip="10.2.7.{0}".format(n)
                result=subprocess.Popen(["ping", "-n", "1", "-w", "200", ip],
                        stdout=limbo, stderr=limbo).wait()
                if result:
                        print ip, "inactive"
                else:
                        print ip, "active"

只需更改ip=为您的方案,将xrange更改为主机范围即可。


4

我是一个初学者,写了一个用于 ping 多个主机的脚本。如果要 ping 多个主机,您可以使用 ipaddress 模块。

import ipaddress
from subprocess import Popen, PIPE

net4 = ipaddress.ip_network('192.168.2.0/24')
for x in net4.hosts():
    x = str(x)
    hostup = Popen(["ping", "-c1", x], stdout=PIPE)
    output = hostup.communicate()[0]
    val1 = hostup.returncode
 if val1 == 0:
    print(x, "is pinging")
 else:
    print(x, "is not responding")

3
为了同时ping多个主机,你可以使用subprocess.Popen()
#!/usr/bin/env python3
import os
import time
from subprocess import Popen, DEVNULL

p = {} # ip -> process
for n in range(1, 100): # start ping processes
    ip = "127.0.0.%d" % n
    p[ip] = Popen(['ping', '-n', '-w5', '-c3', ip], stdout=DEVNULL)
    #NOTE: you could set stderr=subprocess.STDOUT to ignore stderr also

while p:
    for ip, proc in p.items():
        if proc.poll() is not None: # ping finished
            del p[ip] # remove from the process list
            if proc.returncode == 0:
                print('%s active' % ip)
            elif proc.returncode == 1:
                print('%s no response' % ip)
            else:
                print('%s error' % ip)
            break

如果您可以以root身份运行,则可以使用纯Python ping脚本scapy

from scapy.all import sr, ICMP, IP, L3RawSocket, conf

conf.L3socket = L3RawSocket # for loopback interface
ans, unans = sr(IP(dst="127.0.0.1-99")/ICMP(), verbose=0) # make requests
ans.summary(lambda (s,r): r.sprintf("%IP.src% is alive"))

1
import subprocess
import os
'''
servers.txt contains ip address in following format
192.168.1.1
192.168.1.2
'''
    with open('servers.txt', 'r') as f:
        for ip in f:
            result=subprocess.Popen(["ping", "-c", "1", "-n", "-W", "2",    ip],stdout=f, stderr=f).wait()
            if result:
                print(ip, "inactive")
            else:
                print(ip, "active")

1
Python实际上有一个非常好用的方法,可以“返回网络中可用主机的迭代器”(将strict设置为false会遍历所有IP)。
例如:
import subprocess
import ipaddress

subnet = ipaddress.ip_network('192.168.1.0/24', strict=False)
for i in subnet.hosts():
    i = str(i)
    subprocess.call(["ping", "-c1", "-n", "-i0.1", "-W1", i])

等待间隔(-i0.1)可能对自动化很重要,即使是一秒钟的超时(-t1)也可能在 .0/24 上永远持续。
编辑: 因此,为了跟踪 ICMP(ping)请求,我们可以这样做:
#!/usr/bin/env python

import subprocess
import ipaddress

alive = []
subnet = ipaddress.ip_network('192.168.1.0/23', strict=False)
for i in subnet.hosts():
    i = str(i)
    retval = subprocess.call(["ping", "-c1", "-n", "-i0.1", "-W1", i])
    if retval == 0:
        alive.append(i)
for ip in alive:
    print(ip + " is alive") 

这将返回类似以下内容的东西:
192.168.0.1 is alive
192.168.0.2 is alive
192.168.1.1 is alive
192.168.1.246 is alive

即所有响应ICMP的IP地址范围涵盖整个/23网段-非常酷!

FYI,我在没有将IP地址指定为Unicode之前一直出现错误:subnet = ipaddress.ip_network(u'192.168.1.0/23', strict=False) - blackappy

0
import subprocess,os,threading,time
from queue import Queue
lock=threading.Lock()
_start=time.time()
def check(n):
    with open(os.devnull, "wb") as limbo:
                ip="192.168.21.{0}".format(n)
                result=subprocess.Popen(["ping", "-n", "1", "-w", "300", ip],stdout=limbo, stderr=limbo).wait()
                with lock:                    
                    if not result:
                        print (ip, "active")                       
    
def threader():
    while True:
        worker=q.get()
        check(worker)
        q.task_done()
q=Queue()

for x in range(255):
    t=threading.Thread(target=threader)
    t.daemon=True
    t.start()
for worker in range(1,255):
    q.put(worker)
q.join()
print("Process completed in: ",time.time()-_start)

我认为这个会更好。


谢谢,它对我有用。 为避免错误需要几次机会: 在第二行中导入Queue: import Queue 在第21行中: q = Queue.Queue() - umesh pant
为什么要加上 else: pass - Minek Po1

0

我已经在Python 2.7中使用多线程对上述代码进行了几处修改:

import subprocess,os,threading,time
import Queue

lock=threading.Lock()
_start=time.time()
def check(n):
    with open(os.devnull, "wb") as limbo:
                ip=n
                result=subprocess.Popen(["ping", "-n", "2", "-w", "300", ip],stdout=limbo, stderr=limbo).wait()
                with lock:
                    if not result:
                        print ip, "active"
                    else:
                        print ip, "Inactive"

def threader():
    while True:
        worker=q.get()
        check(worker)
        q.task_done()
q = Queue.Queue()

for x in range(255):
    t=threading.Thread(target=threader)
    t.daemon=True
    t.start()

ip = ["13.45.23.523", "13.35.23.523","23.23.56.346"]
for worker in ip:
    q.put(worker)
q.join()
print("Process completed in: ",time.time()-_start)

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接