Notion-Uptime-Monitorig/services_uptime_monitor.py

#!/usr/bin/env python3
"""
Uptime Monitor for smallmountains.de
Checks service availability and updates the Notion dashboard.

Run via cron every 5 minutes:
    */5 * * * * /usr/local/bin/python3 /path/to/services_uptime_monitor.py >> /path/to/monitor.log 2>&1

Services are configured entirely in the Notion database — no code changes needed
to add, remove, or reconfigure a service.
"""
from __future__ import annotations

import sys
import sqlite3
import socket
import subprocess
import time
from datetime import datetime, timezone, timedelta
from pathlib import Path

import requests
from notion_client import Client
from notion_client.errors import APIResponseError

# ── Configuration ─────────────────────────────────────────────────────────────
NOTION_TOKEN           = "secret_b7PiPL2FqC9QEikqkAEWOht7LmzPMIJMWTzUPWwbw4H"
NOTION_DATA_SOURCE_ID  = "22174dd2-e6fc-4dc9-ac86-a5d614c995bd"  # Services data source
UPTIME_PAGE_ID         = "38210a5f51bd807bae1edb699d9591e8"       # Uptime Tracker page

DB_PATH      = Path(__file__).parent / "uptime_history.db"
HTTP_TIMEOUT = 10   # seconds
UDP_TIMEOUT  = 5    # seconds


# ── Fetch services from Notion ─────────────────────────────────────────────────
def fetch_services(notion: Client) -> list[dict]:
    """
    Read the Services database and return a list of service dicts.

    Each HTTP service dict:  {"name", "notion_page_id", "type": "http", "url"}
    Each UDP service dict:   {"name", "notion_page_id", "type": "udp", "host", "port"}

    Rows missing a name or URL are skipped with a warning.
    Check Type defaults to HTTP when the field is left blank.
    """
    response = notion.data_sources.query(NOTION_DATA_SOURCE_ID)
    services = []

    for page in response["results"]:
        props = page["properties"]

        # Service name
        title_arr = props.get("Service", {}).get("title", [])
        name = title_arr[0]["plain_text"].strip() if title_arr else ""
        if not name:
            continue

        # URL / endpoint
        url = (props.get("URL") or {}).get("url") or ""
        if not url:
            print(f"  Skipping '{name}': no URL configured in Notion")
            continue

        # Check Type (default to HTTP when blank)
        select = ((props.get("Check Type") or {}).get("select")) or {}
        check_type = select.get("name", "HTTP").upper()

        service: dict = {
            "name": name,
            "notion_page_id": page["id"],
            "type": check_type.lower(),
        }

        if check_type == "UDP":
            # URL field stores "host:port"
            host, sep, port_str = url.rpartition(":")
            service["host"] = host if sep else url
            service["port"] = int(port_str) if port_str.isdigit() else 34197
        else:
            service["url"] = url

        services.append(service)

    if not services:
        raise RuntimeError(
            "No services found in Notion database. "
            "Check that NOTION_DATABASE_ID is correct and PyBot has access."
        )

    return services


# ── SQLite ─────────────────────────────────────────────────────────────────────
def init_db(db_path: Path) -> sqlite3.Connection:
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS checks (
            id               INTEGER PRIMARY KEY AUTOINCREMENT,
            service_name     TEXT    NOT NULL,
            checked_at       TEXT    NOT NULL,
            is_online        INTEGER NOT NULL,
            response_time_ms REAL
        )
    """)
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_service_time ON checks (service_name, checked_at)"
    )
    conn.commit()
    return conn


def prune_old_records(conn: sqlite3.Connection):
    cutoff = (datetime.now(timezone.utc) - timedelta(days=35)).isoformat()
    conn.execute("DELETE FROM checks WHERE checked_at < ?", (cutoff,))
    conn.commit()


def compute_uptime(conn: sqlite3.Connection, service_name: str, hours: int) -> float | None:
    """Return fraction 0.0–1.0 for Notion's percent format, or None if no data."""
    cutoff = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat()
    row = conn.execute(
        "SELECT COUNT(*), COALESCE(SUM(is_online), 0) FROM checks "
        "WHERE service_name = ? AND checked_at >= ?",
        (service_name, cutoff),
    ).fetchone()
    total, online = row
    if total == 0:
        return None
    return online / total


# ── Service Checks ─────────────────────────────────────────────────────────────
def check_http(url: str) -> tuple[bool, float | None]:
    try:
        start = time.monotonic()
        resp = requests.get(url, timeout=HTTP_TIMEOUT, allow_redirects=True)
        elapsed_ms = (time.monotonic() - start) * 1000
        return resp.status_code < 500, round(elapsed_ms, 1)
    except requests.RequestException:
        return False, None


