go-cqhttp權(quán)限管理系統(tǒng)的實(shí)現(xiàn)代碼
權(quán)限管理系統(tǒng)
一、 概述
在寫好我們的智能聊天功能之后,大家有沒有感覺很煩呢?感覺這個(gè)機(jī)器人在群里面一直被艾特,一直被戳一戳。那么,我們有沒有一種方法,使得其在群里面的權(quán)限可控呢?
或許大家看到這個(gè)問題就想到了一個(gè)方法,那就是通過python文件,但是使用python文件保存的話有一個(gè)缺點(diǎn),那就是修改配置文件后,需要重新運(yùn)行我們的項(xiàng)目,這會(huì)讓我們覺得很麻煩!
那么,還有沒有更好的方法呢?給大家一分鐘時(shí)間思考……好,大家思考出來(lái)了嗎?我的想法是,將權(quán)限存儲(chǔ)到數(shù)據(jù)庫(kù)中,當(dāng)我們需要調(diào)用這個(gè)功能的時(shí)候,通過調(diào)用數(shù)據(jù)庫(kù),來(lái)判斷是否有進(jìn)行這項(xiàng)功能的權(quán)限!這里,我們選擇的是mysql
數(shù)據(jù)庫(kù),關(guān)于對(duì)數(shù)據(jù)庫(kù)的操作,我已經(jīng)給大家準(zhǔn)備好了!
- mysql數(shù)據(jù)庫(kù)的安裝:linux 中安裝數(shù)據(jù)庫(kù)的方法
- mysql基本語(yǔ)法:關(guān)于數(shù)據(jù)庫(kù)操作的基本語(yǔ)法
- SQLAchemy 常用操作:使用ORM操作數(shù)據(jù)庫(kù)
看到SQLAchemy
,就應(yīng)該有小伙伴要說了,既然我們看了mysql
的基本語(yǔ)法,我們完全可以通過SQL語(yǔ)句來(lái)操作我們的數(shù)據(jù)庫(kù),為什么需要使用ORM
來(lái)操作數(shù)據(jù)庫(kù)呢?其實(shí)不然,我們使用SQL
操作數(shù)據(jù)庫(kù)的話,就無(wú)法處理一些高并發(fā)的操作了,會(huì)造成嚴(yán)重的阻塞,使用SQLAchemy
可以創(chuàng)建數(shù)據(jù)庫(kù)連接池,緩解服務(wù)器的壓力,同時(shí)ORM
對(duì)于一些不是很熟悉SQL
的小伙伴來(lái)說比較友好。
好了,相信大家也看完了上面對(duì)數(shù)據(jù)庫(kù)的操作的文章了,同時(shí)也看完了我寫的關(guān)于flask
的全部文章了,廢話不多說了,來(lái)開始實(shí)現(xiàn)我們的權(quán)限管理系統(tǒng)
展示一下我的目錄結(jié)構(gòu):
|-- App | |-- exts.py | |-- __init__.py | |-- models.py | |-- script.py | |-- settings.py | `-- views | |-- goCqhttp.py | |-- __init__.py |-- app.py
二、 創(chuàng)建表
1、 創(chuàng)建
在實(shí)現(xiàn)權(quán)限管理系統(tǒng)之前,我們肯定是需要?jiǎng)?chuàng)建相應(yīng)的數(shù)據(jù)表來(lái)存儲(chǔ)我們的數(shù)據(jù)的
我們使用flask-sqlachemy
來(lái)實(shí)現(xiàn)orm
,同時(shí)快速生成表
#!/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): # 創(chuàng)建一個(gè)群的表 __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) # 是否開啟,進(jìn)行邏輯刪除 auth = db.Column(db.SmallInteger, default=0) # 在群里面的的地位,默認(rèn)為群成員,1為管理員,2為群主 def __init__(self, name, qqId): self.name = name self.qqId = qqId class GroupAuthority(db.Model): # 創(chuàng)建一個(gè)權(quán)限管理表 __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) # 戳一戳功能,默認(rèn)開啟 smallFunction = db.Column(db.INTEGER, default=1) # 是否開啟小功能,如疫情數(shù)據(jù)查詢等 dailyBrief = db.Column(db.INTEGER, default=0) # 是否開啟每日簡(jiǎn)報(bào)功能 groupId = db.Column(db.INTEGER, db.ForeignKey("group.id", ondelete="CASCADE")) # 外鍵約束,同時(shí)進(jìn)行級(jí)聯(lián)刪除 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
表作為主表,用來(lái)存放各群的數(shù)據(jù),同時(shí)將機(jī)器人對(duì)各群的權(quán)限存儲(chǔ)到另一張表中,因?yàn)槭且粚?duì)一的關(guān)系,我們使用同時(shí)對(duì)表使用外鍵約束來(lái)關(guān)聯(lián)主表以及子表
2、 生成
把表的結(jié)構(gòu)創(chuàng)建完后,我們來(lái)創(chuàng)建主程序,用來(lái)生成我們創(chuàng)建的表
在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() # 操作數(shù)據(jù)庫(kù) 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: """基礎(chǔ)的配置字""" 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", ] # 超級(jí)用戶,其可以私發(fā)消息給機(jī)器人,管理機(jī)器人對(duì)所有群的權(quán)限,當(dāng)然也可以存儲(chǔ)到數(shù)據(jù)庫(kù)中,這里為了方便,存儲(chǔ)在配置文件中,大家可根據(jù)這篇文章執(zhí)行實(shí)現(xiàn) ADMIN = 3500515050 # 開發(fā)者 class Help: """幫助文檔""" ADMIN_HELP = """開發(fā)者命令提示: 1 添加群:\n/admin:add QQ群號(hào) QQ群名 2. 刪除功能:\n/admin:delete QQ群號(hào) 3 查找功能:\n/admin:get QQ群號(hào) 4. 獲取所有群:\n/admin:show 5. 修改權(quán)限命令:\n/admin:changeAuth QQ群號(hào) |聊天功能|入群歡迎|管理群|戳一戳|拓展功能|定時(shí)功能|(比如110011)\n 如果還是有問題,請(qǐng)與開發(fā)人員聯(lián)系哦!""" GROUP_ADMIN = """[CQ:at,qq=%d]命令提示: 1. 查看群權(quán)限:\n/admin:get 2. 修改群權(quán)限:\n/admin:change |聊天功能|入群歡迎|管理群|戳一戳|拓展功能|定時(shí)功能|(比如#admin:change# 110011)\n 如果還是有問題,請(qǐng)與開發(fā)人員聯(lián)系哦!""" class ProductConfig(Config, Mes, Url): """生產(chǎn)環(huán)境配置""" pass class DevelopConfig(Config, Mes, Url): """開發(fā)環(huán)境配置""" 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 # 后面主要是通過這個(gè)接口來(lái)對(duì)消息的分發(fā)
在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()
然后,我們的基本框架就搭建好了,就將我們寫好的模型映射到數(shù)據(jù)庫(kù)中
3、 映射
在項(xiàng)目的根目錄運(yùn)行這個(gè)代碼:
flask db init # 初始化操作 flask db migrate # 遷移操作 flask db upgrade # 更新操作
然后,我們的數(shù)據(jù)表就更新完成了
三、 增刪改查
首先,我們先不考慮如何監(jiān)控我們的QQ消息,通過消息來(lái)操控我們的數(shù)據(jù)庫(kù),我們先來(lái)實(shí)現(xiàn)通過代碼直接對(duì)權(quán)限數(shù)據(jù)的修改
為了方便,我們就直接在script.py
中寫入我們的函數(shù),可以根據(jù)文章的順序來(lái)添加
1、 群管理
1.1 增加群
from flask import current_app from App.models import Group, GroupAuthority from base64 import b64decode # 對(duì)中文進(jìn)行編碼 async def add(gid, nick): # 傳入QQ群號(hào)和qq群的昵稱就可以添加了 session = current_app.config["db"].session data = session.query(Group).filter_by(qqId=gid).first() # 首先查看一下數(shù)據(jù)庫(kù)中是否收錄了這個(gè)群 if data is None: # 如果沒有則添加,同時(shí),也檢查其是否有邏輯刪除 g = Group(nick, gid) # 創(chuàng)建一個(gè)群 g.group2auth = [GroupAuthority()] # 給群賦予默認(rèn)的權(quán)限 session.add(g) # 添加到數(shù)據(jù)庫(kù)中 session.commit() # 提交事務(wù) return { "status": 200 } # 返回狀態(tài)碼 # 如果群存在,直接修改群名,以及使其可以檢測(cè)到 data.isDetect = True data.name = nick session.commit() return { "status": 300, "error": "該群號(hào)已存在!", }
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 "該群不存在"
這里使用的是邏輯刪的刪除,使得程序無(wú)法訪問到該數(shù)據(jù),即代表刪除效果,同時(shí),數(shù)據(jù)也保存了下來(lái)
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、 權(quán)限管理
2.1 展示權(quán)限
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 "在該群不支持,請(qǐng)與開發(fā)者聯(lián)系!" 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. 已開啟定時(shí)發(fā)消息功能" if data.group2auth[0].dailyBrief else "6. 未開啟定時(shí)發(fā)消息功能" # 是否開啟每日自動(dòng)播報(bào)的功能 return f"{name}\n{chat}\n{welcome}\n{banTalk}\n{click}\n{smallFunction}\n{dailyBrief}\n"
2.2 修改權(quán)限
async def changeAuth(gid, data, ty): """ 傳入群號(hào),要修改的數(shù)據(jù)其為6位數(shù)字,如:"011011",每一個(gè)數(shù)字代表一個(gè)權(quán)限,1為真|0為假 最后傳入的是調(diào)用這個(gè)函數(shù)的類型,是群管理員還是超級(jí)管理員 """ db = current_app.config["db"] session = db.session # 獲取數(shù)據(jù)庫(kù)連接 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) # 獲取到對(duì)應(yīng)的權(quán)限數(shù)字 group = session.query(Group).filter(db.and_(Group.qqId == gid, Group.isDetect)).first() # 獲取對(duì)象 if group: group.group2auth = [GroupAuthority(True, *data)] # 使用元組拆包的方式來(lái)傳參 session.commit() # 提交數(shù)據(jù) _ = await Admin.get(gid) # 獲取到群里面的權(quán)限 ret = f"[CQ:at,qq=%d]設(shè)置成功,設(shè)置后的權(quán)限為:\n{_}" if ty == "group" else f"設(shè)置成功,設(shè)置后的權(quán)限為:\n{_}" else: ret = "該群不支持機(jī)器人" except Exception as e: ret = "[CQ:at,qq=%d]設(shè)置失敗,請(qǐng)查看幫助文檔!" if ty == "group" else "設(shè)置失敗,請(qǐng)查看幫助文檔!" return ret
那么,我們就把我們的權(quán)限管理大概的搭建好了,我們只需要再對(duì)消息進(jìn)行檢測(cè),使其機(jī)器人可以解析命令就完成了
四、 獲取命令
1、 消息分發(fā)
這里,我們來(lái)結(jié)合我們的權(quán)限系統(tǒng),使得機(jī)器人可以對(duì)修改權(quán)限的命令作出回應(yīng)
# !/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 # 用于發(fā)送信息的函數(shù) async def send(id, message, ty): """ 用于發(fā)送消息的函數(shù) :param id: qq號(hào),或者qq群號(hào) :param message: 發(fā)送的消息 :param ty: 傳入的 :return: None """ async with httpx.AsyncClient(base_url="http://127.0.0.1:5700") as client: # 如果發(fā)送的為私聊消息 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): # 這里對(duì)消息進(jìn)行分發(fā),暫時(shí)先設(shè)置一個(gè)簡(jiǎn)單的分發(fā) _ = request.json if _.get("message_type") == "private": # 說明有好友發(fā)送信息過來(lái) uid = _["sender"]["user_id"] # 獲取發(fā)信息的好友qq號(hào) message = _["raw_message"] # 獲取發(fā)送過來(lái)的消息 if message.startswith("/admin:") and str(uid) in current_app.config["SUPER_USER"]: # 其為超級(jí)用戶的命令,通過正則解析出命令,并且做出回應(yīng),那么我們就需要?jiǎng)?chuàng)建一個(gè)函數(shù),來(lái)解析我們的命令,這個(gè)函數(shù)我們后面來(lái)實(shí)現(xiàn) asyncio.run(super_command(uid, message)) # 處理私聊的信息 elif _.get("message_type") == "group" and message.startswith("/admin:") and resp["sender"]["role"] in ["owner", "admin"]: # 獲取群消息,并且,其為群主或者管理員發(fā)送的消息的話 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: # 如果群里面不支持機(jī)器人的話,直接返回 return uid = _["sender"]["user_id"] # 獲取發(fā)信息的好友qq號(hào) message = _.get("raw_message") gid = resp["group_id"] # 獲取群號(hào) asyncio.run(admin_command(uid, gid, message)) # 這里也是先用一個(gè)函數(shù)名來(lái)占位,我們后面來(lái)實(shí)現(xiàn)這個(gè)函數(shù)
2、 解析命令
在 goCqhttp.py
后面繼續(xù)添加以下內(nèi)容
超級(jí)管理員的命令解析
async def super_command(uid, message): # 處理開發(fā)者的命令 com_1 = re.findall(r"/admin:(.*)", message.split()[0])[0] try: com_2 = message.split()[1] # qq群號(hào) 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群的權(quán)限信息 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(): # 修改權(quán)限 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: # 如果都不是的話,發(fā)送幫助文檔 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] # 把命令解析出來(lái) if com_1 == "get": # 獲取本群的權(quán)限信息 ret = await get(gid) await send(gid, ret, ty="group") elif com_1 == "change": # 修改權(quán)限信息 data = list(message.split()[1]) ret = await changeAuth(gid, data, "group") await send(gid, ret % uid, "group") else: # 默認(rèn)為幫助文檔 ret = current_app.config["GROUP_ADMIN"] % uid await send(gid, ret, "group")
最后,我們的權(quán)限系統(tǒng)就做完啦!大家可以運(yùn)行項(xiàng)目嘗試一下哦!python app.py runserver -p5701
同時(shí)里面的每一個(gè)權(quán)限我也會(huì)慢慢的分布實(shí)現(xiàn)!
到此這篇關(guān)于go-cqhttp權(quán)限管理的文章就介紹到這了,更多相關(guān)go cqhttp權(quán)限管理內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
jenkins構(gòu)建go及java項(xiàng)目的方法
這篇文章主要介紹了jenkins構(gòu)建go及java項(xiàng)目,本文通過圖文實(shí)例相結(jié)合給大家介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值了,需要的朋友可以參考下2021-04-04Go語(yǔ)言實(shí)現(xiàn)UDP版聊天小工具的示例詳解
這篇文章主要為大家詳細(xì)介紹了如何利用Go語(yǔ)言實(shí)現(xiàn)聊天小工具(UDP版),文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2023-03-03Go語(yǔ)言擴(kuò)展原語(yǔ)之Semaphore的用法詳解
Go語(yǔ)言的擴(kuò)展包中提供了帶權(quán)重的信號(hào)量?semaphore.Weighted,讓我們可以按照不同的權(quán)重管理資源的訪問,下面小編就來(lái)和大家聊聊它們的具體用法吧2023-07-07使用Go實(shí)現(xiàn)一個(gè)百行聊天服務(wù)器的示例代碼
前段時(shí)間, redis作者整了個(gè)c語(yǔ)言版本的聊天服務(wù)器,代碼量攏共不過百行,于是, 心血來(lái)潮下, 我也整了個(gè)Go語(yǔ)言版本, 簡(jiǎn)單來(lái)說就是實(shí)現(xiàn)了一個(gè)聊天室的功能,文中通過代碼示例給大家介紹的非常詳細(xì),需要的朋友可以參考下2023-12-12