""" Matrix Bot 服务 - 使用 nio 库处理消息收发（支持加密） """ import asyncio import logging import os from typing import Optional, Callable from nio import AsyncClient, RoomMessageText, MegolmEvent from models import SessionLocal, User, Conversation, Message, SystemConfig from services.conversation_service import ConversationService from services import ai_service logger = logging.getLogger(__name__) MAIN_USER_ID = "main_user" STORE_PATH = "/home/xian/.openclaw/workspace-coder/works/ai-chat/matrix_store" class MatrixBot: def __init__(self): self.client: Optional[AsyncClient] = None self.homeserver: str = "" self.user_id: str = "" self.password: str = "" self.is_running: bool = False self.on_message_callback: Optional[Callable] = None self.last_room_id: str = "" async def init_from_config(self): """从数据库配置初始化""" db = SessionLocal() try: configs = {c.key: c.value for c in db.query(SystemConfig).all()} self.homeserver = configs.get('matrix_homeserver', 'http://matrix.tphai.com') self.user_id = configs.get('matrix_username', '@tester:matrix.tphai.com') self.password = configs.get('matrix_password', 'tester12345@!') logger.info(f"Matrix配置: homeserver={self.homeserver}, user={self.user_id}") if self.user_id and self.password: # 创建存储目录 os.makedirs(STORE_PATH, exist_ok=True) # 使用加密存储 self.client = AsyncClient( self.homeserver, self.user_id, store_path=STORE_PATH ) self.is_running = True logger.info(f"Matrix nio 客户端已初始化，存储路径: {STORE_PATH}") return True finally: db.close() return False async def start_sync(self, message_handler: Callable = None): """开始同步消息""" if not self.is_running or not self.client: logger.warning("Matrix未连接") return self.on_message_callback = message_handler # 登录 try: login_resp = await self.client.login(self.password) logger.info(f"Matrix登录成功: {login_resp}") # 加载加密存储 logger.info("加载加密存储...") # 第一次同步获取房间密钥 sync_resp = await self.client.sync(full_state=True) logger.info(f"初始同步完成: next_batch={sync_resp.next_batch}") # 注册消息处理器 - 普通文本消息 self.client.add_event_callback(self._handle_nio_message, RoomMessageText) # 注册加密消息处理器（解密后） self.client.add_event_callback(self._handle_encrypted_message, MegolmEvent) logger.info("Matrix消息回调已注册") # 使用 sync_forever 自动处理事件（后台任务） asyncio.create_task(self._run_sync_forever()) logger.info("Matrix nio sync_forever 任务已启动") except Exception as e: logger.error(f"Matrix启动失败: {e}") import traceback logger.error(traceback.format_exc()) async def _run_sync_forever(self): """运行 sync_forever，自动处理所有事件""" try: logger.info("开始 Matrix sync_forever...") # 不使用 full_state，只做增量同步 await self.client.sync_forever(timeout=30000) except Exception as e: logger.error(f"sync_forever 错误: {e}") import traceback logger.error(traceback.format_exc()) self.is_running = False async def _handle_encrypted_message(self, room, event): """处理加密消息（MegolmEvent）""" logger.info(f"收到加密消息: [{room.room_id}] event_id={event.event_id}") # 打印 event 的所有属性来调试 logger.info(f"MegolmEvent属性: {dir(event)}") logger.info(f"event.source: {event.source}") logger.info(f"event.sender: {event.sender}") logger.info(f"event.session_id: {event.session_id if hasattr(event, 'session_id') else 'N/A'}") # 尝试从不同来源获取消息内容 try: body = None sender = event.sender # 方法1: 从 source 获取 if hasattr(event, 'source') and event.source: content = event.source.get('content', {}) body = content.get('body', '') sender = event.source.get('sender', event.sender) logger.info(f"从source获取: body={body[:50] if body else 'empty'}") # 方法2: 直接尝试 event.body (nio可能已解密) if hasattr(event, 'body') and event.body: body = event.body logger.info(f"从event.body获取: {body[:50]}") # 方法3: 检查 room 中的最近事件 if not body: # 从 room timeline 查找解密后的消息 for timeline_event in room.timeline: if hasattr(timeline_event, 'event_id') and timeline_event.event_id == event.event_id: if hasattr(timeline_event, 'body'): body = timeline_event.body logger.info(f"从room.timeline获取: {body[:50] if body else 'empty'}") if body: logger.info(f"解密成功: sender={sender}, body={body[:50]}") await self._process_message(room, sender, body, event.event_id) else: logger.warning(f"加密消息无法解密，event_id={event.event_id}") # 尝试手动解密 try: decrypted = await self.client.decrypt_event(event) if decrypted and hasattr(decrypted, 'body'): body = decrypted.body sender = decrypted.sender logger.info(f"手动解密成功: {body[:50]}") await self._process_message(room, sender, body, event.event_id) except Exception as de: logger.error(f"手动解密失败: {de}") except Exception as e: logger.error(f"处理加密消息失败: {e}") import traceback logger.error(traceback.format_exc()) async def _handle_nio_message(self, room, event): """处理 nio 收到的消息（普通文本）""" # 忽略自己发送的消息 sender = event.sender if sender == self.user_id: logger.debug(f"忽略自己发送的消息: {sender}") return message_text = event.body.strip() logger.info(f"Matrix收到文本消息: [{room.room_id}] {sender}: {message_text}") # 调用统一处理方法 await self._process_message(room, sender, message_text, event.event_id) async def _process_message(self, room, sender, message_text, event_id): """处理消息的核心逻辑""" # 忽略自己发送的消息 if sender == self.user_id: logger.debug(f"忽略自己发送的消息: {sender}") return # 保存房间ID self.last_room_id = room.room_id db = SessionLocal() try: conv_service = ConversationService(db) # 使用固定主用户 main_user = conv_service.get_or_create_user( MAIN_USER_ID, display_name="主用户", user_type='web' ) # 处理 /new 命令 if message_text == "/new": conversation = conv_service.create_conversation(main_user.id) await self.send_message(room.room_id, "✅ 已创建新会话") logger.info(f"Matrix创建新会话: {conversation.conversation_id}") if self.on_message_callback: await self.on_message_callback( action="new_conversation", conversation_id=conversation.conversation_id, room_id=room.room_id ) return # 获取最新会话 conversations = conv_service.get_user_conversations(main_user.id) if not conversations: conversation = conv_service.create_conversation(main_user.id) else: conversation = conversations[0] # 保存用户消息 user_msg = conv_service.add_message( conversation_id=conversation.id, role='user', content=message_text, source='matrix', extra_data={ 'event_id': event_id, 'room_id': room.room_id, 'sender': sender } ) # 发送"正在输入"状态 try: await self.client.room_typing(room.room_id, typing_state=True) except Exception as e: logger.warning(f"发送输入状态失败: {e}") # 获取AI回复 messages = conv_service.get_messages(conversation.id) ai_response = await ai_service.chat(messages) # 保存AI回复 conv_service.add_message( conversation_id=conversation.id, role='assistant', content=ai_response, source='matrix' ) # 发送回复 await self.send_message(room.room_id, ai_response) # 关闭"正在输入"状态 try: await self.client.room_typing(room.room_id, typing_state=False) except Exception as e: logger.warning(f"关闭输入状态失败: {e}") # 调用回调（用于网页同步） if self.on_message_callback: await self.on_message_callback( action="chat", conversation_id=conversation.conversation_id, user_message=message_text, room_id=room.room_id, message_id=user_msg.id ) logger.info(f"Matrix回复已发送: {ai_response[:50]}") except Exception as e: logger.error(f"处理Matrix消息失败: {e}") import traceback logger.error(traceback.format_exc()) await self.send_message(room.room_id, f"处理消息时出错: {str(e)}") try: await self.client.room_typing(room.room_id, typing_state=False) except: pass finally: db.close() async def send_message(self, room_id: str, message: str): """发送消息到Matrix房间""" if not self.client: logger.error("Matrix客户端未初始化") return False try: await self.client.room_send( room_id, "m.room.message", { "msgtype": "m.text", "body": message } ) logger.info(f"Matrix消息发送成功: {room_id}") return True except Exception as e: logger.error(f"发送Matrix消息错误: {e}") import traceback logger.error(traceback.format_exc()) return False async def disconnect(self): """断开连接""" self.is_running = False if self.client: await self.client.close() # 全局实例 matrix_bot = MatrixBot()