From 9ed4771730d8e2061df78d30745d3539f11e5608 Mon Sep 17 00:00:00 2001 From: SXP-Simon Date: Thu, 26 Mar 2026 19:16:41 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=8C=E5=96=84=E8=87=AA=E5=8A=A8?= =?UTF-8?q?=E5=88=86=E6=9E=90=E5=90=8D=E5=8D=95=E8=B0=83=E5=BA=A6=E4=B8=8E?= =?UTF-8?q?=E9=9D=A2=E6=9D=BF=E9=85=8D=E7=BD=AE=E8=AF=B4=E6=98=8E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 重构自动分析调度流程：以分层名单解析目标群组，统一报告入口，并按群组模式分发传统/增量任务。 - 新增定时名单与增量名单配置能力：补充 scheduled_group_list(_mode)、incremental_group_list(_mode)、incremental_fallback_enabled。 - 优化失败兜底策略：增量最终报告在超时、异常或无有效结果时，可回退至传统全量分析。 - 完善配置与文档说明：在 _conf_schema.json 与 README 中补充面板操作指引、判定顺序与边界场景说明。 - 兼容旧配置行为：保留废弃接口，并在配置管理中统一名单匹配语义（UMO/群号兼容）。 --- README.md | 59 ++- _conf_schema.json | 70 +++- main.py | 2 +- src/infrastructure/config/config_manager.py | 98 ++++- .../scheduler/auto_scheduler.py | 394 ++++++++---------- 5 files changed, 380 insertions(+), 243 deletions(-) diff --git a/README.md b/README.md index 29b6ff1..f5a0d7a 100644 --- a/README.md +++ b/README.md @@ -97,11 +97,64 @@ _✨ 一个基于 AstrBot 的智能群聊分析插件，支持 **QQ (OneBot)** | 配置项 | 说明 | 备注 | |--------|------|--------| -| 启用自动分析 | 启用定时触发自动分析功能。需要在群聊列表中添加群号或者使用 `/分析设置 enable` 启用。 | 默认关闭，建议开启 | -| 启用增量分析 | 启用增量分析模式，定时分批分析消息，最终生成全天报告。 | 适合消息量大的群聊 | +| 定时分析名单模式 + 列表 | 控制哪些群参与定时任务（报告时间点触发）。 | `whitelist + 空列表` 表示不注册定时任务 | +| 增量分析名单模式 + 列表 | 控制哪些群走增量模式，其他群走传统全量。 | `whitelist + 空列表` 表示不启用增量周期任务 | | PDF格式的报告 | 初次使用需要使用 `/安装PDF` 命令安装依赖。需重启 AstrBot 生效。 | 输出格式需设为 PDF | | 自定义LLM服务 | 用户可自行选取个人提供的服务商。 | 留空则回退到默认服务商 | +### 分析黑白名单配置说明（小白能懂） + +下面只讲“在面板里怎么点”。 + +#### 自动分析的判定顺序（很重要） + +系统会按下面顺序判断，前一关没过就直接停止： + +1. 基础群权限（`basic`） +2. 定时分析名单（`auto_analysis`） +3. 增量名单（`incremental`，只决定模式，不决定放行） + +一句话版： +`basic` 决定“能不能参与自动分析” -> `auto_analysis` 决定“会不会自动触发” -> `incremental` 决定“触发后用哪种分析方式”。 + +#### 场景 A：只让一个群自动出报告（最常用） + +1. 在插件配置面板找到 `定时分析设置`。 +2. 把 `定时分析名单模式` 设为 `whitelist`。 +3. 在 `定时分析群列表` 里添加你的目标群（建议粘贴 `/sid` 拿到的完整会话ID）。 +4. 在 `自动分析时间列表` 里填时间（例如 `09:00`、`21:30`）。 +5. 去 `增量分析设置`： +6. 把 `增量分析名单模式` 设为 `whitelist`，并保持 `增量分析群列表` 为空。 +这样就是“这个群会自动跑，但走普通分析，不走增量”。 + +#### 场景 B：除了某个群，其他群都自动跑 + +1. 在 `定时分析设置` 里把 `定时分析名单模式` 设为 `blacklist`。 +2. 在 `定时分析群列表` 里填“不要自动跑”的那个群。 +3. 在 `自动分析时间列表` 里填每天自动运行时间。 +4. 如果你希望其他群默认走增量： +5. 在 `增量分析设置` 把 `增量分析名单模式` 设为 `blacklist`，并把 `增量分析群列表` 留空。 + +#### 场景 C：Telegram 用户怎么填最稳 + +1. 在面板里需要填群的地方，尽量填完整会话ID（例如 `telegram2:GroupMessage:-1001234567890`）。 +2. 不建议新手只填纯群号，容易填错平台。 +3. 先在群里执行 `/sid`，复制结果粘贴到列表里就行。 + +#### 最容易踩坑的 3 点 + +- `定时分析名单模式` 选 `whitelist` 时，如果 `定时分析群列表` 为空，任务不会自动跑。 +- `增量分析名单模式` 选 `whitelist` 时，如果 `增量分析群列表` 为空，增量不会生效，会走普通分析。 +- `增量失败自动回退全量分析` 建议保持开启，这样增量异常时也能尽量产出报告。 + +#### 你可能会问（关键边界） + +- 不在“定时白名单”里，但在“增量白名单”里，会触发吗？ +不会。因为会先被“定时白名单”拦住，进不到增量判断。 + +- 不在“基础群权限”里，但你开了定时分析，会触发吗？ +不会。基础群权限是第一关，不通过就不会进入后续流程。 + > [!IMPORTANT] > **多平台配置注意**： > - **自动发现**: 插件会自动发现已登录的 Bot 实例。 @@ -214,7 +267,7 @@ _✨ 一个基于 AstrBot 的智能群聊分析插件，支持 **QQ (OneBot)** - **自动去重**：智能识别重复话题和金句，合并生成最终报告。 **启用方法：** -- 在插件配置中开启 `incremental_enabled`，并根据需要调整分析间隔 (`incremental_interval_minutes`) 和消息阈值。 +- 通过 `incremental_group_list_mode + incremental_group_list` 指定哪些群走增量模式，并根据需要调整分析间隔 (`incremental_interval_minutes`) 和消息阈值。 ## 常见问题 (FAQ) diff --git a/_conf_schema.json b/_conf_schema.json index ae5d059..14a29b5 100644 --- a/_conf_schema.json +++ b/_conf_schema.json @@ -2,7 +2,7 @@ "basic": { "description": "基础设置", "type": "object", - "hint": "群聊分析插件的基础配置项，包括群聊权限、消息数量、输出格式等", + "hint": "群聊分析插件的基础配置项，包括群聊权限、消息数量、输出格式等。注意：基础群权限是第一道门槛，不通过就不会进入定时/增量流程。", "items": { "group_list_mode": { "description": "群聊权限模式", @@ -13,13 +13,13 @@ "none" ], "default": "none", - "hint": "whitelist: 仅允许列表内群；blacklist: 拒绝列表内群；none: 不限制" + "hint": "先决定“哪些群可以用这个插件”。这是第一道门槛：不通过这里，后面的定时/增量都不会触发。whitelist=只允许列表里的群；blacklist=屏蔽列表里的群；none=所有群都可以。示例：whitelist + [\"onebot:GroupMessage:123456\"] 表示只有这个群能用。" }, "group_list": { "type": "list", "description": "群组白/黑名单列表", "default": [], - "hint": "黑白名单模式下使用的群组列表。支持填写 AstrBot UMO (如 xxxxx:GroupMessage:123456) 或 纯群号 (如 123456，将尝试自动匹配)。可以使用 /sid 命令查看当前会话的 UMO。", + "hint": "要填的群列表。可以填完整会话ID（如 onebot:GroupMessage:123456）或只填群号（如 123456）。新手建议优先填完整会话ID，更不容易填错。可用 /sid 获取当前会话ID。", "items": { "type": "string" } @@ -98,23 +98,36 @@ } }, "auto_analysis": { - "description": "自动分析设置", + "description": "定时分析设置", "type": "object", - "hint": "定时自动分析相关配置，包括启用开关、分析时间和并发数", + "hint": "配置每日自动分析的时间点和目标群聊范围。只有先通过基础群权限的群，才会在这里继续判断。", "items": { - "enable_auto_analysis": { - "type": "bool", - "description": "启用定时自动分析", - "default": false, - "hint": "是否在指定时间自动进行群聊分析，启用需要填写下面的机器人ID号，并且确保群聊号在 enabled_groups 配置中，否则获取不到实例，不会自动分析" - }, "auto_analysis_time": { "type": "list", "description": "自动分析时间列表", "default": [ "23:00" ], - "hint": "每日自动分析的时间点列表，格式HH:MM（例如 23:00 ），支持多个时间点，增量分析也基于此时间点进行分析和报告生成", + "hint": "每天几点自动出报告，格式是 HH:MM。可填多个时间，例如 [\"09:00\", \"21:30\"]。", + "items": { + "type": "string" + } + }, + "scheduled_group_list_mode": { + "type": "string", + "description": "定时分析名单模式", + "options": [ + "whitelist", + "blacklist" + ], + "default": "whitelist", + "hint": "定时任务要作用到哪些群。whitelist=只跑下面列表里的群；blacklist=除了下面列表，其他都跑。注意：这里只对“已通过基础群权限”的群生效。" + }, + "scheduled_group_list": { + "type": "list", + "description": "定时分析群列表", + "default": [], + "hint": "定时任务群列表。示例1：mode=whitelist + [\"onebot:GroupMessage:123456\"] => 只跑这个群。示例2：mode=blacklist + [\"onebot:GroupMessage:123456\"] => 这个群不跑，其他群跑。注意：whitelist + 空列表 = 不会开启定时任务。即使某群在增量名单里，只要不在定时白名单里，也不会自动触发。", "items": { "type": "string" } @@ -257,13 +270,26 @@ "incremental": { "description": "增量分析设置", "type": "object", - "hint": "增量分析模式相关配置，启用后插件会在一天内多次进行小规模分析并累积结果，不开启则以基础设置的参数为标准；适合用于消息量特别大的群聊和有足够 Token 折腾的用户，尽力做到原本的单次全量分析模式的上位替代（如果把消息阈值拉到足够大），提升分析覆盖率和24小时活跃图表准确性", + "hint": "增量分析通过在活跃时段分步提取摘要，能显著提高分析深度和覆盖率。这里仅决定“已通过前两关的群”是走增量还是普通模式。", "items": { - "incremental_enabled": { - "type": "bool", - "description": "启用增量分析模式", - "default": false, - "hint": "启用后，插件会在一天内多次进行小规模分析并累积结果，最终在报告时间汇总输出。适用于消息量大的群聊，可显著提升分析覆盖率和24小时活跃图表准确性。关闭时保持传统的单次全量分析模式。" + "incremental_group_list_mode": { + "type": "string", + "description": "增量分析名单模式", + "options": [ + "whitelist", + "blacklist" + ], + "default": "whitelist", + "hint": "哪些群用“增量分析”。whitelist=只有列表里的群用增量；blacklist=除了列表外都用增量。没命中的群会走普通模式。注意：增量名单不负责放行群，只负责选模式。" + }, + "incremental_group_list": { + "type": "list", + "description": "增量分析群列表", + "default": [], + "hint": "增量分析群列表。示例1：mode=whitelist + [\"onebot:GroupMessage:123456\"] => 只有这个群增量。示例2：mode=blacklist + [\"onebot:GroupMessage:123456\"] => 这个群不用增量，其他群用增量。注意：whitelist + 空列表 = 不启用增量。只有先通过基础群权限 + 定时名单的群，才会在这里判断。", + "items": { + "type": "string" + } }, "incremental_report_immediately": { "type": "bool", @@ -324,6 +350,12 @@ "description": "多群增量分析交错间隔（秒）", "default": 2, "hint": "当同时对多个群进行增量分析时，每个群之间的延迟间隔（秒），用于均匀分散 API 请求压力。" + }, + "incremental_fallback_enabled": { + "type": "bool", + "description": "增量失败自动回退全量分析", + "default": true, + "hint": "建议开启。开启后，如果增量分析失败，会自动改用普通分析，避免当天没报告。" } } }, @@ -484,4 +516,4 @@ } } } -} \ No newline at end of file +} diff --git a/main.py b/main.py index a3a05df..6b5c81c 100644 --- a/main.py +++ b/main.py @@ -871,7 +871,7 @@ class GroupDailyAnalysis(Star): mode = self.config_manager.get_group_list_mode() auto_status = ( - "已启用" if self.config_manager.get_enable_auto_analysis() else "未启用" + "已启用" if self.config_manager.is_auto_analysis_enabled() else "未启用" ) auto_time = self.config_manager.get_auto_analysis_time() diff --git a/src/infrastructure/config/config_manager.py b/src/infrastructure/config/config_manager.py index b671192..78622d1 100644 --- a/src/infrastructure/config/config_manager.py +++ b/src/infrastructure/config/config_manager.py @@ -137,8 +137,13 @@ class ConfigManager: return val if isinstance(val, list) else ["09:00"] def get_enable_auto_analysis(self) -> bool: - """获取是否启用自动分析""" - return self._get_group("auto_analysis").get("enable_auto_analysis", False) + """ + 获取是否启用自动分析（兼容旧接口）。 + + 旧版本使用 auto_analysis.enable_auto_analysis 布尔值； + 新版本改为由 scheduled_group_list_mode + scheduled_group_list 推导。 + """ + return self.is_auto_analysis_enabled() def get_output_format(self) -> str: """获取输出格式""" @@ -390,15 +395,75 @@ class ConfigManager: self.config.save_config() def set_auto_analysis_time(self, time_val: str | list[str]): - """设置自动分析时间""" + """设置自动分析时间点""" self._ensure_group("auto_analysis")["auto_analysis_time"] = time_val self.config.save_config() - def set_enable_auto_analysis(self, enabled: bool): - """设置是否启用自动分析""" - self._ensure_group("auto_analysis")["enable_auto_analysis"] = enabled + def is_auto_analysis_enabled(self) -> bool: + """ + 判断自动分析功能是否通过名单“按需开启”。 + 逻辑：如果是白名单模式且名单不为空，或者为黑名单模式，则视为开启。 + """ + mode = self.get_scheduled_group_list_mode() + lst = self.get_scheduled_group_list() + return (mode == "whitelist" and len(lst) > 0) or (mode == "blacklist") + + def get_scheduled_group_list_mode(self) -> str: + """获取定时分析名单模式 (whitelist/blacklist)""" + return self._get_group("auto_analysis").get( + "scheduled_group_list_mode", "whitelist" + ) + + def set_scheduled_group_list_mode(self, mode: str): + """设置定时分析名单模式""" + self._ensure_group("auto_analysis")["scheduled_group_list_mode"] = mode self.config.save_config() + def get_scheduled_group_list(self) -> list[str]: + """获取定时分析目标群列表""" + return self._get_group("auto_analysis").get("scheduled_group_list", []) + + def set_scheduled_group_list(self, groups: list[str]): + """设置定时分析目标群列表""" + self._ensure_group("auto_analysis")["scheduled_group_list"] = groups + self.config.save_config() + + def is_group_in_filtered_list( + self, group_umo_or_id: str, mode: str, group_list: list + ) -> bool: + """ + 通用的名单判定逻辑。 + + 逻辑如下： + - whitelist 模式： + - 如果列表为空，则视为“此级别未开启”。 + - 如果不为空，仅在列表中的通过。 + - blacklist 模式： + - 在列表中的不通过。 + - 如果列表为空，则全部通过。 + """ + group_list = [str(x).strip() for x in group_list] + target = str(group_umo_or_id).strip() + + # 兼容 UMO 匹配 (如果列表里写的是 ID，UMO 也能匹配上) + def match_umo(umo: str, item: str) -> bool: + if umo == item: + return True + if ":" in umo and umo.split(":")[-1] == item: + return True + return False + + if mode == "whitelist": + if not group_list: + # 白名单为空：此级别不开启 (按需开启逻辑) + return False + return any(match_umo(target, x) for x in group_list) + else: # blacklist + if not group_list: + # 黑名单为空：全通过 + return True + return not any(match_umo(target, x) for x in group_list) + def set_min_messages_threshold(self, threshold: int): """设置最小消息阈值""" self._ensure_group("basic")["min_messages_threshold"] = threshold @@ -509,8 +574,25 @@ class ConfigManager: # ========== 增量分析配置 ========== def get_incremental_enabled(self) -> bool: - """获取是否启用增量分析模式""" - return self._get_group("incremental").get("incremental_enabled", False) + """获取是否开启了增量分析（由名单状态决定）""" + mode = self.get_incremental_group_list_mode() + lst = self.get_incremental_group_list() + # 如果是白名单且不为空，或者是黑名单模式，则视为功能“开启” + return (mode == "whitelist" and len(lst) > 0) or (mode == "blacklist") + + def get_incremental_group_list_mode(self) -> str: + """获取增量分析名单模式 (whitelist/blacklist)""" + return self._get_group("incremental").get( + "incremental_group_list_mode", "whitelist" + ) + + def get_incremental_group_list(self) -> list[str]: + """获取增量分析群列表""" + return self._get_group("incremental").get("incremental_group_list", []) + + def get_incremental_fallback_enabled(self) -> bool: + """获取增量分析失败回退到全量分析的开关（默认启用）""" + return self._get_group("incremental").get("incremental_fallback_enabled", True) def get_incremental_report_immediately(self) -> bool: """获取是否启用增量分析立即发送报告（调试用）""" diff --git a/src/infrastructure/scheduler/auto_scheduler.py b/src/infrastructure/scheduler/auto_scheduler.py index 691df9f..ac3760b 100644 --- a/src/infrastructure/scheduler/auto_scheduler.py +++ b/src/infrastructure/scheduler/auto_scheduler.py @@ -145,75 +145,76 @@ class AutoScheduler: # ================================================================ def schedule_jobs(self, context): - """注册定时任务，根据配置选择传统模式或增量模式""" - # 先清理旧任务 + """根据分层名单配置注册定时任务。""" + # 首先清理之前的任务 self.unschedule_jobs(context) - # unschedule_jobs 会将 _terminating 置为 True（用于关闭场景）， - # 但 schedule_jobs 表示插件仍在运行，需要重置终止标志位 + # unschedule_jobs 会将 _terminating 设为 True (用于关闭场景), + # 但 schedule_jobs 意味着插件仍在运行；因此需要重置此标志位 self._terminating = False - if not self.config_manager.get_enable_auto_analysis(): - logger.info("自动分析功能未启用，不注册定时任务") + if not self.config_manager.is_auto_analysis_enabled(): + logger.info("定时分析名单为空且为白名单模式，不注册定时任务。") return scheduler = context.cron_manager.scheduler - # 根据增量模式开关决定调度策略 - if self.config_manager.get_incremental_enabled(): - logger.info("增量分析模式已启用，注册增量调度任务") - self._schedule_incremental_jobs(scheduler) - else: - logger.info("使用传统分析模式，注册定时分析任务") - self._schedule_traditional_jobs(scheduler) + # 1. 注册核心报告生成任务（涵盖全量分析与增量总结报告） + # 每个配置的时间点都会触发一次解析 + logger.info("注册定时分析报告任务...") + self._schedule_report_time_jobs(scheduler) - def _schedule_traditional_jobs(self, scheduler): - """注册传统模式的定时任务（在配置的时间点执行完整分析）""" + # 2. 只有在增量功能总开关开启时，才注册全天候的增量提取任务 + if self.config_manager.get_incremental_enabled(): + logger.info("增量分析功能已开启，正在注册全天增量提取任务...") + self._schedule_incremental_cron_jobs(scheduler) + else: + logger.info("增量分析总开关未启用，仅执行传统定时全量分析。") + + def _schedule_report_time_jobs(self, scheduler): + """在配置的时间点注册报告生成任务。 + + 这些任务根据运行时解析出的生效模式，决定执行传统的全量分析还是增量汇报。 + """ time_config = self.config_manager.get_auto_analysis_time() if isinstance(time_config, str): time_config = [time_config] for i, t_str in enumerate(time_config): try: - # t_str 格式: "HH:MM" t_str = str(t_str).replace("：", ":").strip() hour, minute = t_str.split(":") - # 创建 CronTrigger trigger = CronTrigger(hour=int(hour), minute=int(minute)) - - # 任务 ID job_id = f"astrbot_plugin_qq_group_daily_analysis_trigger_{i}" - # 添加任务 scheduler.add_job( - self._run_auto_analysis, + self._run_scheduled_report, trigger=trigger, id=job_id, replace_existing=True, misfire_grace_time=60, ) self.scheduler_job_ids.append(job_id) - logger.info(f"已注册定时自动分析任务: {t_str} (Job ID: {job_id})") + logger.info(f"已注册定时报告任务: {t_str} (Job ID: {job_id})") except Exception as e: logger.error(f"注册定时任务失败 ({t_str}): {e}") - def _schedule_incremental_jobs(self, scheduler): + def _schedule_incremental_cron_jobs(self, scheduler): """ - 注册增量模式的定时任务。 + 在活跃时段注册增量分析定时任务。 - 在活跃时段内按固定间隔注册增量分析任务， - 并在配置的报告时间点注册最终报告生成任务。 + 这类任务仅执行增量数据的提取；而报告生成阶段在配置的每日分析时间点进行。 """ active_start_hour = self.config_manager.get_incremental_active_start_hour() active_end_hour = self.config_manager.get_incremental_active_end_hour() interval_minutes = self.config_manager.get_incremental_interval_minutes() max_daily = self.config_manager.get_incremental_max_daily_analyses() - # 计算增量分析触发时间点 + # 计算活跃时段内的触发时间点 trigger_times = [] - current_minutes = active_start_hour * 60 # 从活跃开始小时的 :00 开始 + current_minutes = active_start_hour * 60 end_minutes = active_end_hour * 60 while current_minutes < end_minutes and len(trigger_times) < max_daily: @@ -242,35 +243,7 @@ class AutoScheduler: except Exception as e: logger.error(f"注册增量分析任务失败 ({hour:02d}:{minute:02d}): {e}") - # 注册最终报告生成任务（使用配置的自动分析时间点） - time_config = self.config_manager.get_auto_analysis_time() - if isinstance(time_config, str): - time_config = [time_config] - - for i, t_str in enumerate(time_config): - try: - t_str = str(t_str).replace("：", ":").strip() - hour_str, minute_str = t_str.split(":") - - trigger = CronTrigger(hour=int(hour_str), minute=int(minute_str)) - job_id = f"incremental_final_report_{i}" - - scheduler.add_job( - self._run_incremental_final_report, - trigger=trigger, - id=job_id, - replace_existing=True, - misfire_grace_time=60, - ) - self.scheduler_job_ids.append(job_id) - logger.info(f"已注册增量最终报告任务: {t_str} (Job ID: {job_id})") - except Exception as e: - logger.error(f"注册增量最终报告任务失败 ({t_str}): {e}") - - logger.info( - f"增量调度注册完成: {len(trigger_times)} 个增量分析任务, " - f"{len(time_config)} 个最终报告任务" - ) + logger.info(f"增量调度注册完成: {len(trigger_times)} 个增量分析任务") def unschedule_jobs(self, context): """取消定时任务""" @@ -296,119 +269,147 @@ class AutoScheduler: self.scheduler_job_ids.clear() # ================================================================ - # 共享辅助方法：获取启用的群聊目标 + # 共享辅助方法：解析定时分析目标 # ================================================================ - async def _get_enabled_targets(self) -> set[tuple[str, str]]: + async def _get_scheduled_targets( + self, mode_filter: str | None = None + ) -> list[tuple[str, str, str]]: """ - 获取所有启用分析的群聊目标。 - 确保每个群组 ID 在一次调度任务中只出现一次（多平台/多适配器去重）。 + 根据分层过滤逻辑判定所有应参与计划分析的目标群组及其分析策略。 + + 判定过程： + 1. 准入层：群组必须在基础设置的允许名单内。 + 2. 定时层：群组需通过定时分析名单的过滤。 + 3. 模式层：如果群组在增量名单内，则使用增量模式，否则使用默认策略。 + + 参数： + mode_filter: 如果提供，则只返回匹配指定模式的目标 (traditional 或 incremental)。 """ - group_list_mode = self.config_manager.get_group_list_mode() - - # 使用字典记录 group_id -> platform_id 实现去重 - # 优先级：先发现先处理 - targets_map: dict[str, str] = {} - - # 1. 通过 API 获取所有群组（自动发现） - logger.info(f"自动分析使用 {group_list_mode} 模式，正在扫描所有平台的群列表...") + # 获取基础信息 all_groups = await self._get_all_groups() - for platform_id, group_id in all_groups: - group_id_str = str(group_id) - if group_id_str in targets_map: + # 预加载所有配置名单和模式 + sched_list = self.config_manager.get_scheduled_group_list() + sched_list_mode = self.config_manager.get_scheduled_group_list_mode() + + incr_list = self.config_manager.get_incremental_group_list() + incr_list_mode = self.config_manager.get_incremental_group_list_mode() + + result = [] + + # 遍历所有平台上的群组 + for platform_id, group_id_orig in all_groups: + group_id = str(group_id_orig) + umo = f"{platform_id}:GroupMessage:{group_id}" + + # 1. 准入层判定 (基础黑白名单) + if not self.config_manager.is_group_allowed(umo): continue - # 权限检查 - umo = f"{platform_id}:GroupMessage:{group_id_str}" - if self.config_manager.is_group_allowed(umo): - targets_map[group_id_str] = platform_id + # 2. 定时层判定 (定时分析黑白名单) + if not self.config_manager.is_group_in_filtered_list( + umo, sched_list_mode, sched_list + ): + continue - # 2. 白名单模式补全 - if group_list_mode == "whitelist": - whitelist_config = self.config_manager.get_group_list() - for item in whitelist_config: - item = str(item).strip() - if ":" in item: - parts = item.split(":") - if len(parts) >= 3: - p_id = parts[0] - g_id = parts[-1] - if g_id not in targets_map: - if self.bot_manager.get_bot_instance(p_id): - targets_map[g_id] = p_id + # 3. 模式层判定 (增量黑白名单) + # 3. 模式层判定 (增量黑白名单) + if self.config_manager.is_group_in_filtered_list(umo, incr_list_mode, incr_list): + # 如果在增量名单内，则执行增量模式 + effective_mode = "incremental" + else: + # 不在增量名单内，则执行普通模式 + effective_mode = "traditional" - # 转换为集合形式返回 - enabled_targets = set(targets_map.items()) - logger.info(f"扫码完成：共有 {len(enabled_targets)} 个群聊目标将执行分析任务") - return enabled_targets + # 4. 模式过滤 (如果函数调用者要求过滤) + if mode_filter and effective_mode != mode_filter: + continue + + result.append((group_id, platform_id, effective_mode)) + + logger.info( + f"分层调度解析完成：符合条件的群组共 {len(result)} 个" + + (f" (模式过滤: {mode_filter})" if mode_filter else "") + ) + return result # ================================================================ - # 传统模式：自动分析 + # 统一报告调度入口 # ================================================================ - async def _run_auto_analysis(self): - """执行传统自动分析 - 并发处理所有群聊""" + async def _run_scheduled_report(self): + """统一的定时分析入口。 + + 在配置的时间点触发，解析所有目标群并根据其分析模式分发任务： + - traditional: 执行全量拉取分析并发送报告 + - incremental: 执行增量最终报告阶段（合并并汇报） + """ if self._terminating: return try: - logger.info("开始执行自动群聊分析（并发模式）") + logger.info("定时报告触发 — 开始解析调度目标") - enabled_targets = await self._get_enabled_targets() + all_targets = await self._get_scheduled_targets() - if not enabled_targets: - logger.info("没有启用的群聊需要分析") + if not all_targets: + logger.info("没有配置的群聊需要定时分析") return - # 转为列表以便索引 - target_list = list(enabled_targets) - max_concurrent = self.config_manager.get_max_concurrent_tasks() sem = asyncio.Semaphore(max_concurrent) - logger.info(f"自动分析任务入口并发限制: {max_concurrent}") + logger.info( + f"定时报告: {len(all_targets)} 个目标 (并发限制: {max_concurrent})" + ) - async def throttled_analysis(gid, pid): + async def dispatch_group(gid, pid, mode): async with sem: - return await self._perform_auto_analysis_for_group_with_timeout( - gid, pid - ) + if mode == "incremental": + return await self._perform_incremental_final_report_for_group_with_timeout( + gid, pid + ) + else: + return await self._perform_auto_analysis_for_group_with_timeout( + gid, pid + ) - analysis_tasks = [] - for gid, pid in target_list: + tasks = [] + for gid, pid, mode in all_targets: if self._terminating: - logger.info("检测到插件正在停止，取消后续分析任务创建") + logger.info("检测到插件正在停止，取消后续任务创建") break task = asyncio.create_task( - throttled_analysis(gid, pid), - name=f"analysis_group_{gid}", + dispatch_group(gid, pid, mode), + name=f"report_{mode}_{gid}", ) - analysis_tasks.append(task) + tasks.append(task) - # 并发执行所有分析任务 - results = await asyncio.gather(*analysis_tasks, return_exceptions=True) + results = await asyncio.gather(*tasks, return_exceptions=True) - # 统计执行结果 + # 统计结果 success_count = 0 skip_count = 0 error_count = 0 for i, result in enumerate(results): - gid, _ = target_list[i] + gid, _, _ = all_targets[i] if isinstance(result, DuplicateGroupTaskError): - # 锁冲突导致的跳过 skip_count += 1 elif isinstance(result, Exception): - logger.error(f"群 {gid} 分析任务异常: {result}") + logger.error(f"群 {gid} 定时报告任务异常: {result}") error_count += 1 + elif isinstance(result, dict) and not result.get("success", True): + skip_count += 1 else: success_count += 1 logger.info( - f"并发分析完成 - 成功: {success_count}, 跳过: {skip_count}, 失败: {error_count}, 总计: {len(target_list)}" + f"定时报告完成 — 成功: {success_count}, 跳过: {skip_count}, " + f"失败: {error_count}, 总计: {len(all_targets)}" ) except Exception as e: - logger.error(f"自动分析执行失败: {e}", exc_info=True) + logger.error(f"定时报告执行失败: {e}", exc_info=True) async def _perform_auto_analysis_for_group_with_timeout( self, group_id: str, target_platform_id: str | None = None @@ -487,19 +488,20 @@ class AutoScheduler: # ================================================================ async def _run_incremental_analysis(self): - """执行增量分析任务 - 并发处理模式""" + """为所有目标模式设定为 incremental 的群执行增量分析任务。""" if self._terminating: return try: logger.info("开始执行自动增量分析（并发模式）") - enabled_targets = await self._get_enabled_targets() + # 仅选取模式为 incremental 的目标群 + incr_targets = await self._get_scheduled_targets(mode_filter="incremental") - if not enabled_targets: - logger.info("没有启用的群聊需要增量分析") + if not incr_targets: + logger.info("没有配置为增量模式的群聊需要增量分析") return - target_list = list(enabled_targets) + target_list = incr_targets stagger = self.config_manager.get_incremental_stagger_seconds() max_concurrent = self.config_manager.get_max_concurrent_tasks() @@ -508,11 +510,9 @@ class AutoScheduler: f"(并发限制: {max_concurrent}, 交错间隔: {stagger}秒)" ) - # 任务粒度的入口并发限制，保护本地资源（DB/内存） sem = asyncio.Semaphore(max_concurrent) async def staggered_incremental(idx, gid, pid): - # 按索引交错延迟，均匀分散 API 压力 if idx > 0 and stagger > 0: await asyncio.sleep(stagger * idx) @@ -523,13 +523,12 @@ class AutoScheduler: ) ) - # 检查是否需要立即发送报告（调试模式） + # 为调试提供的立即上报选项 if self.config_manager.get_incremental_report_immediately(): if isinstance(result, dict) and result.get("success"): logger.info( f"增量分析立即报告模式生效，正在为群 {gid} 生成报告..." ) - # 立即生成最终报告 await self._perform_incremental_final_report_for_group_with_timeout( gid, pid ) @@ -537,7 +536,7 @@ class AutoScheduler: return result analysis_tasks = [] - for idx, (gid, pid) in enumerate(target_list): + for idx, (gid, pid, _mode) in enumerate(target_list): if self._terminating: logger.info("检测到插件正在停止，取消后续增量分析任务创建") break @@ -547,18 +546,15 @@ class AutoScheduler: ) analysis_tasks.append(task) - # 并发执行所有增量分析任务 results = await asyncio.gather(*analysis_tasks, return_exceptions=True) - # 统计执行结果 success_count = 0 skip_count = 0 error_count = 0 for i, result in enumerate(results): - gid, _ = target_list[i] + gid, _, _ = target_list[i] if isinstance(result, DuplicateGroupTaskError): - # 锁冲突导致的跳过 skip_count += 1 elif isinstance(result, Exception): logger.error(f"群 {gid} 增量分析任务异常: {result}") @@ -650,86 +646,17 @@ class AutoScheduler: logger.debug(f"群 {group_id} 增量分析流程结束") # ================================================================ - # 增量模式：最终报告生成 + # 增量最终报告（单群）与回退逻辑 # ================================================================ - async def _run_incremental_final_report(self): - """基于当天增量累积数据生成并发送最终报告""" - if self._terminating: - return - try: - logger.info("开始生成增量最终报告（交错并发模式）") - - enabled_targets = await self._get_enabled_targets() - - if not enabled_targets: - logger.info("没有启用的群聊需要生成最终报告") - return - - target_list = list(enabled_targets) - stagger = self.config_manager.get_incremental_stagger_seconds() - max_concurrent = self.config_manager.get_max_concurrent_tasks() - - logger.info( - f"将为 {len(target_list)} 个群聊生成增量最终报告 " - f"(并发限制: {max_concurrent}, 交错间隔: {stagger}秒)" - ) - - sem = asyncio.Semaphore(max_concurrent) - - async def staggered_final_report(idx, gid, pid): - if idx > 0 and stagger > 0: - await asyncio.sleep(stagger * idx) - - async with sem: - return await self._perform_incremental_final_report_for_group_with_timeout( - gid, pid - ) - - report_tasks = [] - for idx, (gid, pid) in enumerate(target_list): - if self._terminating: - logger.info("检测到插件正在停止，取消后续最终报告任务创建") - break - task = asyncio.create_task( - staggered_final_report(idx, gid, pid), - name=f"final_report_group_{gid}", - ) - report_tasks.append(task) - - # 并发执行所有最终报告任务 - results = await asyncio.gather(*report_tasks, return_exceptions=True) - - # 统计执行结果 - success_count = 0 - skip_count = 0 - error_count = 0 - - for i, result in enumerate(results): - gid, _ = target_list[i] - if isinstance(result, DuplicateGroupTaskError): - # 锁冲突导致的跳过 - skip_count += 1 - elif isinstance(result, Exception): - logger.error(f"群 {gid} 最终报告任务异常: {result}") - error_count += 1 - elif isinstance(result, dict) and not result.get("success", True): - skip_count += 1 - else: - success_count += 1 - - logger.info( - f"增量最终报告完成 - 成功: {success_count}, 跳过: {skip_count}, " - f"失败: {error_count}, 总计: {len(target_list)}" - ) - - except Exception as e: - logger.error(f"增量最终报告执行失败: {e}", exc_info=True) - async def _perform_incremental_final_report_for_group_with_timeout( self, group_id: str, target_platform_id: str | None = None ): - """为指定群生成增量最终报告（带超时控制，20分钟）""" + """带超时及回退机制的增量最终报告生成。 + + 若增量汇报失败（非 '消息不足' 或 '正在运行' 导致的）， + 且启用了自动回退，则将该群转由传统模式执行全量分析。 + """ try: result = await asyncio.wait_for( self._perform_incremental_final_report_for_group( @@ -737,14 +664,57 @@ class AutoScheduler: ), timeout=1200, ) + + # 判定是否需要触发回退 (例如：无增量数据等) + if isinstance(result, dict) and not result.get("success"): + reason = result.get("reason", "") + if reason in ("below_threshold", "already_running"): + return result # 正常跳过，无需回退 + if self.config_manager.get_incremental_fallback_enabled(): + logger.warning( + f"群 {group_id} 增量最终报告失败 (reason={reason})，" + f"正在回退到传统全量分析..." + ) + return await self._fallback_to_traditional( + group_id, target_platform_id + ) + return result + except asyncio.TimeoutError: - logger.error(f"群 {group_id} 最终报告超时（20分钟），跳过") + logger.error(f"群 {group_id} 最终报告超时（20分钟）") + if self.config_manager.get_incremental_fallback_enabled(): + logger.warning(f"群 {group_id} 增量报告超时，正在回退到传统全量分析...") + return await self._fallback_to_traditional(group_id, target_platform_id) return {"success": False, "reason": "timeout"} + except Exception as e: logger.error(f"群 {group_id} 最终报告任务执行失败: {e}") + if self.config_manager.get_incremental_fallback_enabled(): + logger.warning(f"群 {group_id} 增量报告异常，正在回退到传统全量分析...") + return await self._fallback_to_traditional(group_id, target_platform_id) return {"success": False, "reason": str(e)} + async def _fallback_to_traditional( + self, group_id: str, target_platform_id: str | None = None + ): + """回退操作：在增量报告失败时，执行传统的全量拉取分析。""" + try: + logger.info( + f"⬆️ 群 {group_id} 回退到传统全量分析 " + f"(Platform: {target_platform_id or 'Auto'})" + ) + await self._perform_auto_analysis_for_group_with_timeout( + group_id, target_platform_id + ) + return {"success": True, "fallback": True} + except Exception as fallback_err: + logger.error( + f"群 {group_id} 回退传统分析也失败: {fallback_err}", + exc_info=True, + ) + return {"success": False, "reason": f"fallback_failed: {fallback_err}"} + async def _perform_incremental_final_report_for_group( self, group_id: str, target_platform_id: str | None = None ):