【零基础QQ机器人开发三】程序上云篇

前言：本文为大家带来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 = leveldef 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, modelstencent_translate_AppId = 填入appid
tencent_secret_id = '填入id'
tencent_secret_key = '填入key'class MultiFunc:# 翻译功能@staticmethoddef 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.TargetTextexcept TencentCloudSDKException as err:print(err)# 上传图片到图床功能@staticmethoddef 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并返回结果@staticmethoddef 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# 监控日志@staticmethoddef 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 jsonclass Robot:def __init__(self):self.msg = ''def ai_talk(self):msg = self.msgdef 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_dataresponse = urllib.request.urlopen(url).read().decode('utf-8')response_msg = eval(response)['content'].replace('{br}', '\n')return response_msgif self.msg['message_type'] == 'private':return self.sdMsg(qykApi(msg), self.msg['user_id'], 'private')else:return False# ----------------------------------------------------# 接收消息函数 需要循环执行，返回值字典格式def revMsg(self):  # json or NoneListenSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)ListenSocket.bind(('localhost', 5701))ListenSocket.listen(100)HttpResponseHeader = '''HTTP/1.1 200 OKContent-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 Nonetry:Request = Client.recv(1024).decode(encoding='utf-8')rev_json = request2json(Request)Client.sendall(HttpResponseHeader.encode(encoding='utf-8'))Client.close()self.msg = rev_jsonexcept:passreturn rev_jsondef 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 Falseif rev.json()['status'] == 'ok':return Truereturn 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 timeCONTROQID = 这里填入控制机器人的QQif __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 resbot = 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_objelse: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':continueelse:continuebot.sdMsg(res, bot.msg['user_id'], 'private')