""" 群组统计服务 用于统计群组中好友的发言次数 """ import time from typing import Optional, Dict, List, Tuple from loguru import logger from app.services.redis_queue import redis_queue from app.models.contact import Contact from app.models.database import get_db from config import settings class GroupStatsService: """群组统计服务""" def __init__(self): self.stats_key_prefix = "group_stats:" self.stats_expiry = 24 * 60 * 60 # 24小时过期 def _get_group_stats_key(self, group_id: str) -> str: """ 获取群组统计的Redis键名 Args: group_id: 群组ID Returns: Redis键名 """ return f"{self.stats_key_prefix}{group_id}" def increment_user_message_count(self, group_id: str, user_id: str) -> bool: """ 增加用户在群组中的发言次数 Args: group_id: 群组ID user_id: 用户ID Returns: 操作成功返回True，失败返回False """ try: stats_key = self._get_group_stats_key(group_id) # 使用Redis的HINCRBY命令增加计数 redis_queue.redis_client.hincrby(stats_key, user_id, 1) # 设置过期时间 redis_queue.redis_client.expire(stats_key, self.stats_expiry) logger.debug(f"用户发言次数已增加: group={group_id}, user={user_id}") return True except Exception as e: logger.error( f"增加用户发言次数失败: group={group_id}, user={user_id}, error={str(e)}" ) return False def get_group_message_stats(self, group_id: str) -> Dict[str, int]: """ 获取群组中所有用户的发言次数统计 Args: group_id: 群组ID Returns: 用户ID到发言次数的映射字典 """ try: stats_key = self._get_group_stats_key(group_id) stats = redis_queue.redis_client.hgetall(stats_key) # 将字节字符串转换为普通字符串，并转换计数为整数 result = {} for user_id, count in stats.items(): if isinstance(user_id, bytes): user_id = user_id.decode("utf-8") if isinstance(count, bytes): count = count.decode("utf-8") result[user_id] = int(count) logger.debug(f"获取群组发言统计: group={group_id}, stats={result}") return result except Exception as e: logger.error(f"获取群组发言统计失败: group={group_id}, error={str(e)}") return {} def get_most_active_user_nickname(self, group_id: str) -> str: """ 获取群组中发言次数最多的用户昵称 Args: group_id: 群组ID Returns: 发言次数最多的用户昵称，如果没有找到或发言次数都相同或都为0，则返回默认客服名称 """ try: # 获取群组发言统计 stats = self.get_group_message_stats(group_id) if not stats: logger.info(f"群组无发言统计数据，使用默认客服名称: group={group_id}") return settings.customer_service_default_name # 过滤掉发言次数为0的用户 active_stats = { user_id: count for user_id, count in stats.items() if count > 0 } if not active_stats: logger.info(f"群组无有效发言数据，使用默认客服名称: group={group_id}") return settings.customer_service_default_name # 检查是否所有用户发言次数都相同 counts = list(active_stats.values()) if len(set(counts)) == 1: logger.info( f"群组所有用户发言次数相同，使用默认客服名称: group={group_id}" ) return settings.customer_service_default_name # 找到发言次数最多的用户 most_active_user_id = max(active_stats, key=active_stats.get) max_count = active_stats[most_active_user_id] logger.info( f"找到最活跃用户: group={group_id}, user={most_active_user_id}, count={max_count}" ) # 根据用户ID查找昵称 nickname = self._get_user_nickname(most_active_user_id) if nickname: logger.info( f"获取到最活跃用户昵称: group={group_id}, user={most_active_user_id}, nickname={nickname}" ) return nickname else: logger.warning( f"未找到最活跃用户昵称，使用默认客服名称: group={group_id}, user={most_active_user_id}" ) return settings.customer_service_default_name except Exception as e: logger.error(f"获取最活跃用户昵称失败: group={group_id}, error={str(e)}") return settings.customer_service_default_name def _get_user_nickname(self, user_id: str) -> Optional[str]: """ 根据用户ID获取昵称 Args: user_id: 用户ID Returns: 用户昵称，如果未找到返回None """ try: with next(get_db()) as db: contact = db.query(Contact).filter(Contact.wc_id == user_id).first() if contact and contact.nick_name: return contact.nick_name else: logger.warning(f"未找到用户昵称: user_id={user_id}") return None except Exception as e: logger.error(f"查询用户昵称异常: user_id={user_id}, error={str(e)}") return None def get_group_stats_summary(self, group_id: str) -> Dict: """ 获取群组统计摘要信息 Args: group_id: 群组ID Returns: 包含统计摘要的字典 """ try: stats = self.get_group_message_stats(group_id) most_active_nickname = self.get_most_active_user_nickname(group_id) # 计算总发言次数 total_messages = sum(stats.values()) # 获取活跃用户数量（发言次数>0） active_users = len([count for count in stats.values() if count > 0]) # 构建用户昵称统计 user_stats = [] for user_id, count in stats.items(): if count > 0: nickname = self._get_user_nickname(user_id) user_stats.append( { "user_id": user_id, "nickname": nickname or "未知用户", "message_count": count, } ) # 按发言次数排序 user_stats.sort(key=lambda x: x["message_count"], reverse=True) return { "group_id": group_id, "total_messages": total_messages, "active_users": active_users, "most_active_nickname": most_active_nickname, "user_stats": user_stats, } except Exception as e: logger.error(f"获取群组统计摘要失败: group={group_id}, error={str(e)}") return { "group_id": group_id, "total_messages": 0, "active_users": 0, "most_active_nickname": settings.customer_service_default_name, "user_stats": [], } def clear_group_stats(self, group_id: str) -> bool: """ 清空指定群组的统计数据 Args: group_id: 群组ID Returns: 操作成功返回True，失败返回False """ try: stats_key = self._get_group_stats_key(group_id) redis_queue.redis_client.delete(stats_key) logger.info(f"已清空群组统计数据: group={group_id}") return True except Exception as e: logger.error(f"清空群组统计数据失败: group={group_id}, error={str(e)}") return False # 全局群组统计服务实例 group_stats_service = GroupStatsService()