From b4eec978fe73e5f3a798efd19736af462d5f8462 Mon Sep 17 00:00:00 2001 From: SXP-Simon Date: Thu, 12 Feb 2026 13:24:55 +0800 Subject: [PATCH] =?UTF-8?q?refactor(Telegram):=20=E9=87=8D=E6=9E=84=20Tele?= =?UTF-8?q?gram=20=E9=80=BB=E8=BE=91=E5=B9=B6=E6=8F=90=E5=8F=96=E6=B6=88?= =?UTF-8?q?=E6=81=AF=E5=A4=84=E7=90=86=E4=B8=8E=E6=B3=A8=E5=86=8C=E8=A1=A8?= =?UTF-8?q?=E6=9C=8D=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 提取消息处理服务及 Telegram 注册表，显著精简主插件逻辑。 - Telegram 群组获取回退逻辑下沉至适配器，移除调度中心硬编码。 - 优化依赖注入机制，使适配器层级可访问插件 KV 存储。 - 修正消息内容定义兼容性并清理冗余代码。 --- main.py | 428 +++--------------- .../services/message_processing_service.py | 269 +++++++++++ .../persistence/telegram_group_registry.py | 102 +++++ .../platform/adapters/telegram_adapter.py | 52 ++- src/infrastructure/platform/bot_manager.py | 6 + .../scheduler/auto_scheduler.py | 39 +- 6 files changed, 473 insertions(+), 423 deletions(-) create mode 100644 src/application/services/message_processing_service.py create mode 100644 src/infrastructure/persistence/telegram_group_registry.py diff --git a/main.py b/main.py index 87f54f6..d831b16 100644 --- a/main.py +++ b/main.py @@ -7,9 +7,6 @@ QQ群日常分析插件 import asyncio import os -import re -from collections import Counter -from datetime import datetime, timezone from astrbot.api import AstrBotConfig, logger from astrbot.api.event import AstrMessageEvent, filter @@ -33,12 +30,17 @@ from .src.infrastructure.scheduler.auto_scheduler import AutoScheduler from .src.infrastructure.scheduler.retry import RetryManager from .src.utils.pdf_utils import PDFInstaller +from .src.infrastructure.persistence.telegram_group_registry import ( + TelegramGroupRegistry, +) +from .src.application.services.message_processing_service import ( + MessageProcessingService, +) + class QQGroupDailyAnalysis(Star): """QQ群日常分析插件主类""" - _TG_GROUP_REGISTRY_KV_KEY = "telegram_seen_groups_v1" - def __init__(self, context: Context, config: AstrBotConfig): super().__init__(context) self.config = config @@ -47,9 +49,13 @@ class QQGroupDailyAnalysis(Star): self.config_manager = ConfigManager(config) self.bot_manager = BotManager(self.config_manager) self.bot_manager.set_context(context) + self.bot_manager.set_plugin_instance(self) self.history_manager = HistoryManager(self) self.report_generator = ReportGenerator(self.config_manager) + # Telegram 注册表 (持久层) + self.telegram_group_registry = TelegramGroupRegistry(self) + # 2. 领域层 self.statistics_service = StatisticsService() self.analysis_domain_service = AnalysisDomainService() @@ -74,6 +80,11 @@ class QQGroupDailyAnalysis(Star): incremental_merge_service=self.incremental_merge_service, ) + # 消息处理服务 + self.message_processing_service = MessageProcessingService( + context, self.telegram_group_registry + ) + # 调度与重试 self.retry_manager = RetryManager( self.bot_manager, self.html_render, self.report_generator @@ -89,7 +100,6 @@ class QQGroupDailyAnalysis(Star): ) self._initialized = False - self._tg_registry_lock: asyncio.Lock | None = None # 异步注册任务，处理插件重载情况 asyncio.create_task(self._run_initialization("Plugin Reload/Init")) @@ -110,7 +120,7 @@ class QQGroupDailyAnalysis(Star): def _resolve_template_preview_path(self, template_name: str) -> str | None: """解析模板预览图路径（兼容新旧命名和目录）""" plugin_root = os.path.dirname(__file__) - template_base_dir = self._resolve_template_base_dir() + candidate_paths = [ os.path.join(plugin_root, "assets", f"{template_name}-demo.jpg"), ] @@ -188,380 +198,14 @@ class QQGroupDailyAnalysis(Star): self.bot_manager = None self.report_generator = None self.config_manager = None + self.message_processing_service = None + self.telegram_group_registry = None logger.info("QQ群日常分析插件资源清理完成") except Exception as e: logger.error(f"插件资源清理失败: {e}") - # ==================== 消息历史存储（统一方法，可复用） ==================== - - async def _store_message_to_history(self, event: AstrMessageEvent) -> None: - """ - 将消息存储到 AstrBot 的 message_history_manager - - 这是一个可复用的统一方法，支持所有通过 context 机制存储消息的平台。 - 不使用 fallback 值 - 如果获取不到必要数据会抛出异常。 - - Args: - event: AstrBot 消息事件 - - Raises: - ValueError: 当必要数据（group_id, sender_id, platform_id）无法获取时 - RuntimeError: 当消息内容为空时 - """ - # 1. 获取群组 ID（必需） - group_id = self._get_group_id_from_event(event) - if not group_id: - raise ValueError("无法获取群组 ID，拒绝存储消息") - - # 2. 获取发送者 ID（必需） - sender_id = event.get_sender_id() - if not sender_id: - raise ValueError(f"群 {group_id}: 无法获取发送者 ID，拒绝存储消息") - sender_id = str(sender_id) - - # 3. 获取发送者名称（昵称优先，必要时回退） - sender_name = self._resolve_sender_name(event, sender_id) - - # 4. 获取平台 ID（必需） - platform_id = event.get_platform_id() - if not platform_id: - raise ValueError(f"群 {group_id}: 无法获取平台 ID，拒绝存储消息") - - # 5. 提取消息内容 - message_parts = self._extract_message_parts(event) - if not message_parts: - raise RuntimeError( - f"群 {group_id}: 消息内容为空 (sender={sender_name})，拒绝存储" - ) - - # 6. 提取事件消息 ID（用于 Telegram 已见群/话题记录） - msg_obj = getattr(event, "message_obj", None) - event_message_id = str(getattr(msg_obj, "message_id", "") or "") - - # 7. 存储到数据库 - await self.context.message_history_manager.insert( - platform_id=platform_id, - user_id=group_id, - content={"type": "user", "message": message_parts}, - sender_id=sender_id, - sender_name=sender_name, - ) - - # Telegram: 记录已见群/话题，用于自动分析拉群回退 - if self._is_telegram_event(event, platform_id): - try: - await self._upsert_telegram_group_registry( - platform_id=platform_id, - group_id=group_id, - sender_id=sender_id, - sender_name=sender_name, - event_message_id=event_message_id, - ) - except Exception as e: - logger.warning( - "[TGRegistry] Upsert failed: " - f"platform_id={platform_id} group_id={group_id} error={e}" - ) - - logger.debug( - f"[{platform_id}] 已缓存群 {group_id} 的消息 (发送者: {sender_name})" - ) - - @staticmethod - def _is_telegram_event(event: AstrMessageEvent, platform_id: str) -> bool: - """判断当前事件是否为 Telegram 平台。""" - platform_name = str(event.get_platform_name() or "").strip().lower() - if platform_name == "telegram": - return True - return str(platform_id or "").strip().lower().startswith("telegram") - - def _get_tg_registry_lock(self) -> asyncio.Lock: - """懒加载 Telegram 群/话题注册表锁，避免并发读改写覆盖。""" - if self._tg_registry_lock is None: - self._tg_registry_lock = asyncio.Lock() - return self._tg_registry_lock - - async def _upsert_telegram_group_registry( - self, - platform_id: str, - group_id: str, - sender_id: str, - sender_name: str, - event_message_id: str, - ) -> None: - """更新 Telegram 已见群/话题注册表（KV）。""" - lock = self._get_tg_registry_lock() - async with lock: - registry = await self.get_kv_data(self._TG_GROUP_REGISTRY_KV_KEY, {}) - if not isinstance(registry, dict): - registry = {} - - platforms = registry.get("platforms") - if not isinstance(platforms, dict): - platforms = {} - registry["platforms"] = platforms - - platform_key = str(platform_id).strip() - group_key = str(group_id).strip() - - platform_map = platforms.get(platform_key) - if not isinstance(platform_map, dict): - platform_map = {} - platforms[platform_key] = platform_map - - now_iso = datetime.now(timezone.utc).isoformat() - - entry = platform_map.get(group_key) - if not isinstance(entry, dict): - entry = {} - - first_seen = entry.get("first_seen") - if not isinstance(first_seen, str) or not first_seen: - first_seen = now_iso - - entry.update( - { - "first_seen": first_seen, - "last_seen": now_iso, - "last_sender_id": str(sender_id), - "last_sender_name": str(sender_name), - "last_event_message_id": str(event_message_id), - } - ) - platform_map[group_key] = entry - - registry["updated_at"] = now_iso - await self.put_kv_data(self._TG_GROUP_REGISTRY_KV_KEY, registry) - - async def get_telegram_seen_group_ids( - self, platform_id: str | None = None - ) -> list[str]: - """读取 Telegram 已见群/话题列表（给调度器回退使用）。""" - lock = self._get_tg_registry_lock() - async with lock: - registry = await self.get_kv_data(self._TG_GROUP_REGISTRY_KV_KEY, {}) - if not isinstance(registry, dict): - return [] - - platforms = registry.get("platforms") - if not isinstance(platforms, dict): - return [] - - groups: set[str] = set() - if platform_id: - platform_map = platforms.get(str(platform_id).strip(), {}) - if isinstance(platform_map, dict): - groups.update( - str(gid).strip() - for gid in platform_map.keys() - if str(gid).strip() - ) - else: - for platform_map in platforms.values(): - if not isinstance(platform_map, dict): - continue - groups.update( - str(gid).strip() - for gid in platform_map.keys() - if str(gid).strip() - ) - - return sorted(groups) - - @staticmethod - def _is_placeholder_sender_name(name: str | None, sender_id: str) -> bool: - """判断 sender_name 是否为空或占位值。""" - if not name: - return True - normalized = str(name).strip() - if not normalized: - return True - if normalized.lower() in {"unknown", "none", "null", "nil", "undefined"}: - return True - return normalized == str(sender_id).strip() - - def _resolve_sender_name(self, event: AstrMessageEvent, sender_id: str) -> str: - """ - 解析发送者展示名。 - - 优先级： - - Telegram: - 1. raw_message.from_user.full_name - 2. raw_message.from_user.first_name - 3. event.get_sender_name() / message_obj.sender.nickname - 4. raw_message.from_user.username - 5. sender_id - - 其他平台： - 1. event.get_sender_name() - 2. message_obj.sender.nickname - 3. raw_message.from_user.full_name / first_name / username - 4. sender_id（最终回退，避免消息丢失） - """ - platform_name = str(event.get_platform_name() or "").lower() - candidates: list[str | None] = [] - - msg_obj = getattr(event, "message_obj", None) - sender_obj = getattr(msg_obj, "sender", None) - raw_message = getattr(msg_obj, "raw_message", None) - raw_msg_obj = getattr(raw_message, "message", raw_message) - from_user = getattr(raw_msg_obj, "from_user", None) - - # Telegram 特殊策略：优先显示名，不优先 username - if platform_name == "telegram": - if from_user is not None: - candidates.extend( - [ - getattr(from_user, "full_name", None), - getattr(from_user, "first_name", None), - ] - ) - - candidates.append(event.get_sender_name()) - if sender_obj is not None: - candidates.append(getattr(sender_obj, "nickname", None)) - - if from_user is not None: - candidates.append(getattr(from_user, "username", None)) - else: - candidates.append(event.get_sender_name()) - if sender_obj is not None: - candidates.append(getattr(sender_obj, "nickname", None)) - - if from_user is not None: - candidates.extend( - [ - getattr(from_user, "full_name", None), - getattr(from_user, "first_name", None), - getattr(from_user, "username", None), - ] - ) - - for candidate in candidates: - name = str(candidate or "").strip() - if not self._is_placeholder_sender_name(name, sender_id): - return name - - logger.warning( - f"[HistoryStore] 无法解析昵称，回退为 sender_id: {sender_id} " - f"(platform={event.get_platform_id()})" - ) - return sender_id - - def _extract_message_parts(self, event: AstrMessageEvent) -> list[dict]: - """ - 从事件中提取消息内容 - - Returns: - 消息部分列表，格式为 [{"type": "plain", "text": "..."}, ...] - """ - message_parts = [] - message = event.message_obj - - # 先收集 @ 标记，后续用于从 plain 文本中去重 - pending_mentions: Counter[str] = Counter() - if message and hasattr(message, "message"): - for seg in message.message: - if not hasattr(seg, "type"): - continue - if seg.type not in ("At", "at"): - continue - - target = getattr(seg, "target", None) - if target is None: - target = getattr(seg, "qq", None) - if target is None and hasattr(seg, "data"): - target = seg.data.get("qq") or seg.data.get("target") - - target_str = str(target or "").strip() - if target_str: - pending_mentions[target_str] += 1 - - display_name = str(getattr(seg, "name", "") or "").strip() - if display_name and display_name != target_str: - pending_mentions[display_name] += 1 - - if message and hasattr(message, "message"): - for seg in message.message: - if not hasattr(seg, "type"): - continue - - seg_type = seg.type - if seg_type in ("Plain", "text"): - text = getattr(seg, "text", None) - if text is None and hasattr(seg, "data"): - text = seg.data.get("text") - if text: - text = self._strip_known_mentions(text, pending_mentions) - message_parts.append({"type": "plain", "text": text}) - - elif seg_type in ("Image", "image"): - url = getattr(seg, "url", None) - if url is None and hasattr(seg, "data"): - url = seg.data.get("url") - if url: - message_parts.append({"type": "image", "url": url}) - - elif seg_type in ("At", "at"): - target = getattr(seg, "target", None) - if target is None: - target = getattr(seg, "qq", None) - if target is None and hasattr(seg, "data"): - target = seg.data.get("qq") or seg.data.get("target") - if target: - message_parts.append( - { - "type": "at", - "target_id": str(target), - "name": str(getattr(seg, "name", "") or ""), - } - ) - - # 如果没有从消息链提取到内容，尝试使用 message_str - if not message_parts and event.message_str: - message_parts.append({"type": "plain", "text": event.message_str}) - - # 清理空文本段，避免出现仅空格文本 - message_parts = [ - part - for part in message_parts - if not ( - part.get("type") == "plain" and not str(part.get("text", "")).strip() - ) - ] - - return message_parts - - @staticmethod - def _strip_known_mentions(text: str, pending_mentions: Counter[str]) -> str: - """ - 从文本中移除已识别的 @ 提及，避免与结构化 at 段重复。 - """ - cleaned = str(text) - if not cleaned or not pending_mentions: - return cleaned.strip() - - for mention, remaining in list(pending_mentions.items()): - if not mention or remaining <= 0: - continue - - pattern = re.compile(rf"(? 0: - pending_mentions[mention] -= removed - if pending_mentions[mention] <= 0: - pending_mentions.pop(mention, None) - - cleaned = re.sub(r"\s{2,}", " ", cleaned).strip() - return cleaned - # ==================== Telegram 消息拦截器 ==================== @filter.event_message_type(filter.EventMessageType.GROUP_MESSAGE) @@ -570,15 +214,45 @@ class QQGroupDailyAnalysis(Star): """ 拦截 Telegram 群消息并存储到数据库 - 使用统一的 _store_message_to_history 方法存储消息。 + 委托给 MessageProcessingService 处理 """ try: - await self._store_message_to_history(event) + await self.message_processing_service.process_message(event) except (ValueError, RuntimeError) as e: logger.warning(f"[Telegram] 消息存储失败: {e}") except Exception as e: logger.error(f"[Telegram] 消息存储异常: {e}", exc_info=True) + async def get_telegram_seen_group_ids( + self, platform_id: str | None = None + ) -> list[str]: + """读取 Telegram 已见群/话题列表（给调度器回退使用）。""" + return await self.telegram_group_registry.get_all_group_ids(platform_id) + + def _get_group_id_from_event(self, event: AstrMessageEvent) -> str | None: + """从消息事件中安全获取群组 ID""" + # 保留此辅助方法，因为在其他 command 中仍被频繁使用 + try: + group_id = event.get_group_id() + return group_id if group_id else None + except Exception: + return None + + def _get_platform_id_from_event(self, event: AstrMessageEvent) -> str: + """从消息事件中获取平台唯一 ID""" + # 保留此辅助方法，因为在其他 command 中仍被频繁使用 + try: + return event.get_platform_id() + except Exception: + # 后备方案：从元数据获取 + if ( + hasattr(event, "platform_meta") + and event.platform_meta + and hasattr(event.platform_meta, "id") + ): + return event.platform_meta.id + return "default" + @filter.command("群分析", alias={"group_analysis"}) @filter.permission_type(PermissionType.ADMIN) async def analyze_group_daily( diff --git a/src/application/services/message_processing_service.py b/src/application/services/message_processing_service.py new file mode 100644 index 0000000..8086230 --- /dev/null +++ b/src/application/services/message_processing_service.py @@ -0,0 +1,269 @@ +import re +from collections import Counter + +from astrbot.api.event import AstrMessageEvent +from astrbot.api.star import Context + +from ...utils.logger import logger +from ...infrastructure.persistence.telegram_group_registry import TelegramGroupRegistry + + +class MessageProcessingService: + """ + 消息处理服务 + + 负责处理接收到的消息事件： + 1. 解析消息内容（文本、图片、@提及等） + 2. 解析发送者信息（跨平台兼容） + 3. 存储消息历史 + 4. 维护 Telegram 群组注册表（回退机制） + """ + + def __init__(self, context: Context, telegram_registry: TelegramGroupRegistry): + self.context = context + self.telegram_registry = telegram_registry + + async def process_message(self, event: AstrMessageEvent) -> None: + """ + 处理并在历史记录中存储消息。 + + Args: + event: AstrBot 消息事件 + + Raises: + ValueError: 当必要数据无法获取时 + RuntimeError: 当消息内容为空时 + """ + # 1. 获取群组 ID（必需） + group_id = self._get_group_id_from_event(event) + if not group_id: + raise ValueError("无法获取群组 ID，拒绝存储消息") + + # 2. 获取发送者 ID（必需） + sender_id = event.get_sender_id() + if not sender_id: + raise ValueError(f"群 {group_id}: 无法获取发送者 ID，拒绝存储消息") + sender_id = str(sender_id) + + # 3. 获取发送者名称（昵称优先，必要时回退） + sender_name = self._resolve_sender_name(event, sender_id) + + # 4. 获取平台 ID（必需） + platform_id = event.get_platform_id() + if not platform_id: + raise ValueError(f"群 {group_id}: 无法获取平台 ID，拒绝存储消息") + + # 5. 提取消息内容 + message_parts = self._extract_message_parts(event) + if not message_parts: + # 尝试记录一条警告但不中断流程（或者视为错误） + # 原逻辑是抛出 RuntimeError + raise RuntimeError( + f"群 {group_id}: 消息内容为空 (sender={sender_name})，拒绝存储" + ) + + # 6. 提取事件消息 ID（用于 Telegram 已见群/话题记录） + msg_obj = getattr(event, "message_obj", None) + event_message_id = str(getattr(msg_obj, "message_id", "") or "") + + # 7. 存储到数据库 + await self.context.message_history_manager.insert( + platform_id=platform_id, + user_id=group_id, + content={"type": "user", "message": message_parts}, + sender_id=sender_id, + sender_name=sender_name, + ) + + # Telegram: 记录已见群/话题 + if self._is_telegram_event(event, platform_id): + try: + await self.telegram_registry.upsert( + platform_id=platform_id, + group_id=group_id, + sender_id=sender_id, + sender_name=sender_name, + event_message_id=event_message_id, + ) + except Exception as e: + logger.warning( + "[TGRegistry] Upsert failed: " + f"platform_id={platform_id} group_id={group_id} error={e}" + ) + + logger.debug( + f"[{platform_id}] 已缓存群 {group_id} 的消息 (发送者: {sender_name})" + ) + + def _get_group_id_from_event(self, event: AstrMessageEvent) -> str | None: + """从消息事件中安全获取群组 ID""" + try: + group_id = event.get_group_id() + return group_id if group_id else None + except Exception: + return None + + def _resolve_sender_name(self, event: AstrMessageEvent, sender_id: str) -> str: + """解析发送者展示名""" + platform_name = str(event.get_platform_name() or "").lower() + candidates: list[str | None] = [] + + msg_obj = getattr(event, "message_obj", None) + sender_obj = getattr(msg_obj, "sender", None) + raw_message = getattr(msg_obj, "raw_message", None) + raw_msg_obj = getattr(raw_message, "message", raw_message) + from_user = getattr(raw_msg_obj, "from_user", None) + + if platform_name == "telegram": + if from_user is not None: + candidates.extend( + [ + getattr(from_user, "full_name", None), + getattr(from_user, "first_name", None), + ] + ) + candidates.append(event.get_sender_name()) + if sender_obj is not None: + candidates.append(getattr(sender_obj, "nickname", None)) + if from_user is not None: + candidates.append(getattr(from_user, "username", None)) + else: + candidates.append(event.get_sender_name()) + if sender_obj is not None: + candidates.append(getattr(sender_obj, "nickname", None)) + + if from_user is not None: + candidates.extend( + [ + getattr(from_user, "full_name", None), + getattr(from_user, "first_name", None), + getattr(from_user, "username", None), + ] + ) + + for candidate in candidates: + name = str(candidate or "").strip() + if not self._is_placeholder_sender_name(name, sender_id): + return name + + return sender_id + + def _extract_message_parts(self, event: AstrMessageEvent) -> list[dict]: + """从事件中提取消息内容""" + message_parts = [] + message = event.message_obj + + # 收集 @ 标记 + pending_mentions: Counter[str] = Counter() + if message and hasattr(message, "message"): + for seg in message.message: + if not hasattr(seg, "type"): + continue + if seg.type not in ("At", "at"): + continue + + target = getattr(seg, "target", None) or getattr(seg, "qq", None) + if target is None and hasattr(seg, "data"): + target = seg.data.get("qq") or seg.data.get("target") + + target_str = str(target or "").strip() + if target_str: + pending_mentions[target_str] += 1 + + display_name = str(getattr(seg, "name", "") or "").strip() + if display_name and display_name != target_str: + pending_mentions[display_name] += 1 + + if message and hasattr(message, "message"): + for seg in message.message: + if not hasattr(seg, "type"): + continue + + seg_type = seg.type + if seg_type in ("Plain", "text"): + text = getattr(seg, "text", None) + if text is None and hasattr(seg, "data"): + text = seg.data.get("text") + if text: + text = self._strip_known_mentions(text, pending_mentions) + message_parts.append({"type": "plain", "text": text}) + + elif seg_type in ("Image", "image"): + url = getattr(seg, "url", None) or ( + seg.data.get("url") if hasattr(seg, "data") else None + ) + if url: + message_parts.append({"type": "image", "url": url}) + + elif seg_type in ("At", "at"): + target = getattr(seg, "target", None) or getattr(seg, "qq", None) + if target is None and hasattr(seg, "data"): + target = seg.data.get("qq") or seg.data.get("target") + if target: + message_parts.append( + { + "type": "at", + "target_id": str(target), + "name": str(getattr(seg, "name", "") or ""), + } + ) + + if not message_parts and event.message_str: + message_parts.append({"type": "plain", "text": event.message_str}) + + # 清理空文本段 + message_parts = [ + part + for part in message_parts + if not ( + part.get("type") == "plain" and not str(part.get("text", "")).strip() + ) + ] + + return message_parts + + @staticmethod + def _strip_known_mentions(text: str, pending_mentions: Counter[str]) -> str: + """从文本中移除已识别的 @ 提及""" + cleaned = str(text) + if not cleaned or not pending_mentions: + return cleaned.strip() + + for mention, remaining in list(pending_mentions.items()): + if not mention or remaining <= 0: + continue + + pattern = re.compile(rf"(? 0: + pending_mentions[mention] -= removed + if pending_mentions[mention] <= 0: + pending_mentions.pop(mention, None) + + return re.sub(r"\s{2,}", " ", cleaned).strip() + + @staticmethod + def _is_placeholder_sender_name(name: str | None, sender_id: str) -> bool: + """判断 sender_name 是否为占位值""" + if not name: + return True + normalized = str(name).strip() + if not normalized: + return True + if normalized.lower() in {"unknown", "none", "null", "nil", "undefined"}: + return True + return normalized == str(sender_id).strip() + + @staticmethod + def _is_telegram_event(event: AstrMessageEvent, platform_id: str) -> bool: + """判断当前事件是否为 Telegram 平台""" + platform_name = str(event.get_platform_name() or "").strip().lower() + if platform_name == "telegram": + return True + return str(platform_id or "").strip().lower().startswith("telegram") diff --git a/src/infrastructure/persistence/telegram_group_registry.py b/src/infrastructure/persistence/telegram_group_registry.py new file mode 100644 index 0000000..c64ba96 --- /dev/null +++ b/src/infrastructure/persistence/telegram_group_registry.py @@ -0,0 +1,102 @@ +import asyncio +from datetime import datetime, timezone + +from astrbot.api.star import Star + + +class TelegramGroupRegistry: + """ + Telegram 群组/话题注册表 + + 负责管理 Telegram 的已见群组和话题列表，用于在无法通过 API 获取群列表时提供回退支持。 + 数据存储在 AstrBot 的 KV 存储中。 + """ + + _KV_KEY = "telegram_seen_groups_v1" + + def __init__(self, plugin_instance: Star): + self.plugin = plugin_instance + self._lock = asyncio.Lock() + + async def upsert( + self, + platform_id: str, + group_id: str, + sender_id: str, + sender_name: str, + event_message_id: str, + ) -> None: + """更新 Telegram 已见群/话题注册表（KV）。""" + async with self._lock: + registry = await self.plugin.get_kv_data(self._KV_KEY, {}) + if not isinstance(registry, dict): + registry = {} + + platforms = registry.get("platforms") + if not isinstance(platforms, dict): + platforms = {} + registry["platforms"] = platforms + + platform_key = str(platform_id).strip() + group_key = str(group_id).strip() + + platform_map = platforms.get(platform_key) + if not isinstance(platform_map, dict): + platform_map = {} + platforms[platform_key] = platform_map + + now_iso = datetime.now(timezone.utc).isoformat() + + entry = platform_map.get(group_key) + if not isinstance(entry, dict): + entry = {} + + first_seen = entry.get("first_seen") + if not isinstance(first_seen, str) or not first_seen: + first_seen = now_iso + + entry.update( + { + "first_seen": first_seen, + "last_seen": now_iso, + "last_sender_id": str(sender_id), + "last_sender_name": str(sender_name), + "last_event_message_id": str(event_message_id), + } + ) + platform_map[group_key] = entry + + registry["updated_at"] = now_iso + await self.plugin.put_kv_data(self._KV_KEY, registry) + + async def get_all_group_ids(self, platform_id: str | None = None) -> list[str]: + """读取 Telegram 已见群/话题列表。""" + async with self._lock: + registry = await self.plugin.get_kv_data(self._KV_KEY, {}) + if not isinstance(registry, dict): + return [] + + platforms = registry.get("platforms") + if not isinstance(platforms, dict): + return [] + + groups: set[str] = set() + if platform_id: + platform_map = platforms.get(str(platform_id).strip(), {}) + if isinstance(platform_map, dict): + groups.update( + str(gid).strip() + for gid in platform_map.keys() + if str(gid).strip() + ) + else: + for platform_map in platforms.values(): + if not isinstance(platform_map, dict): + continue + groups.update( + str(gid).strip() + for gid in platform_map.keys() + if str(gid).strip() + ) + + return sorted(groups) diff --git a/src/infrastructure/platform/adapters/telegram_adapter.py b/src/infrastructure/platform/adapters/telegram_adapter.py index f0dfc0b..f2e59d1 100644 --- a/src/infrastructure/platform/adapters/telegram_adapter.py +++ b/src/infrastructure/platform/adapters/telegram_adapter.py @@ -65,6 +65,9 @@ class TelegramAdapter(PlatformAdapter): if config: ids = config.get("bot_self_ids", []) self.bot_self_ids = [str(i) for i in ids] if ids else [] + self._plugin_instance = config.get("plugin_instance") + else: + self._plugin_instance = None self._platform_id = str(config.get("platform_id", "")).strip() if config else "" def set_context(self, context: "Context") -> None: @@ -75,6 +78,46 @@ class TelegramAdapter(PlatformAdapter): """ self._context = context + def _init_capabilities(self) -> PlatformCapabilities: + """返回 Telegram 平台能力声明""" + return TELEGRAM_CAPABILITIES + + async def get_group_list(self) -> list[str]: + """ + 获取群组列表 + + Telegram Bot API 不支持直接获取群列表。 + 因此这里尝试结合多种策略： + 1. 尝试调用 API (如果未来支持) + 2. 回退：从插件的 KV 存储中获取已知群组 (需注入插件实例) + """ + groups = [] + client = self._telegram_client + + # 1. 尝试 API (目前 python-telegram-bot 不支持直接列出所有 chat) + # 如果 client 有扩展方法或未来支持，可在此实现 + + # 2. 回退：使用 KV 注册表 + if not groups and self._plugin_instance: + try: + # 检查插件实例是否有 get_telegram_seen_group_ids 方法 + if hasattr(self._plugin_instance, "get_telegram_seen_group_ids"): + kv_groups = await self._plugin_instance.get_telegram_seen_group_ids( + self._platform_id + ) + if kv_groups: + groups.extend(kv_groups) + logger.debug( + f"[Telegram] 通过 KV 回退获取到 {len(kv_groups)} 个群组" + ) + except Exception as e: + logger.warning(f"[Telegram] KV 回退获取群列表失败: {e}") + + if not groups: + logger.debug("[Telegram] 无法获取群列表 (API不支持且无KV记录)") + + return list(set(groups)) + @property def _telegram_client(self) -> "ExtBot | None": """ @@ -621,15 +664,6 @@ class TelegramAdapter(PlatformAdapter): logger.debug(f"[Telegram] 获取群信息失败: {e}") return None - async def get_group_list(self) -> list[str]: - """ - 获取群组列表 - - Telegram Bot API 不支持获取群列表。 - """ - logger.debug("[Telegram] Bot API 不支持获取群列表") - return [] - async def get_member_list(self, group_id: str) -> list[UnifiedMember]: """ 获取成员列表 diff --git a/src/infrastructure/platform/bot_manager.py b/src/infrastructure/platform/bot_manager.py index 28f398c..e7ed3d6 100644 --- a/src/infrastructure/platform/bot_manager.py +++ b/src/infrastructure/platform/bot_manager.py @@ -26,6 +26,7 @@ class BotManager: self._context = None self._is_initialized = False self._default_platform = "default" # 默认平台 + self._plugin_instance = None # 插件实例引用，用于适配器回调 def set_context(self, context): """设置AstrBot上下文，并传递给所有支持的适配器""" @@ -36,6 +37,10 @@ class BotManager: if hasattr(adapter, "set_context"): adapter.set_context(context) + def set_plugin_instance(self, plugin_instance: Any): + """设置插件实例引用""" + self._plugin_instance = plugin_instance + def set_bot_instance(self, bot_instance, platform_id=None, platform_name=None): """ 设置bot实例，支持指定平台ID @@ -56,6 +61,7 @@ class BotManager: adapter_config = { "bot_self_ids": self._bot_self_ids.copy(), "platform_id": str(platform_id), + "plugin_instance": self._plugin_instance, } adapter = PlatformAdapterFactory.create( platform_name, bot_instance, adapter_config diff --git a/src/infrastructure/scheduler/auto_scheduler.py b/src/infrastructure/scheduler/auto_scheduler.py index b6acc6f..c54a2dc 100644 --- a/src/infrastructure/scheduler/auto_scheduler.py +++ b/src/infrastructure/scheduler/auto_scheduler.py @@ -840,33 +840,19 @@ class AutoScheduler: if str(group_id).strip() ] - # 获取平台名称（用于 Telegram 回退判定） + # 获取平台名称（仅用于日志） p_name = None if hasattr(adapter, "get_platform_name"): try: p_name = adapter.get_platform_name() except Exception: p_name = None - if not p_name: - p_name = ( - self.bot_manager._detect_platform_name(bot_instance) - or "unknown" - ) - p_name = str(p_name).strip() - - used_tg_kv_fallback = False - if not groups and p_name.lower() == "telegram": - groups = await self._get_telegram_groups_from_plugin_kv( - str(platform_id) - ) - used_tg_kv_fallback = bool(groups) for group_id in groups: all_groups.add((platform_id, str(group_id))) logger.info( - f"平台 {platform_id} ({p_name}) 成功获取 {len(groups)} 个群组" - + (" (KV回退)" if used_tg_kv_fallback else "") + f"平台 {platform_id} ({p_name or 'unknown'}) 成功获取 {len(groups)} 个群组" ) continue @@ -880,24 +866,3 @@ class AutoScheduler: logger.error(f"平台 {platform_id} 获取群列表异常: {e}") return list(all_groups) - - async def _get_telegram_groups_from_plugin_kv(self, platform_id: str) -> list[str]: - """从插件 KV 获取 Telegram 已见群/话题，作为 get_group_list 的回退。""" - if not self.plugin_instance: - return [] - - getter = getattr(self.plugin_instance, "get_telegram_seen_group_ids", None) - if not callable(getter): - return [] - - try: - groups = await getter(platform_id=platform_id) - return sorted( - {str(group_id).strip() for group_id in groups if str(group_id).strip()} - ) - except Exception as e: - logger.warning( - "[TGRegistry] Scheduler fetch failed: " - f"platform_id={platform_id} error={e}" - ) - return []