目录
- 权限管理系统
- 一、 概述
- 二、 创建表
- 1、 创建
- 2、 生成
- 3、 映射
- 三、 增删改查
- 1、 群管理
- 1.1 增加群
- 1.2 删除群
- 1.3 展示功能
- 2、 权限管理
- 2.1 展示权限
- 2.2 修改权限
- 四、 获取命令
- 1、 消息分发
- 2、 解析命令
权限管理系统
一、 概述
在写好我们的智能聊天功能之后，大家有没有感觉很烦呢？感觉这个机器人在群里面一直被艾特，一直被戳一戳。那么，我们有没有一种方法，使得其在群里面的权限可控呢？
或许大家看到这个问题就想到了一个方法，那就是通过python文件，但是使用python文件保存的话有一个缺点，那就是修改配置文件后，需要重新运行我们的项目，这会让我们觉得很麻烦！
那么，还有没有更好的方法呢？给大家一分钟时间思考……好，大家思考出来了吗？我的想法是，将权限存储到数据库中，当我们需要调用这个功能的时候，通过调用数据库，来判断是否有进行这项功能的权限！这里，我们选择的是mysql数据库，关于对数据库的操作，我已经给大家准备好了！
- mysql数据库的安装：linux 中安装数据库的方法
- mysql基本语法：关于数据库操作的基本语法
- SQLAchemy 常用操作：使用ORM操作数据库
看到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
同时里面的每一个权限我也会慢慢的分布实现！