🎉🍾♻️refactor(core): 重构核心鉴权与缓存机制，引入异步发送队列以提升性能 🚀 (#2088)

* 添加测试插件 * ✨ feat(auth): 添加缓存机制以优化用户和插件数据查询性能 * 🗑️ chore(help): 删除笨蛋检测插件代码 * ✨ feat(bot): 添加对扩展插件的加载支持 * ✨ feat(auth): 优化权限检查和缓存机制，增加用户和插件数据的并行查询 * ✨ feat(cache): 引入运行时缓存机制，优化用户和插件的ban记录管理 * ✨ feat(auth): 更新is_ban函数文档，添加参数和返回值说明 * ``` feat(auth): 使用内存缓存优化权限验证性能 - 移除数据库查询超时控制，改用 LevelUserMemoryCache、BotMemoryCache 和 PluginLimitMemoryCache 进行缓存查询 - 优化 auth_admin、auth_bot、auth_limit 权限验证逻辑，提升响应速度 - 添加 background 参数支持异步发送权限不足提示消息 - 移除 asyncio 依赖，简化代码结构 fix(auth): 修复限制通知频率控制问题 - 实现限制通知冷却机制，避免重复发送相同限制消息 - 添加 AUTH_LIMIT_NOTICE_CD 配置项，默认值为 2 秒 - 使用 FreqLimiter 控制限制通知发送频率 refactor(models): 增强模型数据变更时的缓存同步 - 在 BotConsole、GroupConsole、LevelUser、PluginLimit 模型的 create、update_or_create、save、delete 方法中自动更新对应缓存 - 确保数据库和内存缓存数据一致性 docs(ban_console): 修正文档注释并优化日志信息 - 修正 BanConsole 类中方法的文档字符串，使用标准参数和返回值格式 - 优化调试日志信息，使描述更加清晰准确 ``` * ✨ feat(auth): 更新Limit类以支持PluginLimitSnapshot，优化限制信息处理 * ✨ feat(bot_manage): 优化Bot控制台初始化逻辑，处理IntegrityError异常 * ✨ feat(mmm1): 新增消息推送功能，支持私聊和群聊事件处理 * ✨ feat(group_member_update): 优化群组成员更新逻辑，增加活动跟踪和消息记录功能 * ✨ feat(mmm1): 删除冗余的消息推送功能代码 * ✨ feat(auth): 优化权限检查逻辑，增加模块阻止功能和缓存处理 * ✨ feat(auth): 优化权限检查逻辑，增加快速ban检测和前置检查功能 * ✨ feat(chat_history): 增强消息处理规则，添加时间间隔限制以防止重复消息 ✨ feat(data_source): 引入异步获取群成员信息的功能，优化用户信息更新逻辑 * ✨ feat(send_queue): 添加异步发送队列以优化API调用和速率限制 * ✨ feat(auth): 添加缓存就绪检查以优化权限处理逻辑 * ✨ feat(plugins): 移除不必要的插件加载以简化插件管理 * ✨ feat(group_console): 优化群组获取逻辑，添加缓存检查以提升性能 ✨ 只接收缓存完成之后时间的消息 * ✨ feat(auth): 添加异步任务管理和超载检测，优化权限处理逻辑 ✨ feat(chat_history): 修改规则函数为异步，提升消息处理效率 ✨ feat(group_handle): 增加安全获取群组信息的异步方法，添加超时处理 ✨ feat(record_request): 引入安全获取群组信息的异步方法，优化群邀请处理 ✨ feat(ban_memory_cache): 增强禁言内存缓存，添加负缓存机制 ✨ feat(message_load): 新增消息负载检测功能，优化任务调度 ✨ feat(scheduler): 在调度器中集成消息压力检测，优化任务执行 ✨ feat(send_queue): 引入异步任务管理，优化发送队列处理 * ✨ feat(auth): 添加对 LevelUserSnapshot 和 BotSnapshot 的支持，优化权限检查逻辑 ✨ 格式化 * ✨ feat(db_context): 增强 get_or_create 方法，处理并发创建冲突并回退查询已存在记录 * ✨ feat(bot_manage): 增强 init_bot_console 方法，处理并发创建冲突并回退查询已存在的 bot 数据 * 🚨 auto fix by pre-commit hooks * ✨ feat(runtime_cache): 优化消息处理逻辑，支持 bytes 和 bytearray 类型的联合判断 * ✨ style(runtime_cache): 格式化代码，优化多行表达式的可读性 * 🚨 auto fix by pre-commit hooks * ✨ refactor: 优化类型注解，改进群组成员更新逻辑和平台处理 ✨ 格式化 * 🚨 auto fix by pre-commit hooks * ✨ refactor: 优化代码格式，增强可读性并修复类型注解 * 🚨 auto fix by pre-commit hooks * ✨ refactor: 更新文档注释，增强is_ban函数的可读性 * ✨ refactor: 优化代码结构，移除冗余函数，增强可读性并改进任务调度逻辑 * ✨ refactor: 调整定时任务时间，优化渲染服务的初始化逻辑，增强代码可读性 --------- Co-authored-by: ATTomatoo <1126160939@qq.com> Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
2026-10-18 18:50:05 +08:00 · 2026-01-28 10:15:24 +08:00
parent c9f0a8b9d9
commit 837330e30a
41 changed files with 3656 additions and 763 deletions
@@ -135,7 +135,9 @@ avatar_service = AvatarService()


@scheduler.scheduled_job(
-    "interval", hours=Config.get_config("avatar_cache", "CLEANUP_INTERVAL_HOURS", 24)
+    "cron",
+    hour=4,
+    minute=30,
 )
 async def _run_avatar_cache_cleanup():
    await avatar_service._cleanup_cache()
@@ -81,7 +81,8 @@ import asyncio
 from collections.abc import Callable
 from datetime import datetime
 from functools import wraps
-from typing import Any, ClassVar, Generic, TypeVar, get_type_hints
+from typing import Any, ClassVar, Generic, TypeVar, cast, get_type_hints
+from typing_extensions import Self

 from aiocache import Cache as AioCache
 from aiocache import SimpleMemoryCache
@@ -115,6 +116,8 @@ __all__ = [
    "CacheRoot",
 ]

+from . import runtime_cache as _runtime_cache  # noqa: F401
+
 T = TypeVar("T")
 U = TypeVar("U")

@@ -291,11 +294,11 @@ class CacheManager:
    _dict_caches: ClassVar[dict[str, "CacheDict"]] = {}
    _enabled = False  # 缓存启用标记

-    def __new__(cls) -> "CacheManager":
+    def __new__(cls) -> Self:
        """单例模式"""
        if cls._instance is None:
            cls._instance = super().__new__(cls)
-        return cls._instance
+        return cast(Self, cls._instance)

    @property
    def enabled(self) -> bool:
@@ -47,7 +47,8 @@ class CacheDict(Generic[T]):
            T: 字典值
        """
        if value := self._data.get(key):
-            if self.expire_time(key):
+            if value.expire_time > 0 and value.expire_time < time.time():
+                del self._data[key]
                raise KeyError(f"键 {key} 已过期")
            return value.value
        raise KeyError(f"键 {key} 不存在")
@@ -80,7 +81,13 @@ class CacheDict(Generic[T]):
        返回:
            bool: 是否存在
        """
-        return False if key not in self._data else bool(self.expire_time(key))
+        data = self._data.get(key)
+        if data is None:
+            return False
+        if data.expire_time > 0 and data.expire_time < time.time():
+            del self._data[key]
+            return False
+        return True

    def get(self, key: str, default: Any = None) -> T | None:
        """获取字典项，如果不存在返回默认值
@@ -93,11 +100,11 @@ class CacheDict(Generic[T]):
            Any: 字典值或默认值
        """
        if value := self._data.get(key):
-            if self.expire_time(key):
+            if value.expire_time > 0 and value.expire_time < time.time():
+                del self._data[key]
                return default
-        if not value:
-            return default
-        return default if value.value is None else value.value
+            return default if value.value is None else value.value
+        return default

    def set(self, key: str, value: Any, expire: int | None = None):
        """设置字典项
@@ -126,16 +133,13 @@ class CacheDict(Generic[T]):
        返回:
            Any: 字典值或默认值
        """
-        if key not in self._data:
+        data = self._data.get(key)
+        if data is None:
            return default
-
-        data = self._data.pop(key)
-
-        # 检查是否过期
        if data.expire_time > 0 and data.expire_time < time.time():
            del self._data[key]
            return default
-
+        del self._data[key]
        return data.value

    def clear(self) -> None:
@@ -5,7 +5,11 @@ from typing import Any, ClassVar
 from typing_extensions import Self

 from tortoise.backends.base.client import BaseDBAsyncClient
-from tortoise.exceptions import IntegrityError, MultipleObjectsReturned
+from tortoise.exceptions import (
+    IntegrityError,
+    MultipleObjectsReturned,
+    TransactionManagementError,
+)
 from tortoise.models import Model as TortoiseModel
 from tortoise.transactions import in_transaction

@@ -129,9 +133,23 @@ class Model(TortoiseModel):
        **kwargs: Any,
    ) -> tuple[Self, bool]:
        """获取或创建数据（无锁版本，依赖数据库约束）"""
