python-telegram-bot自动化社交机器人




2023-08-02

blog_main_img

这是个好东西- -

先装依赖

基础安装：

pip install python-telegram-bot

如果需要后台任务、限流或 webhook，可以按项目需要安装附加能力：

pip install "python-telegram-bot[job-queue,rate-limiter,webhooks]"

项目目录可以先这样放：

telegram_bot_demo/
  bot.py
  config.py
  handlers.py
  keyboards.py

小项目放一个文件也能跑，但一旦命令多起来，拆文件会舒服很多。

最小可运行骨架

先写一个能启动、能响应 /start、能复读普通文本的 bot。

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

logging.basicConfig(
    format="%(levelname)s:%(name)s:%(message)s",
    level=logging.INFO,
)


async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.effective_message.reply_text(
        "你好，我是一个用 python-telegram-bot 写的示例 bot。"
    )


async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    text = update.effective_message.text
    await update.effective_message.reply_text(f"收到：{text}")


def main() -> None:
    token = os.environ["TELEGRAM_BOT_TOKEN"]

    app = ApplicationBuilder().token(token).build()
    app.add_handler(CommandHandler("start", start))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))

    app.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == "__main__":
    main()

运行前把 token 放到环境变量里，不要写进源码：

export TELEGRAM_BOT_TOKEN="你的 token"
python bot.py

这里有几个关键点：

ApplicationBuilder().token(token).build() 创建应用。

CommandHandler 处理 /start 这种命令。

MessageHandler 处理普通消息。

回调函数用 async def，里面用 await 调 Telegram API。

python-telegram-bot handler 流程

Handler 的顺序很重要

PTB 会按注册顺序匹配 handler。越具体的 handler 越应该放前面，兜底 handler 放后面。

app.add_handler(CommandHandler("start", start))
app.add_handler(CommandHandler("help", help_command))
app.add_handler(MessageHandler(filters.PHOTO, handle_photo))
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_text))

比如 filters.TEXT & ~filters.COMMAND 是“普通文本但不是命令”。如果你把一个超宽泛的 handler 放太前，后面的处理函数可能根本轮不到。

做一个按钮菜单

Telegram bot 的体验，不应该全靠用户打字。Inline keyboard 能把很多操作变成按钮。

from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
from telegram.ext import CallbackQueryHandler, CommandHandler, ContextTypes


