我給落格輸入法的用戶群添加了個自動回復機器人

最近更改：4月9日, 2022

如題圖，我給落格輸入法的用戶群弄了個機器人，隨著落格輸入法的用戶越來越多，一些慕名而來的新手也多了。很多常見問題重複提問，搞得人焦頭爛額，如果能有個機器人，就像 Siri 那樣，讓它自動捕捉那些關鍵字然後回复這些用戶，豈不美哉？這樣用戶能夠得到精心編輯的答案，而我也能空餘出更多的時間去寫wan代you碼xi。

當然，這樣的機器人我是見過的，所以我想一定有現成的東西，顯然，搜索之下我先找到了這個：HTTPS://github.com/zeruniverse/QQRobot 是作者實驗學習做的一個 QQ 小黃雞，基於 Python，簡單方便，我看中它的地方在於能在 linux 上跑，也就是可以跑在我的 vps 上。

不過這個用起來不太方便，它是通過二維碼掃描來登錄的，而我的vps又沒有圖形端……所以我還得把那個二維碼下載下來，太費事了。後來我通過它的說明鏈接，找到了這個 https://github.com/zeruniverse/QBotWebWrap

他集成了上文所說的 QQRobot，可以在網頁端控制啦！不過可惜的是，配置很複雜，php插件要求的也多，重點是並不穩定，幾乎一上線就掉了。

原理

琢磨了這麼一通之後，我也明白了這些挂機工具的原理，其實就是大家破解了 web QQ 的協議（就是那個 smart qq），雖然高級功能做不到，但聊天的的功能還是妥妥的了，比如能夠監控消息，然後發送給第三方的人工智能接口獲得反饋再給對方發回去這樣。這就是智能聊天機器人的原理啦！

後來，我又找到了一個比較活躍的項目：HTTPS://github.com/Yinzo/SmartQQBot

這個的原理與上文的 QQRobot 一樣，但一直有人維護，也很活躍，項目穩定。沒什麼說的，根據需求安裝包後就能啟動，記得防火牆開 8888 端口，然後可以用蟒蛇跑.PY --沒有-gui --HTTP 啟動，如此就可以了，然後打開vps對應的頁面，同樣掃碼登錄。

定制

SmartQQBot 自帶了幾個插件比如天氣以及一個群內可調戲的自動回复插件，你可以用固定的語法教它，然後就會回复你這樣，這雖然能夠稍微達到我的目的，但還不夠。作為一個專門的問答機器人，怎麼能讓用戶隨便地增刪呢對吧？ SmartQQBot 的另外一個好處是以插件形式來增加功能的，它自帶了幾個插件，比如 Satoru ，我們不用這個，把它關掉。項目介紹裡沒有說明，其實還自帶了一個叫做 tucao 的插件，這個插件與 Satoru 不同的是，它支持在用戶發言的語句中匹配關鍵詞而不是全文匹配！這一點很不錯，如果你願意，可以做到很自然地回復不是嗎？

所以，我們來改這個，SmartQQBot 支持三種消息標籤：

from smart_qq_bot.signals import (

on_all_message, //包括了群消息和私聊消息

on_group_message, //群消息

on_private_message, //私聊消息

)

通過給你的函數賦予不同的標籤，可以讓你的插件對不同的消息作出響應。

首先是改改消息收發，我把群消息的權限減少，不再檢測控制語法，只做回复，同時改為也接受私聊，這有我就可以自己做測試了。所以代碼大概是這樣：

@on_all_message(name='loginput[学习遗忘]')

def loginput(msg, bot):

global core

reply = bot.reply_msg(msg, return_function=True)

group_id = "16788008"

core.load(group_id)

for key in list(core.loginput_dict[group_id].keys()):

if str(key) in msg.content and core.loginput_dict[group_id][key]:

logger.info("RUNTIMELOG loginput pattern detected, replying...")

reply(random.choice(core.loginput_dict[group_id][key]))

return True

return False

由於 tucao 這個插件是針對不同群的群號做處理的，這裡我直接寫成了固定的落格輸入法用戶群的群號，不隱藏了，大家隨便看：）

這裡要注意，雖然本身程序傳進來的群號是 int ，但 tucao 都是按照 str 處理的，所以這裡要寫成字符串。

然後是私聊的控制功能，這裡我除了修改了 tucao 自帶的顯示列表功能外（列表大了根本發不了，太長，我改為了顯示數量），還加入了一個重新加載的功能，這樣就方便我更改後台數據統一添加內容了：

@on_private_message(name='loginput[管理]')

def current_loginput_list(msg, bot):

# webqq接受的消息会以空格结尾

global core

reply = bot.reply_msg(msg, return_function=True)

group_id = "16788008"

match = re.match(r'^(?:!|！)(learn|delete|list|reload)(?:\s?){(.+)}(?:\s?){(.+)}', msg.content)

if match:

core.load(group_id)

logger.info("RUNTIMELOG loginput command detected.")

command = str(match.group(1)).decode('utf8')

key = str(match.group(2)).decode('utf8')

value = str(match.group(3)).decode('utf8')

if command == 'learn':

if group_id not in core.loginput_dict:

core.load(group_id)

if key in core.loginput_dict[group_id] and value not in core.loginput_dict[group_id][key]:

core.loginput_dict[group_id][key].append(value)

else:

core.loginput_dict[group_id][key] = [value]

reply("学习成功！快对我说" + str(key) + "试试吧！")

core.save(group_id)

return True

elif command == 'reload' :

loginput_file_path = LOGINPUT_PATH + str(group_id) + ".loginput"

try:

core.loginput_dict[str(group_id)] = core.readdatafromfile(loginput_file_path)

logger.info("RUNTIMELOG loginput loaded. Now loginput list: {0}".format(str(core.loginput_dict)))

except:

core.loginput_dict[str(group_id)] = dict()

logger.info("RUNTIMELOG loginput file is empty.")

reply("重新加载完成！")

return True

elif command == 'delete':

if key in core.loginput_dict[group_id] and core.loginput_dict[group_id][key].count(value):

core.loginput_dict[group_id][key].remove(value)

reply("呜呜呜我再也不说" + str(value) + "了")

core.save(group_id)

return True

elif command == 'list':

result = 0

for key in list(core.loginput_dict[group_id].keys()):

result += 1

logger.info("RUNTIMELOG Replying the list of loginput for group {}".format(group_id))

reply("总共有关键字" + result + "条")

return True

return

其實關鍵字的檢測就是來自於正則表達式，順著表達式的格式就能猜得出來，添加一條回复就是這樣： !學習 {測試}{測試回复!} 。

那麼刪除呢？格式一樣，由於它可以根據同一個關鍵字進行多條隨機回复，所以如果刪除，一定要對應，不能留空： !刪除 {測試}{測試回复!}

但由於格式要固定，所以查看數量以及重載都是需要用這樣的格式，只不過括號內的內容是無效的，隨便寫： !重裝 {測試}{測試回复!}

最後，由於 tucao 本身用的是 Python 的序列化，保存文件不利於編輯，我們把它改為 JSON 格式，這就需要一個額外的 json 包。

由於我平時使用的都是 Python 3，所以這裡我順手把其他項目裡的代碼拿過來用了，再添加：

import json

import io

def writejson2file(self,obj, filename):

with io.open(filename, 'w', encoding='utf8') as outfile:

data = json.dumps(obj, indent=4, sort_keys=True, ensure_ascii=False)

outfile.write(data)

def readdatafromfile(self,filename):

with io.open(filename, encoding='utf8') as outfile:

return json.load(outfile)

這樣就可以改保存和加載的代碼了：

def save(self, group_id):

"""

:type group_id: int, 用于保存指定群的吐槽存档

"""

global LOGINPUT_PATH

try:

loginput_file_path = LOGINPUT_PATH + str(group_id) + ".loginput"

self.writejson2file(self.loginput_dict[str(group_id)],loginput_file_path)

logger.info("RUNTIMELOG loginput saved. Now loginput list: {0}".format(str(self.loginput_dict)))

except Exception:

logger.error("RUNTIMELOG Fail to save loginput.")

raise IOError("Fail to save loginput.")

def load(self, group_id):

"""

:type group_id: int, 用于读取指定群的吐槽存档

"""

global LOGINPUT_PATH

if str(group_id) in set(self.loginput_dict.keys()):

return

loginput_file_path = LOGINPUT_PATH + str(group_id) + ".loginput"

if not os.path.isdir(LOGINPUT_PATH):

os.makedirs(LOGINPUT_PATH)

if not os.path.exists(loginput_file_path):

with open(loginput_file_path, "w") as tmp:

tmp.close()

try:

self.loginput_dict[str(group_id)] = self.readdatafromfile(loginput_file_path)

logger.info("RUNTIMELOG loginput loaded. Now loginput list: {0}".format(str(self.loginput_dict)))

except :

self.loginput_dict[str(group_id)] = dict()

logger.info("RUNTIMELOG loginput file is empty.")

最後，去配置裡選擇加載這個插件：

vi config/plugin.json

{

"plugin_packages": [],

"plugin_on": [

"basic",

"manager",

"weather",

"loginput"

]

}

然後把修改好的插件放到 smart_qq_plugins 裡就可以了。

插件

如果你需要，這個文件的所有代碼你可以直接在這裡複製，新建一個 loginput.PY 並寫入即可：

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

# -*- coding: utf-8 -*-

import re

import os

import random

import json

import io

from smart_qq_bot.logger import logger

from smart_qq_bot.signals import (

on_all_message,

on_private_message,

)

LOGINPUT_PATH = 'smart_qq_plugins/loginput/'

class LogInputCore(object):

def __init__(self):

self.loginput_dict = dict()

def writejson2file(self,obj, filename):

with io.open(filename, 'w', encoding='utf8') as outfile:

data = json.dumps(obj, indent=4, sort_keys=True, ensure_ascii=False)

outfile.write(data)

def readdatafromfile(self,filename):

with io.open(filename, encoding='utf8') as outfile:

return json.load(outfile)

def save(self, group_id):

"""

:type group_id: int, 用于保存指定群的吐槽存档

"""

global LOGINPUT_PATH

try:

loginput_file_path = LOGINPUT_PATH + str(group_id) + ".loginput"

self.writejson2file(self.loginput_dict[str(group_id)],loginput_file_path)

logger.info("RUNTIMELOG loginput saved. Now loginput list: {0}".format(str(self.loginput_dict)))

except Exception:

logger.error("RUNTIMELOG Fail to save loginput.")

raise IOError("Fail to save loginput.")

def load(self, group_id):

"""

:type group_id: int, 用于读取指定群的吐槽存档

"""

global LOGINPUT_PATH

if str(group_id) in set(self.loginput_dict.keys()):

return

loginput_file_path = LOGINPUT_PATH + str(group_id) + ".loginput"

if not os.path.isdir(LOGINPUT_PATH):

os.makedirs(LOGINPUT_PATH)

if not os.path.exists(loginput_file_path):

with open(loginput_file_path, "w") as tmp:

tmp.close()

try:

self.loginput_dict[str(group_id)] = self.readdatafromfile(loginput_file_path)

logger.info("RUNTIMELOG loginput loaded. Now loginput list: {0}".format(str(self.loginput_dict)))

except :

self.loginput_dict[str(group_id)] = dict()

logger.info("RUNTIMELOG loginput file is empty.")

core = LogInputCore()

@on_all_message(name='loginput[学习遗忘]')

def loginput(msg, bot):

global core

reply = bot.reply_msg(msg, return_function=True)

group_id = "16788008"

core.load(group_id)

for key in list(core.loginput_dict[group_id].keys()):

if str(key) in msg.content and core.loginput_dict[group_id][key]:

logger.info("RUNTIMELOG loginput pattern detected, replying...")

reply(random.choice(core.loginput_dict[group_id][key]))

return True

return False

@on_private_message(name='loginput[管理]')

def current_loginput_list(msg, bot):

# webqq接受的消息会以空格结尾

global core

reply = bot.reply_msg(msg, return_function=True)

group_id = "16788008"

match = re.match(r'^(?:!|！)(learn|delete|list|reload)(?:\s?){(.+)}(?:\s?){(.+)}', msg.content)

if match:

core.load(group_id)

logger.info("RUNTIMELOG loginput command detected.")

command = str(match.group(1)).decode('utf8')

key = str(match.group(2)).decode('utf8')

value = str(match.group(3)).decode('utf8')

if command == 'learn':

if group_id not in core.loginput_dict:

core.load(group_id)

if key in core.loginput_dict[group_id] and value not in core.loginput_dict[group_id][key]:

core.loginput_dict[group_id][key].append(value)

else:

core.loginput_dict[group_id][key] = [value]

reply("学习成功！快对我说" + str(key) + "试试吧！")

core.save(group_id)

return True

elif command == 'reload' :

loginput_file_path = LOGINPUT_PATH + str(group_id) + ".loginput"

try:

core.loginput_dict[str(group_id)] = core.readdatafromfile(loginput_file_path)

logger.info("RUNTIMELOG loginput loaded. Now loginput list: {0}".format(str(core.loginput_dict)))

except:

core.loginput_dict[str(group_id)] = dict()

logger.info("RUNTIMELOG loginput file is empty.")

reply("重新加载完成！")

return True

elif command == 'delete':

if key in core.loginput_dict[group_id] and core.loginput_dict[group_id][key].count(value):

core.loginput_dict[group_id][key].remove(value)

reply("呜呜呜我再也不说" + str(value) + "了")

core.save(group_id)

return True

elif command == 'list':

result = 0

for key in list(core.loginput_dict[group_id].keys()):

result += 1

logger.info("RUNTIMELOG Replying the list of loginput for group {}".format(group_id))

reply("总共有关键字" + result + "条")

return True

return

本文由落格博客原創撰寫：落格博客 » 我給落格輸入法的用戶群添加了個自動回復機器人

轉載請保留出處和原文鏈接：https://www.logcg.com/archives/2507.html

落格博客

我給落格輸入法的用戶群添加了個自動回復機器人

原理

定制

插件

關於作者

R0uter

註釋

發表評論取消回复

我給落格輸入法的用戶群添加了個自動回復機器人

原理

定制

插件

相關文章：

關於作者

R0uter

註釋

發表評論 取消回复

發表評論取消回复