添加 qqlinker_framework 插件

qqlinker_framework/__init__.py

@@ -0,0 +1,70 @@
+# __init__.py
+"""云链群服互通框架 - ToolDelta 插件入口"""
+import asyncio
+import threading
+from tooldelta import Plugin, plugin_entry, ToolDelta
+from .core.host import FrameworkHost
+from .adapters.tooldelta_adapter import ToolDeltaAdapter
+
+
+class QQLinkerFrameworkPlugin(Plugin):
+    """ToolDelta 插件主类，负责启动框架主机及依赖检查。"""
+
+    name = "群服互通框架"
+    version = (1, 0, 0)
+    author = "小石潭记qwq"
+    description = "模块化群服互通框架"
+
+    def __init__(self, frame: ToolDelta):
+        """初始化插件，注册预加载事件。"""
+        super().__init__(frame)
+        self.ListenPreload(self.on_preload)
+        self._framework_thread = None
+        self._host = None
+        self._loop = None
+
+    def on_preload(self):
+        """预加载事件处理：创建适配器、启动后台异步线程。"""
+        data_dir = str(self.data_path)
+
+        adapter = ToolDeltaAdapter(self)
+        self._host = FrameworkHost(adapter, data_path=data_dir)
+
+        pkg_mgr = self._host.package_mgr
+        pkg_mgr.register_requirements({
+            "websocket-client": "websocket",
+            "aiohttp": "aiohttp",
+            "cachetools": "cachetools",
+            "redis": "redis",
+        })
+
+        self._host.register_modules_from_package("qqlinker_framework.modules")
+
+        self._framework_thread = threading.Thread(
+            target=self._run_framework, daemon=True
+        )
+        self._framework_thread.start()
+
+    def _run_framework(self):
+        """在独立线程中创建事件循环并运行框架主机。"""
+        self._loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(self._loop)
+        try:
+            self._loop.run_until_complete(self._host.start())
+            self._loop.run_forever()
+        except Exception:
+            import logging
+            logging.getLogger(__name__).exception("框架运行异常")
+        finally:
+            self._loop.close()
+
+    def on_def(self):
+        """插件卸载时执行，优雅停止框架。"""
+        if self._loop and self._host:
+            asyncio.run_coroutine_threadsafe(self._host.stop(), self._loop)
+            self._loop.call_soon_threadsafe(self._loop.stop)
+        if self._framework_thread and self._framework_thread.is_alive():
+            self._framework_thread.join(timeout=5)
+
+
+entry = plugin_entry(QQLinkerFrameworkPlugin)

qqlinker_framework/adapters/__init__.py

@@ -0,0 +1 @@
+# adapters/__init__.py

qqlinker_framework/adapters/base.py

@@ -0,0 +1,90 @@
+# adapters/base.py
+"""平台适配器抽象接口"""
+from abc import ABC, abstractmethod
+from typing import Callable, List, Optional, Any, Dict
+
+
+class IFrameworkAdapter(ABC):
+    """平台适配器抽象基类，定义所有需要实现的方法。"""
+
+    @abstractmethod
+    def send_game_command(self, cmd: str) -> None:
+        """发送游戏指令。"""
+
+    @abstractmethod
+    def send_game_message(self, target: str, text: str) -> None:
+        """向游戏内目标发送消息。"""
+
+    @abstractmethod
+    def get_online_players(self) -> List[str]:
+        """获取当前在线玩家列表（纯名字列表）。"""
+
+    @abstractmethod
+    def send_group_msg(self, group_id: int, message: str) -> bool:
+        """发送群聊消息。"""
+
+    @abstractmethod
+    def send_private_msg(self, user_id: int, message: str) -> bool:
+        """发送私聊消息。"""
+
+    @abstractmethod
+    def listen_game_chat(
+        self, handler: Callable[[str, str], None]
+    ) -> None:
+        """注册游戏聊天监听。"""
+
+    @abstractmethod
+    def listen_group_message(
+        self, handler: Callable[[Dict[str, Any]], None]
+    ) -> None:
+        """注册群消息监听。"""
+
+    @abstractmethod
+    def listen_player_join(
+        self, handler: Callable[[str], None]
+    ) -> None:
+        """注册玩家加入事件监听。"""
+
+    @abstractmethod
+    def listen_player_leave(
+        self, handler: Callable[[str], None]
+    ) -> None:
+        """注册玩家离开事件监听。"""
+
+    @abstractmethod
+    def register_console_command(
+        self,
+        triggers: List[str],
+        hint: str,
+        usage: str,
+        func: Callable,
+    ) -> None:
+        """注册控制台命令。"""
+
+    @abstractmethod
+    def get_plugin_api(self, name: str) -> Optional[Any]:
+        """获取其他插件的 API 实例。"""
+
+    @abstractmethod
+    def is_user_admin(self, user_id: int, config_mgr) -> bool:
+        """检查用户是否为平台管理员。"""
+
+    @abstractmethod
+    def send_game_command_with_resp(
+        self, cmd: str, timeout: float = 5.0
+    ) -> Optional[str]:
+        """发送游戏指令并等待响应文本，超时返回 None。"""
+
+    @abstractmethod
+    def send_game_command_full(
+        self, cmd: str, timeout: float = 5.0
+    ) -> Optional[Dict[str, Any]]:
+        """发送游戏指令并返回完整响应。
+
+        Returns:
+            None 表示异常或超时，否则返回字典：
+            {
+                "success_count": int,
+                "output": [{"message": str, "parameters": list}, ...]
+            }
+        """

