Python开发Telegram Bot实现服务器监控告警

pip install python-telegram-bot
pip install psutil  # 系统监控
pip install aiohttp  # 异步http
pip install apscheduler  # 定时任务

telegram_bot/
├── bot.py           # 主程序
├── config.py        # 配置
├── monitors/        # 监控模块
│   ├── __init__.py
│   ├── cpu.py
│   ├── memory.py
│   ├── disk.py
│   └── network.py
├── handlers/        # 命令处理
│   ├── __init__.py
│   └── commands.py
└── requirements.txt

# bot.py
from telegram import update
from telegram.ext import application, commandhandler, contexttypes

# 配置
token = "你的bot token"

# 命令处理器
async def start(update: update, context: contexttypes.default_type):
    await update.message.reply_text("你好！我是服务器监控机器人。")

async def help_command(update: update, context: contexttypes.default_type):
    help_text = """
可用命令：
/start - 开始
/help - 帮助
/status - 服务器状态
/cpu - cpu使用率
/memory - 内存使用
/disk - 磁盘使用
    """
    await update.message.reply_text(help_text)

def main():
    # 创建应用
    app = application.builder().token(token).build()
    
    # 注册命令
    app.add_handler(commandhandler("start", start))
    app.add_handler(commandhandler("help", help_command))
    
    # 启动
    print("bot启动中...")
    app.run_polling()

if __name__ == "__main__":
    main()

python bot.py

# 在telegram中：
# 1. 搜索你的bot
# 2. 发送 /start
# 3. 发送 /help

# monitors/cpu.py
import psutil

def get_cpu_info():
    """获取cpu信息"""
    cpu_percent = psutil.cpu_percent(interval=1)
    cpu_count = psutil.cpu_count()
    cpu_freq = psutil.cpu_freq()
    
    # 每核使用率
    per_cpu = psutil.cpu_percent(interval=1, percpu=true)
    
    return {
        "percent": cpu_percent,
        "count": cpu_count,
        "freq": cpu_freq.current if cpu_freq else 0,
        "per_cpu": per_cpu
    }

def format_cpu_info():
    """格式化cpu信息"""
    info = get_cpu_info()
    
    text = f"""
🖥️ cpu状态
━━━━━━━━━━━━━━━
总使用率: {info['percent']}%
核心数量: {info['count']}
当前频率: {info['freq']:.0f} mhz

各核心使用率:
"""
    for i, p in enumerate(info['per_cpu']):
        bar = "█" * int(p / 10) + "░" * (10 - int(p / 10))
        text += f"  核心{i}: [{bar}] {p}%\n"
    
    return text

# monitors/memory.py
import psutil

def get_memory_info():
    """获取内存信息"""
    mem = psutil.virtual_memory()
    swap = psutil.swap_memory()
    
    return {
        "total": mem.total / (1024**3),  # gb
        "used": mem.used / (1024**3),
        "available": mem.available / (1024**3),
        "percent": mem.percent,
        "swap_total": swap.total / (1024**3),
        "swap_used": swap.used / (1024**3),
        "swap_percent": swap.percent
    }

def format_memory_info():
    """格式化内存信息"""
    info = get_memory_info()
    
    bar = "█" * int(info['percent'] / 10) + "░" * (10 - int(info['percent'] / 10))
    
    text = f"""
💾 内存状态
━━━━━━━━━━━━━━━
使用率: [{bar}] {info['percent']}%

物理内存:
  总量: {info['total']:.1f} gb
  已用: {info['used']:.1f} gb
  可用: {info['available']:.1f} gb

交换分区:
  总量: {info['swap_total']:.1f} gb
  已用: {info['swap_used']:.1f} gb ({info['swap_percent']}%)
"""
    return text

# monitors/disk.py
import psutil

def get_disk_info():
    """获取磁盘信息"""
    partitions = []
    for part in psutil.disk_partitions():
        try:
            usage = psutil.disk_usage(part.mountpoint)
            partitions.append({
                "device": part.device,
                "mountpoint": part.mountpoint,
                "total": usage.total / (1024**3),
                "used": usage.used / (1024**3),
                "free": usage.free / (1024**3),
                "percent": usage.percent
            })
        except:
            continue
    return partitions

def format_disk_info():
    """格式化磁盘信息"""
    partitions = get_disk_info()
    
    text = "💿 磁盘状态\n━━━━━━━━━━━━━━━\n"
    
    for p in partitions:
        bar = "█" * int(p['percent'] / 10) + "░" * (10 - int(p['percent'] / 10))
        text += f"""
{p['mountpoint']}
  [{bar}] {p['percent']}%
  已用: {p['used']:.1f} gb / {p['total']:.1f} gb
  剩余: {p['free']:.1f} gb
"""
    return text

