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-03 10:50:53 +08:00
parent 66effdfed7
commit c574989dd4
3 changed files with 142 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
admin/dashboard/server.py
View File

@@ -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

8
admin/dashboard/templates/base.html
View File

@@ -143,6 +143,11 @@
<i class="el-icon-document"></i>
<span slot="title">运行日志</span>
</el-menu-item>
<!-- 在导航菜单中添加通讯录管理入口 -->
<el-menu-item index="10">
<i class="el-icon-notebook-1"></i>
<span slot="title">通讯录管理</span>
</el-menu-item>
</el-menu>
</div>
@@ -204,7 +209,8 @@
'5': '/errors',
'6': '/robot_management',
'7': '/messages',
'9': '/wx_logs'
'9': '/wx_logs',
'10': '/contacts'
};
// 如果当前不在对应页面，则跳转

53
utils/wechat/contact_manager.py
View File

@@ -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)} 个联系人")
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)