qqlinker_framework/adapters/tooldelta_adapter.py

@@ -0,0 +1,209 @@
+# adapters/tooldelta_adapter.py
+"""ToolDelta 平台适配器实现"""
+import logging
+from typing import Callable, Dict, Any, List, Optional
+from tooldelta import Plugin, Player, Chat
+from .base import IFrameworkAdapter
+from services.ws_client import WsClient
+
+
+class ToolDeltaAdapter(IFrameworkAdapter):
+    """基于 ToolDelta 的平台适配器，封装游戏控制、事件监听和 WebSocket 通信。"""
+
+    def __init__(self, plugin_instance: Plugin):
+        self.plugin = plugin_instance
+        self.game_ctrl = plugin_instance.game_ctrl
+        self._config_mgr = None
+
+        self.plugin.ListenChat(self._on_game_chat)
+        self.plugin.ListenPlayerJoin(self._on_player_join)
+        self.plugin.ListenPlayerLeave(self._on_player_leave)
+
+        self._chat_handlers: list[Callable] = []
+        self._player_join_handlers: list[Callable] = []
+        self._player_leave_handlers: list[Callable] = []
+        self._group_message_handlers: list[Callable] = []
+
+        self._ws_client: Optional[WsClient] = None
+        self.event_bus = None
+        self.main_loop = None
+
+    def set_ws_client(self, ws_client: WsClient):
+        """设置 WebSocket 客户端实例。"""
+        self._ws_client = ws_client
+
+    def set_config_mgr(self, config_mgr):
+        """设置配置管理器。"""
+        self._config_mgr = config_mgr
+
+    def send_game_command(self, cmd: str):
+        """发送游戏指令。"""
+        try:
+            self.game_ctrl.sendcmd(cmd)
+        except Exception as e:
+            logging.getLogger(__name__).warning(
+                "游戏命令发送失败: %s, 错误: %s", cmd, e
+            )
+
+    def send_game_message(self, target: str, text: str):
+        """向游戏内目标发送消息。"""
+        try:
+            self.game_ctrl.say_to(target, text)
+        except Exception as e:
+            logging.getLogger(__name__).warning(
+                "游戏消息发送失败, 目标: %s, 错误: %s", target, e
+            )
+
+    def get_online_players(self) -> List[str]:
+        """获取在线玩家列表，自动兼容 ToolDelta 返回的 list 或 dict。"""
+        try:
+            raw = self.game_ctrl.allplayers
+            if isinstance(raw, dict):
+                return list(raw.keys())
+            if isinstance(raw, (list, tuple)):
+                return list(raw)
+            logging.getLogger(__name__).warning(
+                "allplayers 返回了未知类型: %s", type(raw).__name__
+            )
+            return []
+        except Exception as e:
+            logging.getLogger(__name__).error(
+                "获取在线玩家列表异常: %s", e
+            )
+            return []
+
+    def send_group_msg(self, group_id: int, message: str) -> bool:
+        """发送群消息。"""
+        if not self._ws_client:
+            logging.getLogger(__name__).warning("WebSocket 客户端不可用")
+            return False
+        if not self._ws_client.available:
+            logging.getLogger(__name__).warning("WebSocket 未连接")
+            return False
+        return self._ws_client.send_group_msg(group_id, message)
+
+    def send_private_msg(self, user_id: int, message: str) -> bool:
+        """发送私聊消息。"""
+        if not self._ws_client:
+            logging.getLogger(__name__).warning("WebSocket 客户端不可用")
+            return False
+        if not self._ws_client.available:
+            logging.getLogger(__name__).warning("WebSocket 未连接")
+            return False
+        return self._ws_client.send_private_msg(user_id, message)
+
+    def _on_game_chat(self, chat: Chat):
+        """分发游戏聊天事件给所有处理器。"""
+        for h in self._chat_handlers:
+            try:
+                h(chat.player.name, chat.msg)
+            except Exception as e:
+                logging.getLogger(__name__).error("游戏聊天处理器异常: %s", e)
+
+    def _on_player_join(self, player: Player):
+        """分发玩家加入事件。"""
+        for h in self._player_join_handlers:
+            try:
+                h(player.name)
+            except Exception as e:
+                logging.getLogger(__name__).error("玩家加入处理器异常: %s", e)
+
+    def _on_player_leave(self, player: Player):
+        """分发玩家离开事件。"""
+        for h in self._player_leave_handlers:
+            try:
+                h(player.name)
+            except Exception as e:
+                logging.getLogger(__name__).error("玩家离开处理器异常: %s", e)
+
+    def listen_game_chat(self, handler: Callable[[str, str], None]):
+        """注册游戏聊天处理器。"""
+        self._chat_handlers.append(handler)
+
+    def listen_player_join(self, handler: Callable[[str], None]):
+        """注册玩家加入处理器。"""
+        self._player_join_handlers.append(handler)
+
+    def listen_player_leave(self, handler: Callable[[str], None]):
+        """注册玩家离开处理器。"""
+        self._player_leave_handlers.append(handler)
+
+    def listen_group_message(
+        self, handler: Callable[[Dict[str, Any]], None]
+    ):
+        """注册原始群消息处理器。"""
+        self._group_message_handlers.append(handler)
+
+    def trigger_raw_group_handlers(self, data: dict):
+        """触发所有原始群消息处理器。"""
+        for handler in self._group_message_handlers:
+            try:
+                handler(data)
+            except Exception as e:
+                logging.getLogger(__name__).error("原始消息处理器异常: %s", e)
+
+    def register_console_command(
+        self,
+        triggers: List[str],
+        hint: str,
+        usage: str,
+        func: Callable,
+    ):
+        """注册控制台命令。"""
+        self.plugin.frame.add_console_cmd_trigger(triggers, hint, usage, func)
+
+    def get_plugin_api(self, name: str) -> Optional[Any]:
+        """获取其他插件的 API 实例。"""
+        return self.plugin.GetPluginAPI(name)
+
+    def is_user_admin(self, user_id: int, config_mgr=None) -> bool:
+        """检查用户是否为管理员。"""
+        cfg = config_mgr or self._config_mgr
+        if cfg is None:
+            return False
+        admin_list = cfg.get("管理员.管理员QQ", [])
+        try:
+            return user_id in [int(q) for q in admin_list]
+        except (TypeError, ValueError):
+            return False
+
+    def send_game_command_with_resp(
+        self, cmd: str, timeout: float = 5.0
+    ) -> Optional[str]:
+        """发送游戏指令并返回响应文本。"""
+        try:
+            resp = self.game_ctrl.sendwscmd_with_resp(cmd, timeout)
+            if resp and resp.OutputMessages:
+                lines = []
+                for msg in resp.OutputMessages:
+                    if hasattr(msg, "Message"):
+                        lines.append(msg.Message)
+                    else:
+                        lines.append(str(msg))
+                return "\n".join(lines)
+            return ""
+        except Exception as e:
+            logging.getLogger(__name__).error("同步指令执行失败: %s", e)
+            return None
+
+    def send_game_command_full(
+        self, cmd: str, timeout: float = 5.0
+    ) -> Optional[Dict[str, Any]]:
+        """发送游戏指令并返回完整响应（包括 Parameters）。"""
+        try:
+            resp = self.game_ctrl.sendwscmd_with_resp(cmd, timeout)
+            if resp is None:
+                return None
+            output = []
+            for msg in resp.OutputMessages:
+                output.append({
+                    "message": getattr(msg, "Message", ""),
+                    "parameters": getattr(msg, "Parameters", []),
+                })
+            return {
+                "success_count": resp.SuccessCount,
+                "output": output,
+            }
+        except Exception as e:
+            logging.getLogger(__name__).error("完整指令执行失败: %s", e)
+            return None

qqlinker_framework/core/__init__.py

@@ -0,0 +1 @@
+# core/__init__.py