# monitors/network.py
import psutil
import socket

def get_network_info():
    """获取网络信息"""
    # 网络io
    net_io = psutil.net_io_counters()
    
    # ip地址
    hostname = socket.gethostname()
    try:
        ip = socket.gethostbyname(hostname)
    except:
        ip = "未知"
    
    # 网络连接数
    connections = len(psutil.net_connections())
    
    return {
        "hostname": hostname,
        "ip": ip,
        "bytes_sent": net_io.bytes_sent / (1024**2),  # mb
        "bytes_recv": net_io.bytes_recv / (1024**2),
        "packets_sent": net_io.packets_sent,
        "packets_recv": net_io.packets_recv,
        "connections": connections
    }

def format_network_info():
    """格式化网络信息"""
    info = get_network_info()
    
    text = f"""
🌐 网络状态
━━━━━━━━━━━━━━━
主机名: {info['hostname']}
ip地址: {info['ip']}
连接数: {info['connections']}

流量统计:
  发送: {info['bytes_sent']:.1f} mb
  接收: {info['bytes_recv']:.1f} mb
  发送包: {info['packets_sent']}
  接收包: {info['packets_recv']}
"""
    return text

# bot.py
from telegram import update, inlinekeyboardbutton, inlinekeyboardmarkup
from telegram.ext import (
    application, commandhandler, callbackqueryhandler, contexttypes
)
from apscheduler.schedulers.asyncio import asyncioscheduler
import psutil
from datetime import datetime

# 配置
token = "你的bot token"
admin_chat_id = 123456789  # 你的chat id

# 告警阈值
thresholds = {
    "cpu": 80,
    "memory": 85,
    "disk": 90
}

# 监控函数（简化版，集成上面的模块）
def get_system_status():
    cpu = psutil.cpu_percent(interval=1)
    mem = psutil.virtual_memory()
    disk = psutil.disk_usage('/')
    
    return {
        "cpu": cpu,
        "memory": mem.percent,
        "disk": disk.percent,
        "uptime": datetime.now() - datetime.fromtimestamp(psutil.boot_time())
    }

def format_status():
    s = get_system_status()
    
    # 状态图标
    cpu_icon = "🔴" if s['cpu'] > thresholds['cpu'] else "🟢"
    mem_icon = "🔴" if s['memory'] > thresholds['memory'] else "🟢"
    disk_icon = "🔴" if s['disk'] > thresholds['disk'] else "🟢"
    
    text = f"""
📊 服务器状态概览
━━━━━━━━━━━━━━━━━━
{cpu_icon} cpu: {s['cpu']}%
{mem_icon} 内存: {s['memory']}%
{disk_icon} 磁盘: {s['disk']}%

⏱️ 运行时间: {str(s['uptime']).split('.')[0]}
🕐 更新时间: {datetime.now().strftime('%y-%m-%d %h:%m:%s')}
"""
    return text

# 命令处理器
async def start(update: update, context: contexttypes.default_type):
    keyboard = [
        [
            inlinekeyboardbutton("📊 状态", callback_data="status"),
            inlinekeyboardbutton("🖥️ cpu", callback_data="cpu"),
        ],
        [
            inlinekeyboardbutton("💾 内存", callback_data="memory"),
            inlinekeyboardbutton("💿 磁盘", callback_data="disk"),
        ],
        [
            inlinekeyboardbutton("🌐 网络", callback_data="network"),
            inlinekeyboardbutton("🔄 刷新", callback_data="refresh"),
        ]
    ]
    reply_markup = inlinekeyboardmarkup(keyboard)
    
    await update.message.reply_text(
        "🤖 服务器监控机器人\n选择要查看的信息：",
        reply_markup=reply_markup
    )

async def status(update: update, context: contexttypes.default_type):
    await update.message.reply_text(format_status())

async def button_callback(update: update, context: contexttypes.default_type):
    query = update.callback_query
    await query.answer()
    
    if query.data == "status" or query.data == "refresh":
        text = format_status()
    elif query.data == "cpu":
        text = format_cpu_info()
    elif query.data == "memory":
        text = format_memory_info()
    elif query.data == "disk":
        text = format_disk_info()
    elif query.data == "network":
        text = format_network_info()
    else:
        text = "未知命令"
    
    await query.edit_message_text(text=text)

