亲宝软件园·资讯

展开

go cqhttp权限管理

A-L-Kun 人气:0

权限管理系统

一、 概述

在写好我们的智能聊天功能之后,大家有没有感觉很烦呢?感觉这个机器人在群里面一直被艾特,一直被戳一戳。那么,我们有没有一种方法,使得其在群里面的权限可控呢?

或许大家看到这个问题就想到了一个方法,那就是通过python文件,但是使用python文件保存的话有一个缺点,那就是修改配置文件后,需要重新运行我们的项目,这会让我们觉得很麻烦!

那么,还有没有更好的方法呢?给大家一分钟时间思考……好,大家思考出来了吗?我的想法是,将权限存储到数据库中,当我们需要调用这个功能的时候,通过调用数据库,来判断是否有进行这项功能的权限!这里,我们选择的是mysql数据库,关于对数据库的操作,我已经给大家准备好了!

看到SQLAchemy,就应该有小伙伴要说了,既然我们看了mysql的基本语法,我们完全可以通过SQL语句来操作我们的数据库,为什么需要使用ORM来操作数据库呢?其实不然,我们使用SQL操作数据库的话,就无法处理一些高并发的操作了,会造成严重的阻塞,使用SQLAchemy可以创建数据库连接池,缓解服务器的压力,同时ORM对于一些不是很熟悉SQL的小伙伴来说比较友好。

好了,相信大家也看完了上面对数据库的操作的文章了,同时也看完了我写的关于flask的全部文章了,废话不多说了,来开始实现我们的权限管理系统

展示一下我的目录结构:

