Mofox-Core/plugins/napcat_adapter_plugin/plugin.py

import asyncio
import json
import inspect
import websockets as Server
from . import event_types, CONSTS, event_handlers

from typing import List

from src.plugin_system import BasePlugin, BaseEventHandler, register_plugin, EventType, ConfigField
from src.plugin_system.base.base_event import HandlerResult
from src.plugin_system.core.event_manager import event_manager

from src.common.logger import get_logger

from .src.message_chunker import chunker, reassembler
from .src.recv_handler.message_handler import message_handler
from .src.recv_handler.meta_event_handler import meta_event_handler
from .src.recv_handler.notice_handler import notice_handler
from .src.recv_handler.message_sending import message_send_instance
from .src.send_handler import send_handler
from .src.config import global_config
from .src.config.features_config import features_manager
from .src.config.migrate_features import auto_migrate_features
from .src.mmc_com_layer import mmc_start_com, router, mmc_stop_com
from .src.response_pool import put_response, check_timeout_response
from .src.websocket_manager import websocket_manager

logger = get_logger("napcat_adapter")

message_queue = asyncio.Queue()


def get_classes_in_module(module):
    classes = []
    for name, member in inspect.getmembers(module):
        if inspect.isclass(member):
            classes.append(member)
    return classes


async def message_recv(server_connection: Server.ServerConnection):
    await message_handler.set_server_connection(server_connection)
    asyncio.create_task(notice_handler.set_server_connection(server_connection))
    await send_handler.set_server_connection(server_connection)
    async for raw_message in server_connection:
        logger.debug(f"{raw_message[:1500]}..." if (len(raw_message) > 1500) else raw_message)
        decoded_raw_message: dict = json.loads(raw_message)
        try:
            # 首先尝试解析原始消息
            decoded_raw_message: dict = json.loads(raw_message)

            # 检查是否是切片消息 (来自 MMC)
            if chunker.is_chunk_message(decoded_raw_message):
                logger.debug("接收到切片消息，尝试重组")
                # 尝试重组消息
                reassembled_message = await reassembler.add_chunk(decoded_raw_message)
                if reassembled_message:
                    # 重组完成，处理完整消息
                    logger.debug("消息重组完成，处理完整消息")
                    decoded_raw_message = reassembled_message
                else:
                    # 切片尚未完整，继续等待更多切片
                    logger.debug("等待更多切片...")
                    continue

            # 处理完整消息（可能是重组后的，也可能是原本就完整的）
            post_type = decoded_raw_message.get("post_type")
            if post_type in ["meta_event", "message", "notice"]:
                await message_queue.put(decoded_raw_message)
            elif post_type is None:
                await put_response(decoded_raw_message)

        except json.JSONDecodeError as e:
            logger.error(f"消息解析失败: {e}")
            logger.debug(f"原始消息: {raw_message[:500]}...")
        except Exception as e:
            logger.error(f"处理消息时出错: {e}")
            logger.debug(f"原始消息: {raw_message[:500]}...")


async def message_process():
    """消息处理主循环"""
    logger.info("消息处理器已启动")
    try:
        while True:
            try:
                # 使用超时等待，以便能够响应取消请求
                message = await asyncio.wait_for(message_queue.get(), timeout=1.0)

                post_type = message.get("post_type")
                if post_type == "message":
                    await message_handler.handle_raw_message(message)
                elif post_type == "meta_event":
                    await meta_event_handler.handle_meta_event(message)
                elif post_type == "notice":
                    await notice_handler.handle_notice(message)
                else:
                    logger.warning(f"未知的post_type: {post_type}")

                message_queue.task_done()
                await asyncio.sleep(0.05)

            except asyncio.TimeoutError:
                # 超时是正常的，继续循环
                continue
            except asyncio.CancelledError:
                logger.info("消息处理器收到取消信号")
                break
            except Exception as e:
                logger.error(f"处理消息时出错: {e}")
                # 即使出错也标记任务完成，避免队列阻塞
                try:
                    message_queue.task_done()
                except ValueError:
                    pass
                await asyncio.sleep(0.1)

    except asyncio.CancelledError:
        logger.info("消息处理器已停止")
        raise
    except Exception as e:
        logger.error(f"消息处理器异常: {e}")
        raise
    finally:
        logger.info("消息处理器正在清理...")
        # 清空剩余的队列项目
        try:
            while not message_queue.empty():
                try:
                    message_queue.get_nowait()
                    message_queue.task_done()
                except asyncio.QueueEmpty:
                    break
        except Exception as e:
            logger.debug(f"清理消息队列时出错: {e}")


async def napcat_server():
    """启动 Napcat WebSocket 连接（支持正向和反向连接）"""
    mode = global_config.napcat_server.mode
    logger.info(f"正在启动 adapter，连接模式: {mode}")

    try:
        await websocket_manager.start_connection(message_recv)
    except Exception as e:
        logger.error(f"启动 WebSocket 连接失败: {e}")
        raise