# 告警检查
async def check_alerts(context: contexttypes.default_type):
    s = get_system_status()
    alerts = []
    
    if s['cpu'] > thresholds['cpu']:
        alerts.append(f"🔴 cpu使用率过高: {s['cpu']}%")
    if s['memory'] > thresholds['memory']:
        alerts.append(f"🔴 内存使用率过高: {s['memory']}%")
    if s['disk'] > thresholds['disk']:
        alerts.append(f"🔴 磁盘使用率过高: {s['disk']}%")
    
    if alerts:
        alert_text = "⚠️ 服务器告警\n━━━━━━━━━━━━━━━\n" + "\n".join(alerts)
        alert_text += f"\n\n⏰ {datetime.now().strftime('%y-%m-%d %h:%m:%s')}"
        await context.bot.send_message(chat_id=admin_chat_id, text=alert_text)

# 定时报告
async def daily_report(context: contexttypes.default_type):
    text = "📈 每日服务器报告\n" + format_status()
    await context.bot.send_message(chat_id=admin_chat_id, text=text)

def main():
    # 创建应用
    app = application.builder().token(token).build()
    
    # 注册命令
    app.add_handler(commandhandler("start", start))
    app.add_handler(commandhandler("status", status))
    app.add_handler(callbackqueryhandler(button_callback))
    
    # 定时任务
    scheduler = asyncioscheduler()
    # 每分钟检查告警
    scheduler.add_job(
        check_alerts, 'interval', minutes=1,
        args=[app]
    )
    # 每天9点发送报告
    scheduler.add_job(
        daily_report, 'cron', hour=9,
        args=[app]
    )
    scheduler.start()
    
    # 启动
    print("bot启动中...")
    app.run_polling()

if __name__ == "__main__":
    main()

import subprocess

async def exec_command(update: update, context: contexttypes.default_type):
    """执行系统命令（危险！仅限管理员）"""
    # 权限检查
    if update.effective_user.id != admin_chat_id:
        await update.message.reply_text("❌ 无权限")
        return
    
    if not context.args:
        await update.message.reply_text("用法: /exec <命令>")
        return
    
    cmd = " ".join(context.args)
    
    try:
        result = subprocess.run(
            cmd, shell=true, capture_output=true,
            text=true, timeout=30
        )
        output = result.stdout or result.stderr or "（无输出）"
        
        # 限制输出长度
        if len(output) > 4000:
            output = output[:4000] + "\n...(输出过长已截断)"
        
        await update.message.reply_text(f"```\n{output}\n```", parse_mode='markdown')
    except subprocess.timeoutexpired:
        await update.message.reply_text("❌ 命令执行超时")
    except exception as e:
        await update.message.reply_text(f"❌ 执行失败: {e}")

async def processes(update: update, context: contexttypes.default_type):
    """查看进程列表"""
    procs = []
    for p in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):
        try:
            procs.append(p.info)
        except:
            pass
    
    # 按cpu排序
    procs.sort(key=lambda x: x['cpu_percent'] or 0, reverse=true)
    
    text = "📋 进程列表 (top 10 cpu)\n━━━━━━━━━━━━━━━━━━\n"
    for p in procs[:10]:
        text += f"{p['pid']:>6} | {p['cpu_percent']:>5.1f}% | {p['name'][:20]}\n"
    
    await update.message.reply_text(f"```\n{text}\n```", parse_mode='markdown')

import docker

async def docker_status(update: update, context: contexttypes.default_type):
    """docker容器状态"""
    try:
        client = docker.from_env()
        containers = client.containers.list(all=true)
        
        text = "🐳 docker容器状态\n━━━━━━━━━━━━━━━━━━\n"
        
        for c in containers:
            status_icon = "🟢" if c.status == "running" else "🔴"
            text += f"{status_icon} {c.name[:20]}: {c.status}\n"
        
        if not containers:
            text += "没有容器"
        
        await update.message.reply_text(text)
    except exception as e:
        await update.message.reply_text(f"❌ docker连接失败: {e}")

# 使用nohup
nohup python bot.py > bot.log 2>&1 &

# 使用screen
screen -s bot
python bot.py
# ctrl+a, d 退出

# 使用systemd（推荐）

# /etc/systemd/system/telegram-bot.service
[unit]
description=telegram server monitor bot
after=network.target

[service]
type=simple
user=root
workingdirectory=/opt/telegram_bot
execstart=/usr/bin/python3 /opt/telegram_bot/bot.py
restart=always
restartsec=10

[install]
wantedby=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable telegram-bot
sudo systemctl start telegram-bot
sudo systemctl status telegram-bot