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

原文鏈接

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

https://t.me/fpicturebot 點擊鏈接可以體驗一下這個bot的功能。

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

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

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()

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

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

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

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)容可以點擊鏈接查看官方解釋，這里是簡單的中文描述。

category.json

[
    {
        "name": "dynamic",
        "title": "動態(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": "冷知識",
        "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ù)，你就能個性化的設(shè)定用戶鍵盤， reply_markup 參數(shù)需要一個json對象，官方指定為ReplyKeyboardMarkup類型，其中keyboard需要傳遞一個KeyboardButton的數(shù)組。

每個keyboard的成員代表著鍵盤中的行，你可以通過修改每行中KeyboardButton的個數(shù)來排列你的鍵盤，比如我們讓鍵盤每行顯示兩個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)按鈕組成，每個按鈕會附帶一個回調(diào)數(shù)據(jù)，每次點擊按鈕之后會有對應(yīng)的回調(diào)函數(shù)處理。

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

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

    await chat.send_text("請翻頁", 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)))

有時候我們想修改之前已經(jīng)發(fā)送過的消息內(nèi)容，例如當(dāng)用戶點擊了內(nèi)聯(lián)鍵盤，而鍵盤的功能是進(jìn)行翻頁更新消息的內(nèi)容。這時候我們可以使用 editMessageText 功能。例如點擊上面內(nèi)聯(lián)鍵盤中的上一頁按鈕，你可以看到消息的內(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)請求模式感覺更適合在群組中使用，在討論組中輸入@botname + 特定指令，輸入框的上方就會顯示查詢內(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)

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

我使用aiotg編寫的機器人是用來抓取來自游民星空的圖片。

爬蟲功能的實現(xiàn)是用aiohttp發(fā)送web請求，使用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請求稍微包裝了一下，簡潔一些。html中元素的提取就不在贅述，就是找找html中的規(guī)律

指令功能實現(xiàn)需要使用aiotg.bot.command裝飾器進(jìn)行命令注冊，下面列出 /start的功能實現(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="請選擇你喜歡的圖片類型", reply_markup=json.dumps(keyboard))

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

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

看頁面更新了，關(guān)于更新頁面的實現(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ù)注冊，使用chat.edit_text進(jìn)行消息更新。

callback功能也用在了圖片的更新。點擊下一頁更新圖片。

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

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

當(dāng)輸入對應(yīng)類型漢字的前幾個字時，bot會匹配你想看的圖片列表，并羅列出來

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