#!/usr/bin/env python3 # -*- coding: utf-8 -*- import logging from typing import List, Callable, Optional from functools import wraps logger = logging.getLogger(__name__) def plugin_command(name: Optional[str] = None, description: Optional[str] = None, permissions: Optional[List[str]] = None): """ 插件命令装饰器 用法: @plugin_command(name="mycmd", description="我的命令", permissions=["read"]) async def my_command_handler(self, *args): return "命令执行结果" 或者简化版: @plugin_command() async def mycmd(self, *args): '''我的命令描述''' return "命令执行结果" """ def decorator(func: Callable): # 设置命令属性 func._is_plugin_command = True func._command_name = name or func.__name__ # 优先使用装饰器参数,其次使用文档字符串,最后使用默认描述 if description: func._command_description = description elif func.__doc__: # 提取文档字符串的第一行作为描述 doc_lines = [line.strip() for line in func.__doc__.split('\n') if line.strip()] func._command_description = doc_lines[0] if doc_lines else f"命令: {func.__name__}" else: func._command_description = f"命令: {func.__name__}" func._command_permissions = permissions or [] @wraps(func) async def wrapper(self, *args, **kwargs): """包装器确保返回字符串结果并处理异常""" try: logger.debug(f"执行插件命令: {func._command_name}, 参数: {args}") # 调用原始方法 result = await func(self, *args, **kwargs) # 确保返回字符串 if result is None: return "✅ 命令执行完成" elif not isinstance(result, str): return str(result) else: return result except Exception as e: logger.error(f"插件命令执行失败 {func._command_name}: {str(e)}", exc_info=True) return f"❌ 命令执行错误: {str(e)}" return wrapper return decorator def command(name: Optional[str] = None, description: Optional[str] = None): """简化版命令装饰器""" return plugin_command(name=name, description=description) # 同步命令装饰器(不推荐,但提供兼容性) def sync_plugin_command(name: Optional[str] = None, description: Optional[str] = None, permissions: Optional[List[str]] = None): """同步插件命令装饰器""" def decorator(func: Callable): func._is_plugin_command = True func._command_name = name or func.__name__ if description: func._command_description = description elif func.__doc__: doc_lines = [line.strip() for line in func.__doc__.split('\n') if line.strip()] func._command_description = doc_lines[0] if doc_lines else f"命令: {func.__name__}" else: func._command_description = f"命令: {func.__name__}" func._command_permissions = permissions or [] @wraps(func) def wrapper(self, *args, **kwargs): """同步命令包装器""" try: logger.debug(f"执行同步插件命令: {func._command_name}, 参数: {args}") result = func(self, *args, **kwargs) if result is None: return "✅ 命令执行完成" elif not isinstance(result, str): return str(result) else: return result except Exception as e: logger.error(f"同步插件命令执行失败 {func._command_name}: {str(e)}", exc_info=True) return f"❌ 命令执行错误: {str(e)}" return wrapper return decorator