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