Rename fmfuncs -> core (cleaner, semantic)

core/plugin_command_decorator.py

@@ -0,0 +1,111 @@
+#!/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