269 lines
9.4 KiB
Python
269 lines
9.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
联系人管理器 - 提供全局访问联系人信息的单例类
|
|
"""
|
|
|
|
from loguru import logger
|
|
from typing import Dict, Optional, List, Tuple
|
|
|
|
from utils.json_converter import json_to_object
|
|
|
|
|
|
class ContactManager:
|
|
"""联系人管理器单例类"""
|
|
_instance = None
|
|
_contacts: Dict[str, str] = {}
|
|
_group_contacts: Dict[str, str] = {} # 群组联系人
|
|
_personal_contacts: Dict[str, str] = {} # 个人联系人
|
|
_public_contacts: Dict[str, str] = {} # 公共好友
|
|
_official_accounts: Dict[str, str] = {} # 公众号
|
|
_head_images: Dict[str, str] = {} # 头像信息
|
|
_initialized = False
|
|
_logger = logger
|
|
_friends: List[str] = []
|
|
_group_members: List[Dict] = []
|
|
_group_contacts_friends: Dict[str, Dict[str, str]] = {}
|
|
# 定义公共好友列表
|
|
_PUBLIC_FRIENDS = {
|
|
'fmessage': '朋友推荐消息',
|
|
'medianote': '语音记事本',
|
|
'floatbottle': '漂流瓶',
|
|
'mphelper': '公众平台安全助手',
|
|
'filehelper': '文件传输助手',
|
|
'exmail_tool': '腾讯企业邮箱',
|
|
'weixin': '微信团队'
|
|
}
|
|
|
|
def __new__(cls):
|
|
if cls._instance is None:
|
|
cls._instance = super(ContactManager, cls).__new__(cls)
|
|
return cls._instance
|
|
|
|
def __init__(self):
|
|
# 确保初始化代码只执行一次
|
|
if not ContactManager._initialized:
|
|
self._logger.info("初始化联系人管理器")
|
|
ContactManager._initialized = True
|
|
|
|
@classmethod
|
|
def get_instance(cls):
|
|
"""获取单例实例"""
|
|
if cls._instance is None:
|
|
cls._instance = ContactManager()
|
|
return cls._instance
|
|
|
|
def set_contacts(self, contacts: Dict[str, str], friends: List[str], head_imgs: Dict[str, str],
|
|
chatroom_members: List[dict]) -> None:
|
|
"""设置联系人字典
|
|
|
|
Args:
|
|
contacts: 联系人字典，格式为 {"wxid": "NickName"}
|
|
head_imgs: 联系人头像字典，格式为 {"wxid": "http://xxxxx"}
|
|
friends: 好友清单 contact = {
|
|
"wxid": cnt.get("wxid", ""),
|
|
"code": cnt.get("code", ""),
|
|
"remark": cnt.get("remark", ""),
|
|
"name": cnt.get("name", ""),
|
|
"country": cnt.get("country", ""),
|
|
"province": cnt.get("province", ""),
|
|
"city": cnt.get("city", ""),
|
|
"gender": gender}
|
|
chatroom_members: 所有的群成员昵称信息
|
|
"""
|
|
self._contacts = contacts
|
|
self._friends = friends
|
|
self._head_images = head_imgs
|
|
self._group_members = chatroom_members
|
|
self._logger.info(f"联系人信息已更新，共 {len(contacts)} 个联系人")
|
|
# 分类联系人
|
|
self._classify_contacts()
|
|
|
|
def _classify_contacts(self) -> None:
|
|
"""将联系人分类为群组、个人联系人、公共好友和公众号"""
|
|
self._group_contacts = {}
|
|
self._personal_contacts = {}
|
|
self._public_contacts = {}
|
|
self._official_accounts = {}
|
|
|
|
for wxid, nickname in self._contacts.items():
|
|
# 判断是否为公共好友
|
|
if wxid in self._PUBLIC_FRIENDS:
|
|
self._public_contacts[wxid] = self._PUBLIC_FRIENDS.get(wxid, nickname)
|
|
# 判断是否为公众号（wxid以gh_开头）
|
|
elif wxid.startswith('gh_'):
|
|
self._official_accounts[wxid] = nickname
|
|
# 判断是否为群组（wxid以@chatroom结尾）
|
|
elif wxid.endswith('@chatroom'):
|
|
self._group_contacts[wxid] = nickname
|
|
# 确保群ID在字典中存在
|
|
if wxid not in self._group_contacts_friends:
|
|
self._group_contacts_friends[wxid] = {}
|
|
# 获取群成员信息：
|
|
for friend in self._group_members:
|
|
if friend.get('chatroom_id') == wxid:
|
|
self._group_contacts_friends[wxid].update(
|
|
{friend.get('wxid'): friend.get('nick_name', friend.get('wxid'))})
|
|
|
|
else:
|
|
# 判断 frinds 在contacts 里面，将在里面的用户分在
|
|
if wxid in self._friends:
|
|
self._personal_contacts[wxid] = nickname
|
|
self._logger.info(f"联系人分类完成: {len(self._group_contacts)} 个群组, "
|
|
f"{len(self._personal_contacts)} 个个人联系人, "
|
|
f"{len(self._public_contacts)} 个公共好友, "
|
|
f"{len(self._official_accounts)} 个公众号")
|
|
|
|
def get_contacts(self) -> Dict[str, str]:
|
|
"""获取所有联系人
|
|
|
|
Returns:
|
|
联系人字典，格式为 {"wxid": "NickName"}
|
|
"""
|
|
return self._contacts
|
|
|
|
def get_group_contacts(self) -> Dict[str, str]:
|
|
"""获取所有群组联系人
|
|
|
|
Returns:
|
|
群组联系人字典，格式为 {"wxid": "GroupName"}
|
|
"""
|
|
return self._group_contacts
|
|
|
|
def get_personal_contacts(self) -> Dict[str, str]:
|
|
"""获取所有个人联系人
|
|
|
|
Returns:
|
|
个人联系人字典，格式为 {"wxid": "NickName"}
|
|
"""
|
|
return self._personal_contacts
|
|
|
|
def get_public_contacts(self) -> Dict[str, str]:
|
|
"""获取所有公共好友
|
|
|
|
Returns:
|
|
公共好友字典，格式为 {"wxid": "NickName"}
|
|
"""
|
|
return self._public_contacts
|
|
|
|
def get_official_accounts(self) -> Dict[str, str]:
|
|
"""获取所有公众号
|
|
|
|
Returns:
|
|
公众号字典，格式为 {"wxid": "NickName"}
|
|
"""
|
|
return self._official_accounts
|
|
|
|
def get_nickname(self, wxid: str) -> str:
|
|
"""根据微信ID获取昵称
|
|
|
|
Args:
|
|
wxid: 微信ID
|
|
|
|
Returns:
|
|
对应的昵称，如果不存在则返回wxid本身
|
|
"""
|
|
return self._contacts.get(wxid, wxid)
|
|
|
|
def get_all_head_images(self) -> Dict[str, str]:
|
|
"""返回所有的头像信息
|
|
|
|
Returns:
|
|
头像 {"wxid": "http://xxxxx"}
|
|
"""
|
|
return self._head_images
|
|
|
|
def get_head_image(self, wxid: str) -> str:
|
|
"""根据微信ID获取头像
|
|
|
|
Args:
|
|
wxid: 微信ID
|
|
|
|
Returns:
|
|
对应的头像，如果不存在这返回""
|
|
"""
|
|
return self._head_images.get(wxid, "")
|
|
|
|
def update_head_image(self, wxid: str, head_image: str) -> bool:
|
|
"""根据微信ID更新头像
|
|
|
|
Args:
|
|
wxid: 微信ID
|
|
head_image:头像地址
|
|
Returns:
|
|
对应的头像，如果不存在这返回""
|
|
"""
|
|
self._head_images.update({wxid: head_image})
|
|
return True
|
|
|
|
def get_group_name(self, roomid: str, wxid: str) -> str:
|
|
"""
|
|
Args:
|
|
roomid: 群ID
|
|
wxid: 微信ID
|
|
|
|
Returns:
|
|
对应的昵称，如果不存在则返回wxid本身
|
|
"""
|
|
return self._group_contacts_friends.get(roomid, "").get(wxid, "未知昵称")
|
|
|
|
def get_group_members(self, roomid: str) -> Dict[str, str]:
|
|
"""获取指定群的成员列表
|
|
|
|
Args:
|
|
roomid: 群ID
|
|
|
|
Returns:
|
|
群成员字典，格式为 {"wxid": "NickName"}
|
|
"""
|
|
return self._group_contacts_friends.get(roomid, {})
|
|
|
|
def update_group_members(self, roomid: str, wxid: str, nick_name: str) -> bool:
|
|
"""更新指定群的成员列表
|
|
|
|
Args:
|
|
roomid: 群ID
|
|
|
|
Returns:
|
|
群成员字典，格式为 {"wxid": "NickName"}
|
|
"""
|
|
self._group_contacts_friends[roomid].update({wxid: nick_name})
|
|
return True
|
|
|
|
def update_contact(self, wxid: str, nickname: str) -> None:
|
|
"""更新单个联系人信息
|
|
|
|
Args:
|
|
wxid: 微信ID
|
|
nickname: 昵称
|
|
"""
|
|
self._contacts[wxid] = nickname
|
|
# 更新分类
|
|
if wxid in self._PUBLIC_FRIENDS:
|
|
self._public_contacts[wxid] = self._PUBLIC_FRIENDS.get(wxid, nickname)
|
|
elif wxid.startswith('gh_'):
|
|
self._official_accounts[wxid] = nickname
|
|
elif wxid.endswith('@chatroom'):
|
|
self._group_contacts[wxid] = nickname
|
|
# 确保群ID在字典中存在
|
|
if wxid not in self._group_contacts_friends:
|
|
self._group_contacts_friends[wxid] = {}
|
|
# 获取群成员信息：
|
|
for friend in self._group_members:
|
|
if friend.get('chatroom_id') == wxid:
|
|
self._group_contacts_friends[wxid].update(
|
|
{friend.get('wxid'): friend.get('nick_name', friend.get('wxid'))})
|
|
else:
|
|
if wxid in self._friends:
|
|
self._personal_contacts[wxid] = nickname
|
|
self._logger.debug(f"已更新联系人: {wxid} -> {nickname}")
|
|
|
|
def get_contact_statistics(self) -> Tuple[int, int, int, int, int]:
|
|
"""获取联系人统计信息
|
|
|
|
Returns:
|
|
包含总联系人数、群组数、个人联系人数、公共好友数和公众号数的元组
|
|
"""
|
|
return (len(self._contacts), len(self._group_contacts),
|
|
len(self._personal_contacts), len(self._public_contacts),
|
|
len(self._official_accounts))
|