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