摘要：而的異步非阻塞特性能夠完美的解決這一問(wèn)題。爬蟲機(jī)器人功能實(shí)現(xiàn)我使用編寫的機(jī)器人是用來(lái)抓取來(lái)自游民星空的圖片。也是使用裝飾器進(jìn)行回調(diào)函數(shù)注冊(cè)，使用進(jìn)行消息更新。當(dāng)沒有指令時(shí)，會(huì)顯示一些能夠查看的圖片類型。

原文鏈接

aiotg 可以通過(guò)異步調(diào)用telegram api的方式來(lái)構(gòu)建bot，因?yàn)闆Q定開發(fā)一個(gè)爬蟲功能的bot，所以網(wǎng)絡(luò)請(qǐng)求阻塞是比較嚴(yán)重的性能障礙。而asyncio的異步非阻塞特性能夠完美的解決這一問(wèn)題。這篇文章在記錄如何使用aiotg進(jìn)行telegram開發(fā)的同時(shí)，也會(huì)說(shuō)明一些aiohttp的使用方法,這里是項(xiàng)目源碼。如果你覺得不錯(cuò)可以幫忙點(diǎn)一下star

https://t.me/fpicturebot 點(diǎn)擊鏈接可以體驗(yàn)一下這個(gè)bot的功能。

如果讀者之前對(duì)telegram的bot沒有了解，可以查看這篇寫給開發(fā)者的telegram-bots介紹文檔

你可以先學(xué)習(xí)如何新建一個(gè)機(jī)器人

from aiotg import Bot, Chat

config = {
    "api_token": "***********",
    "proxy": "http://127.0.0.1:8118"
}

bot = Bot(**config)

@bot.command(r"/echo (.+)")
def echo(chat: Chat, match):
    return chat.reply(match.group(1))

bot.run()

上面是一個(gè)簡(jiǎn)單的回復(fù)機(jī)器人，當(dāng)你使用指令 /echo+內(nèi)容時(shí)，機(jī)器人會(huì)自動(dòng)回復(fù)給你發(fā)送的內(nèi)容。這里要注意一點(diǎn)，在我這里沒有采用使用 pipenv ( pip ) 安裝aiotg的方法，因?yàn)閜ip平臺(tái)上安裝的是master分支的包，aiotg通過(guò)使用aiohttp來(lái)調(diào)用telegram bot api，在創(chuàng)建一個(gè)bot的時(shí)候沒有提供proxy選項(xiàng)為aiohttp設(shè)置代理，在本地開發(fā)的時(shí)候會(huì)因?yàn)閲?guó)內(nèi)網(wǎng)絡(luò)抽搐出現(xiàn)網(wǎng)絡(luò)連接錯(cuò)誤，所以在這里我使用了aiotg的prxoy分支，直接從github上下載的代碼。在創(chuàng)建Bot對(duì)象的時(shí)候加入proxy選項(xiàng)就能使用本地代理來(lái)進(jìn)行開發(fā)調(diào)試了。

后來(lái)我在aiotg telegram群里建議作者將proxy合并到主分支，后來(lái)作者表示他也有這樣的想法，同時(shí)他也吐槽了一下俄羅斯的網(wǎng)絡(luò)也有很多審查和限制，現(xiàn)在在aiotg里已經(jīng)沒有proxy分支了，在aiotg-0.9.9版本中提供proxy選項(xiàng)，所以大家可以繼續(xù)使用pipenv下載aiotg包。

既然用到aiotg來(lái)開發(fā)就是看中了他的異步特性，下面就列出一個(gè)簡(jiǎn)單的例子

import aiohttp
import json
from aiotg import Bot, Chat

with open("token.conf") as f:
    token = json.loads(f.read())

bot = Bot(**token)

@bot.command("/fetch")
async def async_fecth(chat: Chat, match):
    url = "http://www.gamersky.com/ent/111/"
    async with aiohttp.ClientSession() as sesssion:
        async with sesssion.get(url) as resp:
            info = " version: {}
 status :{}
 method: {}
 url: {}
".format(
                resp.version, resp.status, resp.method, resp.url)
            await chat.send_text(info)

bot.run(debug=True)

關(guān)于自定義鍵盤的內(nèi)容可以點(diǎn)擊鏈接查看官方解釋，這里是簡(jiǎn)單的中文描述。

category.json

[
    {
        "name": "dynamic",
        "title": "動(dòng)態(tài)圖",
        "url": "http://www.gamersky.com/ent/111/"
    },
    {
        "name": "oops",
        "title": "囧圖",
        "url": "http://www.gamersky.com/ent/147/"
    },
    {
        "name": "beauty",
        "title": "福利圖",
        "url": "http://tag.gamersky.com/news/66.html"
    },
    {
        "name": "easy-moment",
        "title": "輕松一刻",
        "url": "http://www.gamersky.com/ent/20503/"
    },
    {
        "name": "trivia",
        "title": "冷知識(shí)",
        "url": "http://www.gamersky.com/ent/198/"
    },
    {
        "name": "cold-tucao",
        "title": "冷吐槽",
        "url": "http://www.gamersky.com/ent/20108/"
    }
]

