| | |
| | | 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 |
| | | |
| | | |
| | |
| | | """获取群组静默结束时间键""" |
| | | 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: |
| | | """ |
| | | 激活指定群组的静默模式 |
| | |
| | | 激活成功返回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 |