def _ping(host: str) -> bool:
    # -W timeout unit differs: milliseconds on macOS, seconds on Linux
    w_arg = "3000" if sys.platform == "darwin" else "3"
    try:
        result = subprocess.run(
            ["ping", "-c", "1", "-W", w_arg, host],
            capture_output=True,
            timeout=6,
        )
        return result.returncode == 0
    except Exception:
        return False


def check_udp(host: str, port: int) -> tuple[bool, None]:
    """
    Send a probe UDP packet.
    - Response received      → online
    - ConnectionRefusedError → offline (ICMP port-unreachable)
    - Timeout                → fall back to ICMP ping
    """
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.settimeout(UDP_TIMEOUT)
        sock.connect((host, port))
        sock.send(b"\x00\x00\x00\x00")
        try:
            sock.recv(1024)
            return True, None
        except socket.timeout:
            return _ping(host), None
        except ConnectionRefusedError:
            return False, None
    except (socket.gaierror, OSError):
        return False, None
    finally:
        try:
            sock.close()
        except Exception:
            pass


def check_service(service: dict) -> tuple[bool, float | None]:
    if service["type"] == "http":
        return check_http(service["url"])
    if service["type"] == "udp":
        return check_udp(service["host"], service["port"])
    return False, None


# ── Notion ─────────────────────────────────────────────────────────────────────
def update_notion_service(
    notion: Client,
    service: dict,
    is_online: bool,
    response_ms: float | None,
    uptime_24h: float | None,
    uptime_7d: float | None,
    uptime_30d: float | None,
):
    now_iso = datetime.now(timezone.utc).isoformat()
    notion.pages.update(
        page_id=service["notion_page_id"],
        properties={
            "Status":             {"select": {"name": "Online" if is_online else "Offline"}},
            "Last Checked":       {"date": {"start": now_iso}},
            "Response Time (ms)": {"number": response_ms},
            "Uptime 24h %":       {"number": uptime_24h},
            "Uptime 7d %":        {"number": uptime_7d},
            "Uptime 30d %":       {"number": uptime_30d},
        },
    )


def update_last_updated_block(notion: Client, page_id: str, timestamp_str: str):
    """Find the callout/paragraph containing 'Last Updated' and refresh its text."""
    try:
        result = notion.blocks.children.list(block_id=page_id)
        for block in result.get("results", []):
            btype = block.get("type")
            if btype not in ("callout", "paragraph", "quote"):
                continue
            rich_text = block.get(btype, {}).get("rich_text", [])
            plain = "".join(rt.get("plain_text", "") for rt in rich_text)
            if "Last Updated" not in plain:
                continue
            notion.blocks.update(
                block_id=block["id"],
                **{
                    btype: {
                        "rich_text": [
                            {
                                "type": "text",
                                "text": {"content": f"🔄 Last Updated: {timestamp_str}"},
                                "annotations": {"bold": False},
                            }
                        ]
                    }
                },
            )
            return
    except APIResponseError as e:
        print(f"  Warning: could not update Last Updated block: {e}")


# ── Main ───────────────────────────────────────────────────────────────────────
def main():
    notion = Client(auth=NOTION_TOKEN)

    services = fetch_services(notion)
    print(f"Loaded {len(services)} service(s) from Notion.")

    conn = init_db(DB_PATH)
    prune_old_records(conn)

    now_utc     = datetime.now(timezone.utc)
    now_iso     = now_utc.isoformat()
    now_display = now_utc.strftime("%Y-%m-%d %H:%M UTC")

    print(f"[{now_display}] Running uptime checks...")

    for service in services:
        is_online, response_ms = check_service(service)
        status_str = "ONLINE " if is_online else "OFFLINE"
        rt_str     = f"{response_ms:.0f}ms" if response_ms is not None else "—"
        print(f"  {service['name']:<20} {status_str}  {rt_str}")

        conn.execute(
            "INSERT INTO checks (service_name, checked_at, is_online, response_time_ms) "
            "VALUES (?, ?, ?, ?)",
            (service["name"], now_iso, int(is_online), response_ms),
        )
        conn.commit()

        uptime_24h = compute_uptime(conn, service["name"], 24)
        uptime_7d  = compute_uptime(conn, service["name"], 7 * 24)
        uptime_30d = compute_uptime(conn, service["name"], 30 * 24)

        try:
            update_notion_service(
                notion, service, is_online, response_ms,
                uptime_24h, uptime_7d, uptime_30d,
            )
        except APIResponseError as e:
            print(f"  Warning: Notion update failed for {service['name']}: {e}")

    update_last_updated_block(notion, UPTIME_PAGE_ID, now_display)

    conn.close()
    print("Done.")


if __name__ == "__main__":
    main()