使用 Python 制作 Telegram 机器人
课程大纲
设置 Telegram 机器人
Telegram 机器人库
课程大纲
必备知识和要求
- 基本的 Python 知识
- 电报账户
关于课程
在本课程中,你将从零开始创建一个机器人,探索所有可能的设置,以充分利用 Bot API 的所有功能,并创建一个真实的应用程序。所有这些都使用 Python 和 Telegram Bot 来实现。
代码/项目 与本课程相关的所有代码片段和项目都可以在目录
中找到projects
注意:如果您要制作与本课程相关的任何教育项目,我鼓励您在此处发送有关该项目的 PR
什么是机器人
- 由应用程序操作的简单电报帐户
- 实际上并非为聊天而设计
- 可以用来简化任务
- 教
- 玩
- 搜索
- 播送
- 提醒
- 连接
- 用户可以通过以下方式与他们互动
- 信息
- 表情符号
- 命令
- 内联请求
设置 Telegram 机器人
- 打开电报并搜索
BotFather
- 发送
/start
命令启动并将帐户与您的帐户关联 - 发送
/help
命令来显示帮助 - 发送
/newbot
命令并按照指示操作 - 发送
/mybots
命令列出你的机器人,选择所需的机器人 - 点击API 令牌
了解其他机器人设置
- 发送
/mybots
命令列出你的机器人,选择所需的机器人 - 点击“机器人设置”,您将看到各种选项。
- 点击“内联模式”。如果您希望机器人除了接受命令之外,还能接受内联消息。
- 现在,您可以点击“返回机器人”按钮,然后点击“编辑机器人”按钮。在这里,您可以自定义您的机器人,例如设置机器人描述(功能、用途等)、设置机器人简介和个人资料图片。
注意:更改个人资料图片时,botfather 希望您上传有效的图像文件。
机器人的局限性
- 只有用户可以发起聊天
- 下载限制为 20MB,上传限制仅为 50MB
- 发送消息限制为每秒 30 条。不过,你可以联系@botsupport提高发送限制。
有了这个限制,机器人就有了另一种权力,可以在拥有管理员权限的频道中编辑自己的消息和他人的消息
从 Telegram 获取消息
有两种方法可以从电报中获取用户消息。
轮询这是一种每隔 X秒询问 Telegram 是否有新消息到来,并从电报中接收消息或稍后询问
的方式。
Web Hooks
是一种电报将所有更新发送到已配置的 URL 的方式,以便在用户发送新消息时立即接受消息。
Bots API 的 HTTP 接口
文档:https://core.telegram.org/bots/api
可用方法:https://core.telegram.org/bots/api#available-methods
基本 URL:https://api.telegram.org/bot[BOT_API_KEY]/[methodName]
获取机器人描述
方法名称:getMe
就我而言,
{
"ok": true,
"result": {
"id": 920296790,
"is_bot": true,
"first_name": "TELE BOT",
"username": "tel_ebot",
"can_join_groups": true,
"can_read_all_group_messages": false,
"supports_inline_queries": false
}
}
收到新消息
可用方法:https://core.telegram.org/bots/api#getting-updates
方法名称:getUpdates
就我而言
{
"ok": true,
"result": [
{
"update_id": 643663200,
"message": {
"message_id": 3,
"from": {
"id": 517852228,
"is_bot": false,
"first_name": "Gurkirat",
"last_name": "Singh",
"username": "username_of_the_sender",
"language_code": "en"
},
"chat": {
"id": 517852228,
"first_name": "Gurkirat",
"last_name": "Singh",
"username": "username_of_the_sender",
"type": "private"
},
"date": 1586779436,
"text": "hello"
}
}
]
}
获取更新的更多方法:https://core.telegram.org/bots/api#getting-updates
向用户发送消息
方法名称:sendMessage
在此示例中,您将看到我创建一个 echo 机器人,它将监听用户发送的所有消息。然后,我将读取最后一条消息(-1
python 列表中的索引),并以“您向我发送了 {MESSAGE_TEXT} ”格式发送相同的消息。
代码一目了然,我也添加了一些注释。不过,如果你不明白,可以留言。
import urllib.request as request
from urllib.error import HTTPError
from http.client import HTTPResponse
from typing import Dict, List, Union
import json
from datetime import datetime
import signal
import os
signal.signal(signal.SIGINT, signal.SIG_DFL)
class TelegramEcho:
def __init__(self, TG_KEY: str):
self.TG_URL = "https://api.telegram.org/bot{key}/{method}"
self.TG_KEY = TG_KEY
self.__last = None
self.__last_time = None
pass
def run(self):
"""
method to handle the incoming message and the send echo message to the user
"""
while True:
try:
# getting the incoming data
incoming = self.__handle_incoming()
# checking if incoming message_id is same as of last, then skip
if self.__last == incoming["message"]["message_id"]:
continue
else:
self.__last = incoming["message"]["message_id"]
# adding more validation to prevent messaging the last message whenever the polling starts
if not self.__last_time:
self.__last_time = incoming["message"]["date"]
continue
elif self.__last_time < incoming["message"]["date"]:
self.__last_time = incoming["message"]["date"]
else:
continue
# finally printing the incoming message
self.__print_incoming(incoming)
# now sending the echo message
outgoing = self.__handle_outgoing(
incoming["message"]["chat"]["id"],
incoming["message"]["text"])
# finally printing the outgoing message
self.__print_outgoing(outgoing)
pass
except (HTTPError, IndexError):
continue
pass
pass
def __handle_incoming(self) -> Dict[str, Union[int, str]]:
"""
method fetch the recent messages
"""
# getting all messages
getUpdates = request.urlopen(
self.TG_URL.format(key=self.TG_KEY, method="getUpdates"))
# parsing results
results: List[Dict[str, Union[int, str]]] = json.loads(
getUpdates.read().decode())["result"]
# getting the last error
return results[-1]
def __print_incoming(self, incoming: Dict[str, Union[int, str]]):
"""
method to print the incoming message on console
"""
print("[<<<] Message Recieved on %s" % datetime.fromtimestamp(
incoming["message"]["date"]).strftime("%Y-%m-%d %H:%M:%S"))
print("\tText: %s" % incoming["message"]["text"])
print("\tFrom: %s" %
incoming["message"]["from"].get("first_name", "") + " " +
incoming["message"]["from"].get("last_name", ""))
print("\tMessage ID: %d" % incoming["message"]["message_id"])
print("-" * os.get_terminal_size().columns)
pass
def __handle_outgoing(self, chat_id: int,
message_txt: str) -> Dict[str, Union[int, str]]:
"""
method to send the echo message to the same chat
"""
# making the post data
_data: Dict[str, Union[int, str]] = {
"chat_id":
chat_id,
"text":
"You sent me \"{MESSAGE_TEXT}\"".format(MESSAGE_TEXT=message_txt)
}
# creating the request
_request: request.Request = request.Request(
self.TG_URL.format(key=self.TG_KEY, method="sendMessage"),
data=json.dumps(_data).encode('utf8'),
headers={"Content-Type": "application/json"})
# sending HTTP request, for sending message to the user
sendMessage: HTTPResponse = request.urlopen(_request)
result: Dict[str, Union[int, str]] = json.loads(
sendMessage.read().decode())["result"]
return result
def __print_outgoing(self, outgoing):
"""
method to print outgoing data on the console
"""
print("[>>>] Message Send on %s" % datetime.fromtimestamp(
outgoing["date"]).strftime("%Y-%m-%d %H:%M:%S"))
print("\tText: %s" % outgoing["text"])
print("\tFrom: %s" % outgoing["from"]["first_name"])
print("\tMessage ID: %d" % outgoing["message_id"])
print("-" * os.get_terminal_size().columns)
pass
pass
if __name__ == "__main__":
tg = TelegramEcho("YOUR API KEY")
tg.run()
运行此代码之前,请确保添加 API 密钥。运行上述代码后,您将获得如下输出:
Telegram 机器人库
正如您所见,Telegram 提供了大量的方法和 API 来通过任何语言处理机器人,开源社区已经开发了一个 python 客户端,使用这些 API 来提高开发人员的工作效率。
我将使用python-telegram-bot。在制作本课程时,它有大约10k 个 star,38 个未解决的问题,最后一次提交是在 2 天前。
安装模块
# via pip
pip install -U python-telegram-bot
# via conda
conda install -c conda-forge python-telegram-bot
使用一览
在此我将实现get_me
方法,返回机器人详细信息
from telegram import Bot
# initializing the bot with API
bot = Bot("API KEY")
# getting the bot details
print(bot.get_me())
真是太神奇了,开发人员封装了所有后端,让其他开发人员可以编写简洁的代码并进行克隆。它不仅提供了简洁的代码,还实现了支付处理、处理程序以及其他超越发送 HTTP 请求的功能。
python-telegram-bot 完整文档:https://python-telegram-bot.readthedocs.io/en/stable/
在命令中使用处理程序
我希望你已经创建了一些命令,现在让我们让机器人处理这些命令
该库python-telegram-bot
提供了各种处理程序,我将使用其中的CommandHandler
from telegram.ext.commandhandler import CommandHandler
from telegram.ext.updater import Updater
from telegram.ext.dispatcher import Dispatcher
from telegram.update import Update
from telegram.ext.callbackcontext import CallbackContext
from telegram.bot import Bot
# initializing an updator
# documentation: https://python-telegram-bot.readthedocs.io/en/latest/telegram.ext.updater.html#telegram.ext.Updater
updater = Updater("API KEY",
use_context=True)
# getting the dispatcher required to handle the command /start and send message back to the user
# documentation: https://python-telegram-bot.readthedocs.io/en/latest/telegram.ext.dispatcher.html#telegram.ext.Dispatcher
dispatcher: Dispatcher = updater.dispatcher
def start(update: Update, context: CallbackContext):
"""
the callback for handling start command
"""
# getting the bot from context
# documentation: https://python-telegram-bot.readthedocs.io/en/latest/telegram.bot.html#telegram-bot
bot: Bot = context.bot
# sending message to the chat from where it has received the message
# documentation: https://python-telegram-bot.readthedocs.io/en/latest/telegram.bot.html#telegram.Bot.send_message
bot.send_message(chat_id=update.effective_chat.id,
text="You have just entered start command")
# register a handler (here command handler)
# documentation: https://python-telegram-bot.readthedocs.io/en/latest/telegram.ext.dispatcher.html#telegram.ext.Dispatcher.add_handler
dispatcher.add_handler(
# it can accept all the telegram.ext.Handler, CommandHandler inherits Handler class
# documentation: https://python-telegram-bot.readthedocs.io/en/latest/telegram.ext.commandhandler.html#telegram-ext-commandhandler
CommandHandler("start", start))
# starting polling updates from Telegram
# documentation: https://python-telegram-bot.readthedocs.io/en/latest/telegram.ext.updater.html#telegram.ext.Updater.start_polling
updater.start_polling()
发布 HTML 消息
本文我将演示如何向用户发送 HTML 格式的消息。我将使用telegram.ParseMode.HTML
解析器和send_message
方法。
仅更新上述代码中的一个 send_message 行
from telegram.ext.commandhandler import CommandHandler
from telegram.ext.updater import Updater
from telegram.ext.dispatcher import Dispatcher
from telegram.update import Update
from telegram.ext.callbackcontext import CallbackContext
from telegram.bot import Bot
from telegram.parsemode import ParseMode
updater = Updater("API KEY",
use_context=True)
dispatcher: Dispatcher = updater.dispatcher
def start(update: Update, context: CallbackContext):
"""
the callback for handling start command
"""
bot: Bot = context.bot
# Added HTML Parser to the existing command handler
# documentation: https://python-telegram-bot.readthedocs.io/en/stable/telegram.parsemode.html#telegram.ParseMode.HTML
bot.send_message(
chat_id=update.effective_chat.id,
text=
"Hello User, You have used <b>start</b> command. Search about developer on google, <a href='https://www.google.com/search?q=tbhaxor'>@tbhaxor</a>",
parse_mode=ParseMode.HTML,
)
dispatcher.add_handler(CommandHandler("start", start))
updater.start_polling()
text
用户将收到的消息将是您在参数中传递的渲染 HTML
使用键盘构建菜单
在这里,你将体验 Telegram 提供的另一个工具——键盘的强大功能。Telegram 提供两种键盘类型:普通键盘和内置键盘。普通键盘用于群组或聊天。内置键盘用于消息。
简易键盘
在这里,我将使用telegram.ReplyKeyboardMarkup
向聊天中添加新键盘并telegram.ReplyKeyboardRemove
从聊天中删除键盘
from telegram.ext.updater import Updater
from telegram.update import Update
from telegram.ext.callbackcontext import CallbackContext
from telegram.ext.commandhandler import CommandHandler
from telegram.replykeyboardmarkup import ReplyKeyboardMarkup
from telegram.replykeyboardremove import ReplyKeyboardRemove
from telegram.ext.messagehandler import MessageHandler
from telegram.ext.filters import Filters
updater = Updater("API KEY", use_context=True)
def start(update: Update, context: CallbackContext):
"""
method to handle the /start command and create keyboard
"""
# defining the keyboard layout
kbd_layout = [['Option 1', 'Option 2'], ['Option 3', 'Option 4'],
["Option 5"]]
# converting layout to markup
# documentation: https://python-telegram-bot.readthedocs.io/en/stable/telegram.replykeyboardmarkup.html
kbd = ReplyKeyboardMarkup(kbd_layout)
# sending the reply so as to activate the keyboard
update.message.reply_text(text="Select Options", reply_markup=kbd)
def remove(update: Update, context: CallbackContext):
"""
method to handle /remove command to remove the keyboard and return back to text reply
"""
# making a reply markup to remove keyboard
# documentation: https://python-telegram-bot.readthedocs.io/en/stable/telegram.replykeyboardremove.html
reply_markup = ReplyKeyboardRemove()
# sending the reply so as to remove the keyboard
update.message.reply_text(text="I'm back.", reply_markup=reply_markup)
pass
def echo(update: Update, context: CallbackContext):
"""
message to handle any "Option [0-9]" Regrex.
"""
# sending the reply message with the selected option
update.message.reply_text("You just clicked on '%s'" % update.message.text)
pass
updater.dispatcher.add_handler(CommandHandler("start", start))
updater.dispatcher.add_handler(CommandHandler("remove", remove))
# adding the message handler with filter to handle the Option [0-9] regex input
# documentation for MessageHandler: https://python-telegram-bot.readthedocs.io/en/stable/telegram.ext.messagehandler.html
# documentation for Filter: https://python-telegram-bot.readthedocs.io/en/stable/telegram.ext.filters.html#telegram.ext.filters.Filters
updater.dispatcher.add_handler(MessageHandler(Filters.regex(r"Option [0-9]"), echo))
updater.start_polling()
内联键盘
在这里我将使用当用户telegram.ext.CallbackQueryHandler
按下telegram.CallbackQuery
telegram.InlineKeyboardButton
from telegram import InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext.updater import Updater
from telegram.ext.commandhandler import CommandHandler
from telegram.ext.callbackqueryhandler import CallbackQueryHandler
from telegram.callbackquery import CallbackQuery
from telegram.ext.callbackcontext import CallbackContext
from telegram.update import Update
from telegram.message import Message
import sys
# creating updater
updater: Updater = Updater("API KEY",
use_context=True)
def error(update: Update, context: CallbackContext):
"""Log Errors caused by Updates."""
sys.stderr.write("ERROR: '%s' caused by '%s'" % context.error, update)
pass
def start(update: Update, context: CallbackContext):
"""
callback method handling /start command
"""
# creating list of input buttons
# documentation: https://python-telegram-bot.readthedocs.io/en/stable/telegram.inlinekeyboardbutton.html
keyboard = [[
InlineKeyboardButton("Option 1", callback_data='1'),
InlineKeyboardButton("Option 2", callback_data='2')
], [InlineKeyboardButton("Option 3", callback_data='3')]]
# creating a reply markup of inline keyboard options
# documentation: https://python-telegram-bot.readthedocs.io/en/stable/telegram.inlinekeyboardmarkup.html
reply_markup = InlineKeyboardMarkup(keyboard)
# sending the message to the current chat id
# documentation: https://python-telegram-bot.readthedocs.io/en/stable/telegram.message.html#telegram.Message.reply_text
update.message.reply_text('Please choose:', reply_markup=reply_markup)
pass
def button(update, context):
"""
callback method handling button press
"""
# getting the callback query
# documentation: https://python-telegram-bot.readthedocs.io/en/stable/telegram.callbackquery.html
query: CallbackQuery = update.callback_query
# CallbackQueries need to be answered, even if no notification to the user is needed
# Some clients may have trouble otherwise. See https://core.telegram.org/bots/api#callbackquery
# documentation: https://python-telegram-bot.readthedocs.io/en/stable/telegram.callbackquery.html#telegram.CallbackQuery.answer
query.answer()
# editing message sent by the bot
# documentation: https://python-telegram-bot.readthedocs.io/en/stable/telegram.callbackquery.html#telegram.CallbackQuery.edit_message_text
query.edit_message_text(text="Selected option: {}".format(query.data))
# adding listeners
updater.dispatcher.add_handler(CommandHandler('start', start)) # handling /start command
updater.dispatcher.add_handler(CallbackQueryHandler(button)) # handling inline buttons pressing
updater.dispatcher.add_error_handler(error) # error handling
# started polling
updater.start_polling()
发送强制回复
收到用户消息后,Telegram 机器人会向用户显示回复界面,就像用户选择回复机器人消息一样。这可以用来创建分步式用户友好界面。
我将使用telegram.ForceReply
发送强制回复
from telegram.forcereply import ForceReply
from telegram.ext.filters import Filters
from telegram.ext.updater import Updater
from telegram.ext.messagehandler import MessageHandler
from telegram.ext.callbackcontext import CallbackContext
from telegram.update import Update
updater = Updater("API KEY", use_context=True)
def echo(update: Update, context: CallbackContext):
# sending the force reply to the user
# documentation: https://python-telegram-bot.readthedocs.io/en/stable/telegram.forcereply.html
update.message.reply_text(reply_markup=ForceReply(selective=True), text="Reply to this message")
pass
updater.dispatcher.add_handler(MessageHandler(Filters.text, echo))
updater.start_polling()
聊天操作
现在假设你创建了一个机器人来处理一些文件/信息,这可能需要一些时间。用户可能会认为机器人坏了而关闭它。Telegram 提供了非常酷的ChatActions选项,可以在用户提交后立即发送消息。
我将telegram.Bot.send_chat_action
使用telegram.ChatAction
from telegram.bot import Bot
from telegram.update import Update
from telegram.ext.updater import Updater
from telegram.ext.callbackcontext import CallbackContext
from telegram.ext.messagehandler import MessageHandler
from telegram.ext.filters import Filters
from telegram.chataction import ChatAction
from time import sleep
updater = Updater("API KEY", use_context=True)
def echo(update: Update, context: CallbackContext):
# sending the chat action, under the name of bot it will show Typing...
# documentation: https://python-telegram-bot.readthedocs.io/en/stable/telegram.chataction.html
context.bot.send_chat_action(chat_id=update.effective_chat.id, action=ChatAction.TYPING)
# simulating some long processing
sleep(3)
# sending reply when it's done
update.message.reply_text(text="Hey ya!! You sent me '%s'" % update.message.text)
updater.dispatcher.add_handler(MessageHandler(Filters.text, echo))
updater.start_polling()
深度链接
Telegram 也支持深度链接。它有助于创建推荐系统,通过机器人推广你的机器人或产品。
我将使用telegram.utils.helpers.create_deep_linked_url
创建深层链接
from telegram.bot import Bot
from telegram.utils.helpers import create_deep_linked_url
from telegram.ext.commandhandler import CommandHandler
from telegram.ext.callbackcontext import CallbackContext
from telegram.update import Update
from telegram.ext.updater import Updater
import re
updater = Updater("API KEY", use_context=True)
def generate(update: Update, context: CallbackContext):
"""
method to create a deep link and send it to the current user for sharing
"""
# generating a sharable link with the payload
# documentation: https://python-telegram-bot.readthedocs.io/en/stable/telegram.utils.helpers.html#telegram.utils.helpers.create_deep_linked_url
url = create_deep_linked_url(context.bot.get_me().username, update.message.chat.username)
update.message.reply_text(text="Share it with your friends: %s.\n Copy the link and share it with them" % url)
pass
def start(update: Update, context: CallbackContext):
"""
method to run on
"""
# extracting the payload with /start
_ = re.findall(r"(?:/start )(.+)", update.message.text)
# checking if it exists and sending message accordingly
if len(_) > 0:
update.message.reply_text(text="You have been refered by: %s" % _[0])
pass
else:
update.message.reply_text(text="Hello, It seems you are new to this bot")
pass
update.message.reply_text(text="Use /generate to create your own referal")
pass
updater.dispatcher.add_handler(CommandHandler("generate", generate))
updater.dispatcher.add_handler(CommandHandler("start", start))
updater.start_polling()
课程结束
感谢您阅读本课程,我鼓励您亲自探索这款出色的 Telegram 客户端。我很高兴看到您在学习本教程后的作品。如果我将来计划添加更多代码片段和项目,您知道在哪里可以找到它。是的,就在这里
如果您需要任何帮助,可以加入他们的电报组https://t.me/pythontelegrambotgroup,或者您可以通过以下任何方式联系我
电子邮件:tbhaxor@gmail.com
Twitter:@tbhaxor
LinkedIn:@tbhaxor