app/services/silence_service.py
@@ -5,8 +5,11 @@
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
@@ -25,6 +28,39 @@
        """获取群组静默结束时间键"""
        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:
        """
        激活指定群组的静默模式
@@ -36,6 +72,11 @@
            激活成功返回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