权限管理系统
一、 概述
在写好我们的智能聊天功能之后，大家有没有感觉很烦呢？感觉这个机器人在群里面一直被艾特，一直被戳一戳。那么，我们有没有一种方法，使得其在群里面的权限可控呢？
或许大家看到这个问题就想到了一个方法，那就是通过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，同时快速生成表
__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)
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"), nullable=False)
auth2group = db.relationship("Group", backref=db.backref("group2auth", uselist=False))
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文件中写入：
__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.config["db"] = db
在settings.py中写入：
__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中写入：
__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中写入：
__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中 写入
__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)
return app
在app.py中写入：
__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):
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 close(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 "该群关闭成功"
async def delete(gid):
"""彻底删除"""
db = current_app.config["db"]
session = db.session
group = session.query(Group).filter(Group.qqId == gid).first()
if group:
session.delete(group.group2auth)
session.delete(group)
session.commit()
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()
chat = "1. 已开启聊天功能" if data.group2auth.chat else "1. 未开启聊天功能"
welcome = "2. 已开启入群欢迎功能" if data.group2auth.welcome else "2. 未开启入群欢迎功能"
banTalk = "3. 已开启管理群功能" if data.group2auth.banTalk else "3. 未开启管理群功能"
click = "4. 已开启戳一戳功能" if data.group2auth.click else "4. 未开启戳一戳功能"
smallFunction = "5. 已开启拓展功能" if data.group2auth.smallFunction else "5. 未开启拓展功能"
dailyBrief = "6. 已开启定时发消息功能" if data.group2auth.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:
session.delete(group.group2auth)
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、 消息分发
这里，我们来结合我们的权限系统，使得机器人可以对修改权限的命令作出回应
__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"]
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"]
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]
except Exception as e:
com_2 = None
if com_1 == "add" and com_2.isdecimal():
com_3 = message.split()[2]
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():
ret = await get(com_2)
await send(uid, ret, ty="private")
elif com_1 == "delete" and com_2.isdecimal():
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
同时里面的每一个权限我也会慢慢的分布实现！
|