Files
abot/admin/dashboard/blueprints/contacts.py
liuwei ee5c1ebadc 只用表情 XML 里的 md5 + len/totallen 作为发送参数。
不再用图片文件大小、文件名等回退逻辑去“猜参数”。
发送接口不再“先返回发送中”,而是直接调用 SendEmoji 并等待结果:
成功才返回 表情发送成功
失败直接返回具体错误(不会再假成功)
2026-04-15 11:36:22 +08:00

698 lines
26 KiB
Python

import asyncio
import os
import re
import threading
import xml.etree.ElementTree as ET
from concurrent.futures import ThreadPoolExecutor
from flask import Blueprint, render_template, jsonify, request, current_app
from .auth import login_required
from loguru import logger
# 创建联系人管理蓝图
contacts_bp = Blueprint('contacts', __name__, url_prefix='/contacts')
# 创建线程池
message_thread_pool = ThreadPoolExecutor(max_workers=10, thread_name_prefix="message_sender_")
# 创建共享的事件循环
shared_loop = None
loop_lock = threading.Lock()
_EMOJI_MD5_RE = re.compile(r'md5\s*=\s*[\"\']([0-9a-fA-F]{16,64})[\"\']', re.IGNORECASE)
_EMOJI_TOTALLEN_RE = re.compile(r'(?:totallen|total_len|len)\s*=\s*[\"\'](\d+)[\"\']', re.IGNORECASE)
def get_or_create_loop():
"""获取或创建共享的事件循环"""
global shared_loop
with loop_lock:
if shared_loop is None:
shared_loop = asyncio.new_event_loop()
# 在新线程中运行事件循环
def run_loop():
asyncio.set_event_loop(shared_loop)
shared_loop.run_forever()
loop_thread = threading.Thread(target=run_loop, daemon=True)
loop_thread.start()
return shared_loop
def send_message_in_thread(func, *args, **kwargs):
"""使用共享事件循环发送消息"""
def run():
try:
loop = get_or_create_loop()
# 创建异步任务
async def send():
try:
await func(*args, **kwargs)
except Exception as e:
logger.error(f"发送消息失败: {e}")
# 在共享事件循环中运行任务
future = asyncio.run_coroutine_threadsafe(send(), loop)
# 等待任务完成,设置超时时间
future.result(timeout=30)
except Exception as e:
logger.error(f"消息发送任务执行失败: {e}")
# 使用线程池提交任务
message_thread_pool.submit(run)
def run_member_context_refresh_in_thread(func, *args, **kwargs):
"""在线程池中异步刷新成员交互摘要,避免阻塞请求线程"""
def run():
try:
func(*args, **kwargs)
except Exception as e:
logger.error(f"成员交互摘要后台刷新失败: {e}")
message_thread_pool.submit(run)
def _safe_text(value):
return "" if value is None else str(value)
def _parse_app_message_payload(content: str):
payload = {
"title": "",
"description": "",
"url": "",
"app_type": ""
}
if not content:
return payload
text = _safe_text(content).strip()
if not text.startswith("<"):
payload["description"] = text
return payload
try:
root = ET.fromstring(text)
payload["title"] = _safe_text(root.findtext('.//title')).strip()
payload["description"] = _safe_text(root.findtext('.//des')).strip()
payload["url"] = _safe_text(root.findtext('.//url')).strip()
payload["app_type"] = _safe_text(root.findtext('.//type')).strip()
except Exception:
payload["description"] = text
return payload
def _parse_sys_message_payload(content: str):
payload = {
"sysmsg_type": "",
"summary": "",
"replace_msg": "",
"session": "",
"msgid": "",
"newmsgid": ""
}
text = _safe_text(content).strip()
if not text.startswith("<sysmsg"):
payload["summary"] = text
return payload
try:
root = ET.fromstring(text)
except Exception:
payload["summary"] = text
return payload
payload["sysmsg_type"] = _safe_text(root.attrib.get("type")).strip()
if payload["sysmsg_type"] == "revokemsg":
revoke_node = root.find("revokemsg")
if revoke_node is not None:
payload["session"] = _safe_text(revoke_node.findtext("session")).strip()
payload["msgid"] = _safe_text(revoke_node.findtext("msgid")).strip()
payload["newmsgid"] = _safe_text(revoke_node.findtext("newmsgid")).strip()
payload["replace_msg"] = _safe_text(revoke_node.findtext("replacemsg")).strip()
payload["summary"] = payload["replace_msg"] or "撤回了一条消息"
return payload
payload["summary"] = _safe_text(root.findtext(".//content")).strip() or text
return payload
def _compact_media_caption(content: str, fallback: str) -> str:
text = _safe_text(content).strip()
if not text:
return fallback
if text.startswith("<"):
return fallback
return text
def _extract_emoji_meta(attachment_url: str, image_path: str):
text = _safe_text(attachment_url)
md5 = ""
total_length = 0
# 只接受 XML 中的参数,不做文件名或文件大小回退,避免参数污染。
if not text.startswith("<"):
return "", 0
try:
root = ET.fromstring(text)
emoji_node = root.find(".//emoji")
if emoji_node is None:
return "", 0
md5 = _safe_text(emoji_node.attrib.get("md5", "")).strip().lower()
for key in ("totallen", "total_len", "totalLen", "len"):
value = _safe_text(emoji_node.attrib.get(key, "")).strip()
if value.isdigit():
total_length = int(value)
break
except Exception:
md5_match = _EMOJI_MD5_RE.search(text)
if md5_match:
md5 = md5_match.group(1).lower()
len_match = _EMOJI_TOTALLEN_RE.search(text)
if len_match:
try:
total_length = int(len_match.group(1))
except Exception:
total_length = 0
return md5, total_length
def _normalize_recent_message(server, raw_message: dict, chat_type: str, target_wxid: str):
sender = _safe_text(raw_message.get("sender")).strip()
message_type = str(raw_message.get("message_type", ""))
content = _safe_text(raw_message.get("content")).strip()
image_path = _safe_text(raw_message.get("image_path")).strip()
attachment_url = _safe_text(raw_message.get("attachment_url")).strip()
message_thumb = _safe_text(raw_message.get("message_thumb")).strip()
self_wxid = _safe_text(getattr(server.robot, "wxid", "") or getattr(server.client, "wxid", "")).strip()
sender_name = sender or "未知发送者"
if chat_type == "group":
sender_name = server.contact_manager.get_group_name(target_wxid, sender) or sender_name
elif sender:
sender_name = server.contact_manager.get_nickname(sender) or sender_name
display_type = "text"
display_content = content
media_url = image_path or attachment_url or message_thumb
link_payload = None
if message_type == "3":
display_type = "image"
display_content = _compact_media_caption(content, "[图片]")
elif message_type in {"47", "1048625", "1090519089"}:
display_type = "image" if media_url else "text"
display_content = _compact_media_caption(content, "[表情]")
elif message_type == "34":
display_type = "voice"
display_content = _compact_media_caption(content, "[语音]")
elif message_type == "43":
display_type = "video"
display_content = _compact_media_caption(content, "[视频]")
elif message_type == "49":
app_payload = _parse_app_message_payload(content)
if app_payload.get("url") or app_payload.get("title"):
display_type = "link"
link_payload = app_payload
display_content = app_payload.get("title") or app_payload.get("description") or "[链接]"
else:
display_type = "text"
display_content = app_payload.get("description") or content or "[应用消息]"
elif message_type in {"10000", "10002"}:
display_type = "system"
sys_payload = _parse_sys_message_payload(content)
display_content = sys_payload.get("summary") or content or "[系统消息]"
link_payload = sys_payload
return {
"timestamp": _safe_text(raw_message.get("timestamp")),
"sender": sender,
"sender_name": sender_name,
"content": content,
"message_type": message_type,
"display_type": display_type,
"display_content": display_content,
"image_path": image_path,
"attachment_url": attachment_url,
"media_url": media_url,
"message_thumb": message_thumb,
"message_id": raw_message.get("message_id"),
"link_payload": link_payload,
"is_self": bool(self_wxid and sender == self_wxid),
"sysmsg_type": (link_payload or {}).get("sysmsg_type", "") if display_type == "system" else "",
}
# 联系人管理页面
@contacts_bp.route('/')
@login_required
def contacts_management():
"""通讯录管理页面"""
return render_template('contacts_management.html')
# API路由
@contacts_bp.route('/api/all', methods=['GET'])
@login_required
def api_contacts_all():
"""获取所有联系人信息API"""
try:
server = current_app.dashboard_server
contacts = server.contact_manager.get_contacts()
return jsonify({
"success": True,
"data": {
"contacts": contacts
}
})
except Exception as e:
logger.error(f"获取所有联系人信息失败: {e}")
return jsonify({"success": False, "error": str(e)}), 500
@contacts_bp.route('/api/statistics', methods=['GET'])
@login_required
def api_contacts_statistics():
"""获取联系人统计信息API"""
try:
server = current_app.dashboard_server
# 使用新的联系人分类方法获取统计信息
total, groups, personal, public, official = server.contact_manager.get_contact_statistics()
return jsonify({
"success": True,
"data": {
"total": total,
"groups": groups,
"personal": personal,
"public": public,
"official": official
}
})
except Exception as e:
logger.error(f"获取联系人统计信息失败: {e}")
return jsonify({"success": False, "error": str(e)}), 500
@contacts_bp.route('/api/groups', methods=['GET'])
@login_required
def api_contacts_groups():
"""获取群组联系人信息API"""
try:
server = current_app.dashboard_server
group_contacts = server.contact_manager.get_group_contacts()
return jsonify({
"success": True,
"data": {
"groups": group_contacts
}
})
except Exception as e:
logger.error(f"获取群组联系人信息失败: {e}")
return jsonify({"success": False, "error": str(e)}), 500
@contacts_bp.route('/api/personal', methods=['GET'])
@login_required
def api_contacts_personal():
"""获取个人联系人信息API"""
try:
server = current_app.dashboard_server
personal_contacts = server.contact_manager.get_personal_contacts()
return jsonify({
"success": True,
"data": {
"personal": personal_contacts
}
})
except Exception as e:
logger.error(f"获取个人联系人信息失败: {e}")
return jsonify({"success": False, "error": str(e)}), 500
@contacts_bp.route('/api/official', methods=['GET'])
@login_required
def api_contacts_official():
"""获取公众号联系人信息API"""
try:
server = current_app.dashboard_server
official_accounts = server.contact_manager.get_official_accounts()
return jsonify({
"success": True,
"data": {
"official": official_accounts
}
})
except Exception as e:
logger.error(f"获取公众号联系人信息失败: {e}")
return jsonify({"success": False, "error": str(e)}), 500
@contacts_bp.route('/api/public', methods=['GET'])
@login_required
def api_contacts_public():
"""获取公共好友信息API"""
try:
server = current_app.dashboard_server
public_contacts = server.contact_manager.get_public_contacts()
return jsonify({
"success": True,
"data": {
"public": public_contacts
}
})
except Exception as e:
logger.error(f"获取公共好友信息失败: {e}")
return jsonify({"success": False, "error": str(e)}), 500
@contacts_bp.route('/api/head_images', methods=['GET'])
@login_required
def api_head_images():
"""获取联系人头像信息API"""
try:
server = current_app.dashboard_server
head_images = server.contact_manager.get_all_head_images()
return jsonify({
"success": True,
"data": {
"head_images": head_images
}
})
except Exception as e:
logger.error(f"获取联系人头像信息失败: {e}")
return jsonify({"success": False, "error": str(e)}), 500
@contacts_bp.route('/api/group_members/<roomid>', methods=['GET'])
@login_required
def api_group_members(roomid):
"""获取指定群的成员列表API
Args:
roomid: 群ID
"""
try:
server = current_app.dashboard_server
group_members = server.contact_db.get_chatroom_small_member_list(roomid)
context_enabled = bool(server.member_context_service) and server.member_context_service.is_group_enabled(roomid)
if context_enabled:
contexts = server.member_context_db.list_group_member_contexts(roomid)
context_map = {item.get("wxid"): item for item in contexts}
for member in group_members:
context = context_map.get(member.get("wxid"), {})
member["activity_level"] = context.get("activity_level", "")
member["response_style_hint"] = context.get("response_style_hint", "")
member["summary_text"] = context.get("summary_text", "")
member["last_profiled_at"] = context.get("last_profiled_at", "")
else:
for member in group_members:
member["activity_level"] = ""
member["response_style_hint"] = ""
member["summary_text"] = ""
member["last_profiled_at"] = ""
return jsonify({
"success": True,
"data": {
"members": group_members,
"member_context_enabled": context_enabled
}
})
except Exception as e:
logger.error(f"获取群成员列表失败: {e}")
return jsonify({"success": False, "error": str(e)}), 500
@contacts_bp.route('/api/group_member_context/<roomid>/<wxid>', methods=['GET'])
@login_required
def api_group_member_context(roomid, wxid):
"""获取群成员交互摘要"""
try:
server = current_app.dashboard_server
if not server.member_context_service:
return jsonify({"success": False, "error": "成员交互摘要插件未加载"}), 503
if not server.member_context_service.is_group_enabled(roomid):
return jsonify({"success": False, "error": "该群未启用成员交互摘要功能"}), 403
context = server.member_context_db.get_member_context(roomid, wxid)
return jsonify({
"success": True,
"data": {
"context": context,
"ready": bool(context)
}
})
except Exception as e:
logger.error(f"获取群成员交互摘要失败: {e}")
return jsonify({"success": False, "error": str(e)}), 500
@contacts_bp.route('/api/group_member_context/refresh', methods=['POST'])
@login_required
def api_refresh_group_member_context():
"""刷新群成员交互摘要"""
try:
server = current_app.dashboard_server
if not server.member_context_service:
return jsonify({"success": False, "error": "成员交互摘要插件未加载"}), 503
data = request.json or {}
roomid = data.get("roomid")
wxid = data.get("wxid")
if roomid and wxid:
if not server.member_context_service.is_group_enabled(roomid):
return jsonify({"success": False, "error": "该群未启用成员交互摘要功能"}), 403
run_member_context_refresh_in_thread(server.member_context_service.refresh_member_context, roomid, wxid)
return jsonify({"success": True, "message": "成员交互摘要刷新任务已提交"})
if roomid:
if not server.member_context_service.is_group_enabled(roomid):
return jsonify({"success": False, "error": "该群未启用成员交互摘要功能"}), 403
run_member_context_refresh_in_thread(server.member_context_service.refresh_group_contexts, roomid)
return jsonify({"success": True, "message": "本群成员交互摘要刷新任务已提交"})
run_member_context_refresh_in_thread(server.member_context_service.refresh_all_chatrooms)
return jsonify({"success": True, "message": "全量成员交互摘要刷新任务已提交"})
except Exception as e:
logger.error(f"刷新群成员交互摘要失败: {e}")
return jsonify({"success": False, "error": str(e)}), 500
@contacts_bp.route('/api/update', methods=['POST'])
@login_required
def api_contacts_update():
"""更新通讯录信息API"""
try:
server = current_app.dashboard_server
# 假设 contact_manager 有 update_contacts 方法用于同步通讯录
result = asyncio.run(server.robot.refresh_contacts_db())
if result:
return jsonify({"success": True, "message": "通讯录更新成功"})
else:
return jsonify({"success": False, "message": "通讯录更新失败"}), 500
except Exception as e:
logger.error(f"更新通讯录失败: {e}")
return jsonify({"success": False, "message": f"更新通讯录失败: {str(e)}"}), 500
@contacts_bp.route('/api/recent_messages', methods=['GET'])
@login_required
def api_recent_messages():
"""获取最近聊天消息"""
try:
server = current_app.dashboard_server
wxid = _safe_text(request.args.get("wxid")).strip()
chat_type = _safe_text(request.args.get("chat_type")).strip() or "personal"
limit = min(max(int(request.args.get("limit", 20)), 1), 50)
if not wxid:
return jsonify({"success": False, "message": "缺少聊天对象"}), 400
if chat_type == "group":
raw_messages = server.message_storage.get_recent_group_chat_messages(wxid, limit=limit)
history_tip = f"最近 {limit} 条群消息"
else:
raw_messages = server.message_storage.get_recent_personal_messages(wxid, limit=limit)
history_tip = f"最近 {limit} 条已归档消息(私聊历史可能不完整)"
messages = [
_normalize_recent_message(server, item, chat_type, wxid)
for item in raw_messages
]
return jsonify({
"success": True,
"data": {
"messages": messages,
"chat_type": chat_type,
"history_tip": history_tip
}
})
except Exception as e:
logger.exception(f"获取最近聊天消息失败: {e}")
return jsonify({"success": False, "message": str(e)}), 500
@contacts_bp.route('/api/emojis', methods=['GET'])
@login_required
def api_emoji_library():
"""获取已下载表情库(从历史消息聚合)。"""
try:
server = current_app.dashboard_server
limit = min(max(int(request.args.get("limit", 200)), 1), 500)
message_storage = getattr(server, "message_storage", None)
if not message_storage:
return jsonify({"success": False, "message": "消息存储未初始化"}), 503
if hasattr(message_storage, "get_recent_emoji_assets"):
records = message_storage.get_recent_emoji_assets(limit=limit)
elif hasattr(message_storage, "message_db") and hasattr(message_storage.message_db, "get_recent_emoji_assets"):
records = message_storage.message_db.get_recent_emoji_assets(limit=limit)
else:
logger.error("当前 message_storage 不支持 get_recent_emoji_assets")
return jsonify({"success": False, "message": "当前消息存储版本不支持表情库"}), 500
dedup = {}
for item in records:
image_path = _safe_text(item.get("image_path")).strip()
if not image_path:
continue
md5, total_length = _extract_emoji_meta(_safe_text(item.get("attachment_url")), image_path)
if not md5 or total_length <= 0:
continue
if md5 in dedup:
continue
dedup[md5] = {
"md5": md5,
"total_length": total_length,
"preview_url": image_path,
"timestamp": _safe_text(item.get("timestamp")),
"group_id": _safe_text(item.get("group_id")),
"message_id": _safe_text(item.get("message_id")),
}
emojis = list(dedup.values())
return jsonify({
"success": True,
"data": {
"emojis": emojis,
"count": len(emojis)
}
})
except Exception as e:
logger.exception(f"获取表情库失败: {e}")
return jsonify({"success": False, "message": str(e)}), 500
@contacts_bp.route('/api/send_message', methods=['POST'])
@login_required
def api_send_message():
"""发送消息API
支持的消息类型:
- text: 文本消息
- image: 图片消息
- voice: 语音消息
- video: 视频消息
- link: 链接消息
"""
try:
data = request.form if request.files else request.json
wxid = data.get('wxid')
msg_type = data.get('type')
content = data.get('content')
if not wxid or not msg_type:
return jsonify({'success': False, 'message': '缺少必要参数'})
# 获取机器人实例
server = current_app.dashboard_server
if not server or not server.client:
return jsonify({'success': False, 'message': '机器人未初始化'})
# 根据消息类型发送消息
if msg_type == 'text':
send_message_in_thread(server.client.send_text_message, wxid, content)
return jsonify({
'success': True,
'message': '消息发送中'
})
elif msg_type == 'image':
if 'file' not in request.files:
return jsonify({'success': False, 'message': '未上传文件'})
file = request.files['file']
send_message_in_thread(server.client.send_image_message, wxid, file.read())
return jsonify({
'success': True,
'message': '消息发送中'
})
elif msg_type == 'voice':
if 'file' not in request.files:
return jsonify({'success': False, 'message': '未上传文件'})
file = request.files['file']
if file.filename.endswith('.mp3'):
format_str = "mp3"
elif file.filename.endswith('.wav'):
format_str = "wav"
else:
return jsonify({
'success': False,
'message': '不支持的音频格式'
})
send_message_in_thread(server.client.send_voice_message, wxid, file.read(), format=format_str)
return jsonify({
'success': True,
'message': '消息发送中'
})
elif msg_type == 'video':
if 'file' not in request.files:
return jsonify({'success': False, 'message': '未上传文件'})
file = request.files['file']
send_message_in_thread(server.client.send_video_message, wxid, file.read())
return jsonify({
'success': True,
'message': '消息发送中'
})
elif msg_type == 'link':
url = content.get('url')
title = content.get('title', '')
description = content.get('description', '')
send_message_in_thread(server.client.send_link_message, wxid, url, title, description)
return jsonify({
'success': True,
'message': '消息发送中'
})
elif msg_type == 'emoji':
if not isinstance(content, dict):
return jsonify({'success': False, 'message': '表情参数格式错误'})
md5 = _safe_text(content.get('md5')).strip().lower()
total_length = int(content.get('total_length') or 0)
if not re.fullmatch(r"[0-9a-f]{16,64}", md5) or total_length <= 0:
return jsonify({'success': False, 'message': '缺少表情 md5 或长度'})
try:
loop = get_or_create_loop()
future = asyncio.run_coroutine_threadsafe(server.client.send_emoji_message(wxid, md5, total_length), loop)
future.result(timeout=20)
except Exception as e:
logger.error(f"发送表情失败 md5={md5} len={total_length} wxid={wxid}: {e}")
return jsonify({'success': False, 'message': f'表情发送失败: {str(e)}'}), 500
return jsonify({
'success': True,
'message': '表情发送成功'
})
else:
return jsonify({'success': False, 'message': '不支持的消息类型'})
except Exception as e:
logger.exception(f"发送消息失败: {e}")
return jsonify({'success': False, 'message': str(e)}), 500