-        result = await super().get_or_create(
-            defaults=defaults, using_db=using_db, **kwargs
-        )
+        try:
+            result = await super().get_or_create(
+                defaults=defaults, using_db=using_db, **kwargs
+            )
+        except IntegrityError:
+            # 并发创建冲突时，回退为查询已存在记录
+            try:
+                if using_db is not None:
+                    obj = await cls.filter(**kwargs).using_db(using_db).get()
+                    result = (obj, False)
+                else:
+                    raise TransactionManagementError("fallback to new transaction")
+            except TransactionManagementError:
+                async with in_transaction() as connection:
+                    obj = await cls.filter(**kwargs).using_db(connection).get()
+                    result = (obj, False)
+
        if cache_type := cls.get_cache_type():
            await CacheRoot.invalidate_cache(cache_type, cls.get_cache_key(result[0]))
        return result
@@ -21,7 +21,7 @@ from typing import (
    get_origin,
    get_type_hints,
 )
-from typing_extensions import override
+from typing_extensions import Self, override

 from httpx import NetworkError, TimeoutException

@@ -421,10 +421,10 @@ class ToolProviderManager:

    _instance: "ToolProviderManager | None" = None

-    def __new__(cls) -> "ToolProviderManager":
+    def __new__(cls) -> Self:
        if cls._instance is None:
            cls._instance = super().__new__(cls)
