Posted inPython

Pythonとpython-telegram-botでTelegramボットを構築する：通知受信とサーバーのリモート操作

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

VPSを運用していて、ディスクが満杯になったときにすぐ通知を受け取りたい、あるいはサーバーにSSHせずにサービスを再起動したい、そんなことはありませんか？Telegram Botは、まさにその問題を解決するために使い始めたもので — 軽量で高速、複雑なWeb UIを構築する必要もありません。

日々の作業のほとんどでPythonを自動化ツールとして使っていて、デプロイスクリプトから監視アラートまで幅広く活用しています。Telegram Botはいわば「コントロールパネル」のような存在で、スマホからコマンドを打てばサーバーがすぐに応答してくれるので、ラップトップを開く必要がありません。

Telegram Botの仕組み

Telegram Botは特別なアカウントで、HTTP APIを通じて制御されます。ユーザーがボットにメッセージを送信すると、TelegramはそのメッセージをTelegramのサーバーに保存します。ボットは次の2つの方法でメッセージを取得できます：

  • Long polling：ボットがAPIに「新しいメッセージはありますか？」と継続的に問い合わせる方式 — シンプルでドメイン不要、テスト環境や小規模VPSに最適。
  • Webhook：Telegramが登録したURLに能動的にメッセージを送信する方式 — HTTPSと実際のドメインが必要で、高負荷の本番環境に適している。

python-telegram-botライブラリはこのHTTP API全体をラップしており、requestsを手動で呼び出す代わりに純粋なPythonで作業できます。注意：バージョン20以降、ライブラリは完全にasyncioベースに移行しており、多くの古いチュートリアルで使われているv13とはコードが大きく異なります。

準備：ボットの作成とトークンの取得

コーディングを始める前に、Telegramの公式ボット管理ツールであるBotFatherを使ってボットを作成する必要があります。

  1. Telegramを開き、@BotFatherを検索する
  2. /newbotコマンドを送信する
  3. 表示名を設定する（例：My Server Monitor）
  4. ユーザー名を設定する — botで終わる必要がある（例：myserver_monitor_bot）
  5. BotFatherが1234567890:ABCdef...形式のトークンを返してくれる — これは秘密にしておくこと

また、ボットが誰に通知を送るかを認識できるよう、自分のChat IDも取得する必要があります。最も手軽な方法は、ボットに任意のメッセージを送信してから、ブラウザで以下のURLにアクセスすることです：

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

返ってくるJSONのmessage.chat.idフィールドを探してください — それがあなたのChat IDです。

インストールとプロジェクト構成

pip install python-telegram-bot==21.0.1

シンプルなディレクトリ構成を作成します：

server-bot/
├── bot.py          # メインロジック
├── config.py       # トークンとChat ID
└── notify.py       # 他のスクリプトからアラートを送信

config.pyファイル：

TOKEN = "1234567890:ABCdefGHIjklMNOpqrSTUvwxYZ"
ALLOWED_CHAT_ID = 987654321  # ユーザーを1人だけ許可する

実践：ステップごとにボットを構築する

ステップ1 — /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(
        "ボットが起動しました。利用可能なコマンド：\n"
        "/status — サーバーの稼働時間を表示\n"
        "/disk — ディスク使用量を確認\n"
        "/restart <service> — サービスを再起動"
    )

def main():
    app = Application.builder().token(config.TOKEN).build()
    app.add_handler(CommandHandler("start", start))
    print("ボット起動中...")
    app.run_polling()

if __name__ == "__main__":
    main()

ステップ2 — セキュリティ：不審者をブロックする

ボットはパブリックで動作するため、ユーザー名を知っている人なら誰でもメッセージを送れます。コマンドを処理する前にChat IDを確認するデコレーターを追加しましょう：

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("アクセス権限がありません。")
            return
        return await func(update, context, *args, **kwargs)
    return wrapped

このデコレーターを全ての機密コマンドに付与します：

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

ステップ3 — サーバー確認コマンド

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)

ステップ4 — 確認付きサービス再起動コマンド

サービスの再起動はダウンタイムを引き起こす可能性がある操作です — 実行前にインラインキーボードを使って確認ステップを追加しましょう：

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("使い方: /restart <サービス名>")
        return

    service = context.args[0]
    keyboard = [[
        InlineKeyboardButton("確認", callback_data=f"restart:{service}"),
        InlineKeyboardButton("キャンセル", callback_data="cancel"),
    ]]
    await update.message.reply_text(
        f"本当に `{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("キャンセルしました。")
        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"`{service}` を正常に再起動しました。")
    else:
        await query.edit_message_text(f"エラー: {result.stderr}")

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))

ステップ5 — cron jobからアクティブにアラートを送信する

これが最もよく使う部分です。監視スクリプトからボットが稼働していなくてもTelegramに通知を送れます — Telegram HTTP APIを直接呼び出すだけです。

# 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

ディスク監視cron jobで使用します — /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>: 残り {free/gb:.1f} GB しかありません！")

systemdで24時間365日ボットを稼働させる

クラッシュ時やVPSの再起動時にボットが自動的に再起動するよう、サービスファイルを作成します：

# /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

まとめ

シンプルなボットから始まって、サーバーの「リモートコントロール」を構築するのに十分な基盤が整いました — 問題発生時のアラート受信、システム状態の確認、サービスの再起動、これら全てをラップトップを開かずにスマホのTelegramから操作できます。

現在、同様のセットアップを使って3台のVPSを同時に監視しています。何か異常があってもダッシュボードを見続けなくてもTelegramがすぐに知らせてくれます。python-telegram-bot v20+では、インラインキーボードやConversationHandlerの多段階フローを追加するのも、今行ったことより難しくありません。

次のステップとして試せること：psutilを組み込んでCPUとRAMの情報をリアルタイムで取得する、またはConversationHandlerを使ってユーザーから複雑な入力が必要な場合に多段階の質問フローを作成する。

Share:
Tags: