go-cqhttp權限管理系統(tǒng)的實現(xiàn)代碼
更新時間：2022年09月15日 10:44:12 作者：A-L-Kun
這篇文章主要介紹了go-cqhttp權限管理,本文通過實例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下
權限管理系統(tǒng)
一、 概述
在寫好我們的智能聊天功能之后，大家有沒有感覺很煩呢？感覺這個機器人在群里面一直被艾特，一直被戳一戳。那么，我們有沒有一種方法，使得其在群里面的權限可控呢？
或許大家看到這個問題就想到了一個方法，那就是通過python文件，但是使用python文件保存的話有一個缺點，那就是修改配置文件后，需要重新運行我們的項目，這會讓我們覺得很麻煩！
那么，還有沒有更好的方法呢？給大家一分鐘時間思考……好，大家思考出來了嗎？我的想法是，將權限存儲到數(shù)據(jù)庫中，當我們需要調用這個功能的時候，通過調用數(shù)據(jù)庫，來判斷是否有進行這項功能的權限！這里，我們選擇的是mysql數(shù)據(jù)庫，關于對數(shù)據(jù)庫的操作，我已經(jīng)給大家準備好了！
- mysql數(shù)據(jù)庫的安裝：linux 中安裝數(shù)據(jù)庫的方法
- mysql基本語法：關于數(shù)據(jù)庫操作的基本語法
- SQLAchemy 常用操作：使用ORM操作數(shù)據(jù)庫
看到SQLAchemy，就應該有小伙伴要說了，既然我們看了mysql的基本語法，我們完全可以通過SQL語句來操作我們的數(shù)據(jù)庫，為什么需要使用ORM來操作數(shù)據(jù)庫呢？其實不然，我們使用SQL操作數(shù)據(jù)庫的話，就無法處理一些高并發(fā)的操作了，會造成嚴重的阻塞，使用SQLAchemy可以創(chuàng)建數(shù)據(jù)庫連接池，緩解服務器的壓力，同時ORM對于一些不是很熟悉SQL的小伙伴來說比較友好。
好了，相信大家也看完了上面對數(shù)據(jù)庫的操作的文章了，同時也看完了我寫的關于flask的全部文章了，廢話不多說了，來開始實現(xiàn)我們的權限管理系統(tǒng)
展示一下我的目錄結構：
|-- App | |-- exts.py | |-- __init__.py | |-- models.py | |-- script.py | |-- settings.py | `-- views | |-- goCqhttp.py | |-- __init__.py |-- app.py
二、 創(chuàng)建表
1、 創(chuàng)建
在實現(xiàn)權限管理系統(tǒng)之前，我們肯定是需要創(chuàng)建相應的數(shù)據(jù)表來存儲我們的數(shù)據(jù)的
我們使用flask-sqlachemy來實現(xiàn)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): # 創(chuàng)建一個群的表
__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): # 創(chuàng)建一個權限管理表
__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) # 是否開啟小功能，如疫情數(shù)據(jù)查詢等
dailyBrief = db.Column(db.INTEGER, default=0) # 是否開啟每日簡報功能
groupId = db.Column(db.INTEGER, db.ForeignKey("group.id", ondelete="CASCADE")) # 外鍵約束，同時進行級聯(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表作為主表，用來存放各群的數(shù)據(jù)，同時將機器人對各群的權限存儲到另一張表中，因為是一對一的關系，我們使用同時對表使用外鍵約束來關聯(lián)主表以及子表
2、 生成
把表的結構創(chuàng)建完后，我們來創(chuàng)建主程序，用來生成我們創(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ù)庫
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", ] # 超級用戶，其可以私發(fā)消息給機器人，管理機器人對所有群的權限，當然也可以存儲到數(shù)據(jù)庫中，這里為了方便，存儲在配置文件中，大家可根據(jù)這篇文章執(zhí)行實現(xiàn)
ADMIN = 3500515050 # 開發(fā)者
class Help:
"""幫助文檔"""
ADMIN_HELP = """開發(fā)者命令提示：
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
如果還是有問題，請與開發(fā)人員聯(lián)系哦！"""
GROUP_ADMIN = """[CQ:at,qq=%d]命令提示：
1. 查看群權限:\n/admin:get
2. 修改群權限：\n/admin:change |聊天功能|入群歡迎|管理群|戳一戳|拓展功能|定時功能|(比如#admin:change# 110011)\n
如果還是有問題，請與開發(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 # 后面主要是通過這個接口來對消息的分發(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ù)庫中
3、 映射
在項目的根目錄運行這個代碼：
flask db init # 初始化操作 flask db migrate # 遷移操作 flask db upgrade # 更新操作
然后，我們的數(shù)據(jù)表就更新完成了
三、 增刪改查
首先，我們先不考慮如何監(jiān)控我們的QQ消息，通過消息來操控我們的數(shù)據(jù)庫，我們先來實現(xiàn)通過代碼直接對權限數(shù)據(jù)的修改
為了方便，我們就直接在script.py中寫入我們的函數(shù)，可以根據(jù)文章的順序來添加
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() # 首先查看一下數(shù)據(jù)庫中是否收錄了這個群
if data is None: # 如果沒有則添加，同時，也檢查其是否有邏輯刪除
g = Group(nick, gid) # 創(chuàng)建一個群
g.group2auth = [GroupAuthority()] # 給群賦予默認的權限
session.add(g) # 添加到數(shù)據(jù)庫中
session.commit() # 提交事務
return {
"status": 200
} # 返回狀態(tài)碼
# 如果群存在，直接修改群名，以及使其可以檢測到
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 "該群不存在"
這里使用的是邏輯刪的刪除，使得程序無法訪問到該數(shù)據(jù)，即代表刪除效果，同時，數(shù)據(jù)也保存了下來
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 "在該群不支持，請與開發(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. 已開啟定時發(fā)消息功能" if data.group2auth[0].dailyBrief else "6. 未開啟定時發(fā)消息功能" # 是否開啟每日自動播報的功能
return f"{name}\n{chat}\n{welcome}\n{banTalk}\n{click}\n{smallFunction}\n{dailyBrief}\n"
2.2 修改權限
async def changeAuth(gid, data, ty):
"""
傳入群號，要修改的數(shù)據(jù)其為6位數(shù)字，如："011011"，每一個數(shù)字代表一個權限，1為真|0為假
最后傳入的是調用這個函數(shù)的類型，是群管理員還是超級管理員
"""
db = current_app.config["db"]
session = db.session # 獲取數(shù)據(jù)庫連接
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) # 獲取到對應的權限數(shù)字
group = session.query(Group).filter(db.and_(Group.qqId == gid, Group.isDetect)).first() # 獲取對象
if group:
group.group2auth = [GroupAuthority(True, *data)] # 使用元組拆包的方式來傳參
session.commit() # 提交數(shù)據(jù)
_ = 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、 消息分發(fā)
這里，我們來結合我們的權限系統(tǒ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號，或者qq群號
: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):
# 這里對消息進行分發(fā)，暫時先設置一個簡單的分發(fā)
_ = request.json
if _.get("message_type") == "private": # 說明有好友發(fā)送信息過來
uid = _["sender"]["user_id"] # 獲取發(fā)信息的好友qq號
message = _["raw_message"] # 獲取發(fā)送過來的消息
if message.startswith("/admin:") and str(uid) in current_app.config["SUPER_USER"]:
# 其為超級用戶的命令，通過正則解析出命令，并且做出回應，那么我們就需要創(chuàng)建一個函數(shù)，來解析我們的命令，這個函數(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:
# 如果群里面不支持機器人的話，直接返回
return
uid = _["sender"]["user_id"] # 獲取發(fā)信息的好友qq號
message = _.get("raw_message")
gid = resp["group_id"] # 獲取群號
asyncio.run(admin_command(uid, gid, message)) # 這里也是先用一個函數(shù)名來占位，我們后面來實現(xiàn)這個函數(shù)
2、 解析命令
在 goCqhttp.py后面繼續(xù)添加以下內容
超級管理員的命令解析
async def super_command(uid, message): # 處理開發(fā)者的命令
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: # 如果都不是的話，發(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] # 把命令解析出來
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")
最后，我們的權限系統(tǒng)就做完啦！大家可以運行項目嘗試一下哦！python app.py runserver -p5701
同時里面的每一個權限我也會慢慢的分布實現(xiàn)！
到此這篇關于go-cqhttp權限管理的文章就介紹到這了,更多相關go cqhttp權限管理內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
這篇文章主要介紹了jenkins構建go及java項目,本文通過圖文實例相結合給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值了，需要的朋友可以參考下2021-04-04
這篇文章主要為大家詳細介紹了如何利用Go語言實現(xiàn)聊天小工具（UDP版），文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2023-03-03
本文詳細講解了Go中的gRPC入門教程，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2022-03-03
這篇文章主要介紹了Golang 運算符及位運算詳解，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-12-12
Go語言的擴展包中提供了帶權重的信號量?semaphore.Weighted，讓我們可以按照不同的權重管理資源的訪問，下面小編就來和大家聊聊它們的具體用法吧2023-07-07
前段時間, redis作者整了個c語言版本的聊天服務器,代碼量攏共不過百行,于是, 心血來潮下, 我也整了個Go語言版本, 簡單來說就是實現(xiàn)了一個聊天室的功能,文中通過代碼示例給大家介紹的非常詳細,需要的朋友可以參考下2023-12-12
這篇文章主要介紹了gRPC中攔截器的使用詳解,本次主要介紹在gRPC中使用攔截器,包括一元攔截器和流式攔截器,在攔截器中添加JWT認證,客戶端登錄之后會獲得token,請求特定的API時候需要帶上token才能訪問,需要的朋友可以參考下2023-10-10
這篇文章主要介紹了golang指數(shù)運算操作，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-12-12
今天小編就為大家分享一篇關于Golang捕獲panic堆棧信息的講解，小編覺得內容挺不錯的，現(xiàn)在分享給大家，具有很好的參考價值，需要的朋友一起跟隨小編來看看吧2019-04-04
這篇文章主要給大家介紹了Golang搭建grpc環(huán)境的流程步驟,文中通過圖文結合的方式給大家講解的非常詳細,對大家了解Golang搭建grpc環(huán)境有一定的幫助,需要的朋友可以參考下2024-03-03