파이썬과 자바스크립트를 사용하여 텔레그램 봇을 구현하고 AWS에 배포하기

python -m venv .venv && source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install python-telegram-bot

import os
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, ContextTypes, filters

TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]  # export TELEGRAM_BOT_TOKEN=123:ABC

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("👋 안녕하세요! 제가 받은 메시지를 다시 보내겠습니다.")

async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(update.message.text)

def main():
    app = ApplicationBuilder().token(TOKEN).build()
    app.add_handler(CommandHandler("start", start))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))
    app.run_polling()  # long polling

if __name__ == "__main__":
    main()

export TELEGRAM_BOT_TOKEN=123456:ABC-DEF...
python bot.py

npm init -y
npm i telegraf

const { Telegraf } = require('telegraf');

const bot = new Telegraf(process.env.BOT_TOKEN); // BOT_TOKEN 환경 변수 설정
bot.start((ctx) => ctx.reply('👋 안녕하세요! 제가 받은 메시지를 다시 보내겠습니다.'));
bot.on('text', (ctx) => ctx.reply(ctx.message.text));

bot.launch();

// 부드러운 종료
process.once('SIGINT', () => bot.stop('SIGINT'));
process.once('SIGTERM', () => bot.stop('SIGTERM'));

export BOT_TOKEN=123456:ABC-DEF...
node index.js

from telegram.ext import CommandHandler

async def help_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("/start – 인사\n/help – 도움")

app.add_handler(CommandHandler("help", help_cmd))

bot.help((ctx) => ctx.reply('/start – 인사\n/help – 도움'));

from telegram import ReplyKeyboardMarkup

async def start(update: Update, context):
    keyboard = [["도움", "소개"]]
    await update.message.reply_text(
        "옵션을 선택하세요:",
        reply_markup=ReplyKeyboardMarkup(keyboard, resize_keyboard=True)
    )

const { Markup } = require('telegraf');
bot.start((ctx) => {
  ctx.reply("옵션을 선택하세요:", Markup.keyboard([["도움", "소개"]]).resize());
});

TOKEN=123456:ABC-DEF...
WEBHOOK_URL="https://your-domain.com/telegram/${TOKEN}"  # 비밀 경로 포함
SECRET="my-secret-42"

curl -X POST "https://api.telegram.org/bot${TOKEN}/setWebhook" \
  -d url="${WEBHOOK_URL}" \
  -d secret_token="${SECRET}" \
  -d drop_pending_updates=true

bot.launch({
  webhook: {
    domain: "your-domain.com",
    port: 8080,
    // 선택적 비밀 헤더
    secretToken: process.env.WEBHOOK_SECRET
  }
});

https://api.telegram.org/bot<token>/getMe
https://api.telegram.org/bot<token>/getUpdates
https://api.telegram.org/bot<token>/sendMessage

import json, os, urllib.request

BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
SECRET_HEADER = os.environ.get("WEBHOOK_SECRET", "")  # Bot API setWebhook secret_token과 일치해야 함

API_BASE = f"https://api.telegram.org/bot{BOT_TOKEN}"

def reply(chat_id: int, text: str):
    data = json.dumps({"chat_id": chat_id, "text": text}).encode()
    req = urllib.request.Request(f"{API_BASE}/sendMessage", data=data, headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req) as resp:
        return resp.read()

def lambda_handler(event, context):
    # 1) Telegram의 비밀 헤더 확인 (webhook 설정 시 설정됨)
    if SECRET_HEADER:
        if event.get("headers", {}).get("X-Telegram-Bot-Api-Secret-Token") != SECRET_HEADER:
            return {"statusCode": 401, "body": "invalid secret"}

    # 2) 업데이트 파싱
    body = event.get("body") or "{}"
    update = json.loads(body)

    message = update.get("message") or update.get("edited_message")
    if not message:
        # 현재는 콜백_쿼리, 인라인_쿼리 등 비메시지 업데이트 무시
        return {"statusCode": 200, "body": "ok"}

    chat_id = message["chat"]["id"]
    text = (message.get("text") or "").strip()

    # 3) 간단한 라우팅
    if text.startswith("/start"):
        reply(chat_id, "👋 AWS Lambda에서 인사드립니다! 어떤 텍스트를 보내면 다시 보내겠습니다.")
    elif text.startswith("/help"):
        reply(chat_id, "/start – 인사\n/help – 도움\n(AWS Lambda에 배포됨)")
    elif text:
        reply(chat_id, text)  # 에코

    return {"statusCode": 200, "body": "ok"}

# 프로젝트 구조
# .
# ├─ handler.py
# └─ requirements.txt   # (이 최소 예제에서는 비워둠)

