AskaEth
e6875f0b4b
- 13-service async plugin framework - Textual TUI with CLI fallback - Plugin hot-reload + permission system - Web management panel (aiohttp) - Bridge-based inter-module communication - 10 regression tests Fixes applied: - PBKDF2-SHA256 auth (was plain SHA256) - Auth bypass removed (was allow-all on fail) - Bare excepts replaced with logged errors - CatFramework/DreamSu -> SenSu naming unified - ServiceManager: health checks + startup_order - Env var credentials (SENSU_ADMIN_PASSWORD etc)
112 lines
4.2 KiB
Python
112 lines
4.2 KiB
Python
#!/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
|