詳解如何使用 Python 操作 Telegram（電報(bào)）機(jī)器人
楔子
Telegram（電報(bào)）相信大家都知道，關(guān)于它的介紹和注冊(cè)方式這里就跳過(guò)了，我假設(shè)你已經(jīng)注冊(cè)好了。本篇文章來(lái)聊一聊 Telegram 提供的機(jī)器人，以及如何用 Python 為機(jī)器人實(shí)現(xiàn)各種各樣的功能。
創(chuàng)建機(jī)器人
首先我們使用瀏覽器打開(kāi) https://web.telegram.org，然后用手機(jī)上的 APP 掃碼登錄。
圖片
登錄之后搜索 BotFather，機(jī)器人需要通過(guò) BotFather 來(lái)創(chuàng)建，當(dāng)然 BotFather 本身也是一個(gè)機(jī)器人，但它同時(shí)管理著其它的機(jī)器人。
我們點(diǎn)擊 BotFather，下面將通過(guò)和它聊天的方式來(lái)創(chuàng)建機(jī)器人，過(guò)程如下。
1）在頁(yè)面中輸入命令 /newbot 并回車，相當(dāng)于給 BotFather 發(fā)指令，表示要?jiǎng)?chuàng)建機(jī)器人。注：命令要以 / 開(kāi)頭。
2）BotFather 收到之后會(huì)將機(jī)器人創(chuàng)建好，并提示我們給機(jī)器人起一個(gè)名字，這里我起名為：古明地覺(jué)。
3）回車之后，BotFather 會(huì)繼續(xù)讓我們給機(jī)器人起一個(gè)用戶名，這個(gè)用戶名會(huì)作為機(jī)器人的唯一標(biāo)識(shí)，用于定位和查找。這里我起名為 Satori_Koishi_bot，注：用戶名必須以 Bot 或 bot 結(jié)尾。
下面來(lái)實(shí)際演示一下。
圖片
我們點(diǎn)擊 t.me/Satori_Koishi_bot，看看結(jié)果如何。
圖片
點(diǎn)擊 t.me/Satori_Koishi_bot 之后，再點(diǎn)擊屏幕中的 start（相當(dāng)于發(fā)送了一條 /start 指令），就可以和機(jī)器人聊天了。因?yàn)槲覀冞€沒(méi)有編寫代碼，來(lái)為機(jī)器人添加相應(yīng)的功能，所以目前不會(huì)有任何事情發(fā)生。
然后我們給自定義的機(jī)器人添加一些描述信息，顯然這依賴于 BotFather。向其發(fā)送 /mybots 指令，會(huì)返回我們創(chuàng)建的所有的機(jī)器人，當(dāng)然這里目前只有一個(gè)。
圖片
我們點(diǎn)擊它，看看結(jié)果：
圖片
里面提供了很多的選項(xiàng)，這里我們?cè)冱c(diǎn)擊 Edit Bot，來(lái)編輯機(jī)器人的相關(guān)信息。
圖片
不難發(fā)現(xiàn)，我們除了給當(dāng)前機(jī)器人一個(gè)名字之外，其它的信息就沒(méi)有了，所以 Telegram 提供了一系列按鈕，供我們進(jìn)行編輯。比如我們點(diǎn)擊 Edit Botpic，編輯頭像。
圖片
然后機(jī)器人的頭像會(huì)發(fā)生改變，當(dāng)然這些都屬于錦上添花的東西，最重要的是 Edit Commands，它是機(jī)器人能夠產(chǎn)生行為的核心，否則當(dāng)前的機(jī)器人就是個(gè)繡花枕頭，中看不中用。
下面我們點(diǎn)擊 Edit Commands，添加一個(gè) /help 命令。
圖片
添加格式為命令 - 描述，可同時(shí)添加多個(gè)。
圖片
目前機(jī)器人便支持了 /help 命令，另外如果點(diǎn)擊 Edit Command 之后再輸入 /empty，那么也可以將機(jī)器人現(xiàn)有的命令清空掉。
雖然 /help 命令有了，但發(fā)送這個(gè)命令之后，機(jī)器人不會(huì)有任何的反應(yīng)，因?yàn)槲覀冞€沒(méi)有給命令綁定相應(yīng)的處理函數(shù)，下面就來(lái)看看如何綁定。當(dāng)然啦，機(jī)器人不光要對(duì)命令做出反應(yīng)，就算是普通的文本、表情、圖片等消息，也應(yīng)該做出反應(yīng)。至于命令本質(zhì)上就是一個(gè)純文本，只不過(guò)它應(yīng)該以 / 開(kāi)頭。
接收消息并處理
我們可以使用 Python 連接 Telegram 機(jī)器人，為它綁定處理函數(shù)，首先需要安裝一個(gè)第三方庫(kù)。
安裝：pip3 install "python-telegram-bot[all]"
然后獲取機(jī)器人的 Token，這個(gè) Token 怎么獲取呢？
圖片
像 BotFather 發(fā)送 /mybots 命令，點(diǎn)擊指定機(jī)器人的 API Token 即可獲取。
圖片
有了這個(gè) Token 之后，就可以和機(jī)器人建立連接了。
import asyncio
import telegram
from telegram.request import HTTPXRequest
# 代理，由于不方便展示，因此我定義在了一個(gè)單獨(dú)的文件中
# 這里的 PROXY 是一個(gè)字符串，類似于 "http://username:password@ip:port"
from proxy import PROXY
BOT_API_TOKEN = "6485526535:AAEvGr9EDqtc4QPehkgohH6gczOTO5RIYRE"
async def main():
# 傳遞機(jī)器人的 Token，內(nèi)部會(huì)自動(dòng)和它建立連接
bot = telegram.Bot(
BOT_API_TOKEN,
# 指定代理
request=HTTPXRequest(proxy=PROXY),
get_updates_request=HTTPXRequest(proxy=PROXY),
)
async with bot:
# 測(cè)試連接是否成功，如果成功，會(huì)返回機(jī)器人的信息
print(await bot.get_me())
asyncio.run(main())
"""
User(api_kwargs={'has_main_web_app': False},
can_connect_to_business=False,
can_join_groups=True,
can_read_all_group_messages=False,
first_name='古明地覺(jué)',
id=6485526535,
is_bot=True,
supports_inline_queries=False,
username='Satori_Koishi_bot')
"""返回值包含了機(jī)器人的具體信息，還是比較簡(jiǎn)單的，只需指定一個(gè) Token 即可訪問(wèn)。當(dāng)然啦，由于網(wǎng)絡(luò)的原因還需要使用代理。
然后通過(guò)該模塊還可以給機(jī)器人發(fā)消息，但這顯然不是我們的重點(diǎn)，因?yàn)橄⒖隙ㄊ峭ㄟ^(guò) APP 或者瀏覽器發(fā)送的。我們要做的是，定義機(jī)器人的回復(fù)邏輯，當(dāng)用戶給它發(fā)消息時(shí)，它應(yīng)該做些什么事情。
先來(lái)一個(gè)簡(jiǎn)單的案例，當(dāng)用戶輸入 /start 命令時(shí)，回復(fù)一段文本。
from telegram import Update
from telegram.ext import ApplicationBuilder, ContextTypes, CommandHandler
from proxy import PROXY
BOT_API_TOKEN = "6485526535:AAEvGr9EDqtc4QPehkgohH6gczOTO5RIYRE"
# 定義一個(gè)處理函數(shù)
# update 封裝了用戶發(fā)送的消息數(shù)據(jù)
# context 則封裝了 Bot 對(duì)象和一些會(huì)話數(shù)據(jù)
# 這兩個(gè)對(duì)象非常重要，后面還會(huì)詳細(xì)說(shuō)
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
# context.bot 便是機(jī)器人，可以調(diào)用它的 send_message 方法回復(fù)消息
await context.bot.send_message(
# 關(guān)于 chat_id 稍后解釋
chat_id=update.message.chat.id,
# 回復(fù)的文本內(nèi)容
text="歡迎來(lái)到地靈殿"
)
# 構(gòu)建一個(gè)應(yīng)用
application = ApplicationBuilder().token(BOT_API_TOKEN).proxy(PROXY).build()
# 創(chuàng)建一個(gè) CommandHandler 實(shí)例，當(dāng)用戶輸入 /start 的時(shí)候，執(zhí)行 start 函數(shù)
start_handler = CommandHandler("start", start)
# 將 start_handler 加到應(yīng)用當(dāng)中
application.add_handler(start_handler)
# 開(kāi)啟無(wú)限循環(huán)，監(jiān)聽(tīng)事件
application.run_polling()我們來(lái)測(cè)試一下：
圖片
顯然結(jié)果是成功的，不過(guò)目前這個(gè)機(jī)器人只能處理 /start 命令，如果希望它支持更多的命令，那么就定義多個(gè) CommandHandler 即可。但是問(wèn)題來(lái)了，如果我們希望這個(gè)機(jī)器人能處理普通文本的話，該怎么辦呢？
from telegram import Update
from telegram.ext import (
ApplicationBuilder, ContextTypes,
MessageHandler, filters
)
from proxy import PROXY
BOT_API_TOKEN = "6485526535:AAEvGr9EDqtc4QPehkgohH6gczOTO5RIYRE"
async def reply(update: Update, context: ContextTypes.DEFAULT_TYPE):
await context.bot.send_message(
chat_id=update.message.chat.id,
# 通過(guò) update.message.text 可以拿到用戶發(fā)送的消息
text=f"古明地覺(jué)已收到，你發(fā)的內(nèi)容是：{update.message.text}"
)
application = ApplicationBuilder().token(BOT_API_TOKEN).proxy(PROXY).build()
# 前面使用了 CommandHandler，它專門用來(lái)處理命令，第一個(gè)參數(shù)應(yīng)該是字符串
# 比如第一個(gè)參數(shù)是 "start"，那么就給機(jī)器人增加了一個(gè)回復(fù) /start 命令的功能
# 而 MessageHandler 可以用于回復(fù)所有類型的消息，比如文本、表情、圖片、視頻等等
# 具體能回復(fù)哪些，通過(guò)第一個(gè)參數(shù)指定。這里表示只要用戶發(fā)送了文本消息，就執(zhí)行 reply 函數(shù)
reply_handler = MessageHandler(filters.TEXT, reply)
application.add_handler(reply_handler)
application.run_polling()測(cè)試一下：
圖片
結(jié)果沒(méi)有問(wèn)題，并且 /start 命令也被當(dāng)成普通的文本處理了，因?yàn)槊畋举|(zhì)上就是一個(gè)文本。然后代碼中的 filters，它里面除了有表示文本類型的 TEXT，還有很多其它類型。
# 命令
filters.COMMAND
# 普通文本（包括 emoji）
filters.TEXT
# Telegram 貼紙包中的貼紙
filters.Sticker.ALL
# 圖片文件
filters.PHOTO
# 音頻文件
filters.AUDIO
# 視頻文件
filters.VIDEO
# 文檔（例如 PDF、DOCX 等等）
filters.Document.ALL
# 語(yǔ)音（使用 Telegram 錄制的語(yǔ)音）
filters.VOICE
# 地理位置
filters.LOCATION
# 聯(lián)系人
filters.CONTACT
# 動(dòng)畫，通常是 GIF
filters.ANIMATION
# 通過(guò) Telegram 的視頻筆記功能錄制的視頻
filters.VIDEO_NOTE
# 如果希望同時(shí)支持多種類型，那么可以使用 | 進(jìn)行連接
# 比如同時(shí)支持 "文本" 和 "圖片"
filters.TEXT | filters.PHOTO
# 當(dāng)然也可以取反，~filters.TEXT 表示除了文本以外的類型
~filters.TEXT
# | 和 ~ 都出現(xiàn)了，顯然還剩下 &，而 & 也是支持的
# 我們知道命令本質(zhì)上就是一個(gè)以 / 開(kāi)頭的文本
# 如果我們希望只處理普通文本，不處理命令，該怎么辦呢？
# 很簡(jiǎn)單，像下面這樣指定即可，此時(shí)以 / 開(kāi)頭的文本（命令）會(huì)被忽略掉
filters.TEXT & ~filters.COMMAND
# 除了以上這些，filters 還支持其它類型，有興趣可以看一下
# 當(dāng)然 filters 還提供了一個(gè) ALL，表示所有類型
filters.ALL然后注意一下里面的 filters.Sticker 和 filters.Document，這兩個(gè)類型比較特殊，它們內(nèi)部還可以細(xì)分，這里我們就不細(xì)分了，直接 .ALL 即可。
我們來(lái)測(cè)試一下，看看這些類型消息都長(zhǎng)什么樣子。
from telegram import Update
from telegram.ext import (
ApplicationBuilder, ContextTypes,
MessageHandler, filters
)
from proxy import PROXY
BOT_API_TOKEN = "6485526535:AAEvGr9EDqtc4QPehkgohH6gczOTO5RIYRE"
async def get_message_type(update: Update, context: ContextTypes.DEFAULT_TYPE):
# 獲取消息
message = update.message
# 獲取消息類型
if message.text:
if message.text[0] == "/":
message_type = "filters.COMMAND"
else:
message_type = "filters.TEXT"
elif message.sticker:
message_type = "filters.Sticker"
elif message.photo:
message_type = "filters.PHOTO"
elif message.audio:
message_type = "filters.AUDIO"
elif message.video:
message_type = "filters.VIDEO"
elif message.document:
message_type = "filters.Document"
elif message.voice:
message_type = "filters.VOICE"
elif message.location:
message_type = "filters.LOCATION"
elif message.contact:
message_type = "filters.CONTACT"
elif message.animation:
message_type = "filters.ANIMATION"
elif message.video_note:
message_type = "filters.VIDEO_NOTE"
else:
message_type = "filters.<OTHER TYPE>"
await context.bot.send_message(
chat_id=update.message.chat.id,
text=f"你發(fā)送的消息的類型是 {message_type}"
)
application = ApplicationBuilder().token(BOT_API_TOKEN).proxy(PROXY).build()
reply_handler = MessageHandler(filters.ALL, get_message_type)
application.add_handler(reply_handler)
application.run_polling()我們發(fā)幾條消息，讓機(jī)器人告訴我們消息的類型。
圖片
至于其它類型，感興趣可以測(cè)試一下。
update 和 context
處理函數(shù)里面有兩個(gè)參數(shù)，分別是 update 和 context。它們非常重要，我們來(lái)打印一下，看看長(zhǎng)什么樣子。
async def reply(update: Update, context: ContextTypes.DEFAULT_TYPE):
pprint(update.to_dict())
await context.bot.send_message(chat_id=update.message.chat.id,
text="不想說(shuō)話")
application = ApplicationBuilder().token(BOT_API_TOKEN).proxy(PROXY).build()
reply_handler = MessageHandler(filters.ALL, reply)
application.add_handler(reply_handler)
application.run_polling()下面發(fā)送一條文本消息。
圖片
然后查看 update.to_dict() 的輸出是什么，為了方便理解，我將字段順序調(diào)整了一下。
{
'message': {
# 是否創(chuàng)建了頻道，因?yàn)槭撬搅?，所以?False
'channel_chat_created': False,
# 聊天照片是否已被刪除，私聊一般也為 False
'delete_chat_photo': False,
# 是否創(chuàng)建了群組，因?yàn)槭撬搅模詾?False
'group_chat_created': False,
# 是否創(chuàng)建了超級(jí)群組，因?yàn)槭撬搅?，所以?False
'supergroup_chat_created': False,
# "發(fā)送者" 發(fā)送的消息
# 因?yàn)榘l(fā)送的是文本，所以這里是 text 字段
'text': '這是一條文本消息',
# 消息發(fā)送的時(shí)間
'date': 1722623118,
# 消息的 ID
'message_id': 84,
# 消息發(fā)送者的信息
'from': {
'first_name': '小云',
'id': 6353481551,
'is_bot': False,
'language_code': 'zh-hans',
'last_name': '同學(xué)'
},
# chat 表示會(huì)話環(huán)境，機(jī)器人要通過(guò) chat 判斷消息應(yīng)該回復(fù)給誰(shuí)
# 因?yàn)槟壳笆呛蜋C(jī)器人私聊，所以機(jī)器人的回復(fù)對(duì)象就是消息的發(fā)送者
# 因此里面的 first_name、last_name、id 和消息發(fā)送者是一致的
# 但如果是群聊，那么里面的 id 字段則表示群組的 id
# 此外還會(huì)包含一個(gè) title 字段，表示群組的名稱
'chat': {
'first_name': '小云',
'last_name': '同學(xué)',
# 不管 chat 的類型是什么，里面一定會(huì)包含 id 字段
# 這個(gè) id 可能是用戶的 id，也可能是群組的 id
# 總之有了這個(gè) id，機(jī)器人就知道要將消息回復(fù)給誰(shuí)
# 所以代碼中的 send_message 方法至少要包含兩個(gè)參數(shù)
# 分別是 chat_id（發(fā)送給誰(shuí)）和 text（發(fā)送的內(nèi)容）
'id': 6353481551,
# chat 的類型，定義在 filters.ChatType 中
# ChatType.PRIVATE：私人對(duì)話
# ChatType.GROUP：普通群組聊天
# ChatType.SUPERGROUP：超級(jí)群組聊天
# ChatType.GROUPS：普通群組聊天或超級(jí)群組聊天
# ChatType.CHANNEL：頻道，用于向訂閱者廣播消息
'type': '<ChatType.PRIVATE>'
},
},
# 每發(fā)送一條消息，會(huì)話都在更新，所以 update_id 表示更新的唯一標(biāo)識(shí)符
# 用于跟蹤更新，以確保消息處理沒(méi)有丟失或重復(fù)
'update_id': 296857735
}以上就是 update.to_dict() 的輸出結(jié)果，當(dāng)用戶向 bot 發(fā)送消息時(shí)，Telegram 服務(wù)器會(huì)將這些數(shù)據(jù)以 JSON 的形式發(fā)送給當(dāng)前的應(yīng)用程序，以便 bot 可以處理和響應(yīng)這些消息。當(dāng)然啦，我們這里使用的庫(kù)會(huì)將數(shù)據(jù)封裝成 Update 對(duì)象，因此獲取數(shù)據(jù)時(shí)，可以有以下兩種獲取方式。
chat_id = update.to_dict()["message"]["chat"]["id"]
chat_id = update.message.chat.id以上是當(dāng)用戶發(fā)送文本消息時(shí)，Telegram 發(fā)送的數(shù)據(jù)，我們?cè)僭囈幌缕渌?，比如上傳一個(gè)文檔。
{
'message': {
'channel_chat_created': False,
'delete_chat_photo': False,
'group_chat_created': False,
'supergroup_chat_created': False,
'chat': {'first_name': '小云',
'id': 6353481551,
'last_name': '同學(xué)',
'type': '<ChatType.PRIVATE>'},
'date': 1722628661,
# 因?yàn)榘l(fā)送的是文檔，所以這里是 document 字段
'document': {'file_id': 'BQACAgUAAxkBAANgZq06NVL6......',
'file_name': 'OpenAI.pdf',
'file_size': 2279632,
'file_unique_id': 'AgADLw8AAn36cFU',
'mime_type': 'application/pdf',
'thumb': {
'file_id': 'AAMCBQADGQEAA2BmrTo1Uv......',
'file_size': 22533,
'file_unique_id': 'AQADLw8AAn36cFVy',
'height': 320,
'width': 243},
'thumbnail': {
'file_id': 'AAMCBQADGQEAA2BmrTo1U......',
'file_size': 22533,
'file_unique_id': 'AQADLw8AAn36cFVy',
'height': 320,
'width': 243}},
'from': {'first_name': '小云',
'id': 6353481551,
'is_bot': False,
'language_code': 'zh-hans',
'last_name': '同學(xué)'},
'message_id': 96,
},
'update_id': 296857741
}至于其它的類型也是類似的，可以自己試一下，比如上傳一段視頻，看看打印的輸出是什么。
不過(guò)還有一個(gè)問(wèn)題，就是當(dāng)用戶上傳音頻、視頻、文檔等，bot 如何獲取它們呢？顯然要依賴?yán)锩娴?file_id。
async def download(update: Update, context: ContextTypes.DEFAULT_TYPE):
document = update.message.document
file_id = document.file_id # 文件 id
file_size = document.file_size # 文件大小
file_name = document.file_name # 文件名
# 用戶上傳的文件會(huì)保存在 Telegram 服務(wù)器，我們可以基于文件 id 獲取
file_obj = await context.bot.get_file(file_id)
# file_obj.file_path 便是文件的地址，直接下載即可
with open(file_name, "wb") as f:
resp = httpx.get(file_obj.file_path, proxy=PROXY)
f.write(resp.content)
await context.bot.send_message(
chat_id=update.message.chat.id,
text=f"{file_name} 下載完畢，大小 {file_size} 字節(jié)"
)
application = ApplicationBuilder().token(BOT_API_TOKEN).proxy(PROXY).build()
download_handler = MessageHandler(filters.Document.ALL, download)
application.add_handler(download_handler)
application.run_polling()我們上傳幾個(gè)文件試試。
圖片
結(jié)果沒(méi)有問(wèn)題，用戶上傳的文件也下載到了本地。
回復(fù)富文本消息
目前機(jī)器人回復(fù)的都是普通的純文本，但也可以回復(fù)富文本消息。
async def rich_msg(update: Update, context: ContextTypes.DEFAULT_TYPE):
message = update.message
if message.text == "baidu":
text = '<a
elif message.text == "zhihu":
text = '<a
elif message.text == "bilibili":
text = '<a >點(diǎn)擊進(jìn)入 B 站頁(yè)面</a>'
else:
text = 'Unsupported Website'
await context.bot.send_message(
chat_id=update.message.chat.id,
text=text,
# 按照 HTML 進(jìn)行解析
parse_mode="HTML"
)測(cè)試一下：
圖片
結(jié)果沒(méi)有問(wèn)題，另外我們看到 a 標(biāo)簽自帶預(yù)覽功能，如果不希望預(yù)覽，那么也可以禁用掉。
圖片
將 disable_web_page_preview 參數(shù)指定為 False，即可禁用 a 標(biāo)簽的預(yù)覽功能。另外發(fā)送的消息除了可以按照 HTML 格式解析，還可以按照 Markdown 格式解析，將 parse_mode 參數(shù)指定為 "Markdown" 或者 "MarkdownV2" 即可。
回復(fù)其它類型的消息
目前機(jī)器人回復(fù)的都是文本，那么能不能回復(fù)音頻、視頻、圖片呢？顯然是可以的，并且它們還可以和文本一起返回。
# 發(fā)送圖片
await context.bot.send_photo(
chat_id=update.message.chat.id,
# 可以是路徑、句柄、bytes 對(duì)象
# 已經(jīng)上傳到 Telegram 服務(wù)器的文件會(huì)有一個(gè) file_id
# 指定 file_id 也是可以的
photo="path/to/image.jpg",
)
# 發(fā)送音頻
await context.bot.send_audio(
chat_id=update.message.chat.id,
# 可以是 路徑、句柄、bytes 對(duì)象、file_id
audio="path/to/audio.mp3"
)
# 發(fā)送視頻
await context.bot.send_video(
chat_id=update.message.chat.id,
# 可以是 路徑、句柄、bytes 對(duì)象、file_id
video="path/to/video.mp4"
)
# 發(fā)送文檔
await context.bot.send_document(
chat_id=update.message.chat.id,
# 可以是 路徑、句柄、bytes 對(duì)象、file_id
document="path/to/document.pdf"
)
# 發(fā)送語(yǔ)音
await context.bot.send_voice(
chat_id=update.message.chat.id,
# 可以是 路徑、句柄、bytes 對(duì)象、file_id
voice=r"path/to/voice.ogg",
)
# 發(fā)送位置
await context.bot.send_location(
chat_id=update.message.chat.id,
latitude=40.4750280, lnotallow=116.2676535
)
# 發(fā)送聯(lián)系人
from telegram import Contact
contact = Contact(
phone_number='+8618510286802',
first_name='芙蘭朵露',
# 以下兩個(gè)參數(shù)也可以不指定
last_name='斯卡雷特',
user_id=5783657687
)
await context.bot.send_contact(
chat_id=update.message.chat.id,
cnotallow=contact
)
# 發(fā)送貼紙
await context.bot.send_sticker(
chat_id=update.message.chat.id,
# 可以是 路徑、句柄、bytes 對(duì)象、file_id
sticker="CAACAgIAAxkBAAO5Zq5kRNKkIGZpH......"
)
# 發(fā)送 GIF
await context.bot.send_animation(
chat_id=update.message.chat.id,
# 可以是 路徑、句柄、bytes 對(duì)象、file_id
animatinotallow="CgACAgIAAxkBAAPBZq5lekVT95I......"
)除了以上這些，還可以發(fā)送其它類型的消息，不過(guò)不常用，有興趣的話可以自己看一下，這些方法都以 send_ 開(kāi)頭。然后我們來(lái)發(fā)幾條消息，測(cè)試一下。
圖片
結(jié)果沒(méi)有問(wèn)題。
媒體組
現(xiàn)在我們已經(jīng)知道如何讓機(jī)器人回復(fù)不同種類的消息了，但如果我想實(shí)現(xiàn)更復(fù)雜的功能，比如同時(shí)發(fā)送多張圖片、多個(gè)視頻，并且還配帶文字，要怎么做呢？可能有人覺(jué)得這還不簡(jiǎn)單，寫個(gè)循環(huán)不就行了，比如要發(fā)送 5 個(gè)視頻，那么調(diào)用 5 次 send_video 方法不就好了。
首先這是一種方法，但循環(huán) 5 次，那么這 5 個(gè)視頻是作為不同的消息分開(kāi)發(fā)送的。更多時(shí)候，我們是希望作為一個(gè)整體發(fā)送，那么此時(shí)可以使用媒體組功能。
from telegram import Update, InputMediaPhoto
from telegram.ext import (
ApplicationBuilder,
ContextTypes,
CommandHandler
)
from proxy import PROXY
BOT_API_TOKEN = "6485526535:AAEvGr9EDqtc4QPehkgohH6gczOTO5RIYRE"
async def send_media_group(update: Update,
context: ContextTypes.DEFAULT_TYPE):
media_group = [
# 可以是 URL、bytes 對(duì)象、文件句柄、file_id
InputMediaPhoto(open('satori1.png', "rb"), captinotallow="古"),
InputMediaPhoto(open('satori2.png', "rb"), captinotallow="明"),
InputMediaPhoto(open('satori3.png', "rb"), captinotallow="地"),
InputMediaPhoto(open('satori4.png', "rb"), captinotallow="覺(jué)")
]
# 發(fā)送媒體組
await context.bot.send_media_group(
chat_id=update.message.chat.id,
media=media_group
)
application = ApplicationBuilder().token(BOT_API_TOKEN).proxy(PROXY).build()
download_handler = CommandHandler("satori", send_media_group)
application.add_handler(download_handler)
application.run_polling()我們輸入命令 /satori，應(yīng)該會(huì)返回 4 張圖片。
圖片
結(jié)果沒(méi)有問(wèn)題，并且這 4 張圖片是整體作為一條消息發(fā)送的。然后我們?cè)诖a中還指定了一個(gè) caption 參數(shù)，它是做什么的呢？我們點(diǎn)擊一下圖片就知道了。
圖片
點(diǎn)擊圖片放大查看時(shí)，captaion 會(huì)顯示在圖片下方。另外，如果發(fā)送了多張圖片，但只有一張圖片指定了 caption 參數(shù)，那么該 caption 會(huì)和圖片一起顯示，我們舉例說(shuō)明。
async def send_media_group(update: Update,
context: ContextTypes.DEFAULT_TYPE):
caption = "+v ?(^_-) 解鎖地靈殿隱藏福利"
media_group = [
# 可以是 URL、bytes 對(duì)象、文件句柄、file_id
InputMediaPhoto(open('satori1.png', "rb")),
InputMediaPhoto(open('satori2.png', "rb")),
InputMediaPhoto(open('satori3.png', "rb"), captinotallow=caption),
InputMediaPhoto(open('satori4.png', "rb"))
]
# 發(fā)送媒體組
await context.bot.send_media_group(
chat_id=update.message.chat.id,
media=media_group
)只有一張圖片指定了 caption 參數(shù)，我們看看效果。
圖片
此時(shí)圖片會(huì)和文字一起顯示，當(dāng)然你也可以不指定 caption 參數(shù)，而是在發(fā)送完圖片之后，再調(diào)用一次 send_message。這種做法也是可以的，只不過(guò)此時(shí)圖片和文字會(huì)作為兩條消息分開(kāi)顯示。
以上是發(fā)送圖片，除了圖片之外還可以發(fā)送音頻、視頻、文檔，并且只支持這 4 種。但要注意：它們不能混在一起發(fā)，只有圖片和視頻可以，我們測(cè)試一下。
from telegram import (
Update,
InputMediaPhoto,
InputMediaAudio,
InputMediaVideo,
InputMediaDocument
)
from telegram.ext import (
ApplicationBuilder,
ContextTypes,
CommandHandler
)
from proxy import PROXY
BOT_API_TOKEN = "6485526535:AAEvGr9EDqtc4QPehkgohH6gczOTO5RIYRE"
async def send_media_group(update: Update,
context: ContextTypes.DEFAULT_TYPE):
video_caption = (
"這游戲我玩不下去了，裝備喂養(yǎng)和貼膜就算了，"
"但自定義詞條我是真忍不了，洗不出來(lái)，根本洗不出來(lái)。"
)
media_group = [
InputMediaPhoto(open("satori1.png", "rb")),
InputMediaVideo(open("DNF 裝備銷毀.mp4", "rb"),
captinotallow=video_caption),
# 也支持發(fā)送音頻和文檔，但不能混在一起
# InputMediaAudio(open("3rd eye.mp3", "rb")),
# InputMediaDocument(open('OpenAI.pdf', 'rb'))
]
# 發(fā)送媒體組
await context.bot.send_media_group(
chat_id=update.message.chat.id,
media=media_group
)
application = ApplicationBuilder().token(BOT_API_TOKEN).proxy(PROXY).build()
download_handler = CommandHandler("test_media_group", send_media_group)
application.add_handler(download_handler)
application.run_polling()測(cè)試一下：
圖片
結(jié)果正常，只是因?yàn)橐曨l和圖片是一起返回的，所以沒(méi)有預(yù)覽功能，需要點(diǎn)擊之后才會(huì)播放。并且我們只給視頻指定了 caption 參數(shù)，所以文字直接顯示在了下方，如果媒體組中有多個(gè) caption，那么就不會(huì)單獨(dú)顯示了，需要點(diǎn)擊放大之后才能看到。
當(dāng)然啦，如果你不需要同時(shí)發(fā)送多個(gè)媒體文件，那么就沒(méi)必要調(diào)用 send_media_group 方法了，直接使用之前的方法即可。
- send_photo；
- send_audio；
- send_video；
- send_document；
這些方法一次性只能發(fā)送一個(gè)媒體文件，比如發(fā)送視頻。
async def send_video(update: Update, context: ContextTypes.DEFAULT_TYPE):
video_caption = (
"這游戲我玩不下去了，裝備喂養(yǎng)和貼膜就算了，"
"但自定義詞條我是真忍不了，洗不出來(lái)，根本洗不出來(lái)。"
)
await context.bot.send_video(
chat_id=update.message.chat.id,
video="DNF 裝備銷毀.mp4",
captinotallow=video_caption,
# 讓 caption 顯示在上方，默認(rèn)顯示在下方
show_caption_above_media=True,
)
application = ApplicationBuilder().token(BOT_API_TOKEN).proxy(PROXY).build()
download_handler = CommandHandler("destroy", send_video)
application.add_handler(download_handler)
application.run_polling()測(cè)試一下：
圖片
怎么樣，是不是很有趣呢？另外 caption 還可以是富文本，只需將 parse_mode 參數(shù)指定為 "HTML"、"Markdown" 或 "MarkdownV2" 即可。
關(guān)于機(jī)器人如何回復(fù)不同種類的消息，以及同時(shí)回復(fù)多條消息，相關(guān)內(nèi)容我們就說(shuō)完了。有了這些功能，我們的機(jī)器人就已經(jīng)很強(qiáng)大了，你也可以把它和公司的業(yè)務(wù)結(jié)合起來(lái)。
比如創(chuàng)建一個(gè)命令：/get，它的功能如下。
圖片
然后在代碼中添加一個(gè) CommandHandler("get", get_table)，便可讓用戶通過(guò) Telegram 查詢數(shù)據(jù)庫(kù)表，當(dāng)然這里只是打個(gè)比方，具體怎么做取決于你的想法。另外多說(shuō)一句，如果你希望輸入 / 之后能像上面那樣有提示，那么需要通過(guò) BotFather 進(jìn)行設(shè)置。
圖片
要強(qiáng)調(diào)的是，這種方式只是起到一個(gè)提示作用，提示機(jī)器人支持 /get 命令。但機(jī)器人實(shí)際上是否支持，取決于代碼中是否為機(jī)器人實(shí)現(xiàn)了 /get。所以當(dāng)我們?cè)诖a中為機(jī)器人添加完命令之后，可以再通過(guò) Edit Commands 進(jìn)行設(shè)置，這樣當(dāng)用戶輸入 / 之后，機(jī)器人有哪些命令以及描述都會(huì)顯示出來(lái)。
當(dāng)然啦，如果你不通過(guò) Edit Commands 進(jìn)行設(shè)置的話，也是可以的，只是用戶輸入 / 之后不會(huì)有提示罷了，但命令是會(huì)回復(fù)的，只要在代碼中實(shí)現(xiàn)了。同理，如果通過(guò) Edit Commands 設(shè)置了，但代碼中沒(méi)實(shí)現(xiàn)，那么該命令也不會(huì)有效果。
自定義按鈕
雖然目前的機(jī)器人已經(jīng)很強(qiáng)大了，但是還不夠，我們看一下 BotFather。
圖片
你會(huì)發(fā)現(xiàn)它下面帶了很多的按鈕，點(diǎn)擊按鈕之后會(huì)執(zhí)行相應(yīng)的邏輯，那我們要怎么實(shí)現(xiàn)這些按鈕呢？
from telegram import (
Update,
InlineKeyboardMarkup,
InlineKeyboardButton,
)
from telegram.ext import (
ApplicationBuilder,
ContextTypes,
CommandHandler
)
from proxy import PROXY
BOT_API_TOKEN = "6485526535:AAEvGr9EDqtc4QPehkgohH6gczOTO5RIYRE"
async def add_button(update: Update, context: ContextTypes.DEFAULT_TYPE):
text = "作為<i>程序猿</i>，你最喜歡哪種編程語(yǔ)言呢？"
# 設(shè)置按鈕
reply_markup = InlineKeyboardMarkup([
# 第一行
[InlineKeyboardButton(text="Python", url="https://www.python.org")],
# 第二行
[InlineKeyboardButton(text="Golang", url="https://golang.org")],
# 第三行
[InlineKeyboardButton(text="Rust", url="https://www.rust-lang.org")],
# 第四行
[InlineKeyboardButton(text="Zig", url="https://ziglang.org")],
])
await context.bot.send_message(
chat_id=update.message.chat.id,
text=text,
parse_mode="HTML",
reply_markup=reply_markup
)
application = ApplicationBuilder().token(BOT_API_TOKEN).proxy(PROXY).build()
download_handler = CommandHandler("language", add_button)
application.add_handler(download_handler)
application.run_polling()測(cè)試一下：
圖片
此時(shí)按鈕就實(shí)現(xiàn)了，由于在 InlineKeyboardButton 里面指定的是 url，所以這是跳轉(zhuǎn)按鈕，點(diǎn)擊之后會(huì)打開(kāi)指定的頁(yè)面。并且按鈕的右上角還有一個(gè)小箭頭，表示按鈕是跳轉(zhuǎn)按鈕。
但除了跳轉(zhuǎn)按鈕之外，還有回調(diào)按鈕，也就是點(diǎn)擊按鈕之后會(huì)執(zhí)行回調(diào)函數(shù)，我們舉例說(shuō)明。
from telegram import (
Update,
InlineKeyboardMarkup,
InlineKeyboardButton,
)
from telegram.ext import (
ApplicationBuilder,
ContextTypes,
CommandHandler,
CallbackQueryHandler,
)
from proxy import PROXY
BOT_API_TOKEN = "6485526535:AAEvGr9EDqtc4QPehkgohH6gczOTO5RIYRE"
async def add_button(update: Update, context: ContextTypes.DEFAULT_TYPE):
text = "o(╥﹏╥)o??╭(╯^╰)╮"
# 設(shè)置按鈕
reply_markup = InlineKeyboardMarkup([
# 第一行，兩個(gè)跳轉(zhuǎn)按鈕
[InlineKeyboardButton(text="百度", url="https://www.baidu.com"),
InlineKeyboardButton(text="谷歌", url="https://www.google.com"),],
# 第二行，兩個(gè)回調(diào)按鈕
[InlineKeyboardButton(text="油管", callback_data="youtube"),
InlineKeyboardButton(text="B站", callback_data="bilibili"),],
])
await context.bot.send_message(
chat_id=update.message.chat.id,
text=text,
reply_markup=reply_markup
)
async def callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
# 當(dāng)點(diǎn)擊回調(diào)按鈕時(shí)，會(huì)執(zhí)行相應(yīng)的回調(diào)函數(shù)
cb_data = update.callback_query.data # 回調(diào)按鈕中指定的 callback_data
if cb_data == "youtube":
text = "歡迎來(lái)到油管"
elif cb_data == "bilibili":
text = "歡迎來(lái)到 B 站"
else:
text = "Unknown Website"
await context.bot.send_message(
# 注意：這里是 update.callback_query.message.chat.id
chat_id=update.callback_query.message.chat.id,
text=text
)
application = ApplicationBuilder().token(BOT_API_TOKEN).proxy(PROXY).build()
# 添加 Handler
application.add_handler(
CommandHandler("website", add_button)
)
# 處理回調(diào)的 Handler，否則點(diǎn)擊按鈕不會(huì)有效果
application.add_handler(
CallbackQueryHandler(callback)
)
application.run_polling()測(cè)試一下效果：
圖片
點(diǎn)擊油管和 B站的時(shí)候會(huì)執(zhí)行回調(diào)函數(shù)，結(jié)果沒(méi)有問(wèn)題。但是我們發(fā)現(xiàn)，這些文字是單獨(dú)發(fā)送的，那可不可以本地修改呢，也就是將按鈕上方的文字替換掉。答案是可以的，我們來(lái)測(cè)試一下。
from telegram import (
Update,
InlineKeyboardMarkup,
InlineKeyboardButton,
)
from telegram.ext import (
ApplicationBuilder,
ContextTypes,
CommandHandler,
CallbackQueryHandler,
)
from proxy import PROXY
BOT_API_TOKEN = "6485526535:AAEvGr9EDqtc4QPehkgohH6gczOTO5RIYRE"
def get_reply_markup():
reply_markup = InlineKeyboardMarkup([
[InlineKeyboardButton(text="古明地覺(jué)", callback_data="satori")],
[InlineKeyboardButton(text="古明地戀", callback_data="koishi")],
[InlineKeyboardButton(text="霧雨魔理沙", callback_data="marisa")],
[InlineKeyboardButton(text="琪露諾", callback_data="cirno")],
])
return reply_markup
async def add_button(update: Update, context: ContextTypes.DEFAULT_TYPE):
text = "點(diǎn)擊想要攻略的角色"
await context.bot.send_message(
chat_id=update.message.chat.id,
text=text,
reply_markup=get_reply_markup()
)
async def callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
cb_data = update.callback_query.data
if cb_data == "satori":
img = "你將要攻略古明地覺(jué)"
elif cb_data == "koishi":
img = "你將要攻略古明地戀"
elif cb_data == "marisa":
img = "你將要攻略霧雨魔理沙"
elif cb_data == "cirno":
img = "你將要攻略琪露諾"
else:
raise RuntimeError("Unreachable")
# 點(diǎn)擊按鈕之后，要對(duì)上方的文字進(jìn)行修改，替換成其它內(nèi)容
# 所以這相當(dāng)于編輯已有消息，既然要編輯，那么除了 chat_id 之外還要指定 message_id
# 因?yàn)槭腔卣{(diào)，所以要多調(diào)用一次 callback_query
message_id = update.callback_query.message.message_id
chat_id = update.callback_query.message.chat.id
# 調(diào)用 edit_message_media 方法，編輯消息
await context.bot.edit_message_text(
text=img,
chat_id=chat_id,
message_id=message_id,
reply_markup=get_reply_markup()
)
application = ApplicationBuilder().token(BOT_API_TOKEN).proxy(PROXY).build()
application.add_handler(
CommandHandler("gogogo", add_button)
)
application.add_handler(
CallbackQueryHandler(callback)
)
application.run_polling()測(cè)試一下：
圖片
然后點(diǎn)擊按鈕，看看文字內(nèi)容有沒(méi)有發(fā)生改變。
圖片
點(diǎn)擊按鈕，文字的內(nèi)容被替換了。所以當(dāng)機(jī)器人回復(fù)一條消息時(shí)，只需知道 chat_id 即可。但如果是修改某條消息，那么除了 chat_id 之外，還要知道 message_id。
修改文字調(diào)用的方法是 edit_message_text，但除了修改文字之外，還可以修改其它內(nèi)容。
圖片
比如修改媒體文件，修改媒體文件的 caption，修改按鈕等等。
修改消息綜合案例
關(guān)于修改消息我們已經(jīng)知道怎么做了，下面來(lái)做一個(gè)綜合案例。假設(shè)當(dāng)前有 N 張圖片，用戶默認(rèn)會(huì)看到第一張，然后點(diǎn)擊按鈕可以查看下一張圖片，當(dāng)然也可以查看上一張。那么這個(gè)需求怎么實(shí)現(xiàn)呢？
from telegram import (
Update,
InlineKeyboardMarkup,
InlineKeyboardButton,
InputMediaPhoto
)
from telegram.ext import (
ApplicationBuilder,
ContextTypes,
CommandHandler,
CallbackQueryHandler,
)
from proxy import PROXY
BOT_API_TOKEN = "6485526535:AAEvGr9EDqtc4QPehkgohH6gczOTO5RIYRE"
# 這里我就用 4 張圖片為例
IMAGES = ["satori.png", "koishi.png", "marisa.png", "cirno.png"]
def get_navigation_buttons(index):
reply_markup = InlineKeyboardMarkup([
[InlineKeyboardButton(text="上一張", callback_data=f"prev:{index}"),
InlineKeyboardButton(text="下一張", callback_data=f"next:{index}")],
])
return reply_markup
async def get_pic(update: Update, context: ContextTypes.DEFAULT_TYPE):
# 默認(rèn)發(fā)送第一張圖片
await context.bot.send_photo(
chat_id=update.message.chat.id,
photo=IMAGES[0],
captinotallow=f"正在瀏覽第 1 / {len(IMAGES)} 張圖片",
reply_markup=get_navigation_buttons(0)
)
async def callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
# 點(diǎn)擊按鈕，觸發(fā)回調(diào)
op, index = update.callback_query.data.split(":")
if op == "prev":
index = (int(index) - 1) % len(IMAGES)
else: # op == "next"
index = (int(index) + 1) % len(IMAGES)
# int(index) 減 1 和加 1 之后，就是上一張圖片和下一張圖片的索引
# 但這里又對(duì) len(IMAGES) 進(jìn)行取模，主要是為了實(shí)現(xiàn)循環(huán)瀏覽
# 比如第一張的上一張會(huì)返回最后一張，最后一張的下一張會(huì)返回第一張
await context.bot.edit_message_media(
chat_id=update.callback_query.message.chat.id,
message_id=update.callback_query.message.message_id,
media=InputMediaPhoto(
open(IMAGES[index], "rb"),
captinotallow=f"正在瀏覽第 {index + 1} / {len(IMAGES)} 張圖片"
),
reply_markup=get_navigation_buttons(index)
)
application = ApplicationBuilder().token(BOT_API_TOKEN).proxy(PROXY).build()
application.add_handler(
CommandHandler("get_pic", get_pic)
)
application.add_handler(
CallbackQueryHandler(callback)
)
application.run_polling()測(cè)試一下：
圖片
此時(shí)點(diǎn)擊按鈕下一張，就會(huì)返回下一張圖片，同理也可以返回上一張圖片。如果已經(jīng)是最后一張圖片了，那么點(diǎn)擊下一張，會(huì)返回第一張圖片。
但問(wèn)題來(lái)了，程序要如何得知用戶正在瀏覽的是第幾張圖片呢？顯然要借助于按鈕。在創(chuàng)建按鈕時(shí)，參數(shù) callback_data 里面保存了 index，當(dāng)點(diǎn)擊下一張或上一張時(shí)，更新 index，返回新的圖片，同時(shí)刷新按鈕。
以上返回的是圖片，你也可以換成視頻，并增加一些點(diǎn)贊、是否喜歡等按鈕。
小結(jié)
以上就是 Python 操作 Telegram 相關(guān)的內(nèi)容，當(dāng)然這里只介紹了一部分，還有一些更復(fù)雜的功能沒(méi)有說(shuō)，比如按鈕的嵌套等等。另外目前是用戶和機(jī)器人一對(duì)一私聊，但我們還可以創(chuàng)建一個(gè)組，讓機(jī)器人回復(fù)組成員的消息。而關(guān)于這些內(nèi)容，后續(xù)有空補(bǔ)上，本文就先到這兒，寫的有點(diǎn)累了。
