21

Python とネットワークに関する簡単に学べる優れたドキュメントが見つかりません。この例では、多数のリモート マシンに ping できる簡単なスクリプトを作成しようとしています。

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

time.sleep(5)そのような単純なスクリプトはマシンに正常に ping を実行しますが、スクリプトが「アクティブ」「応答なし」を返すようにしたいので、時間モジュールも調べる必要があると思います。つまり、break ステートメントがあります。そのため、for の中に while ループが必要だと思います。私は完全に間違った方向に進んでいる可能性があります.

4

9 に答える 9

30

試してみてください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!"
于 2012-08-23T23:21:56.223 に答える
12

このスクリプト：

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使用するとcommunicate()、出力をキャプチャできます。Popen

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

このようにして、コマンドの戻り値を取得し、テキストをキャプチャできます。マニュアルに従うと、柔軟性が必要な場合、これがサブプロセスを操作するための推奨される方法です。

このモジュールの基礎となるプロセスの作成と管理は、Popenクラスによって処理されます。開発者が便利な機能でカバーされていないあまり一般的でないケースを処理できるように、それは多くの柔軟性を提供します。

于 2012-08-23T23:15:10.940 に答える
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 を変更するだけです。

于 2013-09-23T20:27:01.933 に答える
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"))
于 2012-08-24T01:11:44.053 に答える
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")                    
                    else:
                        pass                        

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)

私はこれがより良いものになると思います。

于 2016-04-02T23:56:34.677 に答える