async def menu(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    keyboard = [
        [
            InlineKeyboardButton("查看功能", callback_data="menu:features"),
            InlineKeyboardButton("获取帮助", callback_data="menu:help"),
        ],
        [
            InlineKeyboardButton("项目地址", url="https://github.com/python-telegram-bot/python-telegram-bot")
        ],
    ]
    await update.effective_message.reply_text(
        "请选择一个操作：",
        reply_markup=InlineKeyboardMarkup(keyboard),
    )


async def on_button(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    query = update.callback_query
    await query.answer()

    if query.data == "menu:features":
        await query.edit_message_text("功能：命令、按钮、会话、后台任务、webhook。")
    elif query.data == "menu:help":
        await query.edit_message_text("发送 /start 或 /menu 开始体验。")
    else:
        await query.edit_message_text("这个按钮还没有绑定动作。")


app.add_handler(CommandHandler("menu", menu))
app.add_handler(CallbackQueryHandler(on_button, pattern=r"^menu:"))

按钮回调里建议先 await query.answer()，否则客户端可能一直显示加载状态。callback_data 最好带前缀，比如 menu:、order:、admin:，后续拆 handler 会更清晰。

ConversationHandler：把多步对话写成状态机

复杂一点的 bot 经常要分几步收集信息，比如点餐、报名、配置参数。不要靠一堆 if else 猜用户处在哪一步，ConversationHandler 就是用来写状态机的。

python-telegram-bot 会话流

下面做一个简单点单流程：

from telegram import ReplyKeyboardMarkup, ReplyKeyboardRemove, Update
from telegram.ext import ConversationHandler, CommandHandler, MessageHandler, ContextTypes, filters

CHOOSING, CONFIRMING = range(2)


async def order_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    keyboard = [["咖啡", "甜点"], ["取消"]]
    await update.effective_message.reply_text(
        "想来点什么？",
        reply_markup=ReplyKeyboardMarkup(keyboard, resize_keyboard=True),
    )
    return CHOOSING


async def choose_item(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    item = update.effective_message.text
    if item == "取消":
        return await cancel(update, context)

    context.user_data["item"] = item
    await update.effective_message.reply_text(
        f"你选择了：{item}。回复“确认”提交，或回复“取消”。"
    )
    return CONFIRMING


async def confirm_order(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    text = update.effective_message.text
    if text == "取消":
        return await cancel(update, context)

    if text != "确认":
        await update.effective_message.reply_text("请回复“确认”或“取消”。")
        return CONFIRMING

    item = context.user_data.get("item", "未选择")
    await update.effective_message.reply_text(
        f"已提交：{item}",
        reply_markup=ReplyKeyboardRemove(),
    )
    return ConversationHandler.END


async def cancel(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    context.user_data.pop("item", None)
    await update.effective_message.reply_text(
        "已取消。",
        reply_markup=ReplyKeyboardRemove(),
    )
    return ConversationHandler.END


order_conv = ConversationHandler(
    entry_points=[CommandHandler("order", order_start)],
    states={
        CHOOSING: [MessageHandler(filters.TEXT & ~filters.COMMAND, choose_item)],
        CONFIRMING: [MessageHandler(filters.TEXT & ~filters.COMMAND, confirm_order)],
    },
    fallbacks=[CommandHandler("cancel", cancel)],
)

app.add_handler(order_conv)

context.user_data 很适合存用户会话里的临时数据。它不是业务数据库，重要内容还是应该写到你自己的存储里。

JobQueue：把后台提醒和延迟任务接进来

很多 bot 都需要后台任务，比如延迟提醒、定期拉取状态、批量通知。PTB 的 JobQueue 可以挂在 Application 上使用。

async def send_reminder(context: ContextTypes.DEFAULT_TYPE) -> None:
    job = context.job
    await context.bot.send_message(
        chat_id=job.chat_id,
        text=f"提醒：{job.data}",
    )


async def remind(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    if len(context.args) < 2:
        await update.effective_message.reply_text("用法：/remind 30 喝水")
        return

    delay = int(context.args[0])
    note = " ".join(context.args[1:])
    chat_id = update.effective_chat.id

    context.job_queue.run_once(
        send_reminder,
        when=delay,
        chat_id=chat_id,
        data=note,
        name=f"remind:{chat_id}",
    )

    await update.effective_message.reply_text(f"已安排提醒：{note}")


app.add_handler(CommandHandler("remind", remind))

这里的重点不是“提醒功能”本身，而是把后台任务也放进 PTB 的上下文里，这样你可以直接使用 context.bot、context.job，代码不会散成两套体系。

错误处理：别让异常沉默

bot 长期运行时，网络波动、用户输入异常、第三方接口失败都很常见。统一错误处理可以帮你快速定位问题。

async def on_error(update: object, context: ContextTypes.DEFAULT_TYPE) -> None:
    logging.exception("Handler failed", exc_info=context.error)

    if isinstance(update, Update) and update.effective_message:
        await update.effective_message.reply_text(
            "这一步处理失败了，我已经记录错误。"
        )


app.add_error_handler(on_error)

生产环境里可以把错误发到日志系统或告警通道，但不要把 token、用户隐私、完整请求体随手打到公开日志里。

Webhook：把 bot 做成服务

开发阶段用 polling 很方便；要接入服务化部署，webhook 更适合放在 HTTPS 入口后面。

python-telegram-bot webhook 与部署

import os
from telegram import Update
from telegram.ext import ApplicationBuilder


def main() -> None:
    token = os.environ["TELEGRAM_BOT_TOKEN"]
    public_url = os.environ["PUBLIC_WEBHOOK_URL"]
    port = int(os.environ.get("PORT", "8443"))

    app = ApplicationBuilder().token(token).build()
    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("menu", menu))
    app.add_handler(CallbackQueryHandler(on_button, pattern=r"^menu:"))
    app.add_error_handler(on_error)

    app.run_webhook(
        listen="0.0.0.0",
        port=port,
        url_path=token,
        webhook_url=f"{public_url}/{token}",
        allowed_updates=Update.ALL_TYPES,
    )


if __name__ == "__main__":
    main()

url_path 不要用太容易猜的路径。实际部署时，HTTPS 证书、反向代理、容器健康检查和日志采集都要一起考虑。

配置管理：用 Python 把环境变量收口

配置到处 os.environ[...] 会让代码难维护，可以做一个小配置类。

import os
from dataclasses import dataclass


@dataclass(frozen=True)
class BotConfig:
    token: str
    public_webhook_url: str | None = None
    admin_chat_id: int | None = None


def load_config() -> BotConfig:
    admin_raw = os.environ.get("ADMIN_CHAT_ID")
    admin_chat_id = int(admin_raw) if admin_raw else None

    return BotConfig(
        token=os.environ["TELEGRAM_BOT_TOKEN"],
        public_webhook_url=os.environ.get("PUBLIC_WEBHOOK_URL"),
        admin_chat_id=admin_chat_id,
    )

后续你要切换 polling、webhook、管理员权限、外部接口地址，都可以从这一个入口拿配置。

一个稍微完整的 main

def build_app(config: BotConfig):
    app = ApplicationBuilder().token(config.token).build()

    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("menu", menu))
    app.add_handler(CommandHandler("remind", remind))
    app.add_handler(order_conv)
    app.add_handler(CallbackQueryHandler(on_button, pattern=r"^menu:"))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))
    app.add_error_handler(on_error)

    return app


def main() -> None:
    config = load_config()
    app = build_app(config)

    if config.public_webhook_url:
        app.run_webhook(
            listen="0.0.0.0",
            port=int(os.environ.get("PORT", "8443")),
            url_path=config.token,
            webhook_url=f"{config.public_webhook_url}/{config.token}",
            allowed_updates=Update.ALL_TYPES,
        )
    else:
        app.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == "__main__":
    main()

这个结构的好处是：handler 负责业务，config 负责配置，main 只负责装配。bot 越写越大时，这种边界会很省心。