-        return cls._instance
+        return cast(Self, cls._instance)

    def __init__(self):
        if hasattr(self, "_initialized") and self._initialized:
@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+import time
+
+_OVERLOAD_UNTIL = 0.0
+
+
+def signal_overload(duration: float = 5.0) -> None:
+    """Mark the system as overloaded for a short time window."""
+    global _OVERLOAD_UNTIL
+    if duration <= 0:
+        return
+    now = time.monotonic()
+    until = now + duration
+    if until > _OVERLOAD_UNTIL:
+        _OVERLOAD_UNTIL = until
+
+
+def is_overloaded() -> bool:
+    return time.monotonic() < _OVERLOAD_UNTIL
+
+
+def should_pause_tasks() -> bool:
+    return is_overloaded()
@@ -10,6 +10,7 @@ from collections.abc import Callable
 from datetime import datetime
 from functools import partial
 import random
+import time

 import nonebot
 from nonebot.adapters import Bot
@@ -23,6 +24,7 @@ from pydantic import BaseModel
 from zhenxun.configs.config import Config
 from zhenxun.models.scheduled_job import ScheduledJob
 from zhenxun.services.log import logger
+from zhenxun.services.message_load import should_pause_tasks
 from zhenxun.utils.common_utils import CommonUtils
 from zhenxun.utils.decorator.retry import Retry
 from zhenxun.utils.pydantic_compat import parse_as
