From be316a5cafe366abb9062e93e15866ad121ebb0f Mon Sep 17 00:00:00 2001 From: HibiKier <775757368@qq.com> Date: Wed, 24 Dec 2025 14:54:39 +0800 Subject: [PATCH] refactor: extract is_ban_cached method to BanConsole --- .../hooks/auth/CACHE_OPTIMIZATION.md | 249 ------------ .../hooks/auth/IMMEDIATE_IMPROVEMENTS.md | 373 ------------------ .../hooks/auth/OPTIMIZATION_PROPOSAL.md | 234 ----------- .../builtin_plugins/hooks/auth/auth_ban.py | 22 +- zhenxun/builtin_plugins/mmm.py | 59 +++ zhenxun/models/ban_console.py | 34 ++ zhenxun/plugins/fudu.py | 187 --------- zhenxun/plugins/fudu1.py | 283 ------------- zhenxun/plugins/mute/__init__.py | 20 + zhenxun/plugins/mute/_data_source.py | 122 ++++++ zhenxun/plugins/mute/mute_message.py | 78 ++++ zhenxun/plugins/mute/mute_setting.py | 117 ++++++ zhenxun/utils/http_utils.py | 2 + zhenxun/utils/message.py | 12 + zhenxun/utils/platform.py | 3 +- zhenxun/utils/utils.py | 28 +- zhenxun/utils/withdraw_manage.py | 15 +- 17 files changed, 478 insertions(+), 1360 deletions(-) delete mode 100644 zhenxun/builtin_plugins/hooks/auth/CACHE_OPTIMIZATION.md delete mode 100644 zhenxun/builtin_plugins/hooks/auth/IMMEDIATE_IMPROVEMENTS.md delete mode 100644 zhenxun/builtin_plugins/hooks/auth/OPTIMIZATION_PROPOSAL.md create mode 100644 zhenxun/builtin_plugins/mmm.py delete mode 100644 zhenxun/plugins/fudu.py delete mode 100644 zhenxun/plugins/fudu1.py create mode 100644 zhenxun/plugins/mute/__init__.py create mode 100644 zhenxun/plugins/mute/_data_source.py create mode 100644 zhenxun/plugins/mute/mute_message.py create mode 100644 zhenxun/plugins/mute/mute_setting.py diff --git a/zhenxun/builtin_plugins/hooks/auth/CACHE_OPTIMIZATION.md b/zhenxun/builtin_plugins/hooks/auth/CACHE_OPTIMIZATION.md deleted file mode 100644 index 5f691bfe..00000000 --- a/zhenxun/builtin_plugins/hooks/auth/CACHE_OPTIMIZATION.md +++ /dev/null @@ -1,249 +0,0 @@ -# 权限检查系统 - 缓存优化指南 - -## 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 deleted file mode 100644 index e233acff..00000000 --- a/zhenxun/builtin_plugins/hooks/auth/IMMEDIATE_IMPROVEMENTS.md +++ /dev/null @@ -1,373 +0,0 @@ -# 权限检查系统 - 立即可以实施的改进 - -## 当前问题总结 - -经过分析，当前权限检查系统存在以下主要问题： - -### 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 deleted file mode 100644 index 00db6856..00000000 --- a/zhenxun/builtin_plugins/hooks/auth/OPTIMIZATION_PROPOSAL.md +++ /dev/null @@ -1,234 +0,0 @@ -# 权限检查系统优化方案 - -## 当前设计分析 - -### 优点 -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/auth_ban.py b/zhenxun/builtin_plugins/hooks/auth/auth_ban.py index 55530b7b..efb3027b 100644 --- a/zhenxun/builtin_plugins/hooks/auth/auth_ban.py +++ b/zhenxun/builtin_plugins/hooks/auth/auth_ban.py @@ -9,11 +9,9 @@ from nonebot_plugin_uninfo import Uninfo from zhenxun.configs.config import Config from zhenxun.models.ban_console import BanConsole from zhenxun.models.plugin_info import PluginInfo -from zhenxun.services.cache import CacheRoot -from zhenxun.services.data_access import DataAccess from zhenxun.services.db_context import DB_TIMEOUT_SECONDS from zhenxun.services.log import logger -from zhenxun.utils.enum import CacheType, PluginType +from zhenxun.utils.enum import PluginType from zhenxun.utils.utils import EntityIDs, get_entity_ids from .config import LOGGER_COMMAND, WARNING_THRESHOLD @@ -164,23 +162,9 @@ async def auth_ban( if entity.user_id in bot.config.superusers: return - cache_key = f"{entity.user_id}_{entity.group_id}" - - results = await CacheRoot.get(CacheType.BAN, cache_key) + results = await BanConsole.is_ban_cached(entity.user_id, entity.group_id) if not results: - results = await BanConsole.is_ban(entity.user_id, entity.group_id) - await CacheRoot.set( - CacheType.BAN, - cache_key, - results or DataAccess._NULL_RESULT, - ) - else: - if results == DataAccess._NULL_RESULT: - return - tmp_results: list[BanConsole] = [] - for r in results: - tmp_results.append(CacheRoot._deserialize_value(r, BanConsole)) - results = tmp_results + return for result in results: if not result.user_id and result.group_id: diff --git a/zhenxun/builtin_plugins/mmm.py b/zhenxun/builtin_plugins/mmm.py new file mode 100644 index 00000000..d0817392 --- /dev/null +++ b/zhenxun/builtin_plugins/mmm.py @@ -0,0 +1,59 @@ +import asyncio +import random + +from arclet.alconna import Args +from nonebot import get_driver +from nonebot.adapters.onebot.v11 import ( + Bot, + Event, + GroupMessageEvent, + Message, + PrivateMessageEvent, +) +from nonebot.compat import model_dump, type_validate_python +from nonebot_plugin_alconna import Alconna, on_alconna + +from zhenxun.services.log import logger + +tasks: set["asyncio.Task"] = set() + + +@get_driver().on_shutdown +async def cancel_tasks(): + for task in tasks: + if not task.done(): + task.cancel() + + await asyncio.gather( + *(asyncio.wait_for(task, timeout=10) for task in tasks), + return_exceptions=True, + ) + + +def push_event(bot: Bot, event: PrivateMessageEvent | GroupMessageEvent): + event.message = Message("签到") + event.user_id = random.randint(1, 99999999999) + random.randint(1, 99999999999) + task = asyncio.create_task(bot.handle_event(event)) + task.add_done_callback(tasks.discard) + tasks.add(task) + logger.info(f"发送消息 --> {event.user_id} {event.message}") + return event + + +_matcher = on_alconna( + Alconna("test", Args["n", int]), priority=5, block=True, temp=True +) + + +@_matcher.handle() +async def handle_event(event: Event, bot: Bot, n: int): + for _ in range(n): + data = model_dump(event) + if data.get("message_type") == "private": + data["post_type"] = "message" + push_event(bot, type_validate_python(PrivateMessageEvent, data)) + elif data.get("message_type") == "group": + data["post_type"] = "message" + push_event(bot, type_validate_python(GroupMessageEvent, data)) + await asyncio.sleep(0.1) + logger.info(f"发送消息次数 --> {_ + 1}") diff --git a/zhenxun/models/ban_console.py b/zhenxun/models/ban_console.py index 672a6643..e9cddf54 100644 --- a/zhenxun/models/ban_console.py +++ b/zhenxun/models/ban_console.py @@ -181,6 +181,40 @@ class BanConsole(Model): return results + @classmethod + async def is_ban_cached( + cls, user_id: str | None, group_id: str | None + ) -> list[Self]: + """带缓存的 ban 状态检查 + + 参数: + user_id: 用户id + group_id: 群组id + + 返回: + list[Self]: ban记录列表，空列表表示未被ban + """ + from zhenxun.services.cache import CacheRoot + from zhenxun.services.data_access import DataAccess + from zhenxun.utils.enum import CacheType + + cache_key = f"{user_id}_{group_id}" + + results = await CacheRoot.get(CacheType.BAN, cache_key) + if not results: + results = await cls.is_ban(user_id, group_id) + await CacheRoot.set( + CacheType.BAN, + cache_key, + results or DataAccess._NULL_RESULT, + ) + return results + + if results == DataAccess._NULL_RESULT: + return [] + + return [CacheRoot._deserialize_value(r, cls) for r in results] + @classmethod async def ban( cls, diff --git a/zhenxun/plugins/fudu.py b/zhenxun/plugins/fudu.py deleted file mode 100644 index ea13120c..00000000 --- a/zhenxun/plugins/fudu.py +++ /dev/null @@ -1,187 +0,0 @@ -from collections import OrderedDict -import random -from typing import Any - -from nonebot import on_message -from nonebot.adapters import Event -from nonebot.plugin import PluginMetadata -from nonebot_plugin_alconna import Image, UniMsg -from nonebot_plugin_uninfo import Uninfo - -from zhenxun.configs.config import Config -from zhenxun.configs.path_config import TEMP_PATH -from zhenxun.configs.utils import PluginExtraData, RegisterConfig, Task -from zhenxun.utils.common_utils import CommonUtils -from zhenxun.utils.enum import PluginType -from zhenxun.utils.image_utils import get_download_image_hash -from zhenxun.utils.message import MessageUtils - -__plugin_meta__ = PluginMetadata( - name="复读", - description="群友的本质是什么？是复读机哒！", - usage=""" - usage： - 重复3次相同的消息时会复读 - """.strip(), - extra=PluginExtraData( - author="HibiKier", - version="0.2", - menu_type="其他", - plugin_type=PluginType.DEPENDANT, - tasks=[Task(module="fudu", name="复读")], - ignore_prompt=True, - configs=[ - RegisterConfig( - key="FUDU_PROBABILITY", - value=0.7, - help="复读概率", - default_value=0.7, - type=float, - ), - RegisterConfig( - module="_task", - key="DEFAULT_FUDU", - value=True, - help="被动 复读 进群默认开关状态", - default_value=True, - type=bool, - ), - ], - ).to_dict(), -) - - -class Fudu: - """复读数据管理器，使用 LRU 策略限制内存占用""" - - MAX_GROUPS = 500 # 最大缓存群组数量 - - def __init__(self): - # 使用 OrderedDict 实现 LRU 缓存 - self._data: OrderedDict[str, dict[str, Any]] = OrderedDict() - - def _get_or_create(self, key: str) -> dict[str, Any]: - """获取或创建群组数据，同时维护 LRU 顺序""" - if key in self._data: - # 移动到末尾（最近使用） - self._data.move_to_end(key) - return self._data[key] - - # 如果超出最大限制，删除最旧的条目 - while len(self._data) >= self.MAX_GROUPS: - self._data.popitem(last=False) - - self._data[key] = {"is_repeater": False, "data": []} - return self._data[key] - - def append(self, key: str, content: str) -> None: - """添加消息内容""" - self._get_or_create(key)["data"].append(content) - - def clear(self, key: str) -> None: - """清空群组的复读数据""" - group_data = self._get_or_create(key) - group_data["data"] = [] - group_data["is_repeater"] = False - - def size(self, key: str) -> int: - """获取当前消息数量""" - return len(self._get_or_create(key)["data"]) - - def check(self, key: str, content: str) -> bool: - """检查内容是否与第一条消息相同""" - data_list = self._get_or_create(key)["data"] - return bool(data_list) and data_list[0] == content - - def get_first(self, key: str) -> str | None: - """获取第一条消息内容""" - data_list = self._get_or_create(key)["data"] - return data_list[0] if data_list else None - - def is_repeater(self, key: str) -> bool: - """检查是否已经复读过""" - return self._get_or_create(key)["is_repeater"] - - def set_repeater(self, key: str) -> None: - """标记已复读""" - self._get_or_create(key)["is_repeater"] = True - - -_manager = Fudu() - -base_config = Config.get("fudu") - - -async def rule(message: UniMsg, session: Uninfo, event: Event) -> bool: - """消息匹配规则：仅匹配群聊中的有效消息""" - if not session.group: - return False - if event.is_tome(): - return False - plain_text = message.extract_plain_text().strip() - image_list = [m.url for m in message if isinstance(m, Image) and m.url] - if not plain_text and not image_list: - return False - return not await CommonUtils.task_is_block( - session, "fudu", session.group.id if session.group else None - ) - - -_matcher = on_message(rule=rule, priority=999) - - -@_matcher.handle() -async def _(message: UniMsg, session: Uninfo): - # rule 已经确保 session.group 存在 - group_id = session.group.id # type: ignore - - plain_text = message.extract_plain_text().strip() - image_list = [m.url for m in message if isinstance(m, Image) and m.url] - - # 计算图片哈希（如果有图片） - img_hash = "" - if image_list: - img_hash = await get_download_image_hash(image_list[0], group_id) - - add_msg = f"{plain_text}|-|{img_hash}" - - # 更新复读状态 - if _manager.size(group_id) == 0 or _manager.check(group_id, add_msg): - _manager.append(group_id, add_msg) - else: - _manager.clear(group_id) - _manager.append(group_id, add_msg) - - # 检查是否触发复读 - if _manager.size(group_id) <= 2: - return - if _manager.is_repeater(group_id): - return - if random.random() >= base_config.get("FUDU_PROBABILITY"): - return - - # 20% 概率打断施法 - if random.random() < 0.2: - is_interrupt = plain_text.replace("打断", "").strip() == "施法" - if plain_text.startswith("打断") and is_interrupt: - await MessageUtils.build_message(f"打断{plain_text}").finish() - else: - await MessageUtils.build_message("打断施法！").finish() - - # 执行复读 - _manager.set_repeater(group_id) - - if image_list and plain_text: - result = MessageUtils.build_message( - [plain_text, TEMP_PATH / f"compare_download_{group_id}_img.jpg"] - ) - elif image_list: - result = MessageUtils.build_message( - TEMP_PATH / f"compare_download_{group_id}_img.jpg" - ) - elif plain_text: - result = MessageUtils.build_message(plain_text) - else: - return - - await result.finish() diff --git a/zhenxun/plugins/fudu1.py b/zhenxun/plugins/fudu1.py deleted file mode 100644 index dc053760..00000000 --- a/zhenxun/plugins/fudu1.py +++ /dev/null @@ -1,283 +0,0 @@ -import base64 -from collections import OrderedDict, deque -from pathlib import Path -import random -from typing import Any - -from nonebot import on_message -from nonebot.adapters import Bot, Event -from nonebot.adapters.onebot.v11 import Message as V11Message -from nonebot.adapters.onebot.v11 import MessageSegment -from nonebot.plugin import PluginMetadata -from nonebot.utils import run_sync -from nonebot_plugin_alconna import Image as alcImg -from nonebot_plugin_alconna import UniMsg -from nonebot_plugin_uninfo import Uninfo - -from zhenxun.configs.config import BotConfig, Config -from zhenxun.configs.path_config import DATA_PATH, TEMP_PATH -from zhenxun.configs.utils import PluginExtraData, RegisterConfig, Task -from zhenxun.services.log import logger -from zhenxun.utils.common_utils import CommonUtils -from zhenxun.utils.enum import PluginType -from zhenxun.utils.http_utils import AsyncHttpx -from zhenxun.utils.image_utils import get_img_hash -from zhenxun.utils.message import MessageUtils - -FUDU_IMAGE_PATH = DATA_PATH / "fudu" -FUDU_IMAGE_PATH.mkdir(parents=True, exist_ok=True) -FUDU_CACHE_PATH = TEMP_PATH / "fudu" -FUDU_CACHE_PATH.mkdir(parents=True, exist_ok=True) - -# 支持的图片扩展名 -IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".webp"} - -__plugin_meta__ = PluginMetadata( - name="复读", - description="群友的本质是什么？是复读机哒！", - usage=""" - usage： - 重复3次相同的消息时会复读 - """.strip(), - extra=PluginExtraData( - author="HibiKier", - version="0.3", - menu_type="其他", - plugin_type=PluginType.DEPENDANT, - tasks=[Task(module="fudu", name="复读")], - ignore_prompt=True, - configs=[ - RegisterConfig( - key="FUDU_PROBABILITY", - value=0.8, - help="复读概率", - default_value=0.8, - type=float, - ), - RegisterConfig( - module="_task", - key="DEFAULT_FUDU", - value=True, - help="被动 复读 进群默认开关状态", - default_value=True, - type=bool, - ), - RegisterConfig( - key="FUDU_TRIGGER_COUNT", - value=3, - help="触发复读所需的消息重复次数", - default_value=3, - type=int, - ), - RegisterConfig( - key="FUDU_BREAK_PROBABILITY", - value=0.2, - help="打断复读的概率（基于复读概率）", - default_value=0.2, - type=float, - ), - RegisterConfig( - key="FUDU_BREAK_TEXTS", - value=["打断施法！"], - help="用于打断复读时随机发送的文本列表", - type=list[str], - ), - RegisterConfig( - key="FUDU_BREAK_USE_IMAGE", - value=True, - help="是否启用图片作为打断复读的内容 (图片存放于 data/fudu/ 目录下)", - default_value=True, - type=bool, - ), - RegisterConfig( - key="FUDU_BREAK_USE_TEXT", - value=True, - help="是否启用文本作为打断复读的内容", - default_value=True, - type=bool, - ), - ], - ).to_dict(), -) - - -class Fudu: - """复读数据管理器，使用 LRU 策略限制内存占用""" - - MAX_GROUPS = 500 # 最大缓存群组数量 - - def __init__(self): - self._data: OrderedDict[str, dict[str, Any]] = OrderedDict() - - def _get_or_create(self, key: str) -> dict[str, Any]: - """获取或创建群组数据，同时维护 LRU 顺序""" - if key in self._data: - self._data.move_to_end(key) - return self._data[key] - - # 超出限制时删除最旧的条目 - while len(self._data) >= self.MAX_GROUPS: - self._data.popitem(last=False) - - trigger_count = Config.get_config("fudu", "FUDU_TRIGGER_COUNT", 3) - self._data[key] = { - "is_repeater": False, - "data": deque(maxlen=trigger_count), - "message_obj": None, - "reply_info": None, - } - return self._data[key] - - def append(self, key: str, content: str, msg_obj: Any, reply_info: Any) -> None: - """添加消息内容及原始消息对象""" - group_data = self._get_or_create(key) - group_data["data"].append(content) - group_data["message_obj"] = msg_obj - group_data["reply_info"] = reply_info - - def clear(self, key: str) -> None: - """清空群组的复读数据""" - group_data = self._get_or_create(key) - group_data["data"].clear() - group_data["is_repeater"] = False - group_data["message_obj"] = None - group_data["reply_info"] = None - - def size(self, key: str) -> int: - """获取当前消息数量""" - return len(self._get_or_create(key)["data"]) - - def check(self, key: str, content: str) -> bool: - """检查内容是否与第一条消息相同""" - data_list = self._get_or_create(key)["data"] - return bool(data_list) and data_list[0] == content - - def get_repeat_target(self, key: str) -> tuple[Any, Any]: - """获取要复读的消息对象和回复信息""" - group_data = self._get_or_create(key) - return group_data["message_obj"], group_data["reply_info"] - - def is_repeater(self, key: str) -> bool: - """检查是否已经复读过""" - return self._get_or_create(key)["is_repeater"] - - def set_repeater(self, key: str) -> None: - """标记已复读""" - self._get_or_create(key)["is_repeater"] = True - - -_manager = Fudu() -base_config = Config.get("fudu") - - -def get_break_images() -> list[Path]: - """获取打断复读用的图片列表""" - if not FUDU_IMAGE_PATH.exists(): - return [] - return [ - f - for f in FUDU_IMAGE_PATH.iterdir() - if f.is_file() and f.suffix.lower() in IMAGE_EXTENSIONS - ] - - -async def send_break_response() -> None: - """发送打断复读的响应""" - response_pool: list[str | Path] = [] - - if base_config.get("FUDU_BREAK_USE_TEXT"): - if break_texts := base_config.get("FUDU_BREAK_TEXTS"): - response_pool.extend(break_texts) - - if base_config.get("FUDU_BREAK_USE_IMAGE"): - response_pool.extend(get_break_images()) - - if not response_pool: - response_pool.append("打断施法！") - - response = random.choice(response_pool) - - if isinstance(response, Path): - file_data = f"base64://{base64.b64encode(response.read_bytes()).decode()}" - await _matcher.finish( - MessageSegment("image", {"file": file_data, "sub_type": "1"}) - ) - else: - await MessageUtils.build_message(response).finish() - - -async def rule(message: UniMsg, session: Uninfo, event: Event) -> bool: - """消息匹配规则：仅匹配群聊中的有效消息""" - if not session.group: - return False - if event.is_tome(): - return False - plain_text = message.extract_plain_text() - image_list = [m.url for m in message if isinstance(m, alcImg) and m.url] - if not plain_text and not image_list: - return False - return not await CommonUtils.task_is_block(session, "fudu") - - -_matcher = on_message(rule=rule, priority=999) - - -@_matcher.handle() -async def _(bot: Bot, message: UniMsg, event: Event, session: Uninfo): - # rule 已确保 session.group 存在 - group_id = session.group.id # type: ignore - - raw_message = event.get_message() - reply_info = getattr(event, "reply", None) - plain_text = message.extract_plain_text() - image_list = [m.url for m in message if isinstance(m, alcImg) and m.url] - - # 检测虚空艾特 - if plain_text and plain_text.startswith(f"@可爱的{BotConfig.self_nickname}"): - await MessageUtils.build_message("复制粘贴的虚空艾特？").send(reply_to=True) - - # 计算图片哈希 - img_hash = "" - if image_list: - temp_image_path = FUDU_CACHE_PATH / f"fudu_cache_{group_id}.jpg" - try: - if await AsyncHttpx.download_file(image_list[0], temp_image_path): - img_hash = await run_sync(get_img_hash)(temp_image_path) - except Exception as e: - logger.warning("下载复读图片以获取Hash时出错", "复读", e=e) - - add_msg = f"{plain_text}|-|{img_hash}" - - # 更新复读状态 - if _manager.size(group_id) == 0 or _manager.check(group_id, add_msg): - _manager.append(group_id, add_msg, raw_message, reply_info) - else: - _manager.clear(group_id) - _manager.append(group_id, add_msg, raw_message, reply_info) - - # 检查是否触发复读 - trigger_count = base_config.get("FUDU_TRIGGER_COUNT") - if _manager.size(group_id) < trigger_count: - return - if _manager.is_repeater(group_id): - return - if random.random() >= base_config.get("FUDU_PROBABILITY"): - return - - # 判断是否打断复读 - if random.random() < base_config.get("FUDU_BREAK_PROBABILITY"): - _manager.clear(group_id) - await send_break_response() - - # 执行复读 - message_to_send, reply = _manager.get_repeat_target(group_id) - if not message_to_send: - return - - _manager.set_repeater(group_id) - - # 处理回复消息 - if reply and isinstance(message_to_send, V11Message): - message_to_send.insert(0, MessageSegment.reply(reply.message_id)) - - await _matcher.finish(message_to_send) diff --git a/zhenxun/plugins/mute/__init__.py b/zhenxun/plugins/mute/__init__.py new file mode 100644 index 00000000..f846f4b6 --- /dev/null +++ b/zhenxun/plugins/mute/__init__.py @@ -0,0 +1,20 @@ +from pathlib import Path + +import nonebot +from nonebot.plugin import PluginMetadata + +from zhenxun.configs.utils import PluginExtraData +from zhenxun.utils.enum import PluginType + +__plugin_meta__ = PluginMetadata( + name="刷屏禁言检测", + description="", + usage="", + extra=PluginExtraData( + author="HibiKier", + version="0.1-473ecd8", + plugin_type=PluginType.PARENT, + ).to_dict(), +) + +nonebot.load_plugins(str(Path(__file__).parent.resolve())) diff --git a/zhenxun/plugins/mute/_data_source.py b/zhenxun/plugins/mute/_data_source.py new file mode 100644 index 00000000..5df272a0 --- /dev/null +++ b/zhenxun/plugins/mute/_data_source.py @@ -0,0 +1,122 @@ +import time + +from pydantic import BaseModel +import ujson as json + +from zhenxun.configs.config import Config +from zhenxun.configs.path_config import DATA_PATH + +base_config = Config.get("mute_setting") + + +class GroupData(BaseModel): + count: int + """次数""" + time: int + """检测时长""" + duration: int + """禁言时长""" + message_data: dict = {} + """消息存储""" + + +class MuteManager: + file = DATA_PATH / "group_mute_data.json" + + def __init__(self) -> None: + self._group_data: dict[str, GroupData] = {} + if self.file.exists(): + _data = json.load(open(self.file)) + for gid in _data: + self._group_data[gid] = GroupData( + count=_data[gid]["count"], + time=_data[gid]["time"], + duration=_data[gid]["duration"], + ) + + def get_group_data(self, group_id: str) -> GroupData: + """获取群组数据 + + 参数: + group_id: 群组id + + 返回: + GroupData: GroupData + """ + if group_id not in self._group_data: + self._group_data[group_id] = GroupData( + count=base_config.get("MUTE_DEFAULT_COUNT", 10) or 10, + time=base_config.get("MUTE_DEFAULT_TIME", 7) or 7, + duration=base_config.get("MUTE_DEFAULT_DURATION", 10) or 10, + ) + return self._group_data[group_id] + + def reset(self, user_id: str, group_id: str): + """重置用户检查次数 + + 参数: + user_id: 用户id + group_id: 群组id + """ + if group_data := self._group_data.get(group_id): + if user_id in group_data.message_data: + group_data.message_data[user_id]["count"] = 0 + + def save_data(self): + """保存数据""" + data = {} + for gid in self._group_data: + data[gid] = { + "count": self._group_data[gid].count, + "time": self._group_data[gid].time, + "duration": self._group_data[gid].duration, + } + with open(self.file, "w") as f: + json.dump(data, f, indent=4, ensure_ascii=False) + + def add_message(self, user_id: str, group_id: str, message: str) -> int: + """添加消息 + + 参数: + user_id: 用户id + group_id: 群组id + message: 消息内容 + + 返回: + int: 禁言时长 + """ + if group_id not in self._group_data: + self._group_data[group_id] = GroupData( + count=base_config.get("MUTE_DEFAULT_COUNT"), + time=base_config.get("MUTE_DEFAULT_TIME"), + duration=base_config.get("MUTE_DEFAULT_DURATION"), + ) + group_data = self._group_data[group_id] + if group_data.duration == 0: + return 0 + message_data = group_data.message_data + if not message_data.get(user_id): + message_data[user_id] = { + "time": time.time(), + "count": 1, + "message": message, + } + else: + if message.find(message_data[user_id]["message"]) != -1: + message_data[user_id]["count"] += 1 + else: + message_data[user_id]["time"] = time.time() + message_data[user_id]["count"] = 1 + message_data[user_id]["message"] = message + if time.time() - message_data[user_id]["time"] > group_data.time: + message_data[user_id]["time"] = time.time() + message_data[user_id]["count"] = 1 + if ( + message_data[user_id]["count"] > group_data.count + and time.time() - message_data[user_id]["time"] < group_data.time + ): + return group_data.duration + return 0 + + +mute_manager = MuteManager() diff --git a/zhenxun/plugins/mute/mute_message.py b/zhenxun/plugins/mute/mute_message.py new file mode 100644 index 00000000..f6fa67e9 --- /dev/null +++ b/zhenxun/plugins/mute/mute_message.py @@ -0,0 +1,78 @@ +from nonebot import on_message +from nonebot.adapters import Bot +from nonebot.plugin import PluginMetadata +from nonebot_plugin_alconna import Image, UniMsg +from nonebot_plugin_uninfo import Uninfo + +from zhenxun.configs.config import BotConfig +from zhenxun.configs.utils import PluginExtraData +from zhenxun.models.ban_console import BanConsole +from zhenxun.services.data_access import DataAccess +from zhenxun.services.log import logger +from zhenxun.utils.enum import PluginType +from zhenxun.utils.image_utils import get_download_image_hash +from zhenxun.utils.message import MessageUtils +from zhenxun.utils.platform import PlatformUtils +from zhenxun.utils.utils import FreqLimiter, get_entity_ids + +from ._data_source import mute_manager + +__plugin_meta__ = PluginMetadata( + name="刷屏监听", + description="", + usage="", + extra=PluginExtraData( + author="HibiKier", + version="0.1-473ecd8", + menu_type="其他", + plugin_type=PluginType.DEPENDANT, + ).to_dict(), +) + + +async def rule(session: Uninfo) -> bool: + entity_ids = get_entity_ids(session) + if not session.group: + return False + ban_dao = DataAccess(BanConsole) + if not await ban_dao.safe_get_or_none( + user_id=entity_ids.user_id, group_id=entity_ids.group_id + ): + return False + if not await ban_dao.safe_get_or_none(user_id="", group_id=entity_ids.group_id): + return False + return True + + +_matcher = on_message(rule=rule, priority=1, block=False) + +_flmt = FreqLimiter(30) + + +@_matcher.handle() +async def _(bot: Bot, session: Uninfo, message: UniMsg): + entity_ids = get_entity_ids(session) + plain_text = message.extract_plain_text() + image_list = [m.url for m in message if isinstance(m, Image) and m.url] + img_hash = "" + for url in image_list: + img_hash += await get_download_image_hash(url, "_mute_") + _message = plain_text + img_hash + if duration := mute_manager.add_message( + entity_ids.user_id, entity_ids.group_id or "0", _message + ): + try: + if _flmt.check(entity_ids.user_id): + _flmt.start_cd(entity_ids.user_id) + await PlatformUtils.ban_user( + bot, entity_ids.user_id, entity_ids.group_id or "0", duration + ) + await MessageUtils.build_message( + f"检测到恶意刷屏，{BotConfig.self_nickname}要把你关进小黑屋！" + ).send(at_sender=True) + mute_manager.reset(entity_ids.user_id, entity_ids.group_id or "0") + logger.info( + f"检测刷屏 被禁言 {duration} 分钟", "禁言检查", session=session + ) + except Exception as e: + logger.error("禁言发送错误", "禁言检测", session=session, e=e) diff --git a/zhenxun/plugins/mute/mute_setting.py b/zhenxun/plugins/mute/mute_setting.py new file mode 100644 index 00000000..b1d901fc --- /dev/null +++ b/zhenxun/plugins/mute/mute_setting.py @@ -0,0 +1,117 @@ +from nonebot.plugin import PluginMetadata +from nonebot_plugin_alconna import Alconna, Args, Arparma, Match, Option, on_alconna +from nonebot_plugin_session import EventSession + +from zhenxun.configs.config import BotConfig +from zhenxun.configs.utils import PluginExtraData, RegisterConfig +from zhenxun.services.log import logger +from zhenxun.utils.enum import PluginType +from zhenxun.utils.message import MessageUtils +from zhenxun.utils.rules import ensure_group + +from ._data_source import base_config, mute_manage + +__plugin_meta__ = PluginMetadata( + name="刷屏禁言", + description="刷屏禁言相关操作", + usage=f""" + 刷屏禁言相关操作，需要 {BotConfig.self_nickname} 有群管理员权限 + 指令： + 刷屏设置: 查看当前设置 + -c [count]: 检测最大次数 + -t [time]: 规定时间内 + -d [duration]: 禁言时长 + 示例: + 刷屏设置 -c 10: 设置最大次数为10 + 刷屏设置 -t 100 -d 20: 设置规定时间和禁言时长 + 刷屏设置 -d 10: 设置禁言时长为10 + * 即 X 秒内发送同样消息 N 次，禁言 M 分钟 * + """.strip(), + extra=PluginExtraData( + author="HibiKier", + version="0.1-473ecd8", + menu_type="其他", + plugin_type=PluginType.ADMIN, + admin_level=base_config.get("MUTE_LEVEL", 5), + configs=[ + RegisterConfig( + key="MUTE_LEVEL", + value=5, + help="更改禁言设置的管理权限", + default_value=5, + type=int, + ), + RegisterConfig( + key="MUTE_DEFAULT_COUNT", + value=10, + help="刷屏禁言默认检测次数", + default_value=10, + type=int, + ), + RegisterConfig( + key="MUTE_DEFAULT_TIME", + value=7, + help="刷屏检测默认规定时间", + default_value=7, + type=int, + ), + RegisterConfig( + key="MUTE_DEFAULT_DURATION", + value=10, + help="刷屏检测默禁言时长（分钟）", + default_value=10, + type=int, + ), + ], + ).to_dict(), +) + + +_setting_matcher = on_alconna( + Alconna( + "刷屏设置", + Option("-t|--time", Args["time", int], help_text="检测时长"), + Option("-c|--count", Args["count", int], help_text="检测次数"), + Option("-d|--duration", Args["duration", int], help_text="禁言时长"), + ), + rule=ensure_group, + block=True, + priority=5, +) + + +@_setting_matcher.handle() +async def _( + session: EventSession, + arparma: Arparma, + time: Match[int], + count: Match[int], + duration: Match[int], +): + group_id = session.id2 + if not session.id1 or not group_id: + return + _time = time.result if time.available else None + _count = count.result if count.available else None + _duration = duration.result if duration.available else None + group_data = mute_manage.get_group_data(group_id) + if _time is None and _count is None and _duration is None: + await MessageUtils.build_message( + f"最大次数：{group_data.count} 次\n" + f"规定时间：{group_data.time} 秒\n" + f"禁言时长：{group_data.duration:.2f} 分钟\n" + f"【在规定时间内发送相同消息超过最大次数则禁言\n当禁言时长为0时关闭此功能】" + ).finish(reply_to=True) + if _time is not None: + group_data.time = _time + if _count is not None: + group_data.count = _count + if _duration is not None: + group_data.duration = _duration + await MessageUtils.build_message("设置成功!").send(reply_to=True) + logger.info( + f"设置禁言配置 time: {_time}, count: {_count}, duration: {_duration}", + arparma.header_result, + session=session, + ) + mute_manage.save_data() diff --git a/zhenxun/utils/http_utils.py b/zhenxun/utils/http_utils.py index 8c8e97eb..f482cfa3 100644 --- a/zhenxun/utils/http_utils.py +++ b/zhenxun/utils/http_utils.py @@ -64,6 +64,8 @@ async def _(): _client = get_async_client( headers=get_user_agent(), follow_redirects=True, + limits=httpx.Limits(max_connections=200, max_keepalive_connections=100), + timeout=httpx.Timeout(10), **client_kwargs, ) diff --git a/zhenxun/utils/message.py b/zhenxun/utils/message.py index 5fec2213..4fe666c0 100644 --- a/zhenxun/utils/message.py +++ b/zhenxun/utils/message.py @@ -3,6 +3,7 @@ from io import BytesIO from pathlib import Path import nonebot +from nonebot.adapters import Bot from nonebot.adapters.onebot.v11 import Message, MessageSegment from nonebot_plugin_alconna import ( At, @@ -16,6 +17,7 @@ from nonebot_plugin_alconna import ( Video, Voice, ) +from nonebot_plugin_uninfo import Uninfo from pydantic import BaseModel import ujson as json @@ -104,22 +106,32 @@ class MessageUtils: cls, msg_list: MESSAGE_TYPE | list[MESSAGE_TYPE | list[MESSAGE_TYPE]], format_args: dict | None = None, + auto_forward_msg: Bot | Uninfo | None = None, ) -> UniMessage: """构造消息 参数: msg_list: 消息列表 format_args: 用于格式化字符串的参数字典. + auto_forward_msg: 是否自动转发消息 返回: UniMessage: 构造完成的消息列表 """ + from zhenxun.utils.platform import PlatformUtils + message_list = [] if not isinstance(msg_list, list): msg_list = [msg_list] for m in msg_list: _data = m if isinstance(m, list) else [m] message_list += cls.__build_message(_data, format_args) + if auto_forward_msg and PlatformUtils.is_forward_merge_supported( + auto_forward_msg + ): + message_list = cls.alc_forward_msg( + message_list, auto_forward_msg.self_id, auto_forward_msg.self_id + ) return UniMessage(message_list) @classmethod diff --git a/zhenxun/utils/platform.py b/zhenxun/utils/platform.py index 161c098c..3b77af39 100644 --- a/zhenxun/utils/platform.py +++ b/zhenxun/utils/platform.py @@ -18,7 +18,6 @@ from zhenxun.models.friend_user import FriendUser from zhenxun.models.group_console import GroupConsole from zhenxun.services.log import logger from zhenxun.utils.exception import NotFindSuperuser -from zhenxun.utils.http_utils import AsyncHttpx from zhenxun.utils.message import MessageUtils driver = nonebot.get_driver() @@ -226,6 +225,8 @@ class PlatformUtils: user_id: 用户id platform: 平台 """ + from zhenxun.utils.http_utils import AsyncHttpx + url = None if platform == "qq": if user_id.isdigit(): diff --git a/zhenxun/utils/utils.py b/zhenxun/utils/utils.py index d5921803..a7e19247 100644 --- a/zhenxun/utils/utils.py +++ b/zhenxun/utils/utils.py @@ -9,6 +9,7 @@ from types import TracebackType from typing import Any, ClassVar import httpx +from nonebot_plugin_session import EventSession, Session from nonebot_plugin_uninfo import Uninfo import pypinyin @@ -209,7 +210,7 @@ def is_valid_date(date_text: str, separator: str = "-") -> bool: return False -def get_entity_ids(session: Uninfo) -> EntityIDs: +def get_entity_ids(session: Uninfo | EventSession) -> EntityIDs: """获取用户id，群组id，频道id 参数: @@ -218,16 +219,21 @@ def get_entity_ids(session: Uninfo) -> EntityIDs: 返回: EntityIDs: 用户id，群组id，频道id """ - user_id = session.user.id - group_id = None - channel_id = None - if session.group: - if session.group.parent: - group_id = session.group.parent.id - channel_id = session.group.id - else: - group_id = session.group.id - return EntityIDs(user_id=user_id, group_id=group_id, channel_id=channel_id) + if isinstance(session, Session): + user_id = session.id1 + group_id = session.id2 + channel_id = session.id3 + else: + user_id = session.user.id + group_id = session.group.id if session.group else None + channel_id = session.channel.id if session.channel else None + if session.group: + if session.group.parent: + group_id = session.group.parent.id + channel_id = session.group.id + else: + group_id = session.group.id + return EntityIDs(user_id=user_id or "", group_id=group_id, channel_id=channel_id) def is_number(text: str) -> bool: diff --git a/zhenxun/utils/withdraw_manage.py b/zhenxun/utils/withdraw_manage.py index 3dfc5e80..9b9f4fbe 100644 --- a/zhenxun/utils/withdraw_manage.py +++ b/zhenxun/utils/withdraw_manage.py @@ -8,9 +8,11 @@ from nonebot.adapters import Bot from nonebot.adapters.onebot.v11 import Bot as v11Bot from nonebot.adapters.onebot.v12 import Bot as v12Bot from nonebot_plugin_session import EventSession +from nonebot_plugin_uninfo import Uninfo from ruamel.yaml.comments import CommentedSeq from zhenxun.services.log import logger +from zhenxun.utils.utils import get_entity_ids class WithdrawManager: @@ -18,7 +20,9 @@ class WithdrawManager: _index = 0 @classmethod - def check(cls, session: EventSession, withdraw_time: tuple[int, int]) -> bool: + def check( + cls, session: Uninfo | EventSession, withdraw_time: tuple[int, int] + ) -> bool: """配置项检查 参数: @@ -28,12 +32,17 @@ class WithdrawManager: 返回: bool: 是否允许撤回 """ + entity_ids = get_entity_ids(session) if withdraw_time[0] and withdraw_time[0] > 0: if withdraw_time[1] == 2: return True - if withdraw_time[1] == 1 and (session.id2 or session.id3): + if withdraw_time[1] == 1 and entity_ids.group_id: return True - if withdraw_time[1] == 0 and not session.id2 and not session.id3: + if ( + withdraw_time[1] == 0 + and not entity_ids.group_id + and not entity_ids.channel_id + ): return True return False