main.py

import aiohttp
import json
from aiotg import Bot, Chat

with open("token.json") as t, open("category.json") as c:
    token = json.loads(t.read())
    category = json.loads(c.read())

bot = Bot(**token)

@bot.command("/reply")
async def resply(chat: Chat, match):
    kb = [[item["title"]] for item in category]
    keyboard = {
        "keyboard": kb,
        "resize_keyboard": True
    }
    await chat.send_text(text="看看你的鍵盤", reply_markup=json.dumps(keyboard))

bot.run(debug=True)

只需要在調(diào)用chat的發(fā)送消息函數(shù)中，指定 reply_markup 參數(shù)，你就能個(gè)性化的設(shè)定用戶鍵盤， reply_markup 參數(shù)需要一個(gè)json對(duì)象，官方指定為ReplyKeyboardMarkup類型，其中keyboard需要傳遞一個(gè)KeyboardButton的數(shù)組。

每個(gè)keyboard的成員代表著鍵盤中的行，你可以通過(guò)修改每行中KeyboardButton的個(gè)數(shù)來(lái)排列你的鍵盤，比如我們讓鍵盤每行顯示兩個(gè)KeyboardButton，如下所示

@bot.command("/reply")
async def reply(chat: Chat, match):
    # kb = [[item["title"]] for item in category]
    kb, row = [], -1
    for idx, item in enumerate(category):
        if idx % 2 == 0:
            kb.append([])
            row += 1
        kb[row].append(item["title"])
    keyboard = {
        "keyboard": kb,
        "resize_keyboard": True
    }
    await chat.send_text(text="看看你的鍵盤", reply_markup=json.dumps(keyboard))

內(nèi)聯(lián)鍵盤的意思就是附著在消息上的鍵盤，內(nèi)聯(lián)鍵盤由內(nèi)聯(lián)按鈕組成，每個(gè)按鈕會(huì)附帶一個(gè)回調(diào)數(shù)據(jù)，每次點(diǎn)擊按鈕之后會(huì)有對(duì)應(yīng)的回調(diào)函數(shù)處理。

@bot.command("/inline")
async def inlinekeyboard(chat: Chat, match):

    inlinekeyboardmarkup = {
            "type": "InlineKeyboardMarkup",
            "inline_keyboard": [
                [{"type": "InlineKeyboardButton",
                  "text": "上一頁(yè)",
                  "callback_data": "page-pre"},
                 {"type": "InlineKeyboardButton",
                  "text": "下一頁(yè)",
                  "callback_data": "page-next"}]
                ]
            }

    await chat.send_text("請(qǐng)翻頁(yè)", reply_markup=json.dumps(inlinekeyboardmarkup))

@bot.callback(r"page-(w+)")
async def buttonclick(chat, cq, match):
    await chat.send_text("You clicked {}".format(match.group(1)))

有時(shí)候我們想修改之前已經(jīng)發(fā)送過(guò)的消息內(nèi)容，例如當(dāng)用戶點(diǎn)擊了內(nèi)聯(lián)鍵盤，而鍵盤的功能是進(jìn)行翻頁(yè)更新消息的內(nèi)容。這時(shí)候我們可以使用 editMessageText 功能。例如點(diǎn)擊上面內(nèi)聯(lián)鍵盤中的上一頁(yè)按鈕，你可以看到消息的內(nèi)容被更改了。

@bot.callback(r"page-(w+)")
async def buttonclick(chat, cq, match):
    await chat.edit_text(message_id=chat.message["message_id"], text="消息被修改了")

內(nèi)聯(lián)請(qǐng)求模式感覺更適合在群組中使用，在討論組中輸入@botname + 特定指令，輸入框的上方就會(huì)顯示查詢內(nèi)容，你可以返回給用戶文章類型、圖片類型或者其他類型的查詢信息。官網(wǎng)有更詳細(xì)的內(nèi)容。

@bot.inline
async def inlinemode(iq):
    results = [{
            "type": "article",
            "id": str(index),
            "title": article["title"],
            "input_message_content": { "message_text": article["title"]},
            "description": f"這里是{article["title"]}"
        } for index, article in enumerate(category)]
    await iq.answer(results)

我們?cè)O(shè)定當(dāng)用戶輸入內(nèi)聯(lián)指令時(shí)，bot返回可以選擇的圖片種類，返回結(jié)果的類型是article類型，官方還提供了語(yǔ)音，圖片，gif，視頻，音頻。表情等類型，你可以根據(jù)自己的需要進(jìn)行選擇。

我使用aiotg編寫的機(jī)器人是用來(lái)抓取來(lái)自游民星空的圖片。

