用 Telegram Bot 把伺服器狀態推到手機：從建立 Bot 到監控通報

#!/usr/bin/env python3
# /usr/local/bin/server-monitor.py

import os
import json
import shutil
import urllib.request
import urllib.parse
import subprocess

TOKEN = os.environ.get("TG_TOKEN", "")
CHAT_ID = os.environ.get("TG_CHAT_ID", "")

def send_message(text: str, level: str = "info") -> None:
    emoji = {"error": "🔴", "warning": "🟡", "success": "🟢"}.get(level, "🔵")
    full_text = f"{emoji} {text}"
    
    data = urllib.parse.urlencode({
        "chat_id": CHAT_ID,
        "text": full_text,
        "parse_mode": "Markdown",
    }).encode()
    
    req = urllib.request.Request(
        f"https://api.telegram.org/bot{TOKEN}/sendMessage",
        data=data,
        method="POST"
    )
    try:
        urllib.request.urlopen(req, timeout=10)
    except Exception as e:
        print(f"[ERROR] 傳送 Telegram 通報失敗：{e}")


def check_disk(path: str = "/", threshold: int = 85) -> None:
    total, used, free = shutil.disk_usage(path)
    usage_pct = int(used / total * 100)
    
    if usage_pct >= threshold:
        send_message(
            f"*磁碟使用率警告*\n"
            f"路徑：`{path}`\n"
            f"使用率：{usage_pct}%（閾值：{threshold}%）\n"
            f"剩餘：{free // (1024**3)} GB",
            level="warning" if usage_pct < 95 else "error"
        )


def check_service(service_name: str) -> None:
    result = subprocess.run(
        ["systemctl", "is-active", service_name],
        capture_output=True, text=True
    )
    if result.stdout.strip() != "active":
        send_message(
            f"*服務異常*\n"
            f"服務：`{service_name}`\n"
            f"狀態：`{result.stdout.strip()}`",
            level="error"
        )


def check_memory(threshold: int = 90) -> None:
    with open("/proc/meminfo") as f:
        lines = {k.strip(): v.strip() for k, v in
                 (line.split(":", 1) for line in f)}
    
    total = int(lines["MemTotal"].split()[0])
    available = int(lines["MemAvailable"].split()[0])
    usage_pct = int((1 - available / total) * 100)
    
    if usage_pct >= threshold:
        send_message(
            f"*記憶體使用率警告*\n"
            f"使用率：{usage_pct}%（閾值：{threshold}%）\n"
            f"可用：{available // 1024} MB",
            level="warning"
        )


if __name__ == "__main__":
    check_disk("/", threshold=85)
    check_memory(threshold=90)
    check_service("nginx")
    check_service("postgresql")