AWS Lambda function to forward UptimeRobot up / down events to a Telegram group chat

lambda_function.py

# Needed previous configuration:
# 
# - In UptimeRobot, configure a new **Webhook** Alert Contact with this POST payload:
#   ```
#   {
#     "monitorID": "*monitorID*",
#     "monitorURL": "*monitorURL*",
#     "monitorFriendlyName": "*monitorFriendlyName*",
#     "alertType": "*alertType*",
#     "alertTypeFriendlyName": "*alertTypeFriendlyName*",
#     "alertDetails": "*alertDetails*",
#     "alertDuration": "*alertDuration*",
#     "friendlyMessage": "Monitor is *alertTypeFriendlyName*: *monitorFriendlyName* (*monitorURL*) *alertDetails*. Duration: *alertDuration*"
#   }
#   ```
#  Make sure "Send as JSON (application/json)" is ACTIVE.
# 
# - In AWS API Gateway, create a new REST API. Then, under the / resource, create a method of type POST. Make its integration type "Lambda proxy" and select the Lambda. Then, deploy the API under a `prod` stage and use the provided URL as "URL to notify" in UptimeRobot. It should look something like: `https://abcdef123456.execute-api.us-east-1.amazonaws.com/prod`
# 
# - In the Lambda configuration, under Environment variables, define `TELEGRAM_BOT_TOKEN` and `TELEGRAM_CHAT_ID`


import os
import requests

# Set your Telegram Bot token and chat ID here (or use environment variables)
TELEGRAM_BOT_TOKEN = os.environ.get('TELEGRAM_BOT_TOKEN')
TELEGRAM_CHAT_ID = os.environ.get('TELEGRAM_CHAT_ID')

def send_telegram_message(message):
    """Send a message to the Telegram group."""
    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    data = {
        "chat_id": TELEGRAM_CHAT_ID,
        "text": message,
        "parse_mode": "Markdown",
        "disable_web_page_preview": True
    }
    response = requests.post(url, data=data)
    return response.json()

def seconds_to_human(seconds):
    """Convert seconds to a human-readable time format."""
    days, seconds = divmod(seconds, 86400)
    hours, seconds = divmod(seconds, 3600)
    minutes, seconds = divmod(seconds, 60)
    human_readable = f"{days}d {hours}h {minutes}m {seconds}s"
    return human_readable.strip()

def compose_message(event):
    """Compose a custom message based on the alert type."""
    alert_type = event.get("alertTypeFriendlyName")
    monitor_name = event.get("monitorFriendlyName")
    monitor_url = event.get("monitorURL")
    alert_details = event.get("alertDetails", "No details provided.")
    
    # Use emojis and custom messages for different alert types
    if alert_type == "Down":
        emoji = "🔴‼️😨"
        return f"{emoji} Heads up: {monitor_name} is DOWN\nURL: {monitor_url}\n\nDetails: {alert_details}."
    elif alert_type == "Up":
        emoji = "🟢✅😮‍💨"
        duration_seconds = int(event.get("alertDuration", 0))
        duration_human_readable = seconds_to_human(duration_seconds)
        return f"{emoji} Solved: {monitor_name} is BACK UP\nURL: {monitor_url}\n\nDowntime was {duration_human_readable}."
    else:
        return f"Monitor update for {monitor_name}: {alert_type}\nURL: {monitor_url}.\n\nDetails: {alert_details}."

def lambda_handler(event, context):
    """Process UptimeRobot notification and send message to Telegram."""
    try:
        # Compose the message based on the event
        message = compose_message(event)
        
        # Send the message to Telegram
        response = send_telegram_message(message)
        return {
            "statusCode": 200,
            "body": json.dumps(response)
        }
    except Exception as e:
        return {
            "statusCode": 500,
            "body": json.dumps({"error": str(e)})
        }