""" 静默模式管理服务 """ import time from typing import Optional from loguru import logger from sqlalchemy.orm import Session from app.services.redis_queue import redis_queue from app.models.database import get_db from app.models.contact import Contact from config import settings class SilenceService: """静默模式管理服务""" def __init__(self): self.silence_key_prefix = "ecloud_silence_mode:" self.silence_end_time_key_prefix = "ecloud_silence_end_time:" def _get_group_silence_key(self, group_id: str) -> str: """获取群组静默模式键""" return f"{self.silence_key_prefix}{group_id}" def _get_group_silence_end_time_key(self, group_id: str) -> str: """获取群组静默结束时间键""" return f"{self.silence_end_time_key_prefix}{group_id}" def is_test_group(self, group_id: str) -> bool: """ 检查指定群组是否为测试群组 Args: group_id: 群组ID Returns: 如果是测试群组返回True，否则返回False """ try: # 检查是否配置了测试群组关键字 if not settings.test_group_keywords: return False # 从数据库查询群组信息 with next(get_db()) as db: contact = db.query(Contact).filter(Contact.wc_id == group_id).first() if not contact or not contact.nick_name: return False # 检查群组昵称是否包含任何测试关键字 nick_name = contact.nick_name for keyword in settings.test_group_keywords: if keyword in nick_name: logger.info(f"识别到测试群组: group_id={group_id}, nick_name={nick_name}, keyword={keyword}") return True return False except Exception as e: logger.error(f"检查测试群组异常: group_id={group_id}, error={str(e)}") return False def activate_silence_mode(self, group_id: str) -> bool: """ 激活指定群组的静默模式 Args: group_id: 群组ID Returns: 激活成功返回True，失败返回False """ try: # 检查是否为测试群组，如果是则不激活静默模式 if self.is_test_group(group_id): logger.info(f"测试群组不激活静默模式: group_id={group_id}") return True if not settings.silence_mode_enabled: logger.debug("静默模式功能已禁用") return False # 计算静默结束时间（当前时间 + 配置的分钟数） silence_duration_seconds = settings.silence_duration_minutes * 60 end_time = time.time() + silence_duration_seconds # 获取群组专用的键 group_silence_key = self._get_group_silence_key(group_id) group_end_time_key = self._get_group_silence_end_time_key(group_id) # 设置静默模式标志和结束时间 redis_queue.redis_client.setex( group_silence_key, silence_duration_seconds, "active" ) redis_queue.redis_client.setex( group_end_time_key, silence_duration_seconds, str(end_time) ) logger.info(f"群组静默模式已激活: group_id={group_id}, 持续时间: {settings.silence_duration_minutes} 分钟") return True except Exception as e: logger.error(f"激活群组静默模式失败: group_id={group_id}, error={str(e)}") return False def extend_silence_mode(self, group_id: str) -> bool: """ 延长指定群组的静默模式时间（刷新静默时长） Args: group_id: 群组ID Returns: 延长成功返回True，失败返回False """ try: if not settings.silence_mode_enabled: logger.debug("静默模式功能已禁用") return False # 重新激活静默模式（相当于刷新时长） return self.activate_silence_mode(group_id) except Exception as e: logger.error(f"延长群组静默模式失败: group_id={group_id}, error={str(e)}") return False def is_silence_active(self, group_id: str) -> bool: """ 检查指定群组的静默模式是否激活 Args: group_id: 群组ID Returns: 如果静默模式激活返回True，否则返回False """ try: if not settings.silence_mode_enabled: return False # 检查群组静默模式标志是否存在 group_silence_key = self._get_group_silence_key(group_id) is_active = redis_queue.redis_client.exists(group_silence_key) if is_active: logger.debug(f"群组静默模式当前处于激活状态: group_id={group_id}") return bool(is_active) except Exception as e: logger.error(f"检查群组静默模式状态失败: group_id={group_id}, error={str(e)}") return False def get_silence_remaining_time(self, group_id: str) -> Optional[int]: """ 获取指定群组的静默模式剩余时间（秒） Args: group_id: 群组ID Returns: 剩余时间（秒），如果静默模式未激活返回None """ try: if not self.is_silence_active(group_id): return None # 获取群组静默结束时间 group_end_time_key = self._get_group_silence_end_time_key(group_id) end_time_str = redis_queue.redis_client.get(group_end_time_key) if not end_time_str: return None end_time = float(end_time_str) current_time = time.time() remaining_time = int(end_time - current_time) return max(0, remaining_time) except Exception as e: logger.error(f"获取群组静默模式剩余时间失败: group_id={group_id}, error={str(e)}") return None def deactivate_silence_mode(self, group_id: str = None) -> bool: """ 手动停用静默模式 Args: group_id: 群组ID，如果为None则停用所有群组的静默模式 Returns: 停用成功返回True，失败返回False """ try: if group_id: # 停用指定群组的静默模式 group_silence_key = self._get_group_silence_key(group_id) group_end_time_key = self._get_group_silence_end_time_key(group_id) redis_queue.redis_client.delete(group_silence_key) redis_queue.redis_client.delete(group_end_time_key) logger.info(f"群组静默模式已手动停用: group_id={group_id}") else: # 停用所有群组的静默模式 silence_keys = redis_queue.redis_client.keys(f"{self.silence_key_prefix}*") end_time_keys = redis_queue.redis_client.keys(f"{self.silence_end_time_key_prefix}*") all_keys = silence_keys + end_time_keys if all_keys: redis_queue.redis_client.delete(*all_keys) logger.info("所有群组的静默模式已手动停用") return True except Exception as e: logger.error(f"停用静默模式失败: group_id={group_id}, error={str(e)}") return False def get_silence_status(self, group_id: str = None) -> dict: """ 获取静默模式详细状态信息 Args: group_id: 群组ID，如果为None则返回全局状态概览 Returns: 包含静默模式状态信息的字典 """ try: if group_id: # 获取指定群组的状态 status = { "enabled": settings.silence_mode_enabled, "group_id": group_id, "active": False, "remaining_seconds": None, "remaining_minutes": None, "duration_minutes": settings.silence_duration_minutes } if settings.silence_mode_enabled: status["active"] = self.is_silence_active(group_id) if status["active"]: remaining_seconds = self.get_silence_remaining_time(group_id) if remaining_seconds is not None: status["remaining_seconds"] = remaining_seconds status["remaining_minutes"] = round(remaining_seconds / 60, 1) else: # 获取全局状态概览 active_groups = [] if settings.silence_mode_enabled: # 查找所有激活的群组 silence_keys = redis_queue.redis_client.keys(f"{self.silence_key_prefix}*") for key in silence_keys: group_id_from_key = key.replace(self.silence_key_prefix, "") remaining_time = self.get_silence_remaining_time(group_id_from_key) if remaining_time and remaining_time > 0: active_groups.append({ "group_id": group_id_from_key, "remaining_seconds": remaining_time, "remaining_minutes": round(remaining_time / 60, 1) }) status = { "enabled": settings.silence_mode_enabled, "active_groups_count": len(active_groups), "active_groups": active_groups, "duration_minutes": settings.silence_duration_minutes } return status except Exception as e: logger.error(f"获取静默模式状态失败: group_id={group_id}, error={str(e)}") return { "enabled": False, "active": False, "remaining_seconds": None, "remaining_minutes": None, "duration_minutes": 0, "error": str(e) } # 全局静默服务实例 silence_service = SilenceService()