zhenxun_bot/zhenxun/models/plugin_info.py
HibiKier 8649aaaa54
引入缓存机制 (#1889)
* 添加全局cache

*  构建缓存,hook使用缓存

*  新增数据库Model方法监控

*  数据库添加semaphore锁

* 🩹 优化webapi返回数据

*  添加增量缓存与缓存过期

* 🎨 优化检测代码结构

*  优化hook权限检测性能

* 🐛 添加新异常判断跳过权限检测

*  添加插件limit缓存

* 🎨 代码格式优化

* 🐛  修复代码导入

* 🐛 修复刷新时检查

* 👽 Rename exception for missing database URL in initialization

*  Update default database URL to SQLite in configuration

* 🔧 Update tortoise-orm and aiocache dependencies restrictions; add optional redis and asyncpg support

* 🐛 修复ban检测

* 🐛 修复所有插件关闭时缓存更新

* 🐛 尝试迁移至aiocache

* 🐛 完善aiocache缓存

*  代码性能优化

* 🐛 移除获取封禁缓存时的日志记录

* 🐛 修复缓存类型声明,优化封禁用户处理逻辑

* 🐛 优化LevelUser权限更新逻辑及数据库迁移

*  cache支持redis连接

* 🚨 auto fix by pre-commit hooks

*  :增强获取群组的安全性和准确性。同时,优化了缓存管理中的相关逻辑,确保缓存操作的一致性。

*  feat(auth_limit): 将插件初始化逻辑的启动装饰器更改为优先级管理器

* 🔧 修复日志记录级别

* 🔧 更新数据库连接字符串

* 🔧 更新数据库连接字符串为内存数据库,并优化权限检查逻辑

*  feat(cache): 增加缓存功能配置项,并新增数据访问层以支持缓存逻辑

* ♻️ 重构cache

*  feat(cache): 增强缓存管理,新增缓存字典和缓存列表功能,支持过期时间管理

* 🔧 修复Notebook类中的viewport高度设置,将其从1000调整为10

*  更新插件管理逻辑,替换缓存服务为CacheRoot并优化缓存失效处理

*  更新RegisterConfig类中的type字段

*  修复清理重复记录逻辑,确保检查记录的id属性有效性

*  超级无敌大优化,解决延迟与卡死问题

*  更新封禁功能,增加封禁时长参数和描述,优化插件信息返回结构

*  更新zhenxun_help.py中的viewport高度,将其从453调整为10,以优化页面显示效果

*  优化插件分类逻辑,增加插件ID排序,并更新插件信息返回结构

---------

Co-authored-by: BalconyJH <balconyjh@gmail.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
2025-07-14 22:35:29 +08:00

113 lines
4.4 KiB
Python

from typing_extensions import Self
from tortoise import fields
from zhenxun.models.plugin_limit import PluginLimit # noqa: F401
from zhenxun.services.db_context import Model
from zhenxun.utils.enum import BlockType, CacheType, PluginType
class PluginInfo(Model):
id = fields.IntField(pk=True, generated=True, auto_increment=True)
"""自增id"""
module = fields.CharField(255, description="模块名")
"""模块名"""
module_path = fields.CharField(255, description="模块路径", unique=True)
"""模块路径"""
name = fields.CharField(255, description="插件名称")
"""插件名称"""
status = fields.BooleanField(default=True, description="全局开关状态")
"""全局开关状态"""
block_type: BlockType | None = fields.CharEnumField(
BlockType, default=None, null=True, description="禁用类型"
)
"""禁用类型"""
load_status = fields.BooleanField(default=True, description="加载状态")
"""加载状态"""
author = fields.CharField(255, null=True, description="作者")
"""作者"""
version = fields.CharField(max_length=255, null=True, description="版本")
"""版本"""
level = fields.IntField(default=5, description="所需群权限")
"""所需群权限"""
default_status = fields.BooleanField(default=True, description="进群默认开关状态")
"""进群默认开关状态"""
limit_superuser = fields.BooleanField(default=False, description="是否限制超级用户")
"""是否限制超级用户"""
menu_type = fields.CharField(max_length=255, default="", description="菜单类型")
"""菜单类型"""
plugin_type = fields.CharEnumField(PluginType, null=True, description="插件类型")
"""插件类型"""
cost_gold = fields.IntField(default=0, description="调用插件所需金币")
"""调用插件所需金币"""
plugin_limit = fields.ReverseRelation["PluginLimit"]
"""插件限制"""
admin_level = fields.IntField(default=0, null=True, description="调用所需权限等级")
"""调用所需权限等级"""
ignore_prompt = fields.BooleanField(default=False, description="是否忽略提示")
"""是否忽略阻断提示"""
is_delete = fields.BooleanField(default=False, description="是否删除")
"""是否删除"""
parent = fields.CharField(max_length=255, null=True, description="父插件")
"""父插件"""
is_show = fields.BooleanField(default=True, description="是否显示在帮助中")
"""是否显示在帮助中"""
impression = fields.FloatField(default=0, description="插件好感度限制")
"""插件好感度限制"""
class Meta: # pyright: ignore [reportIncompatibleVariableOverride]
table = "plugin_info"
table_description = "插件基本信息"
cache_type = CacheType.PLUGINS
"""缓存类型"""
cache_key_field = "module"
"""缓存键字段"""
@classmethod
async def get_plugin(
cls, load_status: bool = True, filter_parent: bool = True, **kwargs
) -> Self | None:
"""获取插件列表
参数:
load_status: 加载状态.
filter_parent: 过滤父组件
返回:
Self | None: 插件
"""
if filter_parent:
return await cls.get_or_none(
load_status=load_status, plugin_type__not=PluginType.PARENT, **kwargs
)
return await cls.get_or_none(load_status=load_status, **kwargs)
@classmethod
async def get_plugins(
cls, load_status: bool = True, filter_parent: bool = True, **kwargs
) -> list[Self]:
"""获取插件列表
参数:
load_status: 加载状态.
filter_parent: 过滤父组件
返回:
list[Self]: 插件列表
"""
if filter_parent:
return await cls.filter(
load_status=load_status, plugin_type__not=PluginType.PARENT, **kwargs
).all()
return await cls.filter(load_status=load_status, **kwargs).all()
@classmethod
async def _run_script(cls):
return [
"ALTER TABLE plugin_info ADD COLUMN parent character varying(255);",
"ALTER TABLE plugin_info ADD COLUMN is_show boolean DEFAULT true;",
"ALTER TABLE plugin_info ADD COLUMN ignore_prompt boolean DEFAULT false;",
"ALTER TABLE plugin_info ADD COLUMN impression float DEFAULT 0;",
]