zip -r function.zip handler.py
aws lambda create-function \
  --function-name telegram-bot-webhook \
  --runtime python3.11 \
  --role arn:aws:iam::<ACCOUNT_ID>:role/<LambdaExecutionRole> \
  --handler handler.lambda_handler \
  --timeout 10 \
  --memory-size 256 \
  --zip-file fileb://function.zip \
  --environment Variables='{TELEGRAM_BOT_TOKEN=<TOKEN_FROM_BOTFATHER>,WEBHOOK_SECRET=my-secret-42}'

# HTTP API (REST가 아닌)를 사용하여 지연 시간을 줄임
API_ID=$(aws apigatewayv2 create-api \
  --name telegram-webhook \
  --protocol-type HTTP \
  --target arn:aws:lambda:<REGION>:<ACCOUNT_ID>:function:telegram-bot-webhook \
  --query 'ApiId' --output text)

# 기본 경로 POST /webhook 추가
aws apigatewayv2 create-integration \
  --api-id $API_ID \
  --integration-type AWS_PROXY \
  --integration-uri arn:aws:lambda:<REGION>:<ACCOUNT_ID>:function:telegram-bot-webhook \
  --payload-format-version 2.0

aws apigatewayv2 create-route \
  --api-id $API_ID \
  --route-key "POST /webhook" \
  --target "integrations/$(
    aws apigatewayv2 get-integrations --api-id $API_ID --query 'Items[0].IntegrationId' --output text
  )"

aws apigatewayv2 create-deployment --api-id $API_ID
GW_URL=$(aws apigatewayv2 get-apis --query "Items[?ApiId=='$API_ID'].ApiEndpoint" --output text)
echo $GW_URL  # 예: https://abc123.execute-api.ap-somewhere.amazonaws.com

aws lambda add-permission \
  --function-name telegram-bot-webhook \
  --statement-id apigw \
  --action lambda:InvokeFunction \
  --principal apigateway.amazonaws.com \
  --source-arn "arn:aws:execute-api:<REGION>:<ACCOUNT_ID>:$API_ID/*/*/webhook"

TOKEN=<TOKEN_FROM_BOTFATHER>
WEBHOOK_URL="$GW_URL/webhook"
SECRET="my-secret-42"

curl -X POST "https://api.telegram.org/bot${TOKEN}/setWebhook" \
  -d url="${WEBHOOK_URL}" \
  -d secret_token="${SECRET}" \
  -d drop_pending_updates=true

import os, asyncio
from fastapi import FastAPI, Request, Header
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, ContextTypes, filters

TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
SECRET = os.environ.get("WEBHOOK_SECRET", "")

app = FastAPI()
tg_app = None  # PTB 애플리케이션

async def start_cmd(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("🚀 ECS Fargate + PTB에서 인사드립니다!")

async def echo(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(update.message.text)

@app.on_event("startup")
async def on_startup():
    global tg_app
    tg_app = ApplicationBuilder().token(TOKEN).build()
    tg_app.add_handler(CommandHandler("start", start_cmd))
    tg_app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))
    # run_polling() 실행하지 않음; 웹훅을 수동으로 피드합니다.

@app.post("/telegram/webhook")
async def telegram_webhook(request: Request, x_telegram_bot_api_secret_token: str | None = Header(default=None)):
    if SECRET and (x_telegram_bot_api_secret_token != SECRET):
        return {"ok": True}  # 조용히 무시

    data = await request.json()
    update = Update.de_json(data, tg_app.bot)
    await tg_app.process_update(update)
    return {"ok": True}

FROM python:3.11-slim
WORKDIR /app
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1
RUN pip install --no-cache-dir fastapi uvicorn[standard] python-telegram-bot==21.*
COPY app.py .
EXPOSE 8080
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8080"]

aws ecr create-repository --repository-name telegram-ptb || true
ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
REGION=<REGION>
ECR="$ACCOUNT_ID.dkr.ecr.$REGION.amazonaws.com/telegram-ptb"

aws ecr get-login-password --region $REGION | docker login --username AWS --password-stdin "$ACCOUNT_ID.dkr.ecr.$REGION.amazonaws.com"
docker build -t telegram-ptb .
docker tag telegram-ptb:latest $ECR:latest
docker push $ECR:latest

TOKEN=<TOKEN_FROM_BOTFATHER>
SECRET="my-secret-42"
WEBHOOK_URL="https://your.domain.com/telegram/webhook"

curl -X POST "https://api.telegram.org/bot${TOKEN}/setWebhook" \
  -d url="${WEBHOOK_URL}" \
  -d secret_token="${SECRET}" \
  -d drop_pending_updates=true \
  -d allowed_updates='["message","edited_message","callback_query"]'