From e5320969eefefca65ae8302c3a359c8c5ce3c148 Mon Sep 17 00:00:00 2001 From: Helian Nuits Date: Sun, 23 Nov 2025 13:26:09 +0800 Subject: [PATCH] =?UTF-8?q?feat(auto=5Fscheduler):=20=E6=B7=BB=E5=8A=A0?= =?UTF-8?q?=E8=87=AA=E5=8A=A8=E8=B0=83=E5=BA=A6=E5=99=A8=EF=BC=8C=E4=BD=BF?= =?UTF-8?q?=E7=94=A8=E4=BF=A1=E5=8F=B7=E9=87=8F=EF=BC=88Semaphore=EF=BC=89?= =?UTF-8?q?=E5=B9=B6=E5=8F=91=E9=99=90=E5=88=B6=E5=92=8C=E5=BC=B1=E5=BC=95?= =?UTF-8?q?=E7=94=A8=E5=AD=97=E5=85=B8=EF=BC=88WeakValueDictionary?= =?UTF-8?q?=EF=BC=89=E4=BC=98=E5=8C=96=E9=94=81=E7=AE=A1=E7=90=86=EF=BC=9B?= =?UTF-8?q?=E7=BE=A4=E9=BB=91=E7=99=BD=E5=90=8D=E5=8D=95=E6=9C=BA=E5=88=B6?= =?UTF-8?q?=E3=80=82=20(#50)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(auto_scheduler): 添加自动调度器，使用信号量（Semaphore）并发限制和弱引用字典（WeakValueDictionary）优化锁管理。 * feat(auto_scheduler): 群黑白名单机制 * feat(自动分析最大并发数): 可配置项添加信号量自定义控制 * debug(获取群) * fix(config) * fix(pre-commit): cr --- _conf_schema.json | 19 +++++- main.py | 76 ++++++++++++++------- src/core/config.py | 66 ++++++++++++------ src/scheduler/auto_scheduler.py | 114 ++++++++++++++++++++++++++++---- 4 files changed, 215 insertions(+), 60 deletions(-) diff --git a/_conf_schema.json b/_conf_schema.json index f3ca901..ec25743 100644 --- a/_conf_schema.json +++ b/_conf_schema.json @@ -1,13 +1,26 @@ { - "enabled_groups": { + "group_list_mode": { + "description": "群聊权限模式", + "type": "string", + "options": ["whitelist", "blacklist", "none"], + "default": "none", + "hint": "whitelist: 仅允许列表内群；blacklist: 拒绝列表内群；none: 不限制" + }, + "group_list": { "type": "list", - "description": "启用分析功能的QQ群列表", + "description": "群组名单列表", "default": [], - "hint": "填入允许使用分析功能的QQ群号", + "hint": "黑白名单模式下使用的群号列表", "items": { "type": "string" } }, + "max_concurrent_tasks": { + "type": "int", + "description": "自动分析最大并发数", + "default": 5, + "hint": "同时进行的群聊分析任务数量，建议根据机器性能调整，过高可能导致LLM API RPM 超出限制，卡顿或被风控" + }, "max_messages": { "type": "int", "description": "单次分析的获取最大消息条数基准", diff --git a/main.py b/main.py index 9ba34f2..f517f6b 100644 --- a/main.py +++ b/main.py @@ -159,8 +159,7 @@ class QQGroupDailyAnalysis(Star): bot_manager.update_from_event(event) # 检查群组权限 - enabled_groups = config_manager.get_enabled_groups() - if enabled_groups and group_id not in enabled_groups: + if not config_manager.is_group_allowed(group_id): yield event.plain_result("❌ 此群未启用日常分析功能") return @@ -371,26 +370,56 @@ class QQGroupDailyAnalysis(Star): return if action == "enable": - enabled_groups = config_manager.get_enabled_groups() - if group_id not in enabled_groups: - config_manager.add_enabled_group(group_id) - yield event.plain_result("✅ 已为当前群启用日常分析功能") - - # 重新启动定时任务 - await auto_scheduler.restart_scheduler() + mode = config_manager.get_group_list_mode() + if mode == "whitelist": + glist = config_manager.get_group_list() + if group_id not in glist: + glist.append(group_id) + config_manager.set_group_list(glist) + yield event.plain_result("✅ 已将当前群加入白名单") + # 重新启动定时任务 + await auto_scheduler.restart_scheduler() + else: + yield event.plain_result("ℹ️ 当前群已在白名单中") + elif mode == "blacklist": + glist = config_manager.get_group_list() + if group_id in glist: + glist.remove(group_id) + config_manager.set_group_list(glist) + yield event.plain_result("✅ 已将当前群从黑名单移除") + # 重新启动定时任务 + await auto_scheduler.restart_scheduler() + else: + yield event.plain_result("ℹ️ 当前群不在黑名单中") else: - yield event.plain_result("ℹ️ 当前群已启用日常分析功能") + yield event.plain_result("ℹ️ 当前为无限制模式，所有群聊默认启用") elif action == "disable": - enabled_groups = config_manager.get_enabled_groups() - if group_id in enabled_groups: - config_manager.remove_enabled_group(group_id) - yield event.plain_result("✅ 已为当前群禁用日常分析功能") - - # 重新启动定时任务 - await auto_scheduler.restart_scheduler() + mode = config_manager.get_group_list_mode() + if mode == "whitelist": + glist = config_manager.get_group_list() + if group_id in glist: + glist.remove(group_id) + config_manager.set_group_list(glist) + yield event.plain_result("✅ 已将当前群从白名单移除") + # 重新启动定时任务 + await auto_scheduler.restart_scheduler() + else: + yield event.plain_result("ℹ️ 当前群不在白名单中") + elif mode == "blacklist": + glist = config_manager.get_group_list() + if group_id not in glist: + glist.append(group_id) + config_manager.set_group_list(glist) + yield event.plain_result("✅ 已将当前群加入黑名单") + # 重新启动定时任务 + await auto_scheduler.restart_scheduler() + else: + yield event.plain_result("ℹ️ 当前群已在黑名单中") else: - yield event.plain_result("ℹ️ 当前群未启用日常分析功能") + yield event.plain_result( + "ℹ️ 当前为无限制模式，如需禁用请切换到黑名单模式" + ) elif action == "reload": # 重新启动定时任务 @@ -399,8 +428,7 @@ class QQGroupDailyAnalysis(Star): elif action == "test": # 测试自动分析功能 - enabled_groups = config_manager.get_enabled_groups() - if group_id not in enabled_groups: + if not config_manager.is_group_allowed(group_id): yield event.plain_result("❌ 请先启用当前群的分析功能") return @@ -417,8 +445,10 @@ class QQGroupDailyAnalysis(Star): yield event.plain_result(f"❌ 自动分析测试失败: {str(e)}") else: # status - enabled_groups = config_manager.get_enabled_groups() - status = "已启用" if group_id in enabled_groups else "未启用" + is_allowed = config_manager.is_group_allowed(group_id) + status = "已启用" if is_allowed else "未启用" + mode = config_manager.get_group_list_mode() + auto_status = ( "已启用" if config_manager.get_enable_auto_analysis() else "未启用" ) @@ -429,7 +459,7 @@ class QQGroupDailyAnalysis(Star): min_threshold = config_manager.get_min_messages_threshold() yield event.plain_result(f"""📊 当前群分析功能状态: -• 群分析功能: {status} +• 群分析功能: {status} (模式: {mode}) • 自动分析: {auto_status} ({auto_time}) • 输出格式: {output_format} • PDF 功能: {pdf_status} diff --git a/src/core/config.py b/src/core/config.py index 1826173..69a4928 100644 --- a/src/core/config.py +++ b/src/core/config.py @@ -16,9 +16,37 @@ class ConfigManager: self._pyppeteer_version = None self._check_pyppeteer_availability() - def get_enabled_groups(self) -> list[str]: - """获取启用的群组列表""" - return self.config.get("enabled_groups", []) + def get_group_list_mode(self) -> str: + """获取群组列表模式 (whitelist/blacklist/none)""" + return self.config.get("group_list_mode", "none") + + def get_group_list(self) -> list[str]: + """获取群组列表（用于黑白名单）""" + return self.config.get("group_list", []) + + def is_group_allowed(self, group_id: str) -> bool: + """根据配置的白/黑名单判断是否允许在该群聊中使用""" + mode = self.get_group_list_mode().lower() + if mode not in ("whitelist", "blacklist", "none"): + mode = "none" + + # none模式下，不进行黑白名单检查，由调用方决定（通常是回退到 enabled_groups） + if mode == "none": + return True + + glist = [str(g) for g in self.get_group_list()] + group_id_str = str(group_id) + + if mode == "whitelist": + return group_id_str in glist if glist else False + if mode == "blacklist": + return group_id_str not in glist if glist else True + + return True + + def get_max_concurrent_tasks(self) -> int: + """获取自动分析最大并发数""" + return self.config.get("max_concurrent_tasks", 5) def get_max_messages(self) -> int: """获取最大消息数量""" @@ -203,9 +231,19 @@ class ConfigManager: self.config["output_format"] = format_type self.config.save_config() - def set_enabled_groups(self, groups: list[str]): - """设置启用的群组列表""" - self.config["enabled_groups"] = groups + def set_group_list_mode(self, mode: str): + """设置群组列表模式""" + self.config["group_list_mode"] = mode + self.config.save_config() + + def set_group_list(self, groups: list[str]): + """设置群组列表""" + self.config["group_list"] = groups + self.config.save_config() + + def set_max_concurrent_tasks(self, count: int): + """设置自动分析最大并发数""" + self.config["max_concurrent_tasks"] = count self.config.save_config() def set_max_messages(self, count: int): @@ -273,22 +311,6 @@ class ConfigManager: self.config["pdf_filename_format"] = format_str self.config.save_config() - def add_enabled_group(self, group_id: str): - """添加启用的群组""" - enabled_groups = self.get_enabled_groups() - if group_id not in enabled_groups: - enabled_groups.append(group_id) - self.config["enabled_groups"] = enabled_groups - self.config.save_config() - - def remove_enabled_group(self, group_id: str): - """移除启用的群组""" - enabled_groups = self.get_enabled_groups() - if group_id in enabled_groups: - enabled_groups.remove(group_id) - self.config["enabled_groups"] = enabled_groups - self.config.save_config() - def get_enable_user_card(self) -> bool: """获取是否使用用户群名片""" return self.config.get("enable_user_card", False) diff --git a/src/scheduler/auto_scheduler.py b/src/scheduler/auto_scheduler.py index 5455119..e5fa846 100644 --- a/src/scheduler/auto_scheduler.py +++ b/src/scheduler/auto_scheduler.py @@ -4,6 +4,7 @@ """ import asyncio +import weakref from datetime import datetime, timedelta from astrbot.api import logger @@ -189,7 +190,22 @@ class AutoScheduler: try: logger.info("开始执行自动群聊分析（并发模式）") - enabled_groups = self.config_manager.get_enabled_groups() + # 根据配置确定需要分析的群组 + group_list_mode = self.config_manager.get_group_list_mode() + + # 始终获取所有群组并进行过滤 + logger.info(f"自动分析使用 {group_list_mode} 模式，正在获取群列表...") + all_groups = await self._get_all_groups() + logger.info(f"共获取到 {len(all_groups)} 个群组: {all_groups}") + enabled_groups = [] + for group_id in all_groups: + if self.config_manager.is_group_allowed(group_id): + enabled_groups.append(group_id) + + logger.info( + f"根据 {group_list_mode} 过滤后，共有 {len(enabled_groups)} 个群聊需要分析" + ) + if not enabled_groups: logger.info("没有启用的群聊需要分析") return @@ -199,10 +215,21 @@ class AutoScheduler: ) # 创建并发任务 - 为每个群聊创建独立的分析任务 + # 限制最大并发数 + max_concurrent = self.config_manager.get_max_concurrent_tasks() + logger.info(f"自动分析并发数限制: {max_concurrent}") + sem = asyncio.Semaphore(max_concurrent) + + async def safe_perform_analysis(group_id): + async with sem: + return await self._perform_auto_analysis_for_group_with_timeout( + group_id + ) + analysis_tasks = [] for group_id in enabled_groups: task = asyncio.create_task( - self._perform_auto_analysis_for_group_with_timeout(group_id), + safe_perform_analysis(group_id), name=f"analysis_group_{group_id}", ) analysis_tasks.append(task) @@ -246,12 +273,16 @@ class AutoScheduler: # 为每个群聊使用独立的锁，避免全局锁导致串行化 group_lock_key = f"analysis_{group_id}" if not hasattr(self, "_group_locks"): - self._group_locks = {} + self._group_locks = weakref.WeakValueDictionary() - if group_lock_key not in self._group_locks: - self._group_locks[group_lock_key] = asyncio.Lock() + # 从 WeakValueDictionary 获取锁，如果不存在则创建 + # 注意：必须将锁赋值给局部变量以保持引用，否则可能会被回收 + lock = self._group_locks.get(group_lock_key) + if lock is None: + lock = asyncio.Lock() + self._group_locks[group_lock_key] = lock - async with self._group_locks[group_lock_key]: + async with lock: try: start_time = asyncio.get_event_loop().time() @@ -381,12 +412,71 @@ class AutoScheduler: logger.error(f"群 {group_id} 自动分析执行失败: {e}", exc_info=True) finally: - # 清理群聊锁资源（可选，防止内存泄漏） - if hasattr(self, "_group_locks") and len(self._group_locks) > 50: - old_locks = list(self._group_locks.keys())[:10] - for lock_key in old_locks: - if not self._group_locks[lock_key].locked(): - del self._group_locks[lock_key] + # 锁资源由 WeakValueDictionary 自动管理，无需手动清理 + logger.info(f"群 {group_id} 自动分析完成") + + async def _get_all_groups(self) -> list[str]: + """获取所有bot实例所在的群列表""" + all_groups = set() + + if ( + not hasattr(self.bot_manager, "_bot_instances") + or not self.bot_manager._bot_instances + ): + return [] + + for platform_id, bot_instance in self.bot_manager._bot_instances.items(): + try: + # 尝试使用 call_action 获取群列表 + call_action_func = None + if hasattr(bot_instance, "call_action"): + call_action_func = bot_instance.call_action + elif hasattr(bot_instance, "api") and hasattr( + bot_instance.api, "call_action" + ): + call_action_func = bot_instance.api.call_action + + if call_action_func: + # 尝试 OneBot v11 get_group_list + try: + result = await call_action_func("get_group_list") + logger.debug( + f"平台 {platform_id} get_group_list 返回类型: {type(result)}" + ) + + # 处理可能的字典返回 (e.g. {'data': [...], 'retcode': 0}) + if ( + isinstance(result, dict) + and "data" in result + and isinstance(result["data"], list) + ): + logger.debug("检测到字典格式返回，提取 data 字段") + result = result["data"] + + if isinstance(result, list): + for group in result: + if isinstance(group, dict) and "group_id" in group: + all_groups.add(str(group["group_id"])) + logger.info( + f"平台 {platform_id} 成功获取 {len(result)} 个群组" + ) + else: + logger.warning( + f"平台 {platform_id} get_group_list 返回格式非列表: {result}" + ) + except Exception as e: + logger.debug( + f"平台 {platform_id} 获取群列表失败 (get_group_list): {e}" + ) + + # 如果需要，尝试其他方法（例如针对其他协议） + # 目前专注于 OneBot v11，因为它是最常见的 + else: + logger.debug(f"平台 {platform_id} 的 bot 实例没有 call_action 方法") + except Exception as e: + logger.error(f"平台 {platform_id} 获取群列表异常: {e}") + + return list(all_groups) async def _send_analysis_report(self, group_id: str, analysis_result: dict): """发送分析报告到群"""