hubian
0d25cdc344
- 使用 matrix-nio 库替代 HTTP API - 支持解密加密消息（MegolmEvent） - 添加 matrix_password 配置项 - 发送'正在输入'状态提示
217 lines
7.7 KiB
Python
217 lines
7.7 KiB
Python
"""
|
|
Matrix Bot 服务 - 使用 nio 库处理消息收发（支持加密）
|
|
"""
|
|
import asyncio
|
|
import logging
|
|
from typing import Optional, Callable
|
|
from nio import AsyncClient, RoomMessageText, RoomMessageEmote
|
|
|
|
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"
|
|
|
|
|
|
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:
|
|
self.client = AsyncClient(self.homeserver, self.user_id)
|
|
self.is_running = True
|
|
logger.info("Matrix nio 客户端已初始化")
|
|
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
|
|
|
|
# 注册消息处理器
|
|
self.client.add_event_callback(self._handle_nio_message, RoomMessageText)
|
|
|
|
# 登录
|
|
try:
|
|
login_resp = await self.client.login(self.password)
|
|
logger.info(f"Matrix登录成功: {login_resp}")
|
|
|
|
# 开始同步（后台任务）
|
|
asyncio.create_task(self._sync_loop())
|
|
logger.info("Matrix nio 同步任务已启动")
|
|
except Exception as e:
|
|
logger.error(f"Matrix登录失败: {e}")
|
|
|
|
async def _sync_loop(self):
|
|
"""后台同步循环"""
|
|
while self.is_running:
|
|
try:
|
|
# 同步一次
|
|
sync_resp = await self.client.sync(timeout=30000)
|
|
if sync_resp:
|
|
logger.debug(f"Matrix同步完成: next_batch={sync_resp.next_batch}")
|
|
|
|
# 处理已加入的房间
|
|
for room_id, room in self.client.rooms.items():
|
|
self.last_room_id = room_id
|
|
logger.debug(f"房间: {room_id}, 名称: {room.display_name}")
|
|
|
|
await asyncio.sleep(1)
|
|
except Exception as e:
|
|
logger.error(f"Matrix同步错误: {e}")
|
|
await asyncio.sleep(10)
|
|
|
|
async def _handle_nio_message(self, room, event):
|
|
"""处理 nio 收到的消息"""
|
|
# 忽略自己发送的消息
|
|
sender = event.sender
|
|
if sender == self.user_id:
|
|
return
|
|
|
|
message_text = event.body.strip()
|
|
|
|
logger.info(f"Matrix收到消息: [{room.room_id}] {sender}: {message_text}")
|
|
|
|
# 保存房间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.event_id,
|
|
'room_id': room.room_id,
|
|
'sender': sender
|
|
}
|
|
)
|
|
|
|
# 发送"正在输入"状态
|
|
await self.client.room_typing(room.room_id, typing_state=True)
|
|
|
|
# 获取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)
|
|
|
|
# 关闭"正在输入"状态
|
|
await self.client.room_typing(room.room_id, typing_state=False)
|
|
|
|
# 调用回调（用于网页同步）
|
|
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}")
|
|
await self.send_message(room.room_id, f"处理消息时出错: {str(e)}")
|
|
await self.client.room_typing(room.room_id, typing_state=False)
|
|
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}")
|
|
return False
|
|
|
|
async def disconnect(self):
|
|
"""断开连接"""
|
|
self.is_running = False
|
|
if self.client:
|
|
await self.client.close()
|
|
|
|
|
|
# 全局实例
|
|
matrix_bot = MatrixBot() |