用 Gemini CLI 自動巡檢伺服器叢集：Python + SSH + Telegram 推送實戰

npm install -g @google/gemini-cli
gemini auth   # 用 Google 賬號登入

echo "hello" | gemini

ssh-keygen -t ed25519 -f ~/.ssh/audit_key

ssh-copy-id -i ~/.ssh/audit_key.pub user@server-ip

#!/usr/bin/env python3
"""
每日伺服器叢集巡檢指令碼
- SSH 採集各伺服器指標
- Gemini CLI 分析
- Telegram 推送報告
"""

import subprocess
import json
import requests
from datetime import datetime

# === 配置區 ===
SERVERS = [
    {"name": "aws-hk",  "host": "你的HK伺服器IP", "user": "ubuntu"},
    {"name": "aws-us",  "host": "你的US伺服器IP", "user": "ubuntu"},
    {"name": "home-1",  "host": "192.168.1.10",   "user": "as"},
]

SSH_KEY = "/root/.ssh/audit_key"   # 或你的 key 路徑
TG_TOKEN = "你的TelegramBotToken"
TG_CHAT_ID = "你的ChatID"
# ==============


def ssh_run(host: str, user: str, cmd: str) -> str:
    """在遠端主機執行命令，返回輸出"""
    result = subprocess.run(
        ["ssh", "-i", SSH_KEY, "-o", "StrictHostKeyChecking=no",
         "-o", "ConnectTimeout=10", f"{user}@{host}", cmd],
        capture_output=True, text=True, timeout=30
    )
    return result.stdout.strip() if result.returncode == 0 else f"[ERROR] {result.stderr.strip()}"


def collect_metrics(server: dict) -> dict:
    """採集單臺伺服器指標"""
    name = server["name"]
    host = server["host"]
    user = server["user"]

    print(f"  採集 {name} ({host})...")

    metrics = {
        "name": name,
        "host": host,
        "timestamp": datetime.now().isoformat(),
    }

    # CPU 使用率（1分鐘均值）
    metrics["cpu_load"] = ssh_run(host, user,
        "cat /proc/loadavg | awk '{print $1, $2, $3}'")

    # 記憶體使用
    metrics["memory"] = ssh_run(host, user,
        "free -h | grep Mem | awk '{print \"total:\", $2, \"used:\", $3, \"free:\", $4}'")

    # 磁碟使用（根分割槽）
    metrics["disk"] = ssh_run(host, user,
        "df -h / | tail -1 | awk '{print \"total:\", $2, \"used:\", $3, \"avail:\", $4, \"use%:\", $5}'")

    # 系統執行時間
    metrics["uptime"] = ssh_run(host, user, "uptime -p")

    # 關鍵服務狀態（根據你的服務調整）
    services = ["nginx", "xray", "hysteria-server"]
    service_status = {}
    for svc in services:
        status = ssh_run(host, user,
            f"systemctl is-active {svc} 2>/dev/null || echo 'not-found'")
        service_status[svc] = status
    metrics["services"] = service_status

    # 近期系統錯誤（過去1小時）
    metrics["recent_errors"] = ssh_run(host, user,
        "journalctl -p err --since '1 hour ago' --no-pager -q 2>/dev/null | tail -20")

    # 網路連線數
    metrics["connections"] = ssh_run(host, user,
        "ss -s | grep 'TCP:' | head -3")

    return metrics


def format_report(all_metrics: list) -> str:
    """將採集資料格式化為文字報告"""
    lines = []
    lines.append(f"=== 叢集巡檢報告 {datetime.now().strftime('%Y-%m-%d %H:%M')} ===\n")

    for m in all_metrics:
        lines.append(f"[{m['name']}] {m['host']}")
        lines.append(f"  負載: {m.get('cpu_load', 'N/A')}")
        lines.append(f"  記憶體: {m.get('memory', 'N/A')}")
        lines.append(f"  磁碟: {m.get('disk', 'N/A')}")
        lines.append(f"  執行: {m.get('uptime', 'N/A')}")

        services = m.get('services', {})
        for svc, status in services.items():
            icon = "✓" if status == "active" else "✗"
            lines.append(f"  {icon} {svc}: {status}")

        errors = m.get('recent_errors', '')
        if errors and '[ERROR]' not in errors and len(errors) > 5:
            lines.append(f"  近期錯誤:")
            for line in errors.split('\n')[:5]:
                lines.append(f"    {line}")

        lines.append("")

    return '\n'.join(lines)


