Posted inPython

Xây dựng Telegram Bot với Python và python-telegram-bot: Nhận thông báo và điều khiển server từ xa

Python tutorial - IT technology blog
Python tutorial - IT technology blog

Bạn đang chạy một con VPS và muốn nhận thông báo ngay khi disk đầy, hay cần restart service mà không cần SSH vào server? Telegram Bot là thứ mình đã dùng để giải quyết đúng bài toán đó — nhẹ, nhanh, và không cần build web UI phức tạp.

Mình dùng Python làm automation tool cho hầu hết task hàng ngày, từ deploy script đến monitoring alert. Telegram Bot gần như là “màn hình điều khiển” của mình — gõ lệnh từ điện thoại là server phản hồi ngay, không cần mở laptop.

Telegram Bot hoạt động như thế nào?

Mỗi Telegram Bot là một tài khoản đặc biệt, được điều khiển qua HTTP API. Khi người dùng gửi tin nhắn tới bot, Telegram lưu tin nhắn đó trên server của họ. Bot của bạn lấy tin nhắn về theo một trong hai cách:

  • Long polling: Bot liên tục hỏi API “có tin nhắn mới không?” — đơn giản, không cần domain, phù hợp để test và VPS nhỏ.
  • Webhook: Telegram chủ động gửi tin nhắn tới URL bạn đăng ký — cần HTTPS và domain thật, phù hợp môi trường production tải cao.

Thư viện python-telegram-bot bọc toàn bộ HTTP API này lại, cho bạn làm việc với Python thuần túy thay vì tự gọi requests thủ công. Lưu ý: từ phiên bản 20+, thư viện chuyển sang asyncio hoàn toàn — code khác hẳn v13 mà nhiều tutorial cũ vẫn dùng.

Chuẩn bị: Tạo bot và lấy token

Trước khi code, bạn cần tạo bot qua BotFather — bot chính thức của Telegram để quản lý bot.

  1. Mở Telegram, tìm @BotFather
  2. Gửi lệnh /newbot
  3. Đặt tên hiển thị (VD: My Server Monitor)
  4. Đặt username — phải kết thúc bằng bot (VD: myserver_monitor_bot)
  5. BotFather trả về token dạng 1234567890:ABCdef... — giữ cái này riêng tư

Bạn cũng cần lấy Chat ID của mình để bot biết gửi thông báo cho ai. Cách nhanh nhất: gửi một tin nhắn bất kỳ cho bot, rồi truy cập URL này trên trình duyệt:

https://api.telegram.org/bot<TOKEN>/getUpdates

Tìm trường message.chat.id trong JSON trả về — đó là Chat ID của bạn.

Cài đặt và cấu trúc project

pip install python-telegram-bot==21.0.1

Tạo cấu trúc thư mục gọn:

server-bot/
├── bot.py          # Logic chính
├── config.py       # Token và Chat ID
└── notify.py       # Gửi alert từ script khác

File config.py:

TOKEN = "1234567890:ABCdefGHIjklMNOpqrSTUvwxYZ"
ALLOWED_CHAT_ID = 987654321  # Chỉ cho phép đúng 1 người dùng

Thực hành: Xây dựng bot từng bước

Bước 1 — Bot cơ bản với lệnh /start

from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes
import config

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(
        "Bot đang chạy. Các lệnh có sẵn:\n"
        "/status — Xem uptime server\n"
        "/disk — Kiểm tra dung lượng ổ đĩa\n"
        "/restart <service> — Restart service"
    )

def main():
    app = Application.builder().token(config.TOKEN).build()
    app.add_handler(CommandHandler("start", start))
    print("Bot đang chạy...")
    app.run_polling()

if __name__ == "__main__":
    main()

Bước 2 — Bảo mật: chặn người lạ

Bot chạy public — ai biết username đều có thể nhắn tin. Thêm decorator kiểm tra Chat ID trước khi xử lý lệnh:

from functools import wraps
import config

def restricted(func):
    @wraps(func)
    async def wrapped(update: Update, context: ContextTypes.DEFAULT_TYPE, *args, **kwargs):
        if update.effective_chat.id != config.ALLOWED_CHAT_ID:
            await update.message.reply_text("Không có quyền truy cập.")
            return
        return await func(update, context, *args, **kwargs)
    return wrapped

Gắn decorator này cho mọi lệnh nhạy cảm:

@restricted
async def disk_status(update: Update, context: ContextTypes.DEFAULT_TYPE):
    ...

Bước 3 — Lệnh kiểm tra server

import subprocess
import shutil

