前言：本文为大家带来QQ机器人程序上云的教程，环境搭建请参考下面链接
【0基础QQ机器人开发】基于go-cqhttp的QQ机器人开发教程,仅供自学
【零基础QQ机器人开发二】服务器篇
文章目录
- 程序
- Logger类
- StatuStore类
- MultiFunc类
- QQBot类
- main.py
前言：基础的程序以及windows下的环境配置我们已经在基于go-cqhttp的QQ机器人开发教程中说过了，本次我们使用云服务器中的Linux环境以及python程序来配置
这篇文章是22年写的，可能会有些过时，仅供参考
程序
本程序参考@小锋学长生活大爆炸的程序进行开发。
因为我们的程序是运行在服务器并接通互联网的，所以理论上我们可以把该程序作为一个通入互联网的入口，通过QQ实现所有通过互联网可以实现的功能，比如发送邮件，访问网站等。在这里我们先实现一点简单的功能，以后有了新的创意再更新。
程序结构图如下：
Logger类
用于输出内容到日志中
class Logger:
def __init__(self, level='debug'):
self.level = level
def DebugLog(self, *args):
if self.level == 'debug':
print(*args)
def TraceLog(self, *args):
if self.level == 'trace':
print(*args)
def setDebugLevel(self, level):
self.level = level.lower()
StatuStore类
用于存储当前指令的状态，便于处理下一步命令
class StatusStore:
def __init__(self, from_qq: int = None, is_cmd: bool = False, funcName: str = None, need_second: bool = False, msg: str = None) -> None:
self.from_qq = from_qq # 发送者的QQ号
self.is_cmd = is_cmd # 是否是指令(选择功能)
self.funcName = funcName # 选择的功能的名称
self.need_second = need_second # 是否需要经过两步：先发cmd指令，再发详细内容
self.msg = msg # 本次发送的消息内容
def detail(self):
return self.__dict__
MultiFunc类
保存了相应功能的代码
里面用中文写的参数需要改成自己的
import re
import time
import requests
import json
from github import Github
from tencentcloud.common import credential
from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
from tencentcloud.tmt.v20180321 import tmt_client, models
tencent_translate_AppId = 填入appid
tencent_secret_id = '填入id'
tencent_secret_key = '填入key'
class MultiFunc:
# 翻译功能
@staticmethod
def translate(original: str, convert: str = 'en'):
try:
cred = credential.Credential(tencent_secret_id, tencent_secret_key)
client = tmt_client.TmtClient(cred, "ap-guangzhou")
req = models.TextTranslateRequest()
params = {
"SourceText": original,
"Source": "auto",
"Target": convert,
"ProjectId": tencent_translate_AppId
}
req.from_json_string(json.dumps(params))
resp = client.TextTranslate(req)
return resp.TargetText
except TencentCloudSDKException as err:
print(err)
# 上传图片到图床功能
@staticmethod
def uploadImage(imgMsg):
cmt = '来自QQ'
timestamp = time.strftime("%Y-%m-%d_%H:%M:%S", time.localtime())
p = re.compile(r'=(https.+?)]')
imgUrl = p.findall(imgMsg)
response = requests.get(url=imgUrl[0])
token = '这里填入github仓库的token'
g = Github(token)
# user = g.get_user()
# repo = user.create_repo("q_img")
# 如果是已经存在的仓库，可以使用get_repo来获得repo对象
repo = g.get_repo('175603a/q_img')
# 读取文件并上传
data = response.content
# 不需要进行Base64编码，编码过程在create_file内部已经完成了
try:
repo.create_file(f'{timestamp}.jfif', f'{cmt}', data)
res = 'https://github.com/175603a/q_img/blob/main/' + f'{timestamp}.jfif?raw=true'
except:
repo.create_file(f'{timestamp}.jfif', f'{cmt}', data)
res = 'https://github.com/175603a/q_img/blob/main/' + f'{timestamp}.jfif?raw=true'
return res
# 图片OCR并返回结果
@staticmethod
def imageOCR(imgMsg):
p = re.compile(r'=(.+?.image)')
imgId = p.findall(imgMsg)
url = 'http://127.0.0.1:5700/ocr_image'
data = {
'image': imgId
}
r = requests.post(url=url, data=data)
content = ''
for text in json.loads(r.content)['data']['texts']:
content += text['text'] + '\n'
return content
# 监控日志
@staticmethod
def monitor_crawler():
with open('/root/QQBot/go-cqhttp/nohup.out', 'a', encoding='utf-8') as fp:
res = fp.read()
fp.write('')
return res
QQBot类
封装了机器人这个类，里面有收发消息的功能
import urllib.parse
import requests
import urllib.parse
import urllib.request
import socket
import json
class Robot:
def __init__(self):
self.msg = ''
def ai_talk(self):
msg = self.msg
def qykApi(msg):
base_url = 'http://api.qingyunke.com/api.php?key=free&appid=0&'
data = {
'msg': msg['raw_message']
}
new_data = urllib.parse.urlencode(data)
url = base_url + new_data
response = urllib.request.urlopen(url).read().decode('utf-8')
response_msg = eval(response)['content'].replace('{br}', '\n')
return response_msg
if self.msg['message_type'] == 'private':
return self.sdMsg(qykApi(msg), self.msg['user_id'], 'private')
else:
return False
# ----------------------------------------------------
# 接收消息函数 需要循环执行，返回值字典格式
def revMsg(self): # json or None
ListenSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
ListenSocket.bind(('localhost', 5701))
ListenSocket.listen(100)
HttpResponseHeader = '''HTTP/1.1 200 OK
Content-Type: text/html
'''
Client, Address = ListenSocket.accept()
def request2json(msg):
for i in range(len(msg)):
if msg[i] == "{" and msg[-2] == "}":
return json.loads(msg[i:])
return None
try:
Request = Client.recv(1024).decode(encoding='utf-8')
rev_json = request2json(Request)
Client.sendall(HttpResponseHeader.encode(encoding='utf-8'))
Client.close()
self.msg = rev_json
except:
pass
return rev_json
def sdMsg(self, msg, qq_id, msg_type):
if msg_type == "private":
data = {
'user_id': qq_id,
'message': msg,
'auto_escape': False
}
cq_url = "http://127.0.0.1:5700/send_private_msg"
rev = requests.post(cq_url, data=data)
elif msg_type == "group":
data = {
'group_id': qq_id,
'message': msg,
'auto_escape': False
}
cq_url = "http://127.0.0.1:5700/send_group_msg"
rev = requests.post(cq_url, data=data)
else:
return False
if rev.json()['status'] == 'ok':
return True
return False
main.py
from MultiFunc import MultiFunc
from QQBot import Robot
from Logger import Logger
from StatuStore import StatusStore
import threading
import random
import re
import requests
import ddddocr
import time
CONTROQID = 这里填入控制机器人的QQ
if __name__ == '__main__':
logger = Logger()
status_store = {}
logger.DebugLog('====================开始运行========================')
function_map = {
'翻译': {'function': MultiFunc.translate, 'need_second': True, 'desc': '请输入您要翻译的内容~'},
'上传图片': {'function': MultiFunc.uploadImage, 'need_second': True, 'desc': '请发送图片过来吧~'},
'ocr': {'function': MultiFunc.imageOCR, 'need_second': True, 'desc': '请发送图片过来吧~'},
'监控日志': {'function': MultiFunc.monitor_crawler, 'need_second': False}
}
def choiceFunction(store_obj: StatusStore):
res = ''
if function_map.get(store_obj.funcName):
res = function_map.get(store_obj.funcName)['function'](store_obj.msg)
return res
bot = Robot() # 实例化
while True:
bot.revMsg()
if bot.msg['user_id'] == CONTROQID and bot.msg['message_type'] == 'private':
if bot.msg['message'].strip().lower().startswith('cmd'):
try:
funcName = bot.msg['message'].strip().split('\n')[0].split()[1]
store_obj = StatusStore(from_qq=CONTROQID, is_cmd=True, funcName=funcName)
func_info = function_map.get(funcName)
if not func_info:
res = '指令[{}]暂不支持'.format(funcName)
elif func_info.get('need_second'):
res = '收到你的指令：{}\n{}'.format(funcName, func_info.get(
'desc') or '已进入对应状态, 请继续发送详细内容')
# 添加或更新记录
status_store[CONTROQID] = store_obj
else:
res = '请求结果为：\n' + str(choiceFunction(store_obj))
status_store.pop(CONTROQID, '')
except:
res = 'cmd与命令之间请加空格'
else:
res = '请先发送指令哦...'
store_obj = status_store.get(CONTROQID)
if store_obj and store_obj.is_cmd:
store_obj.msg = bot.msg['message']
res = '请求结果为：\n' + str(choiceFunction(store_obj))
status_store.pop(CONTROQID, '')
elif bot.msg['message_type'] == 'private':
res = bot.ai_talk()
elif bot.msg['message_type'] == 'group':
continue
else:
continue
bot.sdMsg(res, bot.msg['user_id'], 'private')
