×

注意!页面内容来自https://geek-blogs.com/blog/python-telegram-bot/,本站不储存任何内容,为了更好的阅读体验进行在线解析,若有广告出现,请及时反馈。若您觉得侵犯了您的利益,请通知我们进行删除,然后访问 原网页

Python Telegram Bot 开发指南

Telegram 是一款广受欢迎的即时通讯应用,拥有庞大的用户基础。Python Telegram Bot 是一个用于创建 Telegram 机器人的 Python 库,它提供了简单易用的 API,使得开发者可以轻松地构建功能丰富的 Telegram 机器人。本文将详细介绍 Python Telegram Bot 的基础概念、使用方法、常见实践以及最佳实践,帮助读者深入理解并高效使用该库。

目录#

  1. 基础概念
  2. 安装与配置
  3. 使用方法
    • 简单的回声机器人
    • 处理命令
    • 处理消息
  4. 常见实践
    • 发送图片和文件
    • 处理用户输入验证
    • 状态机的使用
  5. 最佳实践
    • 错误处理
    • 日志记录
    • 异步编程
  6. 小结
  7. 参考资料

基础概念#

Telegram 机器人#

Telegram 机器人是一种特殊的 Telegram 用户,它们由程序控制,可以自动执行各种任务。开发者可以通过 Telegram Bot API 与机器人进行交互,实现消息的发送、接收和处理。

Python Telegram Bot 库#

Python Telegram Bot 是一个基于 Telegram Bot API 的 Python 库,它封装了底层的 HTTP 请求,提供了简单易用的面向对象的接口。使用该库,开发者可以快速地创建和部署 Telegram 机器人。

安装与配置#

安装#

使用 pip 命令可以轻松安装 Python Telegram Bot 库:

pip install python-telegram-bot

配置#

在使用 Python Telegram Bot 之前,需要创建一个 Telegram 机器人并获取其 API 令牌。具体步骤如下:

  1. 打开 Telegram,搜索并添加 BotFather 为联系人。
  2. 发送 /newbot 命令给 BotFather,按照提示设置机器人的名称和用户名。
  3. BotFather 会返回一个 API 令牌,将其保存好,后续会用到。

使用方法#

简单的回声机器人#

以下是一个简单的回声机器人示例,它会将用户发送的消息原样返回:

import logging
from telegram.ext import UpdaterCommandHandlerMessageHandlerFilters
 
# 启用日志记录
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
                    level=logging.INFO)
 
logger = logging.getLogger(__name__)
 
# 处理消息的函数
def echo(updatecontext):
    update.message.reply_text(update.message.text)
 
# 错误处理函数
def error(updatecontext):
    logger.warning('Update "%s" caused error "%s"'updatecontext.error)
 
def main():
    # 创建 Updater 对象,并传入 API 令牌
    updater = Updater("YOUR_API_TOKEN"use_context=True)
 
    # 获取 Dispatcher 对象
    dp = updater.dispatcher
 
    # 添加消息处理程序
    dp.add_handler(MessageHandler(Filters.text & ~Filters.commandecho))
 
    # 添加错误处理程序
    dp.add_error_handler(error)
 
    # 启动机器人
    updater.start_polling()
 
    # 运行机器人,直到按下 Ctrl-C 停止
    updater.idle()
 
if __name__ == '__main__':
    main()

YOUR_API_TOKEN 替换为你从 BotFather 处获取的 API 令牌,运行上述代码,你的回声机器人就可以开始工作了。

处理命令#

Telegram 机器人可以处理用户发送的命令,以下是一个处理 /start 命令的示例:

import logging
from telegram.ext import UpdaterCommandHandler
 
# 启用日志记录
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
                    level=logging.INFO)
 
logger = logging.getLogger(__name__)
 
# 处理 /start 命令的函数
def start(updatecontext):
    update.message.reply_text('欢迎使用我的机器人!')
 
# 错误处理函数
def error(updatecontext):
    logger.warning('Update "%s" caused error "%s"'updatecontext.error)
 
def main():
    # 创建 Updater 对象,并传入 API 令牌
    updater = Updater("YOUR_API_TOKEN"use_context=True)
 
    # 获取 Dispatcher 对象
    dp = updater.dispatcher
 
    # 添加命令处理程序
    dp.add_handler(CommandHandler("start"start))
 
    # 添加错误处理程序
    dp.add_error_handler(error)
 
    # 启动机器人
    updater.start_polling()
 
    # 运行机器人,直到按下 Ctrl-C 停止
    updater.idle()
 