def gemini_analyze(report: str) -> str:
    """呼叫 Gemini CLI 分析報告"""
    prompt = f"""以下是伺服器叢集的巡檢資料，請：
1. 識別任何異常或需要關注的問題
2. 對服務狀態給出簡短評估
3. 如果磁碟使用率 > 80% 或服務 inactive，明確標記為警告
4. 給出簡短的建議（如果有問題）
5. 如果一切正常，說明"叢集狀態正常"

回覆請用中文，控制在200字以內。

{report}"""

    result = subprocess.run(
        ["gemini", "--model", "gemini-2.5-flash", "-p", prompt],
        capture_output=True, text=True, timeout=60
    )

    if result.returncode == 0:
        return result.stdout.strip()
    else:
        return f"[Gemini 分析失敗] {result.stderr[:200]}"


def send_telegram(message: str):
    """傳送 Telegram 訊息"""
    url = f"https://api.telegram.org/bot{TG_TOKEN}/sendMessage"

    # Telegram 訊息長度限制 4096，超出則截斷
    if len(message) > 4000:
        message = message[:4000] + "\n...[截斷]"

    resp = requests.post(url, json={
        "chat_id": TG_CHAT_ID,
        "text": message,
        "parse_mode": "HTML"
    }, timeout=15)

    if resp.status_code == 200:
        print("  Telegram 訊息傳送成功")
    else:
        print(f"  Telegram 傳送失敗: {resp.text}")


def main():
    print(f"開始巡檢 {len(SERVERS)} 臺伺服器...")

    # 採集所有伺服器指標
    all_metrics = []
    for server in SERVERS:
        try:
            metrics = collect_metrics(server)
            all_metrics.append(metrics)
        except Exception as e:
            print(f"  採集 {server['name']} 失敗: {e}")
            all_metrics.append({
                "name": server["name"],
                "host": server["host"],
                "error": str(e)
            })

    # 生成文字報告
    report = format_report(all_metrics)
    print("\n--- 原始報告 ---")
    print(report)

    # Gemini 分析
    print("\n呼叫 Gemini 分析...")
    analysis = gemini_analyze(report)
    print(f"分析結果: {analysis}")

    # 組合 Telegram 訊息
    tg_message = f"<b>每日叢集巡檢</b>\n\n{analysis}\n\n<pre>{report[:2000]}</pre>"
    send_telegram(tg_message)

    print("\n巡檢完成")


if __name__ == "__main__":
    main()

crontab -e

# 檢查 Docker 容器狀態
metrics["docker"] = ssh_run(host, user,
    "docker ps --format '{{.Names}}: {{.Status}}' 2>/dev/null | head -10")

# 檢查特定埠是否監聽
metrics["ports"] = ssh_run(host, user,
    "ss -tlnp | grep -E ':(443|80|8080|9090)' | awk '{print $4, $6}'")

# 最近的 nginx 錯誤
metrics["nginx_errors"] = ssh_run(host, user,
    "tail -20 /var/log/nginx/error.log 2>/dev/null | grep -v '^$'")

# 磁碟 inode 使用率（經常被忽略）
metrics["inode"] = ssh_run(host, user,
    "df -i / | tail -1 | awk '{print \"inode use%:\", $5}'")

prompt = f"""今天的報告：\n{today_report}\n\n昨天的報告：\n{yesterday_report}\n\n請對比分析變化..."""

prompt = f"""...特別關注磁碟使用率是否接近滿，以及 xray/hysteria 服務是否線上..."""

prompt = f"""...請以 JSON 格式返回，欄位包括：status（ok/warning/critical）、issues（問題列表）、suggestions（建議列表）..."""