تطبيق روبوت تليجرام باستخدام بايثون وجاوا سكربت مع نشره على 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()  # الاستجواب الطويل

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 = [["Help", "About"]]
    await update.message.reply_text(
        "اختر خيارًا:",
        reply_markup=ReplyKeyboardMarkup(keyboard, resize_keyboard=True)
    )

const { Markup } = require('telegraf');
bot.start((ctx) => {
  ctx.reply("اختر خيارًا:", Markup.keyboard([["Help", "About"]]).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", "")  # يجب أن يتطابق مع secret_token المحدد في Bot API setWebhook

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 (المحدد عند تكوين الويب هوك)
    if SECRET_HEADER:
        if event.get("headers", {}).get("X-Telegram-Bot-Api-Secret-Token") != SECRET_HEADER:
            return {"statusCode": 401, "body": "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"]'