async def graceful_shutdown():
    """优雅关闭所有组件"""
    try:
        logger.info("正在关闭adapter...")

        # 停止消息重组器的清理任务
        try:
            await reassembler.stop_cleanup_task()
        except Exception as e:
            logger.warning(f"停止消息重组器清理任务时出错: {e}")

        # 停止功能管理器文件监控
        try:
            await features_manager.stop_file_watcher()
        except Exception as e:
            logger.warning(f"停止功能管理器文件监控时出错: {e}")

        # 关闭消息处理器（包括消息缓冲器）
        try:
            await message_handler.shutdown()
        except Exception as e:
            logger.warning(f"关闭消息处理器时出错: {e}")

        # 关闭 WebSocket 连接
        try:
            await websocket_manager.stop_connection()
        except Exception as e:
            logger.warning(f"关闭WebSocket连接时出错: {e}")

        # 关闭 MaiBot 连接
        try:
            await mmc_stop_com()
        except Exception as e:
            logger.warning(f"关闭MaiBot连接时出错: {e}")

        # 取消所有剩余任务
        current_task = asyncio.current_task()
        tasks = [t for t in asyncio.all_tasks() if t is not current_task and not t.done()]

        if tasks:
            logger.info(f"正在取消 {len(tasks)} 个剩余任务...")
            for task in tasks:
                task.cancel()

            # 等待任务取消完成，忽略 CancelledError
            try:
                await asyncio.wait_for(asyncio.gather(*tasks, return_exceptions=True), timeout=10)
            except asyncio.TimeoutError:
                logger.warning("部分任务取消超时")
            except Exception as e:
                logger.debug(f"任务取消过程中的异常（可忽略）: {e}")

        logger.info("Adapter已成功关闭")

    except Exception as e:
        logger.error(f"Adapter关闭中出现错误: {e}")
    finally:
        # 确保消息队列被清空
        try:
            while not message_queue.empty():
                try:
                    message_queue.get_nowait()
                    message_queue.task_done()
                except asyncio.QueueEmpty:
                    break
        except Exception:
            pass


class LauchNapcatAdapterHandler(BaseEventHandler):
    """自动启动Adapter"""

    handler_name: str = "launch_napcat_adapter_handler"
    handler_description: str = "自动启动napcat adapter"
    weight: int = 100
    intercept_message: bool = False
    init_subscribe = [EventType.ON_START]

    async def execute(self, kwargs):
        # 执行功能配置迁移（如果需要）
        logger.info("检查功能配置迁移...")
        auto_migrate_features()

        # 启动消息重组器的清理任务
        logger.info("启动消息重组器...")
        await reassembler.start_cleanup_task()

        # 初始化功能管理器
        logger.info("正在初始化功能管理器...")
        features_manager.load_config()
        await features_manager.start_file_watcher(check_interval=2.0)
        logger.info("功能管理器初始化完成")
        logger.info("开始启动Napcat Adapter")
        message_send_instance.maibot_router = router
        # 创建单独的异步任务，防止阻塞主线程
        asyncio.create_task(napcat_server())
        asyncio.create_task(mmc_start_com())
        asyncio.create_task(message_process())
        asyncio.create_task(check_timeout_response())


class StopNapcatAdapterHandler(BaseEventHandler):
    """关闭Adapter"""

    handler_name: str = "stop_napcat_adapter_handler"
    handler_description: str = "关闭napcat adapter"
    weight: int = 100
    intercept_message: bool = False
    init_subscribe = [EventType.ON_STOP]

    async def execute(self, kwargs):
        await graceful_shutdown()
        return


@register_plugin
class NapcatAdapterPlugin(BasePlugin):
    plugin_name = CONSTS.PLUGIN_NAME
    enable_plugin: bool = True
    dependencies: List[str] = []  # 插件依赖列表
    python_dependencies: List[str] = []  # Python包依赖列表
    config_file_name: str = "config.toml"  # 配置文件名

    # 配置节描述
    config_section_descriptions = {"plugin": "插件基本信息"}

    # 配置Schema定义
    config_schema: dict = {
        "plugin": {
            "name": ConfigField(type=str, default="napcat_adapter_plugin", description="插件名称"),
            "version": ConfigField(type=str, default="1.0.0", description="插件版本"),
            "enabled": ConfigField(type=bool, default=False, description="是否启用插件"),
        }
    }

    def register_events(self):
        # 注册事件
        for e in event_types.NapcatEvent.ON_RECEIVED:
            event_manager.register_event(e, allowed_triggers=[self.plugin_name])

        for e in event_types.NapcatEvent.ACCOUNT:
            event_manager.register_event(e, allowed_subscribers=[f"{e.value}_handler"])

        for e in event_types.NapcatEvent.GROUP:
            event_manager.register_event(e, allowed_subscribers=[f"{e.value}_handler"])

        for e in event_types.NapcatEvent.MESSAGE:
            event_manager.register_event(e, allowed_subscribers=[f"{e.value}_handler"])

    def get_plugin_components(self):
        self.register_events()

        components = []
        components.append((LauchNapcatAdapterHandler.get_handler_info(), LauchNapcatAdapterHandler))
        components.append((StopNapcatAdapterHandler.get_handler_info(), StopNapcatAdapterHandler))
        for handler in get_classes_in_module(event_handlers):
            if issubclass(handler, BaseEventHandler):
                components.append((handler.get_handler_info(), handler))
        return components