""" 好友忽略列表管理服务 """ from typing import List, Set, 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 app.services.silence_service import silence_service from config import settings class FriendIgnoreService: """好友忽略列表管理服务""" def __init__(self): self.ignore_list_key = "ecloud_ignore_friends" def _get_wid_by_nickname(self, nickname: str) -> Optional[str]: """ 根据昵称获取w_id Args: nickname: 好友昵称 Returns: 对应的w_id，如果未找到返回None """ try: with next(get_db()) as db: contact = ( db.query(Contact).filter(Contact.nick_name == nickname).first() ) if contact: wc_id = contact.wc_id if contact.work_wc_id: wc_id += f",{contact.work_wc_id}" return wc_id else: logger.warning(f"未找到昵称为 '{nickname}' 的联系人") return None except Exception as e: logger.error(f"根据昵称查找w_id异常: nickname={nickname}, error={str(e)}") return None def _get_whitelist_wids(self) -> List[str]: """ 将配置中的昵称白名单转换为w_id列表 Returns: w_id列表 """ wid_list = [] for nickname in settings.friend_ignore_whitelist: wid = self._get_wid_by_nickname(nickname) if wid: wid_list.append(wid) logger.debug(f"白名单昵称 '{nickname}' 对应w_id: {wid}") else: logger.warning(f"白名单昵称 '{nickname}' 未找到对应的联系人") return wid_list def add_friends_to_ignore_list(self, friends: List[str]) -> bool: """ 将好友w_id添加到Redis忽略列表 Args: friends: 好友w_id列表 Returns: 添加成功返回True，失败返回False """ try: if not friends: logger.info("好友列表为空，无需添加到忽略列表") return True # 批量添加好友w_id到忽略列表（不清空现有列表） redis_queue.redis_client.sadd(self.ignore_list_key, *friends) logger.info(f"已将 {len(friends)} 个好友添加到忽略列表") return True except Exception as e: logger.error(f"添加好友到忽略列表异常: error={str(e)}") return False def is_friend_ignored(self, w_id: str, group_id: Optional[str] = None) -> bool: """ 检查指定w_id是否应该被忽略 逻辑： 1. 如果好友忽略功能未启用，返回False（不忽略） 2. 如果w_id在白名单中，返回False（不忽略） 3. 如果w_id在忽略列表中，但所在群组为测试群组，返回False（不忽略） 4. 如果w_id在忽略列表中，返回True（忽略） 5. 如果w_id不在忽略列表中，返回False（不忽略） Args: w_id: 用户w_id group_id: 群组ID（可选），用于检查是否为测试群组 Returns: 如果应该被忽略返回True，否则返回False """ try: # 检查好友忽略功能是否启用 if not settings.friend_ignore_enabled: logger.debug(f"好友忽略功能已禁用，不忽略消息: w_id={w_id}") return False # 检查是否在白名单中（通过昵称） whitelist_wids = self._get_whitelist_wids() if any(w_id in wids for wids in whitelist_wids): logger.info(f"w_id在白名单中，不忽略消息: w_id={w_id}") return False # 检查是否在忽略列表中 is_in_ignore_list = redis_queue.redis_client.sismember( self.ignore_list_key, w_id ) if is_in_ignore_list: # 如果在忽略列表中，检查是否在测试群组中 if group_id and silence_service.is_test_group(group_id): logger.info( f"测试群组中的好友消息不被忽略: w_id={w_id}, group_id={group_id}" ) return False logger.info(f"w_id在忽略列表中，忽略消息: w_id={w_id}") return is_in_ignore_list except Exception as e: logger.error(f"检查忽略列表异常: w_id={w_id}, error={str(e)}") return False def get_ignore_list(self) -> Set[str]: """ 获取完整的忽略列表 Returns: 忽略列表中的所有w_id集合 """ try: return redis_queue.redis_client.smembers(self.ignore_list_key) except Exception as e: logger.error(f"获取忽略列表异常: error={str(e)}") return set() def remove_friend_from_ignore_list(self, w_id: str) -> bool: """ 从忽略列表中移除指定w_id Args: w_id: 用户w_id Returns: 移除成功返回True，失败返回False """ try: result = redis_queue.redis_client.srem(self.ignore_list_key, w_id) if result: logger.info(f"已从忽略列表中移除: w_id={w_id}") else: logger.info(f"w_id不在忽略列表中: w_id={w_id}") return True except Exception as e: logger.error(f"从忽略列表移除w_id异常: w_id={w_id}, error={str(e)}") return False def clear_ignore_list(self) -> bool: """ 清空忽略列表 Returns: 清空成功返回True，失败返回False """ try: redis_queue.redis_client.delete(self.ignore_list_key) logger.info("已清空忽略列表") return True except Exception as e: logger.error(f"清空忽略列表异常: error={str(e)}") return False def get_ignore_list_count(self) -> int: """ 获取忽略列表中的好友数量 Returns: 忽略列表中的好友数量 """ try: return redis_queue.redis_client.scard(self.ignore_list_key) except Exception as e: logger.error(f"获取忽略列表数量异常: error={str(e)}") return 0 def get_whitelist(self) -> List[str]: """ 获取当前的白名单列表 Returns: 白名单中的w_id列表 """ return settings.friend_ignore_whitelist.copy() def is_whitelist_enabled(self) -> bool: """ 检查白名单功能是否启用 Returns: 如果启用返回True，否则返回False """ return settings.friend_ignore_enabled def get_ignore_status_info(self, w_id: str) -> dict: """ 获取指定w_id的详细忽略状态信息 Args: w_id: 用户w_id Returns: 包含详细状态信息的字典 """ try: # 获取白名单w_id列表 whitelist_wids = self._get_whitelist_wids() info = { "w_id": w_id, "ignore_enabled": settings.friend_ignore_enabled, "in_whitelist": w_id in whitelist_wids, "in_ignore_list": False, "final_ignored": False, "reason": "", "whitelist_nicknames": settings.friend_ignore_whitelist, } if not settings.friend_ignore_enabled: info["reason"] = "好友忽略功能已禁用" return info if w_id in whitelist_wids: info["reason"] = "在白名单中，不会被忽略" return info info["in_ignore_list"] = redis_queue.redis_client.sismember( self.ignore_list_key, w_id ) if info["in_ignore_list"]: info["final_ignored"] = True info["reason"] = "在忽略列表中，会被忽略" else: info["reason"] = "不在忽略列表中，不会被忽略" return info except Exception as e: logger.error(f"获取忽略状态信息异常: w_id={w_id}, error={str(e)}") return {"w_id": w_id, "error": str(e), "final_ignored": False} # 全局好友忽略服务实例 friend_ignore_service = FriendIgnoreService()