用 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（建议列表）..."""