pip install python-telegram-bot
pip install psutil  # 系統(tǒng)監(jiān)控
pip install aiohttp  # 異步HTTP
pip install apscheduler  # 定時任務(wù)

telegram_bot/
├── bot.py           # 主程序
├── config.py        # 配置
├── monitors/        # 監(jiān)控模塊
│   ├── __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("你好！我是服務(wù)器監(jiān)控機器人。")

async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    help_text = """
可用命令：
/start - 開始
/help - 幫助
/status - 服務(wù)器狀態(tài)
/cpu - CPU使用率
/memory - 內(nèi)存使用
/disk - 磁盤使用
    """
    await update.message.reply_text(help_text)

def main():
    # 創(chuàng)建應(yīng)用
    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. 發(fā)送 /start
# 3. 發(fā)送 /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狀態(tài)
━━━━━━━━━━━━━━━
總使用率: {info['percent']}%
核心數(shù)量: {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():
    """獲取內(nèi)存信息"""
    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():
    """格式化內(nèi)存信息"""
    info = get_memory_info()
    
    bar = "█" * int(info['percent'] / 10) + "?" * (10 - int(info['percent'] / 10))
    
    text = f"""
?? 內(nèi)存狀態(tài)
━━━━━━━━━━━━━━━
使用率: [{bar}] {info['percent']}%

物理內(nèi)存:
  總量: {info['total']:.1f} GB
  已用: {info['used']:.1f} GB
  可用: {info['available']:.1f} GB

交換分區(qū):
  總量: {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 = "?? 磁盤狀態(tài)\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():
    """獲取網(wǎng)絡(luò)信息"""
    # 網(wǎng)絡(luò)IO
    net_io = psutil.net_io_counters()
    
    # IP地址
    hostname = socket.gethostname()
    try:
        ip = socket.gethostbyname(hostname)
    except:
        ip = "未知"
    
    # 網(wǎng)絡(luò)連接數(shù)
    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():
    """格式化網(wǎng)絡(luò)信息"""
    info = get_network_info()
    
    text = f"""
?? 網(wǎng)絡(luò)狀態(tài)
━━━━━━━━━━━━━━━
主機名: {info['hostname']}
IP地址: {info['ip']}
連接數(shù): {info['connections']}

流量統(tǒng)計:
  發(fā)送: {info['bytes_sent']:.1f} MB
  接收: {info['bytes_recv']:.1f} MB
  發(fā)送包: {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
}

# 監(jiān)控函數(shù)（簡化版，集成上面的模塊）
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()
    
    # 狀態(tài)圖標
    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"""
?? 服務(wù)器狀態(tài)概覽
━━━━━━━━━━━━━━━━━━
{cpu_icon} CPU: {s['cpu']}%
{mem_icon} 內(nèi)存: {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("?? 狀態(tài)", callback_data="status"),
            InlineKeyboardButton("??? CPU", callback_data="cpu"),
        ],
        [
            InlineKeyboardButton("?? 內(nèi)存", callback_data="memory"),
            InlineKeyboardButton("?? 磁盤", callback_data="disk"),
        ],
        [
            InlineKeyboardButton("?? 網(wǎng)絡(luò)", callback_data="network"),
            InlineKeyboardButton("?? 刷新", callback_data="refresh"),
        ]
    ]
    reply_markup = InlineKeyboardMarkup(keyboard)
    
    await update.message.reply_text(
        "?? 服務(wù)器監(jiān)控機器人\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"?? 內(nèi)存使用率過高: {s['memory']}%")
    if s['disk'] > THRESHOLDS['disk']:
        alerts.append(f"?? 磁盤使用率過高: {s['disk']}%")
    
    if alerts:
        alert_text = "?? 服務(wù)器告警\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 = "?? 每日服務(wù)器報告\n" + format_status()
    await context.bot.send_message(chat_id=ADMIN_CHAT_ID, text=text)

def main():
    # 創(chuàng)建應(yīng)用
    app = Application.builder().token(TOKEN).build()
    
    # 注冊命令
    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("status", status))
    app.add_handler(CallbackQueryHandler(button_callback))
    
    # 定時任務(wù)
    scheduler = AsyncIOScheduler()
    # 每分鐘檢查告警
    scheduler.add_job(
        check_alerts, 'interval', minutes=1,
        args=[app]
    )
    # 每天9點發(fā)送報告
    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):
    """執(zhí)行系統(tǒng)命令（危險！僅限管理員）"""
    # 權(quán)限檢查
    if update.effective_user.id != ADMIN_CHAT_ID:
        await update.message.reply_text("? 無權(quán)限")
        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("? 命令執(zhí)行超時")
    except Exception as e:
        await update.message.reply_text(f"? 執(zhí)行失敗: {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容器狀態(tài)"""
    try:
        client = docker.from_env()
        containers = client.containers.list(all=True)
        
        text = "?? Docker容器狀態(tài)\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