@restricted
async def disk_status(update: Update, context: ContextTypes.DEFAULT_TYPE):
    total, used, free = shutil.disk_usage("/")
    gb = 1024 ** 3
    msg = (
        f"Disk Usage:\n"
        f"Total: {total / gb:.1f} GB\n"
        f"Used:  {used / gb:.1f} GB ({used/total*100:.1f}%)\n"
        f"Free:  {free / gb:.1f} GB"
    )
    await update.message.reply_text(msg)

@restricted
async def system_status(update: Update, context: ContextTypes.DEFAULT_TYPE):
    uptime = subprocess.run(["uptime", "-p"], capture_output=True, text=True)
    load = subprocess.run(["cat", "/proc/loadavg"], capture_output=True, text=True)
    msg = f"Uptime: {uptime.stdout.strip()}\nLoad: {load.stdout.strip()}"
    await update.message.reply_text(msg)

Bước 4 — Lệnh restart service với xác nhận

Restart service là thao tác có thể gây downtime — thêm bước xác nhận qua Inline Keyboard trước khi thực hiện:

from telegram import InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import CallbackQueryHandler

@restricted
async def restart_service(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if not context.args:
        await update.message.reply_text("Dùng: /restart <tên_service>")
        return

    service = context.args[0]
    keyboard = [[
        InlineKeyboardButton("Xác nhận", callback_data=f"restart:{service}"),
        InlineKeyboardButton("Hủy", callback_data="cancel"),
    ]]
    await update.message.reply_text(
        f"Bạn chắc muốn restart `{service}`?",
        reply_markup=InlineKeyboardMarkup(keyboard),
        parse_mode="Markdown"
    )

async def button_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    await query.answer()

    if query.data == "cancel":
        await query.edit_message_text("Đã hủy.")
        return

    _, service = query.data.split(":", 1)
    result = subprocess.run(
        ["sudo", "systemctl", "restart", service],
        capture_output=True, text=True
    )
    if result.returncode == 0:
        await query.edit_message_text(f"Đã restart `{service}` thành công.")
    else:
        await query.edit_message_text(f"Lỗi: {result.stderr}")

Đăng ký handler trong main():

app.add_handler(CommandHandler("disk", disk_status))
app.add_handler(CommandHandler("status", system_status))
app.add_handler(CommandHandler("restart", restart_service))
app.add_handler(CallbackQueryHandler(button_handler))

Bước 5 — Gửi alert chủ động từ cron job

Đây là phần mình dùng nhiều nhất: script monitoring gửi thông báo Telegram mà không cần bot đang chạy — chỉ cần gọi Telegram HTTP API trực tiếp.

# notify.py
import requests
import config

def send_alert(message: str) -> bool:
    url = f"https://api.telegram.org/bot{config.TOKEN}/sendMessage"
    payload = {
        "chat_id": config.ALLOWED_CHAT_ID,
        "text": message,
        "parse_mode": "HTML"
    }
    resp = requests.post(url, json=payload, timeout=10)
    return resp.ok

Dùng trong cron job theo dõi disk — thêm vào /etc/cron.hourly/:

#!/usr/bin/env python3
import shutil, sys
sys.path.insert(0, "/home/ubuntu/server-bot")
from notify import send_alert

_, _, free = shutil.disk_usage("/")
gb = 1024 ** 3

if free / gb < 5:
    send_alert(f"<b>Disk Warning</b>: Chỉ còn {free/gb:.1f} GB trống!")

Chạy bot 24/7 với systemd

Tạo file service để bot tự khởi động lại khi crash hoặc khi VPS reboot:

# /etc/systemd/system/telegram-bot.service
[Unit]
Description=Telegram Server Bot
After=network.target

[Service]
Type=simple
User=ubuntu
WorkingDirectory=/home/ubuntu/server-bot
ExecStart=/home/ubuntu/server-bot/venv/bin/python bot.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target
sudo systemctl enable telegram-bot
sudo systemctl start telegram-bot
sudo systemctl status telegram-bot

Kết luận

Từ một con bot đơn giản, bạn đã có đủ nền tảng để build một “remote control” cho server — nhận alert khi có vấn đề, kiểm tra trạng thái hệ thống, restart service, tất cả từ điện thoại qua Telegram mà không cần mở laptop.

Mình đang dùng setup tương tự để monitor 3 VPS cùng lúc. Mỗi khi có gì bất thường, Telegram báo ngay thay vì mình phải ngồi theo dõi dashboard. Với python-telegram-bot v20+, việc thêm inline keyboard hay ConversationHandler nhiều bước cũng không phức tạp hơn những gì bạn vừa làm.

Bước tiếp theo bạn có thể thử: tích hợp thêm psutil để lấy thông tin CPU và RAM theo thời gian thực, hoặc dùng ConversationHandler để tạo flow hỏi-đáp nhiều bước khi cần input phức tạp hơn từ người dùng.

Share:
Tags: