""" 消息处理核心逻辑 """ import json import time import re from typing import Dict, Any, Optional, List, Tuple from datetime import datetime from sqlalchemy.orm import Session from loguru import logger from app.models.database import get_db from app.models.contact import Contact from app.models.conversation import Conversation from app.services.redis_queue import redis_queue from app.services.ecloud_client import ecloud_client from app.services.dify_client import dify_client from app.services.friend_ignore_service import friend_ignore_service from app.services.silence_service import silence_service from app.services.group_stats_service import group_stats_service from config import settings class MessageProcessor: """消息处理器""" def __init__(self): pass def parse_at_mentions(self, ai_answer: str) -> Tuple[str, List[str]]: """ 解析AI回复中的@字符，提取客服名称 Args: ai_answer: AI回复内容 Returns: (处理后的消息内容, 需要@的客服wcid列表) """ try: # 获取配置的客服名称列表 customer_service_names = settings.customer_service_names # 查找所有@字符后的客服名称 at_pattern = r'@([^\s]+)' matches = re.findall(at_pattern, ai_answer) valid_at_names = [] at_wc_ids = [] for match in matches: # 检查是否在配置的客服名称列表中 if match in customer_service_names: valid_at_names.append(match) logger.info(f"发现有效的@客服名称: {match}") # 如果有有效的@客服名称，查询数据库获取wcid if valid_at_names: with next(get_db()) as db: for name in valid_at_names: # 根据nick_name查找联系人 contact = db.query(Contact).filter(Contact.nick_name == name).first() if contact: at_wc_ids.append(contact.wc_id) logger.info(f"找到客服联系人: name={name}, wc_id={contact.wc_id}") else: logger.warning(f"未找到客服联系人: name={name}") return ai_answer, at_wc_ids except Exception as e: logger.error(f"解析@字符异常: error={str(e)}") return ai_answer, [] def is_valid_group_message(self, callback_data: Dict[str, Any]) -> bool: """ 检查是否是有效的群聊消息 Args: callback_data: 回调数据 Returns: 是否是有效的群聊消息 """ # 检查消息类型是否是群聊消息（80001） message_type = callback_data.get("messageType") if message_type != "80001": logger.info(f"忽略非群聊消息: messageType={message_type}") return False # 检查是否是自己发送的消息 data = callback_data.get("data", {}) if data.get("self", False): logger.info(f"忽略自己发送的消息: fromUser={data.get('fromUser')}") return False # 检查必要字段 if ( not data.get("fromUser") or not data.get("fromGroup") or not data.get("content") ): logger.warning(f"消息缺少必要字段: data={data}") return False # 获取用户和群组信息 from_user = data.get("fromUser") from_group = data.get("fromGroup") # 检查发送者是否在好友忽略列表中 is_friend_ignored = friend_ignore_service.is_friend_ignored(from_user) if is_friend_ignored: logger.info(f"忽略好友发送的消息: fromUser={from_user}, fromGroup={from_group}") # 统计被忽略的好友发言次数（确保被忽略的好友消息也纳入统计） group_stats_service.increment_user_message_count(from_group, from_user) # 激活或延长该群组的静默模式 if silence_service.is_silence_active(from_group): # 如果该群组静默模式已激活，延长时间 silence_service.extend_silence_mode(from_group) logger.info(f"好友消息被忽略，群组静默模式时间已刷新: fromUser={from_user}, fromGroup={from_group}") else: # 如果该群组静默模式未激活，激活静默模式 silence_service.activate_silence_mode(from_group) logger.info(f"好友消息被忽略，群组静默模式已激活: fromUser={from_user}, fromGroup={from_group}") return False # # 统计正常处理的好友发言次数 # group_stats_service.increment_user_message_count(from_group, from_user) # 检查该群组的静默模式是否激活（在好友忽略检查之后） if silence_service.is_silence_active(from_group): logger.info(f"群组静默模式激活中，忽略消息: fromGroup={from_group}") return False return True def enqueue_callback_message(self, callback_data: Dict[str, Any]) -> bool: """ 将回调消息加入队列 Args: callback_data: 回调数据 Returns: 是否成功加入队列 """ # 验证消息有效性 if not self.is_valid_group_message(callback_data): return False data = callback_data.get("data", {}) from_user = data.get("fromUser") # 将整个回调内容转成JSON字符串存储到Redis队列 return redis_queue.enqueue_message(from_user, callback_data) def ensure_contact_exists(self, from_group: str, w_id: str, db: Session) -> bool: """ 确保联系人信息存在于数据库中 Args: from_group: 群组ID w_id: 登录实例标识 db: 数据库会话 Returns: 是否成功确保联系人存在 """ try: # 检查数据库中是否已存在该联系人 existing_contact = ( db.query(Contact).filter(Contact.wc_id == from_group).first() ) if existing_contact: logger.info(f"联系人已存在: wc_id={from_group}") return True # 调用E云管家API获取联系人信息 contact_info = ecloud_client.get_contact_info(w_id, from_group) if not contact_info: logger.error(f"无法获取联系人信息: wc_id={from_group}") return False # 保存联系人信息到数据库 new_contact = Contact( wc_id=from_group, user_name=contact_info.get("userName"), nick_name=contact_info.get("nickName"), remark=contact_info.get("remark"), signature=contact_info.get("signature"), sex=contact_info.get("sex"), alias_name=contact_info.get("aliasName"), country=contact_info.get("country"), big_head=contact_info.get("bigHead"), small_head=contact_info.get("smallHead"), label_list=contact_info.get("labelList"), v1=contact_info.get("v1"), ) db.add(new_contact) db.commit() logger.info( f"成功保存联系人信息: wc_id={from_group}, nick_name={contact_info.get('nickName')}" ) return True except Exception as e: logger.error(f"确保联系人存在失败: wc_id={from_group}, error={str(e)}") db.rollback() return False def process_single_message(self, message_data: Dict[str, Any]) -> bool: """ 处理单条消息 Args: message_data: 消息数据 Returns: 是否处理成功 """ try: data = message_data.get("data", {}) from_user = data.get("fromUser") from_group = data.get("fromGroup") content = data.get("content") w_id = data.get("wId") logger.info(f"开始处理消息: from_user={from_user}, from_group={from_group}") # 使用上下文管理器确保数据库会话正确管理 with next(get_db()) as db: # 3.1 确保联系人信息存在 if not self.ensure_contact_exists(from_group, w_id, db): logger.error(f"联系人信息处理失败: from_group={from_group}") return False # 3.2 获取群组中发言次数最多的用户昵称 most_active_nickname = group_stats_service.get_most_active_user_nickname(from_group) logger.info(f"群组最活跃用户昵称: group={from_group}, nickname={most_active_nickname}") # 3.3 获取用户在当前群组的conversation_id conversation_id = redis_queue.get_conversation_id(from_user, from_group) # 调用Dify接口发送消息（根据配置选择模式） dify_response = dify_client.send_message( query=content, user=from_user, conversation_id=conversation_id, nick_name=most_active_nickname ) if not dify_response: logger.error(f"Dify响应失败: from_user={from_user}") return False # 获取AI回答和新的conversation_id ai_answer = dify_response.get("answer", "") new_conversation_id = dify_response.get("conversation_id", "") # 更新Redis中的conversation_id（基于用户+群组） if new_conversation_id: redis_queue.set_conversation_id(from_user, new_conversation_id, from_group, 1800) # 3.4 保存对话记录到数据库 # 按用户、群组和小时分组对话记录 current_time = datetime.now() hour_key = current_time.strftime("%Y%m%d_%H") # 查找当前用户在当前群组当前小时的对话记录 existing_conversation = ( db.query(Conversation) .filter( Conversation.from_user == from_user, Conversation.conversation_id == new_conversation_id, Conversation.group == from_group, Conversation.hour == hour_key ) .first() ) if existing_conversation: # 更新现有记录 - 使用JSON格式追加对话内容（当前用户在当前群组当前小时的对话） try: # 解析现有的content JSON if existing_conversation.content: content_list = json.loads(existing_conversation.content) else: content_list = [] # 追加新的对话内容 content_list.append({ "user": content, "ai": ai_answer }) # 更新记录 existing_conversation.content = json.dumps(content_list, ensure_ascii=False) existing_conversation.is_processed = True logger.info(f"追加到当前用户群组小时对话记录: user={from_user}, group={from_group}, hour={hour_key}, 对话轮次={len(content_list)}") except json.JSONDecodeError as e: logger.error(f"解析现有对话内容JSON失败: {str(e)}, 重新创建") # 如果JSON解析失败，重新创建content content_list = [{"user": content, "ai": ai_answer}] existing_conversation.content = json.dumps(content_list, ensure_ascii=False) existing_conversation.is_processed = True else: # 创建新记录 - 新的用户群组小时对话或首次对话，使用JSON格式存储对话内容 content_list = [{"user": content, "ai": ai_answer}] new_conversation = Conversation( from_user=from_user, conversation_id=new_conversation_id, group=from_group, hour=hour_key, content=json.dumps(content_list, ensure_ascii=False), is_processed=True, ) db.add(new_conversation) logger.info(f"创建新的用户群组小时对话记录: user={from_user}, group={from_group}, hour={hour_key}, 初始对话轮次=1") db.commit() # 发送AI回答到群聊 success = False if ai_answer: # 解析AI回复中的@字符 processed_answer, at_wc_ids = self.parse_at_mentions(ai_answer) # 发送消息，最多重试3次 for attempt in range(3): if at_wc_ids: # 如果有@客服，使用群聊@接口 logger.info(f"使用群聊@接口发送消息: at_wc_ids={at_wc_ids}") if ecloud_client.send_group_at_message( w_id, from_group, processed_answer, at_wc_ids ): success = True break else: # 普通群聊消息 logger.info("使用普通群聊接口发送消息") if ecloud_client.send_group_message( w_id, from_group, processed_answer ): success = True break logger.warning(f"发送AI回答失败，尝试重试 ({attempt + 1}/3): from_user={from_user}") if attempt < 2: # 不是最后一次尝试，等待一段时间再重试 time.sleep(2 ** attempt) # 指数退避 if success: # 更新发送状态 conversation = ( db.query(Conversation) .filter( Conversation.from_user == from_user, Conversation.conversation_id == new_conversation_id, Conversation.group == from_group, Conversation.hour == hour_key ) .first() ) if conversation: conversation.is_sent = True conversation.sent_time = current_time db.commit() logger.info(f"消息处理完成: from_user={from_user}") return True else: logger.error(f"发送AI回答失败: from_user={from_user}") return False except Exception as e: logger.error(f"处理消息异常: from_user={from_user}, error={str(e)}") return False # 全局消息处理器实例 message_processor = MessageProcessor()