"""
|
好友忽略列表管理服务
|
"""
|
|
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
|
|
# 清空现有的忽略列表
|
redis_queue.redis_client.delete(self.ignore_list_key)
|
|
# 批量添加好友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()
|
