180 lines
4.2 KiB
Python
180 lines
4.2 KiB
Python
"""
|
|
技能管理器
|
|
管理所有技能的注册、发现和调用
|
|
"""
|
|
from typing import Dict, Optional, List, Type
|
|
from app.skills.base import BaseSkill
|
|
from app.utils.logger import logger
|
|
|
|
|
|
class SkillManager:
|
|
"""技能管理器"""
|
|
|
|
def __init__(self):
|
|
"""初始化技能管理器"""
|
|
self._skills: Dict[str, BaseSkill] = {}
|
|
logger.info("技能管理器初始化")
|
|
|
|
def register(self, skill: BaseSkill) -> bool:
|
|
"""
|
|
注册技能
|
|
|
|
Args:
|
|
skill: 技能实例
|
|
|
|
Returns:
|
|
是否成功
|
|
"""
|
|
if not skill.name:
|
|
logger.error("技能名称不能为空")
|
|
return False
|
|
|
|
if skill.name in self._skills:
|
|
logger.warning(f"技能已存在，将被覆盖: {skill.name}")
|
|
|
|
self._skills[skill.name] = skill
|
|
logger.info(f"技能注册成功: {skill.name}")
|
|
return True
|
|
|
|
def unregister(self, skill_name: str) -> bool:
|
|
"""
|
|
注销技能
|
|
|
|
Args:
|
|
skill_name: 技能名称
|
|
|
|
Returns:
|
|
是否成功
|
|
"""
|
|
if skill_name in self._skills:
|
|
del self._skills[skill_name]
|
|
logger.info(f"技能注销成功: {skill_name}")
|
|
return True
|
|
|
|
logger.warning(f"技能不存在: {skill_name}")
|
|
return False
|
|
|
|
def get_skill(self, skill_name: str) -> Optional[BaseSkill]:
|
|
"""
|
|
获取技能
|
|
|
|
Args:
|
|
skill_name: 技能名称
|
|
|
|
Returns:
|
|
技能实例或None
|
|
"""
|
|
return self._skills.get(skill_name)
|
|
|
|
def get_all_skills(self) -> List[BaseSkill]:
|
|
"""
|
|
获取所有技能
|
|
|
|
Returns:
|
|
技能列表
|
|
"""
|
|
return list(self._skills.values())
|
|
|
|
def get_enabled_skills(self) -> List[BaseSkill]:
|
|
"""
|
|
获取所有启用的技能
|
|
|
|
Returns:
|
|
启用的技能列表
|
|
"""
|
|
return [skill for skill in self._skills.values() if skill.enabled]
|
|
|
|
async def execute_skill(self, skill_name: str, **kwargs) -> Dict:
|
|
"""
|
|
执行技能
|
|
|
|
Args:
|
|
skill_name: 技能名称
|
|
**kwargs: 技能参数
|
|
|
|
Returns:
|
|
执行结果
|
|
"""
|
|
skill = self.get_skill(skill_name)
|
|
|
|
if not skill:
|
|
return {
|
|
"success": False,
|
|
"error": f"技能不存在: {skill_name}"
|
|
}
|
|
|
|
if not skill.enabled:
|
|
return {
|
|
"success": False,
|
|
"error": f"技能已禁用: {skill_name}"
|
|
}
|
|
|
|
# 验证参数
|
|
valid, error = skill.validate_params(**kwargs)
|
|
if not valid:
|
|
return {
|
|
"success": False,
|
|
"error": error
|
|
}
|
|
|
|
# 执行技能
|
|
try:
|
|
result = await skill.execute(**kwargs)
|
|
return {
|
|
"success": True,
|
|
"data": result
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"技能执行失败 {skill_name}: {e}")
|
|
return {
|
|
"success": False,
|
|
"error": str(e)
|
|
}
|
|
|
|
def enable_skill(self, skill_name: str) -> bool:
|
|
"""
|
|
启用技能
|
|
|
|
Args:
|
|
skill_name: 技能名称
|
|
|
|
Returns:
|
|
是否成功
|
|
"""
|
|
skill = self.get_skill(skill_name)
|
|
if skill:
|
|
skill.enable()
|
|
logger.info(f"技能已启用: {skill_name}")
|
|
return True
|
|
return False
|
|
|
|
def disable_skill(self, skill_name: str) -> bool:
|
|
"""
|
|
禁用技能
|
|
|
|
Args:
|
|
skill_name: 技能名称
|
|
|
|
Returns:
|
|
是否成功
|
|
"""
|
|
skill = self.get_skill(skill_name)
|
|
if skill:
|
|
skill.disable()
|
|
logger.info(f"技能已禁用: {skill_name}")
|
|
return True
|
|
return False
|
|
|
|
def get_skills_info(self) -> List[Dict]:
|
|
"""
|
|
获取所有技能信息
|
|
|
|
Returns:
|
|
技能信息列表
|
|
"""
|
|
return [skill.get_info() for skill in self._skills.values()]
|
|
|
|
|
|
# 创建全局技能管理器实例
|
|
skill_manager = SkillManager()
|