#!/usr/bin/env python3 # -*- coding: utf-8 -*- import logging import asyncio import inspect from typing import Dict, List, Callable, Any import json logger = logging.getLogger(__name__) class PluginNetworkBridge: """插件网络桥接 - 简化插件的网络交互""" def __init__(self, plugin_name: str, internet_service, plugin_bridge): self.plugin_name = plugin_name self.internet_service = internet_service self.plugin_bridge = plugin_bridge self.registered_routes: List[Dict] = [] self.websocket_handlers: List[Dict] = [] logger.debug(f"插件网络桥接初始化: {plugin_name}") def is_network_available(self): """检查网络服务是否可用""" return self.internet_service is not None and hasattr(self.internet_service, 'register_plugin_route') async def register_command_routes(self, plugin_instance, require_auth: bool = True): """自动扫描插件 @plugin_command/cmd_* 方法并注册 REST 端点 每个命令 → POST /api/plugin/{cmd_name} 参数以 JSON body 传入: {"args": [...]} 返回: {"ok": true, "result": "..."} 或 {"ok": false, "error": "..."} """ try: import inspect registered = 0 for attr_name in dir(plugin_instance): if attr_name.startswith("__"): continue method = getattr(plugin_instance, attr_name, None) if not callable(method): continue # 识别 @plugin_command 装饰或 cmd_ 前缀 cmd_name = None if hasattr(method, "_is_plugin_command"): cmd_name = getattr(method, "_command_name", attr_name[4:] if attr_name.startswith("cmd_") else attr_name) elif attr_name.startswith("cmd_"): cmd_name = attr_name[4:] if not cmd_name: continue route_path = f"/api/plugin/{cmd_name}" # 创建闭包捕获 method 和 cmd_name async def _make_handler(_method=method, _cmd_name=cmd_name): from aiohttp import web async def _handler(request): try: body = {} try: body = await request.json() except Exception: pass args = body.get("args", []) if isinstance(args, str): args = [args] kwargs = body.get("kwargs", {}) result = _method(*args, **kwargs) if asyncio.iscoroutine(result): result = await result return web.json_response({"ok": True, "result": str(result)}) except Exception as e: logger.error(f"命令 {_cmd_name} REST 调用失败: {e}") return web.json_response({"ok": False, "error": str(e)}, status=500) return _handler await self.register_http_route( route_path, await _make_handler(), methods=["POST"], require_auth=require_auth, ) logger.debug(f" 自动暴露 REST: POST {route_path}") registered += 1 if registered: logger.info(f"插件 {self.plugin_name} 自动暴露 {registered} 个命令 REST 端点") except Exception as e: logger.warning(f"自动注册命令路由失败 {self.plugin_name}: {e}") async def register_http_route(self, route_path: str, handler: Callable, methods: List[str] = ["GET"], require_auth: bool = True): """注册HTTP路由""" try: if not self.is_network_available(): logger.warning("网络服务不可用,跳过HTTP路由注册") return await self.internet_service.register_plugin_route( self.plugin_name, route_path, handler, methods, require_auth ) self.registered_routes.append({ 'type': 'http', 'path': route_path, 'methods': methods, 'require_auth': require_auth }) logger.debug(f"插件 {self.plugin_name} 注册HTTP路由: {route_path}") except Exception as e: logger.error(f"注册HTTP路由时出错: {str(e)}") # 不抛出异常,让插件继续运行 async def register_websocket(self, ws_path: str, handler: Callable, require_auth: bool = True): """注册WebSocket处理器""" try: if not self.is_network_available(): logger.warning("网络服务不可用,跳过WebSocket注册") return await self.internet_service.register_plugin_websocket( self.plugin_name, ws_path, handler, require_auth ) self.websocket_handlers.append({ 'path': ws_path, 'require_auth': require_auth }) logger.debug(f"插件 {self.plugin_name} 注册WebSocket: {ws_path}") except Exception as e: logger.error(f"注册WebSocket时出错: {str(e)}") # 不抛出异常,让插件继续运行 async def broadcast_websocket(self, message: Dict): """向插件的所有WebSocket连接广播消息""" try: if not self.internet_service: logger.warning("网络服务不可用,无法广播消息") return await self.internet_service.broadcast_to_websockets(self.plugin_name, message) logger.debug(f"插件 {self.plugin_name} WebSocket广播: {len(message)} 字节") except Exception as e: logger.error(f"WebSocket广播时出错: {str(e)}") async def send_data_to_client(self, client_id: str, message: Dict): """向特定客户端发送数据""" try: # 这里可以实现更精确的客户端消息发送 # 目前先使用广播 message['target_client'] = client_id await self.broadcast_websocket(message) except Exception as e: logger.error(f"发送数据到客户端时出错: {str(e)}") def get_network_info(self) -> Dict[str, Any]: """获取网络配置信息""" if not self.internet_service: return { 'plugin_name': self.plugin_name, 'registered_routes': [], 'websocket_handlers': [], 'base_url': '网络服务不可用' } return { 'plugin_name': self.plugin_name, 'registered_routes': self.registered_routes, 'websocket_handlers': self.websocket_handlers, 'base_url': f"http://{self.internet_service.http_host}:{self.internet_service.http_port}/{self.plugin_name}" } async def setup_data_transfer(self, data_handler: Callable): """设置跨端数据传输""" try: # 订阅网络数据接收事件 self.plugin_bridge.subscribe_plugin( self.plugin_name, "network.data.receive", data_handler ) logger.debug(f"插件 {self.plugin_name} 设置跨端数据传输") except Exception as e: logger.error(f"设置数据传输时出错: {str(e)}")