""" chatbot/manager.py - 聊天平台管理器 ===================================== 统一管理所有聊天平台的生命周期和消息路由。 纯 Python 实现，不依赖 Node.js 或额外网关。 """ from __future__ import annotations import asyncio import time from typing import Any, Callable, Dict, List, Optional from core.logger import get_logger from chatbot.base import BaseChatBot, ChatMessage, ChatResponse from config import ChatPlatformConfig logger = get_logger("myagent.chatbot") class ChatBotManager: """ 聊天平台管理器。 功能: - 统一管理多个聊天平台 - 消息路由到主 Agent - 多用户/多会话隔离 - 平台独立的生命周期管理 - 后台异步运行 使用示例: manager = ChatBotManager() manager.setup_platforms(config.chat_platforms, message_handler=handle_message) await manager.start_all() """ # bot 异常后冷却时间（秒），防止 Telegram polling 冲突 _RESTART_COOLDOWN = 5 def __init__(self): self._bots: Dict[str, BaseChatBot] = {} self._bot_tasks: Dict[str, asyncio.Task] = {} # key -> asyncio.Task self._session_map: Dict[str, str] = {} # session_id -> last_message self._message_handler: Optional[Callable] = None self._last_crash: Dict[str, float] = {} # key -> 上次异常时间戳 def get_bot(self, key: str): """根据 key (id 或 platform 名) 获取 bot 实例""" return self._bots.get(key) def setup_platforms( self, platform_configs: List[ChatPlatformConfig], message_handler: Callable, ): """ 初始化所有聊天平台。 [v1.20.7] 修复: 先停止并移除已不存在的/被禁用的 bot， 再创建新启用的 bot。之前只做添加不做移除，导致禁用平台后 bot 仍在后台运行。 Args: platform_configs: 平台配置列表 message_handler: 统一消息处理回调 """ self._message_handler = message_handler # 计算当前应该启用的平台 key 集合 new_keys = set() for cfg in platform_configs: if cfg.enabled: key = cfg.id or cfg.platform new_keys.add(key) # 找出需要移除的（旧的 key 不在新的 key 集合中） removed_keys = [k for k in self._bots if k not in new_keys] for key in removed_keys: bot = self._bots.pop(key, None) task = self._bot_tasks.pop(key, None) if bot: logger.info(f"聊天平台已移除(禁用/删除): {key}") # 尝试停止 bot（同步包装异步） try: loop = asyncio.get_event_loop() if loop.is_running(): asyncio.ensure_future(self._safe_stop(key, bot)) else: loop.run_until_complete(bot.stop()) except Exception: pass if task and not task.done(): task.cancel() logger.info(f"聊天平台 {key} 后台任务已取消") # 创建或更新启用的平台 for cfg in platform_configs: if not cfg.enabled: continue key = cfg.id or cfg.platform # 如果已经存在且配置没变，跳过重建 if key in self._bots: continue # 冷却期检查：刚崩溃的 bot 不要立即重建启动，防止 Telegram polling 冲突 if key in self._last_crash and time.time() - self._last_crash[key] < self._RESTART_COOLDOWN: continue try: bot = self._create_bot(cfg, message_handler) if bot: self._bots[key] = bot logger.info(f"聊天平台已配置: {cfg.display_name or key}") # 如果已经在运行中（start_all 已调用），自动启动新 bot loop = asyncio.get_event_loop() if loop.is_running() and not any( t for t in asyncio.all_tasks(loop) if t.get_name() == f"bot_{key}" ): task = asyncio.ensure_future(self._run_bot(key, bot)) self._bot_tasks[key] = task except Exception as e: logger.error(f"平台 {cfg.display_name or cfg.platform} 初始化失败: {e}") def _create_bot( self, config: ChatPlatformConfig, message_handler: Callable, ) -> Optional[BaseChatBot]: """根据配置创建对应的聊天机器人实例""" platform = config.platform if platform == "telegram": from chatbot.telegram_bot import TelegramBot return TelegramBot( token=config.token, allowed_users=config.allowed_users, message_handler=message_handler, **config.extra, ) elif platform == "discord": from chatbot.discord_bot import DiscordBot return DiscordBot( token=config.token, allowed_users=config.allowed_users, message_handler=message_handler, **config.extra, ) elif platform == "feishu": from chatbot.feishu_bot import FeishuBot return FeishuBot( token=config.token, app_id=config.app_id, app_secret=config.app_secret, allowed_users=config.allowed_users, message_handler=message_handler, **config.extra, ) elif platform == "qq": from chatbot.qq_bot import QQBot return QQBot( token=config.token, allowed_users=config.allowed_users, message_handler=message_handler, **config.extra, ) elif platform == "wechat": from chatbot.wechat_bot import WeChatBot return WeChatBot( app_id=config.app_id, app_secret=config.app_secret, token=config.token, allowed_users=config.allowed_users, message_handler=message_handler, **config.extra, ) elif platform == "whatsapp": from chatbot.whatsapp_bot import WhatsAppBot return WhatsAppBot( token=config.token, app_id=config.app_id, app_secret=config.app_secret, allowed_users=config.allowed_users, message_handler=message_handler, **config.extra, ) else: logger.warning(f"不支持的平台: {platform}") return None async def start_all(self): """启动所有聊天平台""" tasks = [] for name, bot in self._bots.items(): logger.info(f"启动聊天平台: {name}") task = asyncio.create_task(self._run_bot(name, bot), name=f"bot_{name}") self._bot_tasks[name] = task tasks.append(task) await asyncio.gather(*tasks, return_exceptions=True) async def _run_bot(self, name: str, bot: BaseChatBot): """安全运行单个聊天平台""" try: await bot.start() except asyncio.CancelledError: logger.info(f"聊天平台 {name} 已取消") # Cancelled 意味着外部已调用 stop()，不要重复调用 except Exception as e: logger.error(f"聊天平台 {name} 运行异常: {e}") # 异常时必须清理残留的 polling 连接，否则新 bot 会冲突 try: await bot.stop() except Exception: pass # 记录崩溃时间，防止冷却期内立即重启导致 Telegram Conflict self._last_crash[name] = time.time() # 移除 bot 实例和 task，下次热更新（冷却后）会重新创建全新的 bot self._bots.pop(name, None) self._bot_tasks.pop(name, None) async def _safe_stop(self, name: str, bot: BaseChatBot): """安全停止单个 bot（不抛异常）""" try: await bot.stop() logger.info(f"聊天平台 {name} 已停止") except Exception as e: logger.warning(f"停止聊天平台 {name} 异常: {e}") async def stop_platform(self, key: str) -> bool: """[v1.20.7] 停止并移除单个聊天平台""" bot = self._bots.pop(key, None) task = self._bot_tasks.pop(key, None) if not bot: return False await self._safe_stop(key, bot) if task and not task.done(): task.cancel() return True async def stop_all(self): """停止所有聊天平台""" for name, bot in list(self._bots.items()): try: await bot.stop() except Exception as e: logger.error(f"停止 {name} 失败: {e}") self._bots.clear() self._bot_tasks.clear() logger.info("所有聊天平台已停止") async def send_to_all(self, text: str): """向所有平台广播消息""" for bot in self._bots.values(): try: await bot.send_message(ChatResponse(text=text)) except Exception as e: logger.error(f"广播失败: {e}") def get_active_platforms(self) -> List[str]: """获取活跃平台列表""" return list(self._bots.keys()) def get_stats(self) -> Dict[str, Any]: """获取状态""" return { "active_platforms": list(self._bots.keys()), "platform_count": len(self._bots), }