if __name__ == '__main__':
    main()

当用户发送 /start 命令时,机器人会回复“欢迎使用我的机器人!”。

处理消息#

除了处理命令,机器人还可以处理用户发送的普通消息。以下是一个处理图片消息的示例:

import logging
from telegram.ext import UpdaterMessageHandlerFilters
 
# 启用日志记录
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
                    level=logging.INFO)
 
logger = logging.getLogger(__name__)
 
# 处理图片消息的函数
def handle_photo(updatecontext):
    photo = update.message.photo[-1].get_file()
    photo.download('received_photo.jpg')
    update.message.reply_text('已收到图片!')
 
# 错误处理函数
def error(updatecontext):
    logger.warning('Update "%s" caused error "%s"'updatecontext.error)
 
def main():
    # 创建 Updater 对象,并传入 API 令牌
    updater = Updater("YOUR_API_TOKEN"use_context=True)
 
    # 获取 Dispatcher 对象
    dp = updater.dispatcher
 
    # 添加消息处理程序
    dp.add_handler(MessageHandler(Filters.photohandle_photo))
 
    # 添加错误处理程序
    dp.add_error_handler(error)
 
    # 启动机器人
    updater.start_polling()
 
    # 运行机器人,直到按下 Ctrl-C 停止
    updater.idle()
 
if __name__ == '__main__':
    main()

当用户发送图片消息时,机器人会将图片保存为 received_photo.jpg,并回复“已收到图片!”。

常见实践#

发送图片和文件#

以下是一个发送图片的示例:

import logging
from telegram.ext import UpdaterCommandHandler
from telegram import InputFile
 
# 启用日志记录
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
                    level=logging.INFO)
 
logger = logging.getLogger(__name__)
 
# 处理 /sendphoto 命令的函数
def send_photo(updatecontext):
    chat_id = update.message.chat_id
    photo = InputFile('example.jpg')
    context.bot.send_photo(chat_id=chat_idphoto=photo)
 
# 错误处理函数
def error(updatecontext):
    logger.warning('Update "%s" caused error "%s"'updatecontext.error)
 
def main():
    # 创建 Updater 对象,并传入 API 令牌
    updater = Updater("YOUR_API_TOKEN"use_context=True)
 
    # 获取 Dispatcher 对象
    dp = updater.dispatcher
 
    # 添加命令处理程序
    dp.add_handler(CommandHandler("sendphoto"send_photo))
 
    # 添加错误处理程序
    dp.add_error_handler(error)
 
    # 启动机器人
    updater.start_polling()
 
    # 运行机器人,直到按下 Ctrl-C 停止
    updater.idle()
 
if __name__ == '__main__':
    main()

当用户发送 /sendphoto 命令时,机器人会发送 example.jpg 图片。

处理用户输入验证#

在处理用户输入时,需要对输入进行验证,以确保数据的合法性。以下是一个验证用户输入是否为整数的示例:

import logging
from telegram.ext import UpdaterCommandHandlerMessageHandlerFilters
 
# 启用日志记录
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
                    level=logging.INFO)
 
logger = logging.getLogger(__name__)
 
# 处理 /input 命令的函数
def input_command(updatecontext):
    update.message.reply_text('请输入一个整数:')
 
# 处理用户输入的函数
def handle_input(updatecontext):
    try:
        num = int(update.message.text)
        update.message.reply_text(f'你输入的整数是:{num}')
    except ValueError:
        update.message.reply_text('输入无效,请输入一个整数!')
 
# 错误处理函数
def error(updatecontext):
    logger.warning('Update "%s" caused error "%s"'updatecontext.error)
 
def main():
    # 创建 Updater 对象,并传入 API 令牌
    updater = Updater("YOUR_API_TOKEN"use_context=True)
 
    # 获取 Dispatcher 对象
    dp = updater.dispatcher
 
    # 添加命令处理程序
    dp.add_handler(CommandHandler("input"input_command))
 
    # 添加消息处理程序
    dp.add_handler(MessageHandler(Filters.text & ~Filters.commandhandle_input))
 
    # 添加错误处理程序
    dp.add_error_handler(error)
 
    # 启动机器人
    updater.start_polling()
 
    # 运行机器人,直到按下 Ctrl-C 停止
    updater.idle()
 
if __name__ == '__main__':
    main()

当用户发送 /input 命令后,机器人会提示用户输入一个整数,然后对用户输入进行验证。

状态机的使用#

