better printing, better timing, unlimited targets
All checks were successful
Python Lint & Test / build (push) Successful in 9s
All checks were successful
Python Lint & Test / build (push) Successful in 9s
This commit is contained in:
parent
74650baf5d
commit
8c4635a93c
59
httpoll.py
59
httpoll.py
@ -3,53 +3,72 @@
|
||||
import argparse
|
||||
import requests
|
||||
import urllib3
|
||||
from time import time, sleep
|
||||
import time
|
||||
from datetime import datetime
|
||||
import os
|
||||
import sys
|
||||
from rich import print
|
||||
|
||||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||||
|
||||
args = dict()
|
||||
proxy = None
|
||||
proxy = {
|
||||
"http": os.environ.get('http_proxy'),
|
||||
"https": os.environ.get('https_proxy')
|
||||
}
|
||||
|
||||
|
||||
def poll():
|
||||
def poll(url):
|
||||
"""Try to HTTP GET the URL."""
|
||||
time_start = time()
|
||||
status, result = "", ""
|
||||
time_start = time.time()
|
||||
try:
|
||||
r = requests.get(url=args.URL, timeout=10.0,
|
||||
r = requests.get(url=url, timeout=50.0,
|
||||
verify=False, proxies=proxy)
|
||||
status = "ok"
|
||||
status = "[bold green] up[/bold green]"
|
||||
result = str(r)
|
||||
except requests.exceptions.RequestException as e:
|
||||
status = "error"
|
||||
status = "[bold red] down[/bold red]"
|
||||
result = str(e)
|
||||
duration = format(time() - time_start, ".3f")
|
||||
print(f"[time: {duration}] [{status}]\t {result}")
|
||||
return {
|
||||
"duration": (time.time() - time_start),
|
||||
"status": status,
|
||||
"result": result
|
||||
}
|
||||
|
||||
|
||||
def run():
|
||||
def run(global_time_start):
|
||||
"""Runs endlessly but at most once per second."""
|
||||
duration = 1
|
||||
while True:
|
||||
if duration >= 1:
|
||||
time_start = time()
|
||||
poll()
|
||||
else:
|
||||
sleep(1 - duration)
|
||||
duration = time() - time_start
|
||||
time_start = time.time()
|
||||
|
||||
results = []
|
||||
for target in args.URL:
|
||||
results.append(poll(target))
|
||||
|
||||
global_time = datetime.now().replace(microsecond=0) - global_time_start
|
||||
print(f"time: {global_time} | ", end="")
|
||||
for index, result in enumerate(results):
|
||||
print(f"pa{index}: {result['status']}\t |", end="")
|
||||
print()
|
||||
|
||||
time_elapsed = time.time() - time_start
|
||||
time_to_sleep = max(1.0 - time_elapsed, 0)
|
||||
time.sleep(time_to_sleep)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument('URL')
|
||||
parser.add_argument('URL', nargs='*')
|
||||
parser.add_argument('--no-proxy', action="store_true")
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.no_proxy:
|
||||
proxy = {'no': 'pass'}
|
||||
|
||||
run()
|
||||
global_time_start = datetime.now().replace(microsecond=0)
|
||||
run(global_time_start)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
exit()
|
||||
sys.exit()
|
||||
|
Loading…
Reference in New Issue
Block a user