diff --git a/admin/dashboard/server.py b/admin/dashboard/server.py
index 6622bc2..ed636b4 100644
--- a/admin/dashboard/server.py
+++ b/admin/dashboard/server.py
@@ -180,6 +180,90 @@ class DashboardServer:
@login_required
def robot_management():
return render_template('robot_management.html')
+
+ # 添加通讯录管理页面路由
+ @app.route('/contacts')
+ @login_required
+ def contacts_management():
+ """通讯录管理页面"""
+ return render_template('contacts_management.html')
+
+ # 添加通讯录相关API
+ @app.route('/api/contacts/all', methods=['GET'])
+ @login_required
+ def api_contacts_all():
+ """获取所有联系人信息API"""
+ try:
+ contacts = self.contact_manager.get_contacts()
+ return jsonify({
+ "success": True,
+ "data": {
+ "contacts": contacts
+ }
+ })
+ except Exception as e:
+ self.logger.error(f"获取所有联系人信息失败: {e}")
+ return jsonify({"success": False, "error": str(e)}), 500
+
+ @app.route('/api/contacts/groups', methods=['GET'])
+ @login_required
+ def api_contacts_groups():
+ """获取群组联系人信息API"""
+ try:
+ contacts = self.contact_manager.get_contacts()
+ group_contacts = {wxid: name for wxid, name in contacts.items() if '@@' in wxid or '@chatroom' in wxid}
+
+ return jsonify({
+ "success": True,
+ "data": {
+ "groups": group_contacts
+ }
+ })
+ except Exception as e:
+ self.logger.error(f"获取群组联系人信息失败: {e}")
+ return jsonify({"success": False, "error": str(e)}), 500
+
+ @app.route('/api/contacts/personal', methods=['GET'])
+ @login_required
+ def api_contacts_personal():
+ """获取个人联系人信息API"""
+ try:
+ contacts = self.contact_manager.get_contacts()
+ personal_contacts = {wxid: name for wxid, name in contacts.items()
+ if '@@' not in wxid and '@chatroom' not in wxid}
+
+ return jsonify({
+ "success": True,
+ "data": {
+ "personal": personal_contacts
+ }
+ })
+ except Exception as e:
+ self.logger.error(f"获取个人联系人信息失败: {e}")
+ return jsonify({"success": False, "error": str(e)}), 500
+
+ @app.route('/api/contacts/statistics', methods=['GET'])
+ @login_required
+ def api_contacts_statistics():
+ """获取联系人统计信息API"""
+ try:
+ contacts = self.contact_manager.get_contacts()
+ group_contacts = {wxid: name for wxid, name in contacts.items()
+ if '@@' in wxid or '@chatroom' in wxid}
+ personal_contacts = {wxid: name for wxid, name in contacts.items()
+ if '@@' not in wxid and '@chatroom' not in wxid}
+
+ return jsonify({
+ "success": True,
+ "data": {
+ "total": len(contacts),
+ "groups": len(group_contacts),
+ "personal": len(personal_contacts)
+ }
+ })
+ except Exception as e:
+ self.logger.error(f"获取联系人统计信息失败: {e}")
+ return jsonify({"success": False, "error": str(e)}), 500
@app.route('/api/robot/groups')
@login_required
diff --git a/admin/dashboard/templates/base.html b/admin/dashboard/templates/base.html
index eff0f98..9c675a7 100644
--- a/admin/dashboard/templates/base.html
+++ b/admin/dashboard/templates/base.html
@@ -143,6 +143,11 @@
运行日志
+
+
+
+ 通讯录管理
+
@@ -204,7 +209,8 @@
'5': '/errors',
'6': '/robot_management',
'7': '/messages',
- '9': '/wx_logs'
+ '9': '/wx_logs',
+ '10': '/contacts'
};
// 如果当前不在对应页面，则跳转
diff --git a/utils/wechat/contact_manager.py b/utils/wechat/contact_manager.py
index 7994e61..7f7d2c1 100644
--- a/utils/wechat/contact_manager.py
+++ b/utils/wechat/contact_manager.py
@@ -3,13 +3,15 @@
联系人管理器 - 提供全局访问联系人信息的单例类
"""
import logging
-from typing import Dict, Optional
+from typing import Dict, Optional, List, Tuple
class ContactManager:
"""联系人管理器单例类"""
_instance = None
_contacts: Dict[str, str] = {}
+ _group_contacts: Dict[str, str] = {} # 群组联系人
+ _personal_contacts: Dict[str, str] = {} # 个人联系人
_initialized = False
_logger = logging.getLogger("ContactManager")
@@ -39,6 +41,22 @@ class ContactManager:
"""
self._contacts = contacts
self._logger.info(f"联系人信息已更新，共 {len(contacts)} 个联系人")
+ # 分类联系人
+ self._classify_contacts()
+
+ def _classify_contacts(self) -> None:
+ """将联系人分类为群组和个人联系人"""
+ self._group_contacts = {}
+ self._personal_contacts = {}
+
+ for wxid, nickname in self._contacts.items():
+ # 微信群ID通常以@@开头
+ if wxid.startswith('@@'):
+ self._group_contacts[wxid] = nickname
+ else:
+ self._personal_contacts[wxid] = nickname
+
+ self._logger.info(f"联系人分类完成: {len(self._group_contacts)} 个群组, {len(self._personal_contacts)} 个个人联系人")
def get_contacts(self) -> Dict[str, str]:
"""获取所有联系人
@@ -48,6 +66,22 @@ class ContactManager:
"""
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_nickname(self, wxid: str) -> str:
"""根据微信ID获取昵称
@@ -67,6 +101,11 @@ class ContactManager:
nickname: 昵称
"""
self._contacts[wxid] = nickname
+ # 更新分类
+ if wxid.endswith('@chatroom'):
+ self._group_contacts[wxid] = nickname
+ else:
+ self._personal_contacts[wxid] = nickname
self._logger.debug(f"已更新联系人: {wxid} -> {nickname}")
def refresh_contacts(self, new_contacts: Dict[str, str]) -> None:
@@ -76,4 +115,14 @@ class ContactManager:
new_contacts: 新的联系人字典
"""
self._contacts = new_contacts
- self._logger.info(f"联系人信息已刷新，共 {len(new_contacts)} 个联系人")
\ No newline at end of file
+ self._logger.info(f"联系人信息已刷新，共 {len(new_contacts)} 个联系人")
+ # 重新分类联系人
+ self._classify_contacts()
+
+ def get_contact_statistics(self) -> Tuple[int, int, int]:
+ """获取联系人统计信息
+
+ Returns:
+ 包含总联系人数、群组数和个人联系人数的元组
+ """
+ return len(self._contacts), len(self._group_contacts), len(self._personal_contacts)
\ No newline at end of file