爬蟲功能的實(shí)現(xiàn)是用aiohttp發(fā)送web請(qǐng)求，使用beautifulsoup進(jìn)行html解析，核心代碼如下

import re
import aiohttp
from bs4 import BeautifulSoup


def aioget(url):
    return aiohttp.request("GET", url)


def filter_img(tag):
    if tag.name != "p":
        return False
    try:
        if tag.attrs["align"] == "center":
            for child in tag.contents:
                if child.name == "img" or child.name == "a":
                    return True
        return False
    except KeyError:
        if "style" in tag.attrs:
            return True
        else:
            return False
            
            
async def fetch_img(url):
    async with aioget(url) as resp:
        resp_text = await resp.text()
        content = BeautifulSoup(resp_text, "lxml")
        imgs = content.find(class_="Mid2L_con").findAll(filter_img)
        results = []
        for img in imgs:
            try:
                results.append({
                    "src":  img.find("img").attrs["src"],
                    "desc": "
".join(list(img.stripped_strings))
                })
            except AttributeError:
                continue
        return results

我將aiohttp的get請(qǐng)求稍微包裝了一下，簡(jiǎn)潔一些。html中元素的提取就不在贅述，就是找找html中的規(guī)律

指令功能實(shí)現(xiàn)需要使用aiotg.bot.command裝飾器進(jìn)行命令注冊(cè)，下面列出 /start的功能實(shí)現(xiàn)

@bot.command(r"/start")
async def list_category(chat: Chat, match):
    kb, row = [], -1
    for idx, item in enumerate(category["name"]):
        if idx % 2 == 0:
            kb.append([])
            row += 1
        kb[row].append(item)
    keyboard = {
        "keyboard": kb,
        "resize_keyboard": True
    }
    await chat.send_text(text="請(qǐng)選擇你喜歡的圖片類型", reply_markup=json.dumps(keyboard))

關(guān)于自定義鍵盤部分在上文中已經(jīng)講過(guò)，讀者可以自己編碼實(shí)現(xiàn)

讀者可以看到在消息上附有頁(yè)面切換按鈕，每個(gè)按鈕會(huì)帶著一個(gè)callbackdata，當(dāng)點(diǎn)擊按鈕會(huì)調(diào)用相應(yīng)的callback函數(shù)進(jìn)行處理，這里點(diǎn)擊下一頁(yè)時(shí)會(huì)進(jìn)行翻頁(yè)。

看頁(yè)面更新了，關(guān)于更新頁(yè)面的實(shí)現(xiàn)在上面也講到了如何進(jìn)行消息更新。

@bot.callback(r"page-(?Pw+)-(?Pd+)")
async def change_lists(chat: Chat, cq, match):
    req_name = match.group("name")
    page = match.group("page")
    url = category[req_name]
    text, markup = await format_message(req_name, url, page)
    await chat.edit_text(message_id=chat.message["message_id"], text=text, markup=markup)

也是使用裝飾器進(jìn)行回調(diào)函數(shù)注冊(cè)，使用chat.edit_text進(jìn)行消息更新。

callback功能也用在了圖片的更新。點(diǎn)擊下一頁(yè)更新圖片。

當(dāng)用戶在輸入框中輸入@botusername+指令時(shí)，會(huì)在輸入框上顯示查詢內(nèi)容。

當(dāng)沒有指令時(shí)，會(huì)顯示一些能夠查看的圖片類型。

當(dāng)輸入對(duì)應(yīng)類型漢字的前幾個(gè)字時(shí)，bot會(huì)匹配你想看的圖片列表，并羅列出來(lái)

@bot.inline(r"([u4e00-u9fa5]+)")
async def inline_name(iq, match):
    req = match.group(1)
    req_name = match_category(req.strip(), category["name"])
    ptype = "G" if req_name == "dynamic" else "P"
    if req_name is None:
        return
    results, _ = await fetch_lists(category[req_name])
    c_results = [{
            "type": "article",
            "id": str(index),
            "title": item["title"],
            "input_message_content": {
                "message_text": "/" + ptype + item["date"] + "_" + item["key"]
            },
            "description": item["desc"]
        } for index, item in enumerate(results)]
    await iq.answer(c_results)

當(dāng)發(fā)送給用戶圖片時(shí)，telegram會(huì)返回一個(gè)和圖片對(duì)應(yīng)的file_id, 當(dāng)再次發(fā)送相同的圖片時(shí)，只需要在調(diào)用send_photo時(shí)，將photo參數(shù)賦值為file_id即可，所以每次使用爬蟲進(jìn)行抓取圖片時(shí)，將圖片的fild_id存在redis中，用戶請(qǐng)求圖片時(shí)，如果圖片之前已經(jīng)抓取過(guò)，這時(shí)候只要從redis中取出file_id，再調(diào)用send_photo即可。