@@ -32,6 +34,7 @@ from .types import ExecutionPolicy, ScheduleContext

 JOB_PREFIX = "zhenxun_schedule_"
 SCHEDULE_CONCURRENCY_KEY = "all_groups_concurrency_limit"
+_LAST_PRESSURE_SKIP = 0.0


 class APSchedulerAdapter:
@@ -362,6 +365,14 @@ async def _execute_job(
        logger.error("执行持久化任务时 schedule_id 不能为空。")
        return

+    global _LAST_PRESSURE_SKIP
+    if should_pause_tasks():
+        now = time.time()
+        if now - _LAST_PRESSURE_SKIP > 30:
+            _LAST_PRESSURE_SKIP = now
+            logger.info("scheduler paused due to message pressure")
+        return
+
    scheduler_manager._running_tasks.add(schedule_id)
    try:
        schedule = await ScheduleRepository.get_by_id(schedule_id)
@@ -0,0 +1,78 @@
+import asyncio
+import time
+from typing import Any
+
+import nonebot
+from nonebot.adapters import Bot
+
+from zhenxun.services.log import logger
+
+_SEND_APIS = {"send_msg", "send_like"}
+_QUEUE: asyncio.Queue[tuple[Bot, str, dict[str, Any], asyncio.Future]] = asyncio.Queue()
+_WORKERS = 3
+_MIN_INTERVAL = 0.05
+_SEND_LOCK = asyncio.Lock()
+_LAST_SEND_TS = 0.0
+_API_SEMAPHORE = asyncio.Semaphore(3)
+_ORIG_CALL_API = Bot.call_api
+_PATCHED = False
+_WORKER_TASKS: list[asyncio.Task] = []
+
+
+async def _rate_limit():
+    global _LAST_SEND_TS
+    async with _SEND_LOCK:
+        now = time.monotonic()
+        wait = _MIN_INTERVAL - (now - _LAST_SEND_TS)
+        if wait > 0:
+            await asyncio.sleep(wait)
+        _LAST_SEND_TS = time.monotonic()
+
+
+async def _worker(worker_id: int):
+    while True:
+        bot, api, data, future = await _QUEUE.get()
+        try:
+            await _rate_limit()
+            async with _API_SEMAPHORE:
+                result = await _ORIG_CALL_API(bot, api, **data)
+            if not future.done():
+                future.set_result(result)
+        except Exception as exc:
+            if not future.done():
+                future.set_exception(exc)
+            logger.warning(
+                f"send queue failed: {api}",
+                "SendQueue",
+                target=getattr(bot, "self_id", None),
+                e=exc,
+            )
+        finally:
+            _QUEUE.task_done()
+
+
+async def _queued_call_api(self: Bot, api: str, **data: Any):
+    if api not in _SEND_APIS:
+        return await _ORIG_CALL_API(self, api, **data)
+    loop = asyncio.get_running_loop()
+    future: asyncio.Future = loop.create_future()
+    await _QUEUE.put((self, api, data, future))
+    return await future
+
+
+def patch_send_queue() -> None:
+    global _PATCHED
+    if _PATCHED:
+        return
+    Bot.call_api = _queued_call_api  # type: ignore[assignment]
+    _PATCHED = True
+
+
+driver = nonebot.get_driver()
+
+
+@driver.on_startup
+async def _start_send_queue():
+    patch_send_queue()
+    for idx in range(_WORKERS):
+        _WORKER_TASKS.append(asyncio.create_task(_worker(idx)))