Jak pingować witrynę lub adres IP za pomocą Pythona?
Jak pingować witrynę lub adres IP za pomocą Pythona?
Odpowiedzi:
Zobacz ten ping w czystym Pythonie autorstwa Matthew Dixona Cowlesa i Jensa Diemera . Pamiętaj też, że Python wymaga roota do tworzenia gniazd ICMP (tj. Pingowania) w Linuksie.
import ping, socket
try:
ping.verbose_ping('www.google.com', count=3)
delay = ping.Ping('www.wikipedia.org', timeout=2000).do()
except socket.error, e:
print "Ping Error:", e
Sam kod źródłowy jest łatwy do odczytania, zobacz implementacje verbose_ping
i Ping.do
dla inspiracji.
ping
używa, time.clock
które nie dają niczego użytecznego na moim Linuksie. timeit.default_timer
(to jest równe time.time
na moim komputerze) działa. time.clock
-> timeit.default_timer
gist.github.com/255009
W zależności od tego, co chcesz osiągnąć, prawdopodobnie najłatwiej jest wywołać systemowe polecenie ping.
Najlepszym sposobem na to jest użycie modułu podprocesu, chociaż musisz pamiętać, że polecenie ping jest różne w różnych systemach operacyjnych!
import subprocess
host = "www.google.com"
ping = subprocess.Popen(
["ping", "-c", "4", host],
stdout = subprocess.PIPE,
stderr = subprocess.PIPE
)
out, error = ping.communicate()
print out
Nie musisz martwić się o znaki ucieczki pociskiem. Na przykład..
host = "google.com; `echo test`
..will nie wykonać polecenia echo.
Teraz, aby faktycznie uzyskać wyniki pingowania, możesz przeanalizować out
zmienną. Przykładowe dane wyjściowe:
round-trip min/avg/max/stddev = 248.139/249.474/250.530/0.896 ms
Przykładowe wyrażenie regularne:
import re
matcher = re.compile("round-trip min/avg/max/stddev = (\d+.\d+)/(\d+.\d+)/(\d+.\d+)/(\d+.\d+)")
print matcher.search(out).groups()
# ('248.139', '249.474', '250.530', '0.896')
Ponownie pamiętaj, że dane wyjściowe będą się różnić w zależności od systemu operacyjnego (a nawet wersji ping
). To nie jest idealne rozwiązanie, ale będzie działać dobrze w wielu sytuacjach (jeśli znasz maszyny, na których będzie działać skrypt)
out
zawiera zakodowane \ n, co wydaje się przeszkadzać w dopasowywaniu:matcher = re.compile("\nround-trip min/avg/max/stddev = (\d+.\d+)/(\d+.\d+)/(\d+.\d+)/(\d+.\d+)")
matcher.search
zamiast tego bez zmiany wyrażenia regularnego .
Możesz znaleźć prezentację Noah Gift Tworzenie narzędzi wiersza poleceń Agile w Pythonie . Łączy w nim podproces, kolejkę i wątki, aby opracować rozwiązanie, które jest w stanie jednocześnie pingować hosty i przyspieszyć proces. Poniżej znajduje się podstawowa wersja, zanim doda parsowanie wiersza poleceń i kilka innych funkcji. Kod do tej wersji i innych można znaleźć tutaj
#!/usr/bin/env python2.5
from threading import Thread
import subprocess
from Queue import Queue
num_threads = 4
queue = Queue()
ips = ["10.0.1.1", "10.0.1.3", "10.0.1.11", "10.0.1.51"]
#wraps system ping command
def pinger(i, q):
"""Pings subnet"""
while True:
ip = q.get()
print "Thread %s: Pinging %s" % (i, ip)
ret = subprocess.call("ping -c 1 %s" % ip,
shell=True,
stdout=open('/dev/null', 'w'),
stderr=subprocess.STDOUT)
if ret == 0:
print "%s: is alive" % ip
else:
print "%s: did not respond" % ip
q.task_done()
#Spawn thread pool
for i in range(num_threads):
worker = Thread(target=pinger, args=(i, queue))
worker.setDaemon(True)
worker.start()
#Place work in queue
for ip in ips:
queue.put(ip)
#Wait until worker threads are done to exit
queue.join()
Jest także autorem: Python for Unix and Linux System Administration
http://ecx.images-amazon.com/images/I/515qmR%2B4sjL._SL500_AA240_.jpg
Trudno powiedzieć, jakie jest twoje pytanie, ale jest kilka alternatyw.
Jeśli masz na myśli dosłowne wykonanie żądania przy użyciu protokołu ICMP ping, możesz pobrać bibliotekę ICMP i bezpośrednio wykonać żądanie ping. Google „Python ICMP”, aby znaleźć takie rzeczy, jak ten icmplib . Możesz spojrzeć na scapy .
Będzie to znacznie szybsze niż użycie os.system("ping " + ip )
.
Jeśli masz zamiar generalnie „pingować” urządzenie, aby sprawdzić, czy jest włączone, możesz użyć protokołu echa na porcie 7.
W przypadku echo używasz biblioteki gniazd do otwierania adresu IP i portu 7. Piszesz coś na tym porcie, wysyłasz znak powrotu karetki ( "\r\n"
), a następnie czytasz odpowiedź.
Jeśli chcesz „pingować” witrynę sieci Web, aby sprawdzić, czy jest ona uruchomiona, musisz użyć protokołu http na porcie 80.
W przypadku lub prawidłowego sprawdzania serwera WWW używasz urllib2 do otwierania określonego adresu URL. (/index.html
jest zawsze popularne) i przeczytaj odpowiedź.
Nadal istnieje więcej potencjalnych znaczeń „ping”, w tym „traceroute” i „finger”.
Coś podobnego zrobiłem w ten sposób, jako inspirację:
import urllib
import threading
import time
def pinger_urllib(host):
"""
helper function timing the retrival of index.html
TODO: should there be a 1MB bogus file?
"""
t1 = time.time()
urllib.urlopen(host + '/index.html').read()
return (time.time() - t1) * 1000.0
def task(m):
"""
the actual task
"""
delay = float(pinger_urllib(m))
print '%-30s %5.0f [ms]' % (m, delay)
# parallelization
tasks = []
URLs = ['google.com', 'wikipedia.org']
for m in URLs:
t = threading.Thread(target=task, args=(m,))
t.start()
tasks.append(t)
# synchronization point
for t in tasks:
t.join()
subprocess
/index.html
; w każdej witrynie, w której faktycznie byłby wywołany dokument index.html
, znajdowałby się on właśnie tam, w katalogu głównym serwera. Zamiast tego, jako prepend http://
lub https://
do gospodarza
Oto krótki fragment używający subprocess
. check_call
Metoda albo zwraca 0 w przypadku powodzenia lub zgłasza wyjątek. W ten sposób nie muszę analizować danych wyjściowych polecenia ping. Używam shlex
do dzielenia argumentów wiersza poleceń.
import subprocess
import shlex
command_line = "ping -c 1 www.google.comsldjkflksj"
args = shlex.split(command_line)
try:
subprocess.check_call(args,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
print "Website is there."
except subprocess.CalledProcessError:
print "Couldn't get a ping."
-c
jest -n
, a logika dotycząca kodu zwrotnego jest inna)
czytaj nazwę pliku, plik zawiera jeden adres URL w każdym wierszu, na przykład:
http://www.poolsaboveground.com/apache/hadoop/core/
http://mirrors.sonic.net/apache/hadoop/core/
użyj polecenia:
python url.py urls.txt
uzyskać wynik:
Round Trip Time: 253 ms - mirrors.sonic.net
Round Trip Time: 245 ms - www.globalish.com
Round Trip Time: 327 ms - www.poolsaboveground.com
kod źródłowy (url.py):
import re
import sys
import urlparse
from subprocess import Popen, PIPE
from threading import Thread
class Pinger(object):
def __init__(self, hosts):
for host in hosts:
hostname = urlparse.urlparse(host).hostname
if hostname:
pa = PingAgent(hostname)
pa.start()
else:
continue
class PingAgent(Thread):
def __init__(self, host):
Thread.__init__(self)
self.host = host
def run(self):
p = Popen('ping -n 1 ' + self.host, stdout=PIPE)
m = re.search('Average = (.*)ms', p.stdout.read())
if m: print 'Round Trip Time: %s ms -' % m.group(1), self.host
else: print 'Error: Invalid Response -', self.host
if __name__ == '__main__':
with open(sys.argv[1]) as f:
content = f.readlines()
Pinger(content)
import subprocess as s
ip=raw_input("Enter the IP/Domain name:")
if(s.call(["ping",ip])==0):
print "your IP is alive"
else:
print "Check ur IP"
Jeśli chcesz czegoś faktycznie w Pythonie, czym możesz się bawić, spójrz na Scapy:
from scapy.all import *
request = IP(dst="www.google.com")/ICMP()
answer = sr1(request)
To moim zdaniem znacznie lepsze (iw pełni wieloplatformowe) niż niektóre funky wywołania podprocesu. Możesz także mieć tyle informacji o odpowiedzi (identyfikator sekwencji .....), ile chcesz, ile masz samego pakietu.
Rozwijam bibliotekę, która moim zdaniem może ci pomóc. Nazywa się icmplib (niezwiązany z żadnym innym kodem o tej samej nazwie, który można znaleźć w Internecie) i jest czystą implementacją protokołu ICMP w Pythonie.
Jest całkowicie zorientowany obiektowo i ma proste funkcje, takie jak klasyczny ping, multiping i traceroute, a także niskopoziomowe klasy i gniazda dla tych, którzy chcą tworzyć aplikacje w oparciu o protokół ICMP.
Oto kilka innych najważniejszych informacji:
Aby go zainstalować (wymagany Python 3.6+):
pip3 install icmplib
Oto prosty przykład funkcji ping:
host = ping('1.1.1.1', count=4, interval=1, timeout=2, privileged=True)
if host.is_alive:
print(f'{host.address} is alive! avg_rtt={host.avg_rtt} ms')
else:
print(f'{host.address} is dead')
Ustaw parametr „uprzywilejowany” na Fałsz, jeśli chcesz używać biblioteki bez uprawnień administratora.
Pełną dokumentację można znaleźć na stronie projektu: https://github.com/ValentinBELYN/icmplib
Mam nadzieję, że ta biblioteka okaże się przydatna.
użycie systemowego polecenia ping do pingowania listy hostów:
import re
from subprocess import Popen, PIPE
from threading import Thread
class Pinger(object):
def __init__(self, hosts):
for host in hosts:
pa = PingAgent(host)
pa.start()
class PingAgent(Thread):
def __init__(self, host):
Thread.__init__(self)
self.host = host
def run(self):
p = Popen('ping -n 1 ' + self.host, stdout=PIPE)
m = re.search('Average = (.*)ms', p.stdout.read())
if m: print 'Round Trip Time: %s ms -' % m.group(1), self.host
else: print 'Error: Invalid Response -', self.host
if __name__ == '__main__':
hosts = [
'www.pylot.org',
'www.goldb.org',
'www.google.com',
'www.yahoo.com',
'www.techcrunch.com',
'www.this_one_wont_work.com'
]
Pinger(hosts)
p = Popen('ping -n 1 ' + self.host, stdout=PIPE)
Powinno być p = Popen(['ping','-n','1','self.host'], stdout=PIPE)
użycie polecenia ping podprocesu, aby zdekodować ping, ponieważ odpowiedź jest binarna:
import subprocess
ping_response = subprocess.Popen(["ping", "-a", "google.com"], stdout=subprocess.PIPE).stdout.read()
result = ping_response.decode('utf-8')
print(result)
możesz spróbować z gniazda, aby uzyskać ip strony i użyć scrapy do wykonania polecenia ping icmp na ip.
import gevent
from gevent import monkey
# monkey.patch_all() should be executed before any library that will
# standard library
monkey.patch_all()
import socket
from scapy.all import IP, ICMP, sr1
def ping_site(fqdn):
ip = socket.gethostbyaddr(fqdn)[-1][0]
print(fqdn, ip, '\n')
icmp = IP(dst=ip)/ICMP()
resp = sr1(icmp, timeout=10)
if resp:
return (fqdn, False)
else:
return (fqdn, True)
sites = ['www.google.com', 'www.baidu.com', 'www.bing.com']
jobs = [gevent.spawn(ping_site, fqdn) for fqdn in sites]
gevent.joinall(jobs)
print([job.value for job in jobs])
Użyj tego, co zostało przetestowane w Pythonie 2.7 i działa dobrze, zwraca czas ping w milisekundach, jeśli się powiedzie, i zwróci False w przypadku niepowodzenia.
import platform,subproccess,re
def Ping(hostname,timeout):
if platform.system() == "Windows":
command="ping "+hostname+" -n 1 -w "+str(timeout*1000)
else:
command="ping -i "+str(timeout)+" -c 1 " + hostname
proccess = subprocess.Popen(command, stdout=subprocess.PIPE)
matches=re.match('.*time=([0-9]+)ms.*', proccess.stdout.read(),re.DOTALL)
if matches:
return matches.group(1)
else:
return False
command
jest to ciąg zawierający wszystkie argumenty zamiast listy, więc wyzwala command not found
pełny ciąg w systemie Linux.