状态机可以帮助管理机器人的复杂交互流程。以下是一个简单的状态机示例,用于引导用户完成一个简单的问卷调查:

import logging
from telegram.ext import UpdaterCommandHandlerMessageHandlerFiltersConversationHandler
 
# 定义状态
QUESTION_1QUESTION_2 = range(2)
 
# 启用日志记录
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
                    level=logging.INFO)
 
logger = logging.getLogger(__name__)
 
# 处理 /survey 命令的函数
def start_survey(updatecontext):
    update.message.reply_text('问题 1:你喜欢 Python 吗?(是/否)')
    return QUESTION_1
 
# 处理问题 1 的回答
def handle_question_1(updatecontext):
    answer = update.message.text
    if answer.lower() in ['是''否']:
        context.user_data['answer_1'] = answer
        update.message.reply_text('问题 2:你使用 Python 多久了?')
        return QUESTION_2
    else:
        update.message.reply_text('输入无效,请回答 是 或 否。')
        return QUESTION_1
 
# 处理问题 2 的回答
def handle_question_2(updatecontext):
    answer = update.message.text
    context.user_data['answer_2'] = answer
    update.message.reply_text('感谢参与问卷调查!你的回答是:'
                              f'问题 1:{context.user_data["answer_1"]},'
                              f'问题 2:{context.user_data["answer_2"]}')
    return ConversationHandler.END
 
# 处理 /cancel 命令
def cancel(updatecontext):
    update.message.reply_text('问卷调查已取消。')
    return ConversationHandler.END
 
# 错误处理函数
def error(updatecontext):
    logger.warning('Update "%s" caused error "%s"'updatecontext.error)
 
def main():
    # 创建 Updater 对象,并传入 API 令牌
    updater = Updater("YOUR_API_TOKEN"use_context=True)
 
    # 获取 Dispatcher 对象
    dp = updater.dispatcher
 
    # 创建 ConversationHandler
    conv_handler = ConversationHandler(
        entry_points=[CommandHandler('survey'start_survey)],
        states={
            QUESTION_1: [MessageHandler(Filters.text & ~Filters.commandhandle_question_1)],
            QUESTION_2: [MessageHandler(Filters.text & ~Filters.commandhandle_question_2)]
        },
        fallbacks=[CommandHandler('cancel'cancel)]
    )
 
    # 添加 ConversationHandler
    dp.add_handler(conv_handler)
 
    # 添加错误处理程序
    dp.add_error_handler(error)
 
    # 启动机器人
    updater.start_polling()
 
    # 运行机器人,直到按下 Ctrl-C 停止
    updater.idle()
 
if __name__ == '__main__':
    main()

用户发送 /survey 命令后,机器人会引导用户完成问卷调查。

最佳实践#

错误处理#

在开发机器人时,需要对可能出现的错误进行处理,以确保机器人的稳定性。可以使用 add_error_handler 方法添加错误处理程序,如前面示例中所示。

日志记录#

日志记录可以帮助开发者监控机器人的运行状态,及时发现和解决问题。使用 logging 模块可以方便地实现日志记录,如前面示例中所示。

异步编程#

Python Telegram Bot 支持异步编程,可以使用 asyncio 库来提高机器人的性能。以下是一个简单的异步机器人示例:

import logging
import asyncio
from telegram.ext import ApplicationCommandHandler
 
# 启用日志记录
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
                    level=logging.INFO)
 
logger = logging.getLogger(__name__)
 
# 处理 /start 命令的异步函数
async def start(updatecontext):
    await update.message.reply_text('欢迎使用我的异步机器人!')
 
# 错误处理函数
async def error(updatecontext):
    logger.warning('Update "%s" caused error "%s"'updatecontext.error)
 
async def main():
    # 创建 Application 对象,并传入 API 令牌
    application = Application.builder().token("YOUR_API_TOKEN").build()
 
    # 添加命令处理程序
    application.add_handler(CommandHandler("start"start))
 
    # 添加错误处理程序
    application.add_error_handler(error)
 
    # 启动机器人
    await application.run_polling()
 
if __name__ == '__main__':
    asyncio.run(main())

使用异步编程可以提高机器人的并发处理能力。

小结#

本文介绍了 Python Telegram Bot 的基础概念、使用方法、常见实践以及最佳实践。通过学习本文,读者可以掌握如何创建和部署 Telegram 机器人,处理各种类型的消息和命令,管理复杂的交互流程,以及提高机器人的性能和稳定性。希望本文能帮助读者深入理解并高效使用 Python Telegram Bot。

参考资料#