liuwei_x_x/abot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

调整联系人管理功能

Browse Source
This commit is contained in:
liuwei
2025-04-08 14:30:59 +08:00
parent 2d775c227e
commit fc7ecbac8a
2 changed files with 45 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
robot.py
View File

@@ -58,7 +58,7 @@ class Robot(Job):
# 初始化联系人管理器并设置联系人 # 初始化联系人管理器并设置联系人
self.contact_manager = ContactManager.get_instance() self.contact_manager = ContactManager.get_instance()
self.allContacts = self.get_all_contacts() self.allContacts = self.get_all_contacts()
self.contact_manager.set_contacts(self.allContacts) self.contact_manager.set_contacts(self.allContacts, self.wcf.get_friends())
self.LOG.info(f"DB+REDIS 连接池开始初始化") self.LOG.info(f"DB+REDIS 连接池开始初始化")
# 使用单例模式获取实例 # 使用单例模式获取实例
@@ -545,5 +545,5 @@ class Robot(Job):
def refresh_contacts(self): def refresh_contacts(self):
"""刷新联系人信息""" """刷新联系人信息"""
self.allContacts = self.get_all_contacts() self.allContacts = self.get_all_contacts()
self.contact_manager.refresh_contacts(self.allContacts) self.contact_manager.refresh_contacts(self.allContacts, self.wcf.get_friends())
self.LOG.info("联系人信息已刷新") self.LOG.info("联系人信息已刷新")

41
utils/wechat/contact_manager.py
View File

@@ -16,7 +16,7 @@ class ContactManager:
_official_accounts: Dict[str, str] = {} # 公众号 _official_accounts: Dict[str, str] = {} # 公众号
_initialized = False _initialized = False
_logger = logging.getLogger("ContactManager") _logger = logging.getLogger("ContactManager")
_friends: List[Dict] = []
# 定义公共好友列表 # 定义公共好友列表
_PUBLIC_FRIENDS = { _PUBLIC_FRIENDS = {
'fmessage': '朋友推荐消息', 'fmessage': '朋友推荐消息',
@@ -44,13 +44,23 @@ class ContactManager:
cls._instance = ContactManager() cls._instance = ContactManager()
return cls._instance return cls._instance
def set_contacts(self, contacts: Dict[str, str]) -> None: def set_contacts(self, contacts: Dict[str, str], friends: List[Dict]) -> None:
"""设置联系人字典 """设置联系人字典
Args: Args:
contacts: 联系人字典，格式为 {"wxid": "NickName"} contacts: 联系人字典，格式为 {"wxid": "NickName"}
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}
""" """
self._contacts = contacts self._contacts = contacts
self._friends = friends
self._logger.info(f"联系人信息已更新，共 {len(contacts)} 个联系人") self._logger.info(f"联系人信息已更新，共 {len(contacts)} 个联系人")
# 分类联系人 # 分类联系人
self._classify_contacts() self._classify_contacts()
@@ -72,14 +82,17 @@ class ContactManager:
# 判断是否为群组（wxid以@chatroom结尾） # 判断是否为群组（wxid以@chatroom结尾）
elif wxid.endswith('@chatroom'): elif wxid.endswith('@chatroom'):
self._group_contacts[wxid] = nickname self._group_contacts[wxid] = nickname
# 其他为普通好友和群成员 # # 其他为普通好友和群成员
else: # else:
self._personal_contacts[wxid] = nickname # self._personal_contacts[wxid] = nickname
for friend in self._friends:
# 过滤掉非好友
self._personal_contacts[friend.get("wxid", "")] = friend.get("name", "")
self._logger.info(f"联系人分类完成: {len(self._group_contacts)} 个群组, " self._logger.info(f"联系人分类完成: {len(self._group_contacts)} 个群组, "
f"{len(self._personal_contacts)} 个个人联系人, " f"{len(self._personal_contacts)} 个个人联系人, "
f"{len(self._public_contacts)} 个公共好友, " f"{len(self._public_contacts)} 个公共好友, "
f"{len(self._official_accounts)} 个公众号") f"{len(self._official_accounts)} 个公众号")
def get_contacts(self) -> Dict[str, str]: def get_contacts(self) -> Dict[str, str]:
"""获取所有联系人 """获取所有联系人
@@ -151,13 +164,23 @@ class ContactManager:
self._personal_contacts[wxid] = nickname self._personal_contacts[wxid] = nickname
self._logger.debug(f"已更新联系人: {wxid} -> {nickname}") self._logger.debug(f"已更新联系人: {wxid} -> {nickname}")
def refresh_contacts(self, new_contacts: Dict[str, str]) -> None: def refresh_contacts(self, new_contacts: Dict[str, str], friends: List[Dict]) -> None:
"""刷新联系人信息 """刷新联系人信息
Args: Args:
new_contacts: 新的联系人字典 new_contacts: 新的联系人字典
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}
""" """
self._contacts = new_contacts self._contacts = new_contacts
self._friends = friends
self._logger.info(f"联系人信息已刷新，共 {len(new_contacts)} 个联系人") self._logger.info(f"联系人信息已刷新，共 {len(new_contacts)} 个联系人")
# 重新分类联系人 # 重新分类联系人
self._classify_contacts() self._classify_contacts()