|-- App
|   |-- exts.py
|   |-- __init__.py
|   |-- models.py
|   |-- script.py
|   |-- settings.py
|   `-- views
|       |-- goCqhttp.py
|       |-- __init__.py
|-- app.py

二、 创建表

1、 创建

在实现权限管理系统之前,我们肯定是需要创建相应的数据表来存储我们的数据的

我们使用flask-sqlachemy来实现orm,同时快速生成表

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "models.py"
__time__ = "2022/9/11 19:56"

from App.exts import db

class Group(db.Model):  # 创建一个群的表
    __tablename__ = "group"
    id = db.Column(db.INTEGER, primary_key=True, autoincrement=True)
    name = db.Column(db.String(200), nullable=True)
    qqId = db.Column(db.String(20), nullable=False, unique=True, index=True)  
    isDetect = db.Column(db.BOOLEAN, default=True)  # 是否开启,进行逻辑删除
    auth = db.Column(db.SmallInteger, default=0)  # 在群里面的的地位,默认为群成员,1为管理员,2为群主

    def __init__(self, name, qqId):
        self.name = name
        self.qqId = qqId

class GroupAuthority(db.Model):  # 创建一个权限管理表
    __tablename__ = "groupAuth"
    id = db.Column(db.INTEGER, primary_key=True, autoincrement=True)
    chat = db.Column(db.INTEGER, default=0, nullable=False)  # 是否开启智能聊天的功能
    welcome = db.Column(db.INTEGER, default=1, nullable=False)  # 是否开启新成员入群欢迎的功能
    banTalk = db.Column(db.INTEGER, default=0)  # 群禁言功能,以及消息撤回功能
    click = db.Column(db.INTEGER, default=1)  # 戳一戳功能,默认开启
    smallFunction = db.Column(db.INTEGER, default=1)  # 是否开启小功能,如疫情数据查询等
    dailyBrief = db.Column(db.INTEGER, default=0)  # 是否开启每日简报功能
    groupId = db.Column(db.INTEGER, db.ForeignKey("group.id", ondelete="CASCADE"))  # 外键约束,同时进行级联删除
    auth2group = db.relationship("Group", backref=db.backref("group2auth"))  # 使用代理

    def __init__(self, is_privade=False, chat=None, welcome=None, bantalk=None, click=None, smallFunction=None,
                 dailyBrief=None):
        if is_privade:
            self.chat = chat
            self.welcome = welcome
            self.banTalk = bantalk
            self.click = click
            self.smallFunction = smallFunction
            self.dailyBrief = dailyBrief

在这里,我使用group表作为主表,用来存放各群的数据,同时将机器人对各群的权限存储到另一张表中,因为是一对一的关系,我们使用同时对表使用外键约束来关联主表以及子表

2、 生成

把表的结构创建完后,我们来创建主程序,用来生成我们创建的表

exts.py文件中写入:

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "exts.py"
__time__ = "2022/9/11 23:39"

from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

db = SQLAlchemy()  # 操作数据库
def init_exts(app):
    db.init_app(app)
    Migrate().init_app(app, db)  # 使用app初始化Migrate
    app.config["db"] = db

settings.py中写入:

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "settings.py"
__time__ = "2022/9/11 19:17"

class Config:
    """基础的配置字"""
    SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:123456@127.0.0.1:3306/bot?charset=utf8"  
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    SQLALCHEMY_ECHO = True
    SUPER_USER = ["3500515050", ]  # 超级用户,其可以私发消息给机器人,管理机器人对所有群的权限,当然也可以存储到数据库中,这里为了方便,存储在配置文件中,大家可根据这篇文章执行实现
    ADMIN = 3500515050  # 开发者

class Help:
    """帮助文档"""
    ADMIN_HELP = """开发者命令提示:
1 添加群:\n/admin:add QQ群号 QQ群名
2. 删除功能:\n/admin:delete QQ群号
3 查找功能:\n/admin:get QQ群号
4. 获取所有群:\n/admin:show
5. 修改权限命令:\n/admin:changeAuth QQ群号 |聊天功能|入群欢迎|管理群|戳一戳|拓展功能|定时功能|(比如110011)\n
如果还是有问题,请与开发人员联系哦!"""
    GROUP_ADMIN = """[CQ:at,qq=%d]命令提示:
1. 查看群权限:\n/admin:get
2. 修改群权限:\n/admin:change |聊天功能|入群欢迎|管理群|戳一戳|拓展功能|定时功能|(比如#admin:change# 110011)\n
如果还是有问题,请与开发人员联系哦!"""

class ProductConfig(Config, Mes, Url):
    """生产环境配置"""
    pass
class DevelopConfig(Config, Mes, Url):
    """开发环境配置"""
    DEBUG = True
envs = {
    "product": ProductConfig,
    "develop": DevelopConfig
}

views/__init__.py中写入:

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "__init__.py.py"
__time__ = "2022/9/11 19:30"

from App.views.goCqhttp import AcceptMes
from flask_restful import Api
def init_app(app):
    api = Api(app)
    api.add_resource(AcceptMes, "/", endpoint="index")

views/goCqhttp.py中写入:

# !/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "goCqhttp.py"
__time__ = "2022/9/11 19:57"

from flask_restful import Resource
from flask import request
import asyncio
from App.events.private import PriChatMes
from App.events.group import GroupChatMes
from App.events.groupAndPri import GroupAndPri
from flask import current_app
from App.models import Group
from App.events.groupAndPri import Command
from App.script.requests_tools import Sender
class AcceptMes(Resource):

    def post(self):
		pass  # 后面主要是通过这个接口来对消息的分发

App/__init__.py中 写入

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "__init__.py.py"
__time__ = "2022/9/11 14:25"

from flask import Flask
from App.settings import envs
from App.views import init_app
from App.exts import init_exts

def create_app(env):
    app = Flask(__name__)
    app.config.from_object(envs.get(env))
    init_exts(app)
    init_app(app)  # 将路由传入app中
    return app

app.py中写入:

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "app.py"
__time__ = "2022/9/11 19:17"

from App import create_app
from flask_script import Manager

app = create_app("develop")
manage = Manager(app)

if __name__ == '__main__':
    manage.run()

然后,我们的基本框架就搭建好了,就将我们写好的模型映射到数据库中

3、 映射

在项目的根目录运行这个代码:

flask db init  # 初始化操作
flask db migrate  # 迁移操作
flask db upgrade  # 更新操作

然后,我们的数据表就更新完成了

三、 增删改查

首先,我们先不考虑如何监控我们的QQ消息,通过消息来操控我们的数据库,我们先来实现通过代码直接对权限数据的修改

为了方便,我们就直接在script.py中写入我们的函数,可以根据文章的顺序来添加

1、 群管理

1.1 增加群

from flask import current_app
from App.models import Group, GroupAuthority
from base64 import b64decode  # 对中文进行编码
async def add(gid, nick):
    # 传入QQ群号和qq群的昵称就可以添加了
    session = current_app.config["db"].session
    data = session.query(Group).filter_by(qqId=gid).first()  # 首先查看一下数据库中是否收录了这个群
    if data is None:  # 如果没有则添加,同时,也检查其是否有逻辑删除
        g = Group(nick, gid)  # 创建一个群
        g.group2auth = [GroupAuthority()]  # 给群赋予默认的权限
        session.add(g)  # 添加到数据库中
        session.commit()  # 提交事务
        return {  
            "status": 200
        }  # 返回状态码
   	# 如果群存在,直接修改群名,以及使其可以检测到
    data.isDetect = True
    data.name = nick
    session.commit()
    return {
        "status": 300,
        "error": "该群号已存在!",
    }  

1.2 删除群

async def delete(gid):
    db = current_app.config["db"]
    session = db.session
    group = session.query(Group).filter(db.and_(Group.qqId == gid, Group.isDetect)).first()
    if group:
        group.isDetect = False
        session.commit()
        return "删除成功"
    return "该群不存在"

这里使用的是逻辑删的删除,使得程序无法访问到该数据,即代表删除效果,同时,数据也保存了下来

1.3 展示功能

async def show():
    session = current_app.config["db"].session
    data = [f"|_ {b64decode(i.name).decode()} _|_ {i.qqId} _|_ {'yes' if i.isDetect else 'no'} _|" for i in
            session.query(Group).all()]
    return "\n".join(data)

2、 权限管理

2.1 展示权限

async def get(gid):
    db = current_app.config["db"]
    session = db.session
    data = session.query(Group).filter(db.and_(Group.qqId == gid, Group.isDetect)).first()
    if data is None:
        return "在该群不支持,请与开发者联系!"
    name = b64decode(data.name).decode()  # qq群名称
    chat = "1. 已开启聊天功能" if data.group2auth[0].chat else "1. 未开启聊天功能"  # 有没有开启智能聊天的功能
    welcome = "2. 已开启入群欢迎功能" if data.group2auth[0].welcome else "2. 未开启入群欢迎功能"  # 是否开启入群欢迎的功能
    banTalk = "3. 已开启管理群功能" if data.group2auth[0].banTalk else "3. 未开启管理群功能"  # 是否开启管理员的功能
    click = "4. 已开启戳一戳功能" if data.group2auth[0].click else "4. 未开启戳一戳功能"  # 戳一戳功能
    smallFunction = "5. 已开启拓展功能" if data.group2auth[0].smallFunction else "5. 未开启拓展功能"  # 是否开启小功能
    dailyBrief = "6. 已开启定时发消息功能" if data.group2auth[0].dailyBrief else "6. 未开启定时发消息功能"  # 是否开启每日自动播报的功能
    return f"{name}\n{chat}\n{welcome}\n{banTalk}\n{click}\n{smallFunction}\n{dailyBrief}\n"

2.2 修改权限

async def changeAuth(gid, data, ty):
    """
    传入群号,要修改的数据其为6位数字,如:"011011",每一个数字代表一个权限,1为真|0为假
 	最后传入的是调用这个函数的类型,是群管理员还是超级管理员
    """
    db = current_app.config["db"]
    session = db.session  # 获取数据库连接
    try:
        for i in data:
            if int(i) not in [0, 1] or len(data) != 6:
                raise ValueError
        data = (int(i) for i in data)  # 获取到对应的权限数字
        group = session.query(Group).filter(db.and_(Group.qqId == gid, Group.isDetect)).first()  # 获取对象
        if group:
            group.group2auth = [GroupAuthority(True, *data)]  # 使用元组拆包的方式来传参
            session.commit()  # 提交数据
            _ = await Admin.get(gid)  # 获取到群里面的权限
            ret = f"[CQ:at,qq=%d]设置成功,设置后的权限为:\n{_}" if ty == "group" else f"设置成功,设置后的权限为:\n{_}"  
        else: 
            ret = "该群不支持机器人"
    except Exception as e:
        ret = "[CQ:at,qq=%d]设置失败,请查看帮助文档!" if ty == "group" else "设置失败,请查看帮助文档!"
    return ret

那么,我们就把我们的权限管理大概的搭建好了,我们只需要再对消息进行检测,使其机器人可以解析命令就完成了

四、 获取命令

1、 消息分发

这里,我们来结合我们的权限系统,使得机器人可以对修改权限的命令作出回应

# !/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "goCqhttp.py"
__time__ = "2022/9/11 19:57"

from flask_restful import Resource
from flask import request
import asyncio
from flask import current_app
from App.models import Group
from App.script import *
from base64 import b64encode

# 用于发送信息的函数
async def send(id, message, ty):
    """
        用于发送消息的函数
        :param id: qq号,或者qq群号
        :param message: 发送的消息
        :param ty: 传入的
        :return: None
        """
    async with httpx.AsyncClient(base_url="http://127.0.0.1:5700") as client:
        # 如果发送的为私聊消息
        if ty == "group":
            params = {
                "group_id": id,
                "message": message,
            }
            await client.get("/send_group_msg", params=params)
            elif ty == "private":
                params = {
                    "user_id": id,
                    "message": message,
                }
                await client.get("/send_private_msg", params=params)
class AcceptMes(Resource):

    def post(self):
        # 这里对消息进行分发,暂时先设置一个简单的分发
        _ = request.json
        if _.get("message_type") == "private":  # 说明有好友发送信息过来
            uid = _["sender"]["user_id"]  # 获取发信息的好友qq号
        	message = _["raw_message"]  # 获取发送过来的消息
            if message.startswith("/admin:") and str(uid) in current_app.config["SUPER_USER"]:
                # 其为超级用户的命令,通过正则解析出命令,并且做出回应,那么我们就需要创建一个函数,来解析我们的命令,这个函数我们后面来实现
            	asyncio.run(super_command(uid, message))  # 处理私聊的信息

        elif _.get("message_type") == "group" and message.startswith("/admin:") and resp["sender"]["role"] in ["owner", "admin"]:
            # 获取群消息,并且,其为群主或者管理员发送的消息的话
            db = current_app.config["db"]
            session = db.session
            group = session.query(Group).filter(db.and_(Group.qqId == _["group_id"], Group.isDetect)).first()
            if not group:
                # 如果群里面不支持机器人的话,直接返回
                return
            uid = _["sender"]["user_id"]  # 获取发信息的好友qq号
            message = _.get("raw_message")
            gid = resp["group_id"]  # 获取群号
            asyncio.run(admin_command(uid, gid, message))  # 这里也是先用一个函数名来占位,我们后面来实现这个函数

2、 解析命令

goCqhttp.py后面继续添加以下内容

超级管理员的命令解析

async def super_command(uid, message):  # 处理开发者的命令
    com_1 = re.findall(r"/admin:(.*)", message.split()[0])[0]
    try:
        com_2 = message.split()[1]  # qq群号
    except Exception as e:
        com_2 = None
    if com_1 == "add" and com_2.isdecimal():  # 这说明其为添加qq群
        com_3 = message.split()[2]  # qq昵称
        ret = await add(com_2, b64encode(com_3.encode()))
        if ret["status"] == 200:
            await send(uid, f"{com_2}添加成功,该群名为{com_3}", "private")
        else:
            await send(uid, ret["error"], "private")
    elif com_1 == "get" and com_2.isdecimal():  # 获取指定的qq群的权限信息
        ret = await get(com_2)
        await send(uid, ret, ty="private")
    elif com_1 == "delete" and com_2.isdecimal():  # 删除qq群
        ret = await delete(com_2)
        await send(uid, ret, ty="private")
    elif com_1 == "changeAuth" and com_2.isdecimal():  # 修改权限
        data = list(message.split()[2])
        ret = await changeAuth(com_2, data, "private")
        await send(uid, ret, "private")
    elif com_1 == "show":  # 展示所有群信息
        ret = await show()
        await send(uid, ret, "private")
    else:  # 如果都不是的话,发送帮助文档
        ret = current_app.config["ADMIN_HELP"]
        await send(uid, ret, "private")

群管理员命令解析

async def admin_command(uid, gid, message):  # 群管理员命令
    com_1 = re.findall(r"/admin:(.*)", message.split()[0])[0]  # 把命令解析出来
    if com_1 == "get":  # 获取本群的权限信息
        ret = await get(gid)
        await send(gid, ret, ty="group")
    elif com_1 == "change":  # 修改权限信息
            data = list(message.split()[1])
        ret = await changeAuth(gid, data, "group")
        await send(gid, ret % uid, "group")
    else:  # 默认为帮助文档
        ret = current_app.config["GROUP_ADMIN"] % uid
        await send(gid, ret, "group")

最后,我们的权限系统就做完啦!大家可以运行项目尝试一下哦!python app.py runserver -p5701

同时里面的每一个权限我也会慢慢的分布实现!

加载全部内容

相关教程
猜你喜欢
用户评论