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