"""
|
静默模式管理服务
|
"""
|
|
import time
|
from typing import Optional
|
from loguru import logger
|
|
from app.services.redis_queue import redis_queue
|
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 activate_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
|
|
# 计算静默结束时间（当前时间 + 配置的分钟数）
|
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()
|
