TelegramBOT / app.py
Manishx's picture
Rename main.py to app.py
4fef8a5 verified
from fastapi import FastAPI, Request, Response
from contextlib import asynccontextmanager
from http import HTTPStatus
from telegram import ForceReply, Update
from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters
# Initialize python telegram bot
ptb = (
Application.builder()
.updater(None)
.token('7163529766:AAGGGlSjVqtcm_b0vJJu7sU9Y7VzSzeYNZg')
.read_timeout(7)
.get_updates_read_timeout(42)
.build()
)
@asynccontextmanager
async def lifespan(_: FastAPI):
await ptb.bot.setWebhook('https://manishx-telegrambot.hf.space/')
async with ptb:
await ptb.start()
yield
await ptb.stop()
# Initialize FastAPI app (similar to Flask)
app = FastAPI(lifespan=lifespan)
@app.get("/")
def read_general():
return {"response": "Started"}
@app.post("/")
async def process_update(request: Request):
req = await request.json()
update = Update.de_json(req, ptb.bot)
await ptb.process_update(update)
return Response(status_code=HTTPStatus.OK)
# Define a few command handlers. These usually take the two arguments update and context.
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Send a message when the command /start is issued."""
user = update.effective_user
await update.message.reply_html(
rf"Hi {user.mention_html()}!",
reply_markup=ForceReply(selective=True),
)
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Send a message when the command /help is issued."""
await update.message.reply_text("Help!")
async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Echo the user message."""
await update.message.reply_text(update.message.text)
# on different commands - answer in Telegram
ptb.add_handler(CommandHandler("start", start))
ptb.add_handler(CommandHandler("help", help_command))
# on non command i.e message - echo the message on Telegram
ptb.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))