Telegram Bot + Webhook 实战指南：从用 Python 造机器人到 Kubernetes 部署

1. 在 Telegram 中搜索 @BotFather
2. 输入 /newbot 命令
3. 输入机器人名称: "My DevOps Bot"
4. 输入机器人用户名: "my_devops_bot" (必须以 _bot 结尾)
5. 领取 API 令牌: 123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11

pip install python-telegram-bot==21.8

# bot_polling.py
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import (
    ApplicationBuilder, CommandHandler, MessageHandler,
    CallbackQueryHandler, ContextTypes, filters
)

TOKEN = "YOUR_BOT_TOKEN"

# /start 命令
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    keyboard = [
        [InlineKeyboardButton("🔍 服务器状态", callback_data="status")],
        [InlineKeyboardButton("📊 指标", callback_data="metrics")],
        [InlineKeyboardButton("🚀 部署", callback_data="deploy")],
    ]
    reply_markup = InlineKeyboardMarkup(keyboard)
    await update.message.reply_text(
        "欢迎来到 DevOps Bot! 🤖\n请选择你想执行的操作:",
        reply_markup=reply_markup
    )

# /status 命令
async def status(update: Update, context: ContextTypes.DEFAULT_TYPE):
    import subprocess
    result = subprocess.run(
        ["kubectl", "get", "nodes", "-o", "wide"],
        capture_output=True, text=True, timeout=10
    )
    await update.message.reply_text(
        f"```\n{result.stdout}\n```",
        parse_mode="MarkdownV2"
    )

# 内联按钮回调
async def button_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    await query.answer()

    if query.data == "status":
        await query.edit_message_text("🔍 正在检查服务器状态...")
        # 执行 kubectl
        result = check_cluster_status()
        await query.edit_message_text(f"✅ 集群状态:\n```\n{result}\n```",
                                       parse_mode="MarkdownV2")
    elif query.data == "metrics":
        await query.edit_message_text("📊 正在采集指标...")
        metrics = get_prometheus_metrics()
        await query.edit_message_text(f"📊 当前指标:\n{metrics}")
    elif query.data == "deploy":
        keyboard = [
            [InlineKeyboardButton("✅ 批准", callback_data="deploy_approve")],
            [InlineKeyboardButton("❌ 取消", callback_data="deploy_cancel")],
        ]
        await query.edit_message_text(
            "🚀 确认要执行部署吗?",
            reply_markup=InlineKeyboardMarkup(keyboard)
        )

# 处理普通消息
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    text = update.message.text
    # AI 回复 (对接 OpenAI)
    from openai import OpenAI
    client = OpenAI()
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a DevOps assistant."},
            {"role": "user", "content": text}
        ]
    )
    await update.message.reply_text(response.choices[0].message.content)

# 运行应用
app = ApplicationBuilder().token(TOKEN).build()
app.add_handler(CommandHandler("start", start))
app.add_handler(CommandHandler("status", status))
app.add_handler(CallbackQueryHandler(button_callback))
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))

print("Bot is running...")
app.run_polling()

# bot_webhook.py
from fastapi import FastAPI, Request
from telegram import Update, Bot
from telegram.ext import Application, CommandHandler, MessageHandler, filters
import uvicorn

TOKEN = "YOUR_BOT_TOKEN"
WEBHOOK_URL = "https://bot.example.com/webhook"

# FastAPI 应用
fastapi_app = FastAPI()

# Telegram Application
tg_app = Application.builder().token(TOKEN).build()

# 注册处理器
async def start(update: Update, context):
    await update.message.reply_text("Hello! 🤖")

async def echo(update: Update, context):
    await update.message.reply_text(f"Echo: {update.message.text}")

tg_app.add_handler(CommandHandler("start", start))
tg_app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))

@fastapi_app.on_event("startup")
async def on_startup():
    await tg_app.initialize()
    await tg_app.start()
    # 设置 Webhook
    await tg_app.bot.set_webhook(
        url=f"{WEBHOOK_URL}/{TOKEN}",
        allowed_updates=["message", "callback_query"],
        drop_pending_updates=True,
    )
    print(f"Webhook set to {WEBHOOK_URL}/{TOKEN}")

@fastapi_app.on_event("shutdown")
async def on_shutdown():
    await tg_app.stop()
    await tg_app.shutdown()

@fastapi_app.post(f"/webhook/{TOKEN}")
async def webhook(request: Request):
    data = await request.json()
    update = Update.de_json(data, tg_app.bot)
    await tg_app.process_update(update)
    return {"ok": True}

@fastapi_app.get("/health")
async def health():
    return {"status": "ok"}

if __name__ == "__main__":
    uvicorn.run(fastapi_app, host="0.0.0.0", port=8080)

# 查看 Webhook 状态
curl "https://api.telegram.org/bot${TOKEN}/getWebhookInfo" | jq

# 删除 Webhook (切换回 Polling 模式时)
curl "https://api.telegram.org/bot${TOKEN}/deleteWebhook"

from telegram.ext import ConversationHandler

# 定义状态
NAME, EMAIL, CONFIRM = range(3)

async def ticket_start(update, context):
    await update.message.reply_text("开始创建工单。请输入标题:")
    return NAME

async def ticket_name(update, context):
    context.user_data["title"] = update.message.text
    await update.message.reply_text("请输入描述:")
    return EMAIL

