加入头像信息

db/contacts_db.py

@@ -652,3 +652,25 @@ class ContactsDBOperator(BaseDBOperator):
         except Exception as e:
             self.LOG.error(f"删除所有联系人信息失败: {e}")
             return False
+    
+    #新增获取所有联系人头像信息接口
+    def get_all_contacts_avatar(self) -> Dict[str, str]:
+        """获取所有联系人头像信息"""
+        try:
+            sql = """
+            SELECT user_name, small_head_img_url
+            FROM t_wechat_contacts
+            union all
+            SELECT wxid as user_name, small_head_img_url
+            FROM t_chatroom_member
+            union all
+            SELECT chatroom_id as user_name, small_head_img_url
+            FROM t_chatrooms
+            """
+            results = self.execute_query(sql)
+            #返回DICT
+            results = {result['user_name']: result['small_head_img_url'] for result in results}
+            return results
+        except Exception as e:
+            self.LOG.error(f"获取所有联系人头像信息失败: {e}")
+            return []

robot.py

@@ -179,7 +179,8 @@ class Robot(Job):
 
             # 登录成功后加载联系人信息
             self.allContacts = self.get_all_contacts()
-            self.contact_manager.set_contacts(self.allContacts)
+            self.head_images =self.get_all_head_images()
+            self.contact_manager.set_contacts(self.allContacts,self.head_images)
 
             self.message_storage = MessageStorage(self.ipad_bot)
 
@@ -371,13 +372,16 @@ class Robot(Job):
                                 wxid = member.get("UserName", "")
                                 nick_name = member.get("NickName", "")
                                 displayName = member.get("DisplayName", "")
+                                small_head_img_url = member.get("SmallHeadImgUrl", "")
                                 #如果displayName不为空，使用displayName
                                 if displayName:
                                     nick_name = displayName
                                 if wxid:
                                     self.allContacts[wxid] = nick_name
 
-                            self.contact_manager.set_contacts(self.allContacts)
+                                self.head_images[wxid] = small_head_img_url
+
+                            self.contact_manager.set_contacts(self.allContacts,self.head_images)
                             self.LOG.info(f"已更新群 {group_id} 的成员信息")
                 except Exception as e:
                     self.LOG.error(f"获取群成员信息失败: {e}")
@@ -508,6 +512,16 @@ class Robot(Job):
             self.LOG.error(f"获取联系人信息失败: {e}")
             return {}
 
+    def get_all_head_images(self) -> dict:
+        """获取所有的联系人头像信息"""
+        try:
+            # 从数据库获取所有联系人的头像信息
+            head_images = self.contacts_db.get_all_contacts_avatar()
+            return head_images
+        except Exception as e:
+            self.LOG.error(f"获取所有联系人头像信息失败: {e}")
+            return {}
+
     async def refresh_contacts_db(self):
         """刷新联系人信息"""
         # 获取用户所有的联系人，并保存到数据库
@@ -551,13 +565,16 @@ class Robot(Job):
                         wxid = member.get("UserName", "")
                         nick_name = member.get("NickName", "")
                         displayName = member.get("DisplayName", "")
+                        small_head_img_url = member.get("SmallHeadImgUrl", "")
                         #如果displayName不为空，使用displayName
                         if displayName:
                             nick_name = displayName
                         if wxid:
                             self.allContacts[wxid] = nick_name
 
-                    self.contact_manager.set_contacts(self.allContacts)
+                        self.head_images[wxid] = small_head_img_url
+
+                    self.contact_manager.set_contacts(self.allContacts,self.head_images)
                     self.LOG.info(f"已更新群 {group_id} 的成员信息")
             else:
                 self.LOG.error(f"获取群 {group_id} 信息失败，证明用户无该群信息，删除群的相关资料。")

utils/wechat/contact_manager.py

@@ -51,7 +51,7 @@ class ContactManager:
             cls._instance = ContactManager()
         return cls._instance
 
-    def set_contacts(self, contacts: Dict[str, str]) -> None:
+    def set_contacts(self, contacts: Dict[str, str], head_imgs: Dict[str, str]) -> None:
         """设置联系人字典
         
         Args:
@@ -69,6 +69,7 @@ class ContactManager:
         """
         self._contacts = contacts
         self._friends = contacts
+        self._head_images = head_imgs
         self._logger.info(f"联系人信息已更新，共 {len(contacts)} 个联系人")
         # 分类联系人
         self._classify_contacts()