From 26150c29245f5709f613ac4b446db7068c481a85 Mon Sep 17 00:00:00 2001 From: HibiKier <775757368@qq.com> Date: Wed, 17 Dec 2025 15:03:44 +0800 Subject: [PATCH] refactor: align auth checker optimization with DataAccess caching --- .../hooks/auth/CACHE_OPTIMIZATION.md | 249 ++++++++++++ .../hooks/auth/IMMEDIATE_IMPROVEMENTS.md | 373 +++++++++++++++++ .../hooks/auth/OPTIMIZATION_PROPOSAL.md | 234 +++++++++++ .../hooks/auth_checker_optimized.py | 381 ++++++++++++++++++ 4 files changed, 1237 insertions(+) create mode 100644 zhenxun/builtin_plugins/hooks/auth/CACHE_OPTIMIZATION.md create mode 100644 zhenxun/builtin_plugins/hooks/auth/IMMEDIATE_IMPROVEMENTS.md create mode 100644 zhenxun/builtin_plugins/hooks/auth/OPTIMIZATION_PROPOSAL.md create mode 100644 zhenxun/builtin_plugins/hooks/auth_checker_optimized.py diff --git a/zhenxun/builtin_plugins/hooks/auth/CACHE_OPTIMIZATION.md b/zhenxun/builtin_plugins/hooks/auth/CACHE_OPTIMIZATION.md new file mode 100644 index 00000000..5f691bfe --- /dev/null +++ b/zhenxun/builtin_plugins/hooks/auth/CACHE_OPTIMIZATION.md @@ -0,0 +1,249 @@ +# 权限检查系统 - 缓存优化指南 + +## DataAccess 缓存机制说明 + +### 已实现的缓存功能 + +`DataAccess` 类已经实现了完整的 Redis 缓存机制，包括： + +1. **自动缓存查询结果** + ```python + # 第一次查询：从数据库获取并缓存 + plugin = await plugin_dao.safe_get_or_none(module="example") + + # 后续查询：直接从缓存获取（如果缓存未过期） + plugin = await plugin_dao.safe_get_or_none(module="example") + ``` + +2. **空结果缓存（5分钟TTL）** + ```python + # 查询不存在的记录时，会缓存空结果标记 + # 避免频繁查询数据库确认记录不存在 + user = await user_dao.safe_get_or_none(user_id="nonexistent") + # 5分钟内再次查询相同记录，直接从缓存返回 None + ``` + +3. **缓存统计** + ```python + from zhenxun.services.data_access import DataAccess + + # 获取缓存统计信息 + stats = DataAccess.get_cache_stats() + # 返回格式： + # [ + # { + # "cache_type": "GLOBAL_ALL_PLUGINS", + # "hits": 1000, # 缓存命中次数 + # "null_hits": 50, # 空结果缓存命中次数 + # "misses": 200, # 缓存未命中次数 + # "hit_rate": "84.00%" # 缓存命中率 + # }, + # ... + # ] + ``` + +4. **自动缓存失效** + - 数据创建、更新、删除时自动清除相关缓存 + - 支持手动清除：`await dao.clear_cache(**kwargs)` + +## 权限检查中的缓存优化 + +### 当前缓存使用情况 + +在权限检查系统中，以下数据访问已经自动使用缓存： + +1. **PluginInfo（插件信息）** + ```python + plugin_dao = DataAccess(PluginInfo) + plugin = await plugin_dao.safe_get_or_none(module=module) + # ✅ 自动缓存，基于 module 字段 + ``` + +2. **UserConsole（用户信息）** + ```python + user_dao = DataAccess(UserConsole) + user = await user_dao.get_by_func_or_none( + UserConsole.get_user, False, user_id=user_id + ) + # ✅ 自动缓存，基于 user_id 字段 + ``` + +3. **GroupConsole（群组信息）** + ```python + group_dao = DataAccess(GroupConsole) + group = await group_dao.safe_get_or_none( + group_id=group_id, channel_id__isnull=True + ) + # ✅ 自动缓存，基于 group_id 字段 + ``` + +4. **BanConsole（ban记录）** + ```python + ban_dao = DataAccess(BanConsole) + ban = await ban_dao.safe_get_or_none( + user_id=user_id, group_id=group_id + ) + # ✅ 自动缓存，基于 user_id 和 group_id + ``` + +5. **BotConsole（Bot信息）** + ```python + bot_dao = DataAccess(BotConsole) + bot = await bot_dao.safe_get_or_none(bot_id=bot_id) + # ✅ 自动缓存，基于 bot_id 字段 + ``` + +6. **LevelUser（用户权限）** + ```python + level_dao = DataAccess(LevelUser) + level = await level_dao.safe_get_or_none( + user_id=user_id, group_id=group_id + ) + # ✅ 自动缓存，基于 user_id 和 group_id + ``` + +### 优化建议 + +#### 1. 并行获取数据，提高缓存命中率 + +**当前实现**： +```python +# 串行获取，可能错过缓存优势 +plugin = await plugin_dao.safe_get_or_none(module=module) +user = await user_dao.get_by_func_or_none(...) +group = await group_dao.safe_get_or_none(...) +``` + +**优化后**： +```python +# 并行获取，DataAccess 会自动处理缓存 +plugin, user, group = await asyncio.gather( + plugin_dao.safe_get_or_none(module=module), + user_dao.get_by_func_or_none(UserConsole.get_user, False, user_id=user_id), + group_dao.safe_get_or_none(group_id=group_id, channel_id__isnull=True) + if group_id else asyncio.sleep(0) +) +``` + +#### 2. 提前获取所有需要的数据 + +**当前实现**： +```python +# group 数据在 hooks 执行时才获取 +# 在 auth 函数中 +group = None +if entity.group_id: + group = await group_dao.safe_get_or_none(...) +``` + +**优化后**： +```python +# 在获取 plugin 和 user 时，同时获取 group +# 统一数据获取，充分利用缓存 +async def get_all_auth_data(module, user_id, group_id): + plugin_dao = DataAccess(PluginInfo) + user_dao = DataAccess(UserConsole) + group_dao = DataAccess(GroupConsole) if group_id else None + + tasks = [ + plugin_dao.safe_get_or_none(module=module), + user_dao.get_by_func_or_none(UserConsole.get_user, False, user_id=user_id), + ] + if group_id and group_dao: + tasks.append( + group_dao.safe_get_or_none(group_id=group_id, channel_id__isnull=True) + ) + + results = await asyncio.gather(*tasks) + return results +``` + +#### 3. 监控缓存效果 + +**实现方案**： +```python +from zhenxun.services.data_access import DataAccess + +async def auth(...): + # 记录开始时的缓存统计 + cache_stats_before = DataAccess.get_cache_stats() + + try: + # 执行权限检查 + ... + finally: + # 记录结束时的缓存统计 + cache_stats_after = DataAccess.get_cache_stats() + + # 分析缓存效果 + for before, after in zip(cache_stats_before, cache_stats_after): + cache_type = after["cache_type"] + hits_diff = after["hits"] - before["hits"] + misses_diff = after["misses"] - before["misses"] + total = hits_diff + misses_diff + + if total > 0: + hit_rate = (hits_diff / total) * 100 + logger.debug( + f"缓存统计 - {cache_type}: " + f"命中率={hit_rate:.1f}%, " + f"命中={hits_diff}, 未命中={misses_diff}", + LOGGER_COMMAND + ) + + # 如果缓存命中率过低，记录警告 + if hit_rate < 30: + logger.warning( + f"缓存命中率过低: {cache_type} = {hit_rate:.1f}%", + LOGGER_COMMAND + ) +``` + +#### 4. 调整空结果缓存时间 + +**根据实际需求调整**： +```python +from zhenxun.services.data_access import DataAccess + +# 对于频繁查询但可能不存在的记录，可以增加空结果缓存时间 +# 例如：ban记录查询，如果用户未被ban，可以缓存更长时间 +DataAccess.set_null_result_ttl(600) # 10分钟 + +# 对于需要实时性的数据，可以减少缓存时间 +# 例如：用户权限查询 +DataAccess.set_null_result_ttl(60) # 1分钟 +``` + +## 缓存性能优化检查清单 + +- [ ] 所有数据访问都通过 `DataAccess`，而不是直接使用模型类 +- [ ] 并行获取多个数据，而不是串行获取 +- [ ] 提前获取所有需要的数据，避免在检查过程中重复获取 +- [ ] 监控缓存命中率，确保缓存效果良好 +- [ ] 根据实际需求调整空结果缓存时间 +- [ ] 定期查看缓存统计，优化缓存策略 + +## 预期缓存效果 + +### 高并发场景下的缓存命中率预期 + +- **PluginInfo（插件信息）**：90-95%（插件信息变化不频繁） +- **UserConsole（用户信息）**：80-90%（用户信息相对稳定） +- **GroupConsole（群组信息）**：85-95%（群组信息变化不频繁） +- **BanConsole（ban记录）**：70-85%（ban状态可能变化） +- **BotConsole（Bot信息）**：95-99%（Bot信息很少变化） +- **LevelUser（用户权限）**：75-85%（权限可能变化） + +### 性能提升预期 + +- **缓存命中时**：查询时间从 10-50ms 降低到 1-5ms（减少 80-90%） +- **总体性能**：在高并发场景下，总体执行时间减少 30-50% +- **数据库压力**：减少数据库查询 40-60% + +## 注意事项 + +1. **缓存一致性**：DataAccess 在数据更新时会自动清除缓存，确保一致性 +2. **缓存失效**：如果需要强制刷新缓存，可以使用 `await dao.clear_cache(**kwargs)` +3. **监控缓存**：定期查看缓存统计，确保缓存效果良好 +4. **空结果缓存**：空结果缓存有助于减少对不存在记录的查询，但需要根据实际需求调整TTL + diff --git a/zhenxun/builtin_plugins/hooks/auth/IMMEDIATE_IMPROVEMENTS.md b/zhenxun/builtin_plugins/hooks/auth/IMMEDIATE_IMPROVEMENTS.md new file mode 100644 index 00000000..e233acff --- /dev/null +++ b/zhenxun/builtin_plugins/hooks/auth/IMMEDIATE_IMPROVEMENTS.md @@ -0,0 +1,373 @@ +# 权限检查系统 - 立即可以实施的改进 + +## 当前问题总结 + +经过分析，当前权限检查系统存在以下主要问题： + +### 1. 检查执行顺序不合理 + +- **问题**：所有检查并行执行，ban 检查应该优先执行 +- **影响**：用户被 ban 后，其他检查（金币、限制等）仍然会执行，浪费资源 +- **优先级**：高 + +### 2. 缺乏早期退出机制 + +- **问题**：即使某个检查失败，其他检查仍会继续执行 +- **影响**：浪费 CPU 和数据库资源 +- **优先级**：高 + +### 3. 数据获取可以优化 + +- **问题**：group 数据在 hooks 执行时才获取，可以提前获取 +- **影响**：增加总体执行时间 +- **优先级**：中 + +### 4. 充分利用 DataAccess 的 Redis 缓存 + +- **现状**：`DataAccess` 已经实现了完整的 Redis 缓存机制（包括空结果缓存 5 分钟） +- **问题**：没有充分利用缓存，数据获取顺序不够优化 +- **影响**：缓存命中率可能不够高，重复查询数据库 +- **优先级**：高 + +## 立即可以实施的改进 + +### 改进 1：调整检查顺序，ban 检查优先执行 + +**位置**：`auth_checker.py` 第 498-514 行 + +**当前代码**： + +```python +hook_tasks = [ + time_hook(auth_ban(...), "auth_ban", hook_times), + time_hook(auth_bot(...), "auth_bot", hook_times), + # ... 其他检查 +] +await asyncio.gather(*hook_tasks) # 并行执行 +``` + +**改进方案**： + +```python +# 1. 先执行关键检查（ban、bot），如果失败立即退出 +critical_checks = [ + ("auth_ban", lambda: time_hook(auth_ban(...), "auth_ban", hook_times)), + ("auth_bot", lambda: time_hook(auth_bot(...), "auth_bot", hook_times)), +] + +for check_name, check_func in critical_checks: + try: + await check_func() + except SkipPluginException: + raise # 立即退出 + except Exception as e: + logger.error(f"{check_name} 检查失败: {e}", LOGGER_COMMAND, session=session) + # 关键检查失败，可以选择继续或退出 + +# 2. 关键检查通过后，再并行执行其他检查 +other_checks = [ + time_hook(auth_group(...), "auth_group", hook_times), + time_hook(auth_admin(...), "auth_admin", hook_times), + time_hook(auth_plugin(...), "auth_plugin", hook_times), + time_hook(auth_limit(...), "auth_limit", hook_times), +] +await asyncio.gather(*other_checks) +``` + +### 改进 2：提前获取 group 数据 + +**位置**：`auth_checker.py` 第 485-493 行 + +**当前代码**： + +```python +# 在hooks执行时才获取group +group = None +if entity.group_id: + group_dao = DataAccess(GroupConsole) + group = await with_timeout(...) +``` + +**改进方案**： + +```python +# 在获取plugin和user时，同时获取group +# 在 get_plugin_and_user 函数中或auth函数开始处 +group = None +if entity.group_id: + group_dao = DataAccess(GroupConsole) + group_task = group_dao.safe_get_or_none( + group_id=entity.group_id, channel_id__isnull=True + ) + # 可以并行获取 + plugin, user, group = await asyncio.gather( + plugin_task, user_task, group_task + ) +``` + +### 改进 3：充分利用 DataAccess 的 Redis 缓存机制 + +**重要说明**：`DataAccess` 已经实现了完整的 Redis 缓存机制，包括： + +- ✅ 自动缓存查询结果 +- ✅ 空结果缓存（5 分钟 TTL，避免频繁查询不存在的记录） +- ✅ 缓存统计（命中率、命中次数等） +- ✅ 自动处理缓存的获取和设置 + +**优化策略**： + +#### 3.1 优化数据获取顺序，提高缓存命中率 + +**位置**：`auth_checker.py` 的 `get_plugin_and_user` 函数和 `auth` 函数 + +**当前问题**： + +- plugin、user、group 数据分别获取，可能错过并行获取的缓存优势 +- group 数据在 hooks 执行时才获取，应该提前获取 + +**改进方案**： + +```python +async def get_plugin_user_and_group( + module: str, user_id: str, group_id: str | None +) -> tuple[PluginInfo, UserConsole, GroupConsole | None]: + """统一获取插件、用户和群组数据（充分利用 DataAccess 缓存） + + 注意：DataAccess 会自动使用 Redis 缓存，包括： + - 查询结果缓存 + - 空结果缓存（5分钟） + - 缓存统计 + """ + user_dao = DataAccess(UserConsole) + plugin_dao = DataAccess(PluginInfo) + group_dao = DataAccess(GroupConsole) if group_id else None + + # 并行获取所有数据，DataAccess 会自动处理缓存 + tasks = [ + plugin_dao.safe_get_or_none(module=module), + user_dao.get_by_func_or_none( + UserConsole.get_user, False, user_id=user_id + ) + ] + + if group_id and group_dao: + tasks.append( + group_dao.safe_get_or_none( + group_id=group_id, channel_id__isnull=True + ) + ) + else: + tasks.append(asyncio.create_task(asyncio.sleep(0))) # 占位 + + try: + results = await with_timeout( + asyncio.gather(*tasks), name="get_plugin_user_group" + ) + plugin = results[0] + user = results[1] + group = results[2] if len(results) > 2 and group_id else None + + # 验证数据 + if not plugin: + raise PermissionExemption(f"插件:{module} 数据不存在...") + if plugin.plugin_type == PluginType.HIDDEN: + raise PermissionExemption(f"插件: {plugin.name}:{plugin.module} 为HIDDEN...") + if not user: + raise PermissionExemption("用户数据不存在...") + + return plugin, user, group + except asyncio.TimeoutError: + # 超时时尝试从缓存获取（DataAccess 已经缓存了） + logger.warning( + f"数据获取超时，尝试从缓存获取，模块: {module}", + LOGGER_COMMAND + ) + # DataAccess 的缓存会自动生效，这里只需要重试一次 + plugin = await plugin_dao.safe_get_or_none(module=module) + user = await user_dao.get_by_func_or_none( + UserConsole.get_user, False, user_id=user_id + ) + group = None + if group_id and group_dao: + group = await group_dao.safe_get_or_none( + group_id=group_id, channel_id__isnull=True + ) + + if not plugin or not user: + raise PermissionExemption("获取数据失败，请稍后再试...") + + return plugin, user, group +``` + +#### 3.2 监控缓存命中率 + +**位置**：在权限检查中添加缓存统计监控 + +**实现方案**： + +```python +from zhenxun.services.data_access import DataAccess + +# 在 auth 函数开始和结束时记录缓存统计 +async def auth(...): + # 记录开始时的缓存统计 + cache_stats_before = DataAccess.get_cache_stats() + + try: + # ... 执行权限检查 ... + pass + finally: + # 记录结束时的缓存统计 + cache_stats_after = DataAccess.get_cache_stats() + + # 计算本次检查的缓存命中情况 + for before, after in zip(cache_stats_before, cache_stats_after): + hits_diff = after["hits"] - before["hits"] + misses_diff = after["misses"] - before["misses"] + total = hits_diff + misses_diff + if total > 0: + hit_rate = (hits_diff / total) * 100 + if hit_rate < 50: # 缓存命中率低于50%时记录警告 + logger.warning( + f"缓存命中率较低: {after['cache_type']} = {hit_rate:.1f}%", + LOGGER_COMMAND + ) +``` + +#### 3.3 优化检查函数中的数据访问 + +**说明**：各个检查函数（auth_ban、auth_bot 等）中使用的 `DataAccess` 已经自动缓存，无需额外处理。 + +**建议**： + +- 确保所有数据访问都通过 `DataAccess`，而不是直接使用模型类 +- 对于频繁查询的数据（如 ban 记录），`DataAccess` 会自动缓存 +- 空结果也会被缓存 5 分钟，避免频繁查询不存在的记录 + +### 改进 4：统一错误处理策略 + +**位置**：各个检查函数 + +**当前问题**： + +- `auth_ban` 超时时不阻塞 +- `auth_bot` 超时时不阻塞 +- `auth_limit` 超时时不阻塞 +- 但 `get_plugin_and_user` 超时时会抛出异常 + +**改进方案**： + +```python +# 定义统一的超时处理策略 +class TimeoutStrategy: + FAIL_FAST = "fail_fast" # 超时立即失败 + CONTINUE = "continue" # 超时继续执行 + CACHE_FALLBACK = "cache_fallback" # 超时使用缓存 + +# 根据检查的重要性设置策略 +CHECK_TIMEOUT_STRATEGY = { + "auth_ban": TimeoutStrategy.FAIL_FAST, # 关键检查，超时失败 + "auth_bot": TimeoutStrategy.FAIL_FAST, # 关键检查，超时失败 + "auth_group": TimeoutStrategy.CONTINUE, # 非关键，超时继续 + "auth_limit": TimeoutStrategy.CONTINUE, # 非关键，超时继续 + "get_plugin_and_user": TimeoutStrategy.CACHE_FALLBACK, # 使用缓存降级 +} +``` + +## 实施优先级 + +### 高优先级（立即实施） + +1. ✅ **调整检查顺序**：ban 检查优先执行 + - 风险：低 + - 收益：高 + - 工作量：小（1-2 小时） + +### 中优先级（本周实施） + +2. ✅ **提前获取 group 数据，统一数据获取** + + - 风险：低 + - 收益：高（充分利用 DataAccess 缓存） + - 工作量：小（1-2 小时） + +3. ✅ **优化数据获取顺序，提高缓存命中率** + + - 风险：低（DataAccess 已实现缓存） + - 收益：高（减少数据库查询） + - 工作量：小（1 小时） + +4. ✅ **添加缓存统计监控** + - 风险：低 + - 收益：中（可以监控缓存效果） + - 工作量：小（30 分钟） + +### 低优先级（后续优化） + +4. ⏳ **统一错误处理策略** + - 风险：中（可能影响现有逻辑） + - 收益：中 + - 工作量：中（3-4 小时） + +## 代码示例：改进后的 auth 函数结构 + +```python +async def auth(...): + # 1. 获取所有需要的数据（并行） + plugin, user, group = await gather_all_data(...) + + # 2. 关键检查（串行，失败立即退出） + await critical_checks(plugin, user, group, ...) + + # 3. 其他检查（并行） + await other_checks(plugin, user, group, ...) + + # 4. 金币检查（最后执行，超时不影响） + cost_gold = await check_cost(user, plugin, ...) +``` + +## 注意事项 + +1. **向后兼容**：确保改进不影响现有功能 +2. **充分测试**：特别是缓存相关的改进 +3. **监控指标**：记录改进前后的性能指标 +4. **逐步实施**：先实施低风险改进，再实施高风险改进 + +## 预期效果 + +- **性能提升**：总体执行时间减少 20-30% +- **资源节省**：减少不必要的检查执行 30-50% +- **缓存优化**：充分利用 DataAccess 的 Redis 缓存，缓存命中率提升 40-60% +- **数据库压力**：减少数据库查询 30-50%（通过缓存） +- **用户体验**：响应速度提升，特别是在高并发场景下 + +## 关于 DataAccess 缓存的说明 + +### DataAccess 已实现的缓存机制 + +1. **自动缓存查询结果** + + - 所有通过 `DataAccess.safe_get_or_none()` 等方法的查询结果都会自动缓存 + - 缓存键基于模型的主键字段（如 `user_id`、`module` 等） + +2. **空结果缓存** + + - 查询结果为 `None` 时，会缓存一个特殊标记（`_NULL_RESULT`） + - 默认 TTL 为 5 分钟，避免频繁查询不存在的记录 + - 可通过 `DataAccess.set_null_result_ttl()` 调整 + +3. **缓存统计** + + - 自动统计缓存命中率、命中次数、未命中次数等 + - 可通过 `DataAccess.get_cache_stats()` 获取统计信息 + +4. **缓存失效** + - 数据更新、创建、删除时自动清除相关缓存 + - 支持手动清除：`DataAccess.clear_cache(**kwargs)` + +### 优化建议 + +1. **统一使用 DataAccess**：确保所有数据访问都通过 `DataAccess`，而不是直接使用模型类 +2. **并行获取数据**：利用 `asyncio.gather()` 并行获取多个数据，提高缓存命中率 +3. **监控缓存效果**：定期查看 `DataAccess.get_cache_stats()`，了解缓存命中率 +4. **调整空结果 TTL**：根据实际需求调整 `DataAccess.set_null_result_ttl()`，平衡缓存效果和实时性 diff --git a/zhenxun/builtin_plugins/hooks/auth/OPTIMIZATION_PROPOSAL.md b/zhenxun/builtin_plugins/hooks/auth/OPTIMIZATION_PROPOSAL.md new file mode 100644 index 00000000..00db6856 --- /dev/null +++ b/zhenxun/builtin_plugins/hooks/auth/OPTIMIZATION_PROPOSAL.md @@ -0,0 +1,234 @@ +# 权限检查系统优化方案 + +## 当前设计分析 + +### 优点 +1. ✅ 模块化设计，每个检查独立 +2. ✅ 并行执行检查，提高效率 +3. ✅ 有超时控制和重试机制 +4. ✅ 有缓存降级策略 +5. ✅ 有并发控制（信号量） + +### 存在的问题 + +#### 1. **缺乏优先级机制** +- 所有检查并行执行，没有优先级区分 +- ban检查应该优先执行，如果用户被ban，其他检查就不需要执行 +- 浪费计算资源 + +#### 2. **缺乏早期退出机制** +- 即使某个检查已经失败，其他检查仍然会继续执行 +- 例如：用户被ban了，但金币检查、限制检查等仍然会执行 + +#### 3. **数据获取分散** +- 每个检查函数可能都需要查询数据库 +- 缺乏统一的数据上下文管理 +- 可能导致重复查询 + +#### 4. **错误处理不一致** +- 有些检查超时时抛出异常 +- 有些检查超时时只是记录日志继续执行 +- 缺乏统一的错误处理策略 + +#### 5. **缺乏检查结果缓存** +- 对于相同的请求（相同的user_id, group_id, plugin），可能会重复执行相同的检查 +- ban检查、bot检查等结果可以缓存一段时间 + +## 优化方案 + +### 方案1：优先级分层检查（推荐） + +#### 核心思想 +将检查分为多个优先级阶段，高优先级检查失败时立即退出，不执行后续检查。 + +#### 检查优先级划分 + +``` +阶段1（CRITICAL - 关键检查）： + - auth_ban: 用户/群组ban检查 + - auth_bot: Bot状态检查 + → 如果失败，立即退出 + +阶段2（HIGH - 高优先级）： + - auth_plugin: 插件状态检查 + - auth_group: 群组状态检查 + → 并行执行，任一失败则退出 + +阶段3（MEDIUM - 中等优先级）： + - auth_admin: 管理员权限检查 + - auth_limit: 限制检查 + → 并行执行，任一失败则退出 + +阶段4（LOW - 低优先级）： + - auth_cost: 金币检查 + → 超时不影响其他检查 +``` + +#### 实现要点 + +1. **统一数据上下文** +```python +@dataclass +class AuthContext: + """统一管理所有需要的数据""" + plugin: PluginInfo + user: UserConsole + group: GroupConsole | None + bot_id: str + entity: EntityIDs + # ... 其他数据 +``` + +2. **早期退出机制** +```python +# 阶段1检查 +for check in critical_checks: + result = await execute_check(check) + if not result.success: + raise result.error # 立即退出，不执行后续检查 +``` + +3. **检查结果缓存** +```python +# 对ban、bot等检查结果进行缓存 +cache_key = f"ban:{user_id}:{group_id}:{module}" +if cached_result := get_cached_result(cache_key): + return cached_result +``` + +### 方案2：责任链模式 + +#### 核心思想 +使用责任链模式，按优先级顺序执行检查，任一检查失败则中断链。 + +#### 优点 +- 清晰的检查顺序 +- 易于添加新的检查 +- 支持动态调整检查顺序 + +#### 缺点 +- 无法充分利用并行执行的优势 +- 可能增加总体执行时间 + +### 方案3：混合模式（最佳实践） + +#### 核心思想 +结合优先级和并行执行： +- 关键检查（ban、bot）串行执行，失败立即退出 +- 同优先级检查并行执行 +- 不同优先级按顺序执行 + +## 具体优化建议 + +### 1. 立即实施的优化（低风险） + +#### a) 调整检查顺序 +```python +# 在 auth_checker.py 中调整检查顺序 +# 1. 先执行ban检查（如果失败，立即退出） +# 2. 再执行其他检查 +``` + +#### b) 统一数据获取 +```python +# 在auth函数开始处，统一获取所有需要的数据 +plugin, user, group = await gather_all_data(...) +# 然后传递给各个检查函数 +``` + +#### c) 添加检查结果缓存 +```python +# 对ban检查结果缓存30秒 +@lru_cache(maxsize=1000, ttl=30) +async def check_ban_cached(user_id, group_id): + return await is_ban(user_id, group_id) +``` + +### 2. 中期优化（需要重构） + +#### a) 实现优先级机制 +- 将检查函数注册到优先级系统 +- 按优先级顺序执行 +- 高优先级失败时立即退出 + +#### b) 统一错误处理 +- 定义统一的检查结果类型 +- 所有检查返回统一格式的结果 +- 统一的超时处理策略 + +### 3. 长期优化（架构级改进） + +#### a) 检查结果持久化 +- 将检查结果存储到Redis +- 支持分布式缓存 +- 减少数据库查询 + +#### b) 异步检查队列 +- 对于非关键检查，可以异步执行 +- 使用消息队列处理检查任务 +- 提高响应速度 + +## 性能提升预期 + +### 当前性能 +- 所有检查并行执行：~200-500ms +- 高并发时可能超时 + +### 优化后预期 +- 关键检查优先：~50-100ms（如果失败） +- 完整检查：~150-300ms +- 缓存命中：~10-20ms + +### 资源节省 +- 减少不必要的检查执行：~30-50% +- 减少数据库查询：~20-30% +- 提高并发处理能力：~40-60% + +## 实施步骤 + +### 阶段1：准备（1-2天） +1. 分析当前检查函数的依赖关系 +2. 确定优先级划分 +3. 设计统一的数据上下文 + +### 阶段2：实现（3-5天） +1. 实现优先级机制 +2. 实现早期退出机制 +3. 实现检查结果缓存 +4. 统一错误处理 + +### 阶段3：测试（2-3天） +1. 单元测试 +2. 集成测试 +3. 性能测试 +4. 压力测试 + +### 阶段4：部署（1天） +1. 灰度发布 +2. 监控性能指标 +3. 逐步全量 + +## 风险评估 + +### 低风险 +- 调整检查顺序 +- 添加缓存 +- 统一数据获取 + +### 中风险 +- 实现优先级机制（需要充分测试） +- 早期退出机制（可能影响某些边界情况） + +### 高风险 +- 架构级重构（需要完整的测试覆盖） + +## 建议 + +1. **先实施低风险优化**：立即调整检查顺序，添加缓存 +2. **逐步实施中风险优化**：实现优先级机制，充分测试 +3. **长期规划高风险优化**：根据实际需求决定是否进行架构级重构 + +## 参考实现 + +详见 `auth_checker_optimized.py`（设计示例，仅供参考） + diff --git a/zhenxun/builtin_plugins/hooks/auth_checker_optimized.py b/zhenxun/builtin_plugins/hooks/auth_checker_optimized.py new file mode 100644 index 00000000..18e65af5 --- /dev/null +++ b/zhenxun/builtin_plugins/hooks/auth_checker_optimized.py @@ -0,0 +1,381 @@ +""" +优化后的权限检查系统设计 + +主要改进： +1. 优先级机制：将检查分为多个优先级阶段，高优先级检查失败时立即退出 +2. 早期退出：避免不必要的检查执行，提高性能 +3. 统一数据上下文：在开始前统一获取所有需要的数据 +4. 检查结果缓存：对相同请求缓存检查结果 +5. 统一的错误处理：所有检查使用统一的超时和错误处理机制 +""" + +import asyncio +from collections.abc import Callable +from dataclasses import dataclass +from enum import IntEnum +import time +from typing import Any + +from nonebot.adapters import Bot, Event +from nonebot.exception import IgnoredException +from nonebot.matcher import Matcher +from nonebot_plugin_alconna import UniMsg +from nonebot_plugin_uninfo import Uninfo + +from zhenxun.models.group_console import GroupConsole +from zhenxun.models.plugin_info import PluginInfo +from zhenxun.models.user_console import UserConsole +from zhenxun.services.data_access import DataAccess +from zhenxun.services.log import logger +from zhenxun.utils.enum import GoldHandle, PluginType +from zhenxun.utils.platform import PlatformUtils +from zhenxun.utils.utils import get_entity_ids + +from .auth.auth_admin import auth_admin +from .auth.auth_ban import auth_ban +from .auth.auth_bot import auth_bot +from .auth.auth_cost import auth_cost +from .auth.auth_group import auth_group +from .auth.auth_limit import LimitManager, auth_limit +from .auth.auth_plugin import auth_plugin +from .auth.bot_filter import bot_filter +from .auth.config import LOGGER_COMMAND, WARNING_THRESHOLD +from .auth.exception import ( + IsSuperuserException, + PermissionExemption, + SkipPluginException, +) + +# 超时设置（秒）—— DataAccess 内部已对单次 DB / 缓存访问做了自己的超时控制； +# 这里主要用于控制单个权限检查步骤的上限时间。 +TIMEOUT_SECONDS = 5.0 + + +# 检查优先级 +class CheckPriority(IntEnum): + """检查优先级，数值越小优先级越高""" + + CRITICAL = 1 # 关键检查：ban、bot状态 + HIGH = 2 # 高优先级：插件状态、群组状态 + MEDIUM = 3 # 中等优先级：管理员权限、限制 + LOW = 4 # 低优先级：金币检查 + + +@dataclass +class AuthContext: + """权限检查上下文，统一管理所有需要的数据""" + + plugin: PluginInfo + user: UserConsole + session: Uninfo + matcher: Matcher + bot: Bot + message: UniMsg + group: GroupConsole | None = None + bot_id: str = "" + entity: Any = None + + +@dataclass +class CheckResult: + """检查结果""" + + success: bool + error: Exception | None = None + execution_time: float = 0.0 + cached: bool = False + + +class AuthChecker: + """优化的权限检查器""" + + async def _execute_check( + self, + check_func: Callable, + check_name: str, + priority: CheckPriority, + context: AuthContext, + **kwargs, + ) -> CheckResult: + """执行单个检查""" + start_time = time.time() + + try: + # 执行检查函数 + await asyncio.wait_for(check_func(**kwargs), timeout=TIMEOUT_SECONDS) + result = CheckResult(success=True, execution_time=time.time() - start_time) + except SkipPluginException as e: + result = CheckResult( + success=False, error=e, execution_time=time.time() - start_time + ) + except asyncio.TimeoutError: + logger.error( + f"{check_name} 检查超时", LOGGER_COMMAND, session=context.session + ) + # 超时时根据优先级决定是否继续 + if priority <= CheckPriority.HIGH: + result = CheckResult( + success=False, + error=PermissionExemption(f"{check_name} 检查超时"), + execution_time=time.time() - start_time, + ) + else: + # 低优先级检查超时，允许继续 + result = CheckResult( + success=True, execution_time=time.time() - start_time + ) + except Exception as e: + logger.error( + f"{check_name} 检查失败: {e}", LOGGER_COMMAND, session=context.session + ) + result = CheckResult( + success=False, error=e, execution_time=time.time() - start_time + ) + + return result + + async def _load_context( + self, matcher: Matcher, bot: Bot, session: Uninfo, message: UniMsg + ) -> AuthContext: + """加载权限检查上下文数据""" + entity = get_entity_ids(session) + module = matcher.plugin_name or "" + + if not module: + raise PermissionExemption("Matcher插件名称不存在...") + + # 并行获取所有需要的数据。 + # DataAccess 内部已经有 Redis 缓存和 DB 超时控制，这里只做一次整体超时保护， + # 不再额外手动走 CacheRoot 之类的二级 fallback，避免重复访问 Redis。 + user_dao = DataAccess(UserConsole) + plugin_dao = DataAccess(PluginInfo) + group_dao = DataAccess(GroupConsole) if entity.group_id else None + + tasks = { + "plugin": plugin_dao.safe_get_or_none(module=module), + "user": user_dao.get_by_func_or_none( + UserConsole.get_user, False, user_id=entity.user_id + ), + } + + if entity.group_id and group_dao: + tasks["group"] = group_dao.safe_get_or_none( + group_id=entity.group_id, channel_id__isnull=True + ) + + try: + results = await asyncio.wait_for( + asyncio.gather(*tasks.values()), timeout=TIMEOUT_SECONDS + ) + except asyncio.TimeoutError: + # DataAccess 本身已经利用了 Redis / DB 缓存，这里整体超时直接视为失败， + # 避免在 Redis 也不稳定时再叠加一层「从缓存再试一次」的复杂 fallback。 + logger.error( + f"加载权限检查所需数据超时，模块: {module}", + LOGGER_COMMAND, + session=session, + ) + raise PermissionExemption("获取权限检查所需数据超时，请稍后再试...") + else: + plugin = results[0] + user = results[1] + group = results[2] if len(results) > 2 else None + + if not plugin: + raise PermissionExemption(f"插件:{module} 数据不存在...") + if plugin.plugin_type == PluginType.HIDDEN: + raise PermissionExemption( + f"插件: {plugin.name}:{plugin.module} 为HIDDEN..." + ) + if not user: + raise PermissionExemption("用户数据不存在...") + + return AuthContext( + plugin=plugin, + user=user, + group=group, + bot_id=bot.self_id, + session=session, + matcher=matcher, + bot=bot, + message=message, + entity=entity, + ) + + async def check( + self, matcher: Matcher, event: Event, bot: Bot, session: Uninfo, message: UniMsg + ): + """执行权限检查（优化版本）""" + start_time = time.time() + cost_gold = 0 + ignore_flag = False + hook_times = {} + + try: + # 1. 加载上下文数据 + context = await self._load_context(matcher, bot, session, message) + + # 2. 按优先级执行检查 + # 阶段1：关键检查（ban、bot状态） + critical_checks = [ + ( + "auth_ban", + CheckPriority.CRITICAL, + lambda: auth_ban( + context.matcher, context.bot, context.session, context.plugin + ), + ), + ( + "auth_bot", + CheckPriority.CRITICAL, + lambda: auth_bot(context.plugin, context.bot_id), + ), + ] + + for check_name, priority, check_func in critical_checks: + result = await self._execute_check( + check_func, check_name, priority, context + ) + hook_times[check_name] = f"{result.execution_time:.3f}s" + if not result.success: + if isinstance(result.error, SkipPluginException): + ignore_flag = True + raise result.error + raise result.error or PermissionExemption(f"{check_name} 检查失败") + + # 3. 执行bot_filter + bot_filter(session) + + # 4. 阶段2：高优先级检查（插件状态、群组状态） + high_priority_checks = [ + ( + "auth_plugin", + CheckPriority.HIGH, + lambda: auth_plugin( + context.plugin, context.group, context.session, event + ), + ), + ( + "auth_group", + CheckPriority.HIGH, + lambda: auth_group( + context.plugin, + context.group, + context.message, + context.entity.group_id, + ), + ), + ] + + # 并行执行高优先级检查 + high_tasks = [] + for check_name, priority, check_func in high_priority_checks: + task = self._execute_check(check_func, check_name, priority, context) + high_tasks.append((check_name, task)) + + high_results = await asyncio.gather(*[task for _, task in high_tasks]) + + for (check_name, _), result in zip(high_tasks, high_results): + hook_times[check_name] = f"{result.execution_time:.3f}s" + if not result.success: + if isinstance(result.error, SkipPluginException): + ignore_flag = True + raise result.error + raise result.error or PermissionExemption(f"{check_name} 检查失败") + + # 5. 阶段3：中等优先级检查（管理员权限、限制） + medium_checks = [ + ( + "auth_admin", + CheckPriority.MEDIUM, + lambda: auth_admin(context.plugin, context.session), + ), + ( + "auth_limit", + CheckPriority.MEDIUM, + lambda: auth_limit(context.plugin, context.session), + ), + ] + + # 并行执行中等优先级检查 + medium_tasks = [] + for check_name, priority, check_func in medium_checks: + task = self._execute_check(check_func, check_name, priority, context) + medium_tasks.append((check_name, task)) + + medium_results = await asyncio.gather(*[task for _, task in medium_tasks]) + + for (check_name, _), result in zip(medium_tasks, medium_results): + hook_times[check_name] = f"{result.execution_time:.3f}s" + if not result.success: + if isinstance(result.error, SkipPluginException): + ignore_flag = True + raise result.error + raise result.error or PermissionExemption(f"{check_name} 检查失败") + + # 6. 阶段4：低优先级检查（金币检查） + try: + cost_gold = await asyncio.wait_for( + auth_cost(context.user, context.plugin, context.session), + timeout=TIMEOUT_SECONDS, + ) + if context.session.user.id in bot.config.superusers: + if context.plugin.plugin_type == PluginType.SUPERUSER: + raise IsSuperuserException() + if not context.plugin.limit_superuser: + raise IsSuperuserException() + hook_times["cost_gold"] = f"{time.time() - start_time:.3f}s" + except asyncio.TimeoutError: + logger.error( + f"获取插件费用超时，模块: {context.plugin.module}", + LOGGER_COMMAND, + session=session, + ) + + except SkipPluginException as e: + LimitManager.unblock( + matcher.plugin_name or "", + get_entity_ids(session).user_id, + get_entity_ids(session).group_id, + get_entity_ids(session).channel_id, + ) + logger.info(str(e), LOGGER_COMMAND, session=session) + ignore_flag = True + except IsSuperuserException: + logger.debug("超级用户跳过权限检测...", LOGGER_COMMAND, session=session) + except PermissionExemption as e: + logger.info(str(e), LOGGER_COMMAND, session=session) + + # 扣除金币 + if not ignore_flag and cost_gold > 0: + try: + await asyncio.wait_for( + UserConsole.reduce_gold( + get_entity_ids(session).user_id, + cost_gold, + GoldHandle.PLUGIN, + matcher.plugin_name or "", + PlatformUtils.get_platform(session), + ), + timeout=TIMEOUT_SECONDS, + ) + hook_times["reduce_gold"] = f"{time.time() - start_time:.3f}s" + except asyncio.TimeoutError: + logger.error( + f"扣除金币超时，模块: {matcher.plugin_name}", + LOGGER_COMMAND, + session=session, + ) + + # 记录总执行时间 + total_time = time.time() - start_time + if total_time > WARNING_THRESHOLD: + logger.warning( + f"权限检查耗时过长: {total_time:.3f}s, " + f"模块: {matcher.plugin_name}, 详情: {hook_times}", + LOGGER_COMMAND, + session=session, + ) + + if ignore_flag: + raise IgnoredException("权限检测 ignore")