from fastapi import FastAPI, BackgroundTasks
from fastapi.responses import HTMLResponse
import time, collections
import telegram
# application can do a get request to an API endpoint with a name and a timeout
# if that name does not reappear within the timeout,
# this script will send a telegram message to inform of the problem
app = FastAPI()
bot = telegram.Bot(token='2050623789:AAHw9ocGmAwrytlYgQoFWZkj3k1WZxP5X_E')
#timeouts = collections.defaultdict(float)
timeouts = dict()
timer = collections.namedtuple("timer","timestamp active")
def send_message(text):
bot.send_message(text=text,chat_id=2045728184,parse_mode=telegram.ParseMode.MARKDOWN)
def sleeper(name: str, delay : float):
time.sleep(delay)
try:
if (timeouts[name].timestamp < time.time() and timeouts[name].active):
print(f'Error! "{name}" did not respond')
send_message(f'*Error*: "{name}" did not ping in the last {delay} s')
### disable timer to avoid multiple messages:
timeouts[name].active = False
except KeyError:
print(f"Entry {name} requested, but not in dict! Continuing.")
### no entry, must have been deleted by a previous run
@app.get("/")
async def root():
return "go to /heartbeat"
@app.get("/heartbeat")
async def heartbeat(name: str, timeout_s: float, background_tasks: BackgroundTasks):
timestamp = time.time() + timeout_s
entry = timer(timestamp,active=True)
try:
if(not timeouts[name].active):
send_message(f'*Info*: {name} revived.')
except KeyError:
send_message(f'*Info*: {name} registered itself.')
timeouts[name] = entry
background_tasks.add_task(sleeper,name,timeout_s)
return "OK"
@app.get("/status")
async def print_status():
html_content = """
Uptime monitor
Uptime status:
Name
Active
TTL
"""
for name,t in timeouts.items():
html_content += f'
{name}
{"active" if t.active else "disabled"}
{(str(round(t.timestamp - time.time()))+"s") if t.active else ""}