abot/admin/dashboard/blueprints/contacts.py

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()
_PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".."))
_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

    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

    if not md5 and image_path:
        filename = os.path.basename(_safe_text(image_path))
        stem = os.path.splitext(filename)[0]
        if re.fullmatch(r"[0-9a-fA-F]{16,64}", stem):
            md5 = stem.lower()

    if total_length <= 0 and image_path and image_path.startswith("/static/"):
        abs_path = os.path.join(_PROJECT_ROOT, image_path.lstrip("/").replace("/", os.sep))
        if os.path.isfile(abs_path):
            try:
                total_length = int(os.path.getsize(abs_path))
            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)
        records = server.message_storage.message_db.get_recent_emoji_assets(limit=limit)
        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 md5 or total_length <= 0:
                return jsonify({'success': False, 'message': '缺少表情 md5 或长度'})
            send_message_in_thread(server.client.send_emoji_message, wxid, md5, total_length)
            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