async def ticket_desc(update, context):
    context.user_data["description"] = update.message.text
    title = context.user_data["title"]
    desc = context.user_data["description"]

    keyboard = [
        [InlineKeyboardButton("✅ 创建", callback_data="confirm_yes")],
        [InlineKeyboardButton("❌ 取消", callback_data="confirm_no")],
    ]
    await update.message.reply_text(
        f"📋 工单确认:\n标题: {title}\n描述: {desc}",
        reply_markup=InlineKeyboardMarkup(keyboard)
    )
    return CONFIRM

conv_handler = ConversationHandler(
    entry_points=[CommandHandler("ticket", ticket_start)],
    states={
        NAME: [MessageHandler(filters.TEXT & ~filters.COMMAND, ticket_name)],
        EMAIL: [MessageHandler(filters.TEXT & ~filters.COMMAND, ticket_desc)],
        CONFIRM: [CallbackQueryHandler(ticket_confirm)],
    },
    fallbacks=[CommandHandler("cancel", cancel)],
)

from telegram.ext import ApplicationBuilder

async def daily_report(context: ContextTypes.DEFAULT_TYPE):
    """每天上午 9 点的报告"""
    chat_id = context.job.data["chat_id"]

    # 采集指标
    cpu = get_cluster_cpu()
    memory = get_cluster_memory()
    pods = get_pod_count()

    message = (
        "📊 Daily Cluster Report\n"
        f"━━━━━━━━━━━━━━\n"
        f"🖥 CPU: {cpu}%\n"
        f"💾 Memory: {memory}%\n"
        f"📦 Pods: {pods}\n"
        f"━━━━━━━━━━━━━━"
    )
    await context.bot.send_message(chat_id=chat_id, text=message)

# 注册定时任务
async def setup_schedule(update, context):
    chat_id = update.effective_chat.id
    context.job_queue.run_daily(
        daily_report,
        time=datetime.time(hour=9, minute=0, tzinfo=ZoneInfo("Asia/Seoul")),
        data={"chat_id": chat_id},
        name=f"daily_report_{chat_id}"
    )
    await update.message.reply_text("✅ 每天上午 9 点会给你发送报告!")

async def handle_document(update: Update, context):
    """处理文件上传"""
    doc = update.message.document
    file = await context.bot.get_file(doc.file_id)

    # 下载
    local_path = f"/tmp/{doc.file_name}"
    await file.download_to_drive(local_path)

    await update.message.reply_text(f"📁 文件保存完成: {doc.file_name}")

async def send_file(update: Update, context):
    """发送文件"""
    await context.bot.send_document(
        chat_id=update.effective_chat.id,
        document=open("/tmp/report.pdf", "rb"),
        filename="cluster_report.pdf",
        caption="📊 集群报告"
    )

FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8080
CMD ["uvicorn", "bot_webhook:fastapi_app", "--host", "0.0.0.0", "--port", "8080"]

apiVersion: apps/v1
kind: Deployment
metadata:
  name: telegram-bot
  namespace: chatbot
spec:
  replicas: 2
  selector:
    matchLabels:
      app: telegram-bot
  template:
    metadata:
      labels:
        app: telegram-bot
    spec:
      containers:
        - name: bot
          image: registry.example.com/telegram-bot:latest
          ports:
            - containerPort: 8080
          env:
            - name: BOT_TOKEN
              valueFrom:
                secretKeyRef:
                  name: telegram-bot-secret
                  key: token
            - name: WEBHOOK_URL
              value: 'https://bot.example.com'
          resources:
            requests:
              cpu: 100m
              memory: 128Mi
            limits:
              cpu: 500m
              memory: 256Mi
          livenessProbe:
            httpGet:
              path: /health
              port: 8080
            initialDelaySeconds: 10
            periodSeconds: 30
          readinessProbe:
            httpGet:
              path: /health
              port: 8080
            initialDelaySeconds: 5
            periodSeconds: 10
---
apiVersion: v1
kind: Service
metadata:
  name: telegram-bot
  namespace: chatbot
spec:
  selector:
    app: telegram-bot
  ports:
    - port: 80
      targetPort: 8080
---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: telegram-bot
  namespace: chatbot
  annotations:
    cert-manager.io/cluster-issuer: letsencrypt-prod
spec:
  tls:
    - hosts:
        - bot.example.com
      secretName: telegram-bot-tls
  rules:
    - host: bot.example.com
      http:
        paths:
          - path: /webhook
            pathType: Prefix
            backend:
              service:
                name: telegram-bot
                port:
                  number: 80

# 1) 在 Webhook URL 中包含令牌 (防止未授权请求)
WEBHOOK_PATH = f"/webhook/{TOKEN}"

# 2) IP 白名单 (Telegram 服务器 IP)
TELEGRAM_IPS = ["149.154.160.0/20", "91.108.4.0/22"]

from fastapi import Request, HTTPException
import ipaddress

@fastapi_app.middleware("http")
async def check_telegram_ip(request: Request, call_next):
    if request.url.path.startswith("/webhook"):
        client_ip = ipaddress.ip_address(request.client.host)
        allowed = any(
            client_ip in ipaddress.ip_network(net)
            for net in TELEGRAM_IPS
        )
        if not allowed:
            raise HTTPException(status_code=403)
    return await call_next(request)

# 3) Secret Token (Telegram Bot API 6.1+)
await bot.set_webhook(
    url=WEBHOOK_URL,
    secret_token="my-secret-token-123"
)

@fastapi_app.post(WEBHOOK_PATH)
async def webhook(request: Request):
    if request.headers.get("X-Telegram-Bot-Api-Secret-Token") != "my-secret-token-123":
        raise HTTPException(status_code=403)
    # ...