"""
|
消息聚合服务
|
实现同一群组中同一用户短时间内多条消息的聚合功能
|
"""
|
|
import time
|
import threading
|
import json
|
from typing import Dict, List, Optional, Any, Tuple
|
from loguru import logger
|
from dataclasses import dataclass, field
|
from datetime import datetime, timedelta
|
|
from config import settings
|
|
|
@dataclass
|
class PendingMessage:
|
"""待聚合的消息"""
|
message_data: Dict[str, Any]
|
timestamp: float
|
content: str
|
from_user: str
|
from_group: str
|
|
|
@dataclass
|
class MessageAggregation:
|
"""消息聚合对象"""
|
from_user: str
|
from_group: str
|
messages: List[PendingMessage] = field(default_factory=list)
|
first_message_time: float = 0.0
|
last_message_time: float = 0.0
|
timer: Optional[threading.Timer] = None
|
lock: threading.Lock = field(default_factory=threading.Lock)
|
|
def add_message(self, message: PendingMessage):
|
"""添加消息到聚合中"""
|
with self.lock:
|
self.messages.append(message)
|
if not self.first_message_time:
|
self.first_message_time = message.timestamp
|
self.last_message_time = message.timestamp
|
|
def get_aggregated_content(self) -> str:
|
"""获取聚合后的消息内容"""
|
with self.lock:
|
if not self.messages:
|
return ""
|
|
# 按时间戳排序消息
|
sorted_messages = sorted(self.messages, key=lambda x: x.timestamp)
|
|
# 合并消息内容
|
contents = [msg.content for msg in sorted_messages]
|
return "\n".join(contents)
|
|
def get_latest_message_data(self) -> Optional[Dict[str, Any]]:
|
"""获取最新的消息数据作为模板"""
|
with self.lock:
|
if not self.messages:
|
return None
|
|
# 返回最后一条消息的数据作为模板
|
latest_message = max(self.messages, key=lambda x: x.timestamp)
|
return latest_message.message_data.copy()
|
|
def clear(self):
|
"""清空聚合数据"""
|
with self.lock:
|
self.messages.clear()
|
self.first_message_time = 0.0
|
self.last_message_time = 0.0
|
if self.timer:
|
self.timer.cancel()
|
self.timer = None
|
|
|
class MessageAggregatorService:
|
"""消息聚合服务"""
|
|
def __init__(self):
|
self.aggregations: Dict[str, MessageAggregation] = {}
|
self.lock = threading.Lock()
|
self.aggregation_timeout = getattr(settings, 'message_aggregation_timeout', 15) # 默认15秒
|
self.aggregation_enabled = getattr(settings, 'message_aggregation_enabled', True) # 默认启用
|
|
logger.info(f"消息聚合服务初始化: timeout={self.aggregation_timeout}s, enabled={self.aggregation_enabled}")
|
|
def _get_aggregation_key(self, from_user: str, from_group: str) -> str:
|
"""获取聚合键"""
|
return f"{from_group}:{from_user}"
|
|
def should_aggregate_message(self, message_data: Dict[str, Any]) -> bool:
|
"""
|
判断消息是否应该被聚合
|
|
Args:
|
message_data: 消息数据
|
|
Returns:
|
是否应该聚合
|
"""
|
if not self.aggregation_enabled:
|
return False
|
|
# 检查消息类型是否是群聊消息
|
message_type = message_data.get("messageType")
|
if message_type != "80001":
|
return False
|
|
data = message_data.get("data", {})
|
|
# 检查是否是自己发送的消息
|
if data.get("self", False):
|
return False
|
|
# 检查必要字段
|
if not data.get("fromUser") or not data.get("fromGroup") or not data.get("content"):
|
return False
|
|
return True
|
|
def add_message_to_aggregation(self, message_data: Dict[str, Any]) -> Tuple[bool, Optional[Dict[str, Any]]]:
|
"""
|
将消息添加到聚合中
|
|
Args:
|
message_data: 消息数据
|
|
Returns:
|
(是否需要立即处理, 聚合后的消息数据)
|
- 如果返回 (False, None)，表示消息已被聚合，不需要立即处理
|
- 如果返回 (True, aggregated_data)，表示聚合超时，需要处理聚合后的消息
|
"""
|
if not self.should_aggregate_message(message_data):
|
# 不需要聚合，直接处理
|
return True, message_data
|
|
data = message_data.get("data", {})
|
from_user = data.get("fromUser")
|
from_group = data.get("fromGroup")
|
content = data.get("content", "")
|
|
aggregation_key = self._get_aggregation_key(from_user, from_group)
|
current_time = time.time()
|
|
pending_message = PendingMessage(
|
message_data=message_data,
|
timestamp=current_time,
|
content=content,
|
from_user=from_user,
|
from_group=from_group
|
)
|
|
with self.lock:
|
if aggregation_key not in self.aggregations:
|
# 创建新的聚合
|
self.aggregations[aggregation_key] = MessageAggregation(
|
from_user=from_user,
|
from_group=from_group
|
)
|
|
aggregation = self.aggregations[aggregation_key]
|
|
# 取消之前的定时器
|
with aggregation.lock:
|
if aggregation.timer:
|
aggregation.timer.cancel()
|
|
# 添加消息到聚合
|
aggregation.add_message(pending_message)
|
|
# 设置新的定时器
|
timer = threading.Timer(
|
self.aggregation_timeout,
|
self._process_aggregated_messages,
|
args=(aggregation_key,)
|
)
|
|
with aggregation.lock:
|
aggregation.timer = timer
|
|
timer.start()
|
|
logger.info(f"消息已添加到聚合: user={from_user}, group={from_group}, "
|
f"total_messages={len(aggregation.messages)}, timeout={self.aggregation_timeout}s")
|
|
# 消息已被聚合，不需要立即处理
|
return False, None
|
|
def _process_aggregated_messages(self, aggregation_key: str):
|
"""
|
处理聚合的消息（定时器回调）
|
|
Args:
|
aggregation_key: 聚合键
|
"""
|
try:
|
with self.lock:
|
if aggregation_key not in self.aggregations:
|
logger.warning(f"聚合键不存在: {aggregation_key}")
|
return
|
|
aggregation = self.aggregations[aggregation_key]
|
|
# 获取聚合后的消息数据
|
aggregated_content = aggregation.get_aggregated_content()
|
latest_message_data = aggregation.get_latest_message_data()
|
|
if not aggregated_content or not latest_message_data:
|
logger.warning(f"聚合数据为空: {aggregation_key}")
|
self._cleanup_aggregation(aggregation_key)
|
return
|
|
# 创建聚合后的消息数据
|
aggregated_message_data = latest_message_data.copy()
|
aggregated_message_data["data"]["content"] = aggregated_content
|
|
message_count = len(aggregation.messages)
|
logger.info(f"处理聚合消息: user={aggregation.from_user}, group={aggregation.from_group}, "
|
f"message_count={message_count}, aggregated_content_length={len(aggregated_content)}")
|
|
# 清理聚合数据
|
self._cleanup_aggregation(aggregation_key)
|
|
# 调用消息处理器处理聚合后的消息（延迟导入避免循环依赖）
|
from app.services.message_processor import message_processor
|
success = message_processor.process_single_message(aggregated_message_data)
|
|
if success:
|
logger.info(f"聚合消息处理成功: user={aggregation.from_user}, group={aggregation.from_group}")
|
else:
|
logger.error(f"聚合消息处理失败: user={aggregation.from_user}, group={aggregation.from_group}")
|
|
except Exception as e:
|
logger.error(f"处理聚合消息异常: aggregation_key={aggregation_key}, error={str(e)}")
|
self._cleanup_aggregation(aggregation_key)
|
|
def _cleanup_aggregation(self, aggregation_key: str):
|
"""
|
清理聚合数据
|
|
Args:
|
aggregation_key: 聚合键
|
"""
|
try:
|
with self.lock:
|
if aggregation_key in self.aggregations:
|
aggregation = self.aggregations[aggregation_key]
|
aggregation.clear()
|
del self.aggregations[aggregation_key]
|
logger.debug(f"聚合数据已清理: {aggregation_key}")
|
except Exception as e:
|
logger.error(f"清理聚合数据异常: aggregation_key={aggregation_key}, error={str(e)}")
|
|
def get_aggregation_status(self) -> Dict[str, Any]:
|
"""
|
获取聚合状态信息
|
|
Returns:
|
聚合状态信息
|
"""
|
with self.lock:
|
status = {
|
"enabled": self.aggregation_enabled,
|
"timeout": self.aggregation_timeout,
|
"active_aggregations": len(self.aggregations),
|
"aggregations": {}
|
}
|
|
for key, aggregation in self.aggregations.items():
|
with aggregation.lock:
|
status["aggregations"][key] = {
|
"from_user": aggregation.from_user,
|
"from_group": aggregation.from_group,
|
"message_count": len(aggregation.messages),
|
"first_message_time": aggregation.first_message_time,
|
"last_message_time": aggregation.last_message_time,
|
"elapsed_time": time.time() - aggregation.first_message_time if aggregation.first_message_time else 0
|
}
|
|
return status
|
|
def force_process_aggregation(self, from_user: str, from_group: str) -> bool:
|
"""
|
强制处理指定用户和群组的聚合消息
|
|
Args:
|
from_user: 用户ID
|
from_group: 群组ID
|
|
Returns:
|
是否成功处理
|
"""
|
aggregation_key = self._get_aggregation_key(from_user, from_group)
|
|
with self.lock:
|
if aggregation_key not in self.aggregations:
|
logger.warning(f"没有找到待聚合的消息: user={from_user}, group={from_group}")
|
return False
|
|
logger.info(f"强制处理聚合消息: user={from_user}, group={from_group}")
|
self._process_aggregated_messages(aggregation_key)
|
return True
|
|
def stop(self):
|
"""停止聚合服务"""
|
logger.info("正在停止消息聚合服务...")
|
|
with self.lock:
|
for aggregation_key, aggregation in self.aggregations.items():
|
try:
|
with aggregation.lock:
|
if aggregation.timer:
|
aggregation.timer.cancel()
|
logger.debug(f"已取消聚合定时器: {aggregation_key}")
|
except Exception as e:
|
logger.error(f"取消聚合定时器异常: {aggregation_key}, error={str(e)}")
|
|
self.aggregations.clear()
|
|
logger.info("消息聚合服务已停止")
|
|
|
# 全局消息聚合服务实例
|
message_aggregator = MessageAggregatorService()
|
