Server / Website Uptime Updater for Telegram

updater.py

from datetime import datetime
from time import localtime, strftime, sleep
from urllib.parse import urlparse
import telepot
import requests
import subprocess
import platform
import csv
import os

"""
About
----------------

This script will process a list of urls or addresses to see if it's up and 
running. It will log the results to a CSV file and ping you on Telegram if it 
shows that the site is down.

Setup
----------------

You'll need to install a few deps to get this to work.

  ```
  pip install datetime
  pip install requests
  pip install telepot
  ```

Also, include a list of urls called `urls.txt` and place it in the same folder.
Below is an example text file.

  ```
  google.com
  198.100.148.100
  https://www.speedrun.com
  ```

Script options
----------------

logger_level
  Set logging level for console.
  Levels from less verbose to more verbose:
  0: no logging
  1: error 
  2: warn
  3: info
  4: debug
  
timeout
  Set the timout period, in seconds, before checking URLs again.

telegram_token
  Your bot"s token authorization
  
telegram_chat_id
  Your chat id for your bot to send updates to. Type /id in a chat with a bot 
  to get the id.
"""

logger_level = 4
timeout = 30
telegram_token = "---"
telegram_chat_id = "---"

"""
Do not edit below this!
"""

# Initializes the bot
bot = telepot.Bot(telegram_token)

# Console logging functions (thanks Elixir).
class Logger:    
  def debug(text):
    if logger_level == 4:
      time = strftime("%H:%M:%S", localtime())
      print("\x1b[0;36;40m{} [debug] {}\x1b[0m".format(time, text))
    
  def info(text):
    if logger_level >= 3:
      time = strftime("%H:%M:%S", localtime())
      print("{} [info] {}".format(time, text))
    
  def warn(text):
    if logger_level >= 2:
      time = strftime("%H:%M:%S", localtime())
      print("\x1b[0;33;40m{} [warn] {}\x1b[0m".format(time, text))
    
  def error(text):
    if logger_level >= 1:
      time = strftime("%H:%M:%S", localtime())
      print("\x1b[0;31;40m{} [error] {}\x1b[0m".format(time, text))

# Requests a url and returns a dict with status code and reason
# If this fails, it tries pinging the server to see if it's alive
def req(url):
  uri = urlparse(url)
  
  hostname = ""
  if uri.scheme == "":
    hostname = url
    url = "http://{}".format(url)
      
  try:      
    response = requests.get(url, timeout = 5)
    
    log = "{}: {} {}".format(url, response.status_code, response.reason)
    if response.reason == "OK":
      Logger.info(log)
    else:
      Logger.warn(log)
    
    return dict(
      timestamp = datetime.utcnow().isoformat(), 
      url = url, 
      status = response.status_code,
      reason = response.reason
    )
  except:
    reason = ping(hostname)
    if reason == "OK":
      Logger.info("{}: {}".format(hostname, reason))
    else:
      Logger.error("{}: {}".format(hostname, reason))
      
    return dict(
      timestamp = datetime.utcnow().isoformat(), 
      url = hostname, 
      status = -1,
      reason = reason
    )

# Pings an IP or server to see if it's alive, returns status
def ping(hostname):
  command = "ping " + hostname
  if platform.system() == "Linux":
    command = ["/bin/ping", "-c1", hostname]
    
  p = subprocess.Popen(command, stdout = subprocess.PIPE, stderr = subprocess.PIPE)

  status = "OK"
  if not platform.system() == "Linux":
    for line in p.stdout:
      output = line.rstrip().decode("UTF-8")
   
      if (output.endswith("unreachable.")) :
        status = "Unreachable"
        break
      elif (output.startswith("Ping request could not find host")):
        status = "Host not found"
        break
      elif (output.startswith("Request timed out.")) :
        status = "Timed out"
        break
  else:
    output = p.stdout.read().decode("UTF-8")

    if not output.endswith(" ms\n"):
      if output == "":
        status = "Name or service not known"
      else:
        status = output.split(": ")[-1]
      
  return status

# Processes all urls in urls.txt, writes to a CSV and pings the telegram bot 
# if something is down
def url_check_loop():
  resume = False
  open_opts = "w"
  if os.path.exists("log.csv"):
    resume = True
    open_opts = "a"
  
  with open("log.csv", open_opts, encoding = "UTF-8", newline = "") as log:
    w = csv.DictWriter(log, ["timestamp", "url", "status", "reason"])
    if not resume:
      w.writeheader()
      
    with open("urls.txt") as f:
      urls = [x.strip() for x in f.readlines()]
      
      for url in urls:          
        result = req(url)
        w.writerow(result)
        
        if not result["reason"] == "OK":
          bot.sendMessage(telegram_chat_id, "{} is currently down! ({} {})".format(result["url"], result["status"], result["reason"]))

# Bot loop for ID getting
from telepot.loop import MessageLoop
def handle(msg):
  chat_id = msg["chat"]["id"]
  command = msg["text"]

  if command == "/id":
      bot.sendMessage(chat_id, chat_id)

# Starts the bot listener
MessageLoop(bot, handle).run_as_thread()

# Checks urls and sleeps for the specified timeout
while 1:
  url_check_loop()
  sleep(timeout)