如何搭建和部署python-telegram-bot v20 Webhook

Python-Telegram-Bot V20 版本发布引入了重大的结构变化根据文档，所有与网络和I/O相关的逻辑现在都通过协程函数（即 async def ... 函数）来工作特别是，telegram.Bot类中所有向Bot API发出请求的方法现在都是协程函数，需要

python-telegram-bot v20 版本的发布引入了重大的结构性变化。根据文档,

“所有与网络和I/O相关的逻辑现在都通过协程函数来处理（即async def ...函数）。特别是，telegram.Bot类的所有向Bot API发送请求的方法现在都是协程函数，并且需要使用await进行等待。” — python-telegram-bot v20 更新日志

从 python-telegram-bot 版本 13 迁移到版本 20 的过程比我最初预期的要复杂。将同步的def函数转换为async def并在新的协程中添加await是相对容易的。但主要的困难在于找到有关如何在生产环境中构建和部署 python-telegram-bot v20 webhooks的详细文档。

本文将解释我从以下内容迁移的原因和方式:

python-telegram-bot v13 → v20
Flask → FastAPI
Gunicorn → Gunicorn + Uvicorn

为什么升级到 v13 到 v20？

v13.x 及更早版本已经不再受到 python-telegram-bot 开发团队的支持。如果 Telegram API 引入了任何新功能，它们只能在 v20 及以上版本中使用。

为什么使用 Webhook 而不是轮询？

python-telegram-bot 开发团队提供的大多数示例都使用Application.run_polling进行轮询。但是，对于大多数 Telegram 机器人使用情况，通常建议使用 Webhook，因为轮询需要您的机器人不断向 Telegram 的服务器发出请求，这可能会消耗大量资源。另一方面，Webhook 提供了更多功能，可以更快地更新和扩展。

在 python-telegram-bot v20 中使用 Flask 的挑战

使用类似 Flask 的 WSGI（Web服务器网关接口）与 python-telegram-bot v20 有些尴尬。

Flask 是同步的，一次只能处理一个请求的 WSGI（Web服务器网关接口）。但是，您仍然可以使用asyncio.run()在 Flask 中运行异步函数，比如由 python-telegram-bot 开发团队提供的自定义 webhook bot 示例中。

asyncio.run()会启动一个事件循环，并执行给定的协程，直到其完成。如果在处理请求之前或之后有任何异步任务正在运行，这些任务将在单独的事件循环中执行。

# 代码片段来自 https://docs.python-telegram-bot.org/en/v20.6/examples.customwebhookbot.html
webserver = uvicorn.Server(
    config=uvicorn.Config(
        app=WsgiToAsgi(flask_app),
        port=PORT,
        use_colors=False,
        host="127.0.0.1",
    )
)
async with application:
    await application.start()
    await webserver.serve() # 启动机器人的 Web服务器
    await application.stop()

然而，这种实现稍微有些尴尬，因为 Flask 与异步全局变量不兼容。

文档中的示例不适用于生产环境。

通常不建议在生产中使用asyncio.run()作为入口点。asyncio.run()函数是为开发和测试目的而设计的，可能不像 Gunicorn 或 UWSGI 等生产服务器那样提供相同级别的稳健性和可靠性。

这些生产服务器提供了许多额外的功能，如日志记录、监控和健康检查，这些功能对于确保生产应用的稳定性和安全性至关重要。

如果您想要在生产环境中部署机器人，使用 ASGI（异步服务器网关接口）和 ASGI Web 服务器实现会更加清晰。

如何操作 – 迁移和部署

从 Flask（WSGI）迁移到 FastAPI（AGSI）

将 Flask 应用程序迁移到 ASGI 很简单。我选择了 FastAPI，因为在这里找到了一份全面的迁移教程。这两个框架的语法非常相似，这意味着您不需要进行太多的代码更改。

# 从 python-telegram-bot v20application = (    Application.builder()    .updater(None)    .token(<your-bot-token>) # 替换为您的机器人令牌    .read_timeout(7)    .get_updates_read_timeout(42)    .build())# 从 FastAPI@asynccontextmanagerasync def lifespan(app: FastAPI):    async with application:        await application.start()        yield        await application.stop()

Quart 似乎是一个可行的替代方案，但是它不支持使用 Uvicorn 进行部署，而我正在使用 python-telegram-bot 团队提供的脚本对 web 服务器进行调整。

一个实际示例

下面的代码显示了一个使用 FastAPI 构建 python-telegram-bot v20 webhook 的最小示例。当接收到 /start 命令时，此机器人会回复“starting…”。

# main.pyfrom contextlib import asynccontextmanagerfrom http import HTTPStatusfrom telegram import Updatefrom telegram.ext import Application, CommandHandlerfrom telegram.ext._contexttypes import ContextTypesfrom fastapi import FastAPI, Request, Response# 初始化 python telegram botptb = (    Application.builder()    .updater(None)    .token(<your-bot-token>) # 替换为您的机器人令牌    .read_timeout(7)    .get_updates_read_timeout(42)    .build())@asynccontextmanagerasync def lifespan(_: FastAPI):    await ptb.bot.setWebhook(<your-webhook-url>) # 替换为您的 Webhook URL    async with ptb:        await ptb.start()        yield        await ptb.stop()# 初始化 FastAPI 应用程序（类似 Flask）app = FastAPI(lifespan=lifespan)@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)# 示例处理器async def start(update, _: ContextTypes.DEFAULT_TYPE):    """当命令 /start 被发送时，发送一条消息。"""    await update.message.reply_text("starting...")ptb.add_handler(CommandHandler("start", start))

要启动机器人，请安装所有所需的依赖项，然后运行 start 命令：gunicorn main:app -k uvicorn.workers.UvicornWorker。

此代码片段是从一个真实的正在生产中的 Telegram 机器人改编而来。在这里查看 @cron_telebot 的源代码，了解其实现方式。您可以根据您的用例自由修改脚本。