295 lines
12 KiB
Python
295 lines
12 KiB
Python
"""
|
||
技能执行器 - 根据命令自动选择合适的Skill执行
|
||
"""
|
||
|
||
import logging
|
||
import time
|
||
import sys
|
||
import os
|
||
from typing import Dict, Any, Optional
|
||
|
||
# 兼容包内导入和独立运行
|
||
_agent_dir = os.path.dirname(os.path.abspath(__file__))
|
||
if _agent_dir not in sys.path:
|
||
sys.path.insert(0, _agent_dir)
|
||
|
||
try:
|
||
from skills import SKILL_REGISTRY, VoiceControlSkill, AppManagerSkill, SearchSkill
|
||
from skill_bus import SkillChatBus
|
||
except ImportError:
|
||
from .skills import SKILL_REGISTRY, VoiceControlSkill, AppManagerSkill, SearchSkill
|
||
from .skill_bus import SkillChatBus
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class SkillExecutor:
|
||
"""技能执行器"""
|
||
|
||
def __init__(self, device, anti_ban_ctx=None):
|
||
"""
|
||
初始化
|
||
|
||
Args:
|
||
device: uiautomator2设备对象
|
||
anti_ban_ctx: 防护上下文 dict
|
||
"""
|
||
self.device = device
|
||
self.anti_ban_ctx = anti_ban_ctx or {}
|
||
self.skill_bus = SkillChatBus()
|
||
self.skills = {}
|
||
self._init_skills()
|
||
|
||
def _init_skills(self):
|
||
"""初始化所有技能，注入共享聊天总线和防护上下文"""
|
||
for name, skill_class in SKILL_REGISTRY.items():
|
||
try:
|
||
sig = getattr(skill_class, "__init__")
|
||
varnames = sig.__code__.co_varnames
|
||
kwargs = {}
|
||
if "bus" in varnames:
|
||
kwargs["bus"] = self.skill_bus
|
||
if "anti_ban_ctx" in varnames:
|
||
kwargs["anti_ban_ctx"] = self.anti_ban_ctx
|
||
self.skills[name] = skill_class(self.device, **kwargs)
|
||
except Exception as e:
|
||
logger.error(f"初始化技能失败 {name}: {e}")
|
||
|
||
def execute_command(self, command: str, context: Dict[str, Any] = None) -> Dict[str, Any]:
|
||
"""
|
||
执行命令（自动选择合适的Skill）
|
||
|
||
Args:
|
||
command: 命令文本（如"打开微信给张三发消息：你好"）
|
||
context: 上下文信息
|
||
|
||
Returns:
|
||
执行结果
|
||
"""
|
||
try:
|
||
self.skill_bus.clear()
|
||
self.skill_bus.append("executor", f"开始执行: {command[:80]}")
|
||
voice_skill = self.skills.get("voice_control")
|
||
if not voice_skill:
|
||
voice_skill = VoiceControlSkill(self.device)
|
||
|
||
# 解析命令
|
||
parsed = voice_skill.parse_voice_command(command)
|
||
|
||
if not parsed.get("success"):
|
||
return {
|
||
"success": False,
|
||
"error": f"无法理解命令: {command}",
|
||
"command": command
|
||
}
|
||
|
||
# 根据命令类型选择技能
|
||
actions = parsed.get("actions", [])
|
||
|
||
if not actions:
|
||
return {
|
||
"success": False,
|
||
"error": "未找到可执行的操作",
|
||
"command": command
|
||
}
|
||
|
||
# 执行操作序列
|
||
results = []
|
||
current_skill = None
|
||
|
||
for action_data in actions:
|
||
action = action_data.get("action")
|
||
params = action_data.get("params", {})
|
||
|
||
try:
|
||
# 打开应用
|
||
if action == "open_app":
|
||
app_manager = self.skills.get("app_manager")
|
||
if not app_manager:
|
||
app_manager = AppManagerSkill(self.device)
|
||
|
||
package = params.get("package")
|
||
name = params.get("name", "")
|
||
|
||
result = app_manager.open_app(name if name else package)
|
||
results.append(result)
|
||
if package == "com.tencent.mm":
|
||
current_skill = self.skills.get("wechat")
|
||
elif package == "com.ss.android.ugc.aweme":
|
||
current_skill = self.skills.get("douyin")
|
||
elif package == "com.xingin.xhs":
|
||
current_skill = self.skills.get("xhs")
|
||
elif package == "com.taobao.idlefish":
|
||
current_skill = self.skills.get("xianyu")
|
||
elif package == "cn.soulapp.android":
|
||
current_skill = self.skills.get("soul")
|
||
else:
|
||
current_skill = None
|
||
self.skill_bus.append(
|
||
current_skill.NAME if current_skill else "app_manager",
|
||
f"open_app {name or package}",
|
||
data=result if isinstance(result, dict) else {"result": result},
|
||
)
|
||
|
||
# 搜索
|
||
elif action == "search":
|
||
if current_skill:
|
||
# 使用当前APP的技能搜索
|
||
keyword = params.get("keyword", "")
|
||
result = current_skill.search(keyword)
|
||
else:
|
||
# 使用通用搜索技能
|
||
search_skill = self.skills.get("search")
|
||
if not search_skill:
|
||
search_skill = SearchSkill(self.device)
|
||
keyword = params.get("keyword", "")
|
||
result = search_skill.search_in_app(keyword)
|
||
|
||
results.append(result)
|
||
sender = current_skill.NAME if current_skill else "search"
|
||
self.skill_bus.append(sender, f"search {keyword}", data=result if isinstance(result, dict) else {})
|
||
|
||
# 其他操作使用当前技能或基础操作
|
||
elif action == "back":
|
||
if current_skill:
|
||
current_skill.back()
|
||
else:
|
||
self.device.press("back")
|
||
results.append({"action": action, "success": True})
|
||
sender = current_skill.NAME if current_skill else "device"
|
||
self.skill_bus.append(sender, "back")
|
||
|
||
elif action == "home":
|
||
if current_skill:
|
||
current_skill.home()
|
||
else:
|
||
self.device.press("home")
|
||
results.append({"action": action, "success": True})
|
||
sender = current_skill.NAME if current_skill else "device"
|
||
self.skill_bus.append(sender, "home")
|
||
|
||
elif action == "swipe":
|
||
direction = params.get("direction", "up")
|
||
scale = params.get("scale", 0.5)
|
||
if current_skill:
|
||
current_skill.swipe(direction, scale)
|
||
else:
|
||
self.device.swipe_ext(direction, scale=scale)
|
||
results.append({"action": action, "success": True})
|
||
sender = current_skill.NAME if current_skill else "device"
|
||
self.skill_bus.append(sender, f"swipe {direction}")
|
||
|
||
elif action == "screenshot":
|
||
if current_skill:
|
||
filepath = current_skill.screenshot_to_file()
|
||
else:
|
||
filepath = f"/sdcard/screenshot_{int(time.time() * 1000)}.png"
|
||
self.device.screenshot(filepath)
|
||
results.append({
|
||
"action": action,
|
||
"success": True,
|
||
"filepath": filepath
|
||
})
|
||
sender = current_skill.NAME if current_skill else "device"
|
||
self.skill_bus.append(sender, "screenshot", data={"filepath": filepath})
|
||
|
||
elif action == "wait":
|
||
seconds = params.get("seconds", 1)
|
||
import time
|
||
time.sleep(seconds)
|
||
results.append({"action": action, "success": True})
|
||
self.skill_bus.append("executor", f"wait {seconds}s")
|
||
|
||
else:
|
||
results.append({
|
||
"action": action,
|
||
"success": False,
|
||
"error": f"未知操作: {action}"
|
||
})
|
||
|
||
except Exception as e:
|
||
logger.error(f"执行操作失败 {action}: {e}")
|
||
results.append({
|
||
"action": action,
|
||
"success": False,
|
||
"error": str(e)
|
||
})
|
||
|
||
success_count = sum(1 for r in results if r.get("success", False))
|
||
self.skill_bus.append("executor", f"完成 {success_count}/{len(results)} 步", data={"results": results})
|
||
return {
|
||
"success": success_count > 0,
|
||
"command": command,
|
||
"total_actions": len(results),
|
||
"success_count": success_count,
|
||
"results": results,
|
||
"skill_chat": self.skill_bus.get_messages(),
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"执行命令失败: {e}")
|
||
return {
|
||
"success": False,
|
||
"error": str(e),
|
||
"command": command
|
||
}
|
||
|
||
def execute_wechat_task(self, task: str) -> Dict[str, Any]:
|
||
"""
|
||
执行微信任务（智能解析）
|
||
|
||
Args:
|
||
task: 任务描述（如"给张三发消息：你好"）
|
||
|
||
Returns:
|
||
执行结果
|
||
"""
|
||
wechat_skill = self.skills.get("wechat")
|
||
if not wechat_skill:
|
||
wechat_skill = SKILL_REGISTRY["wechat"](self.device)
|
||
|
||
# 尝试使用复合任务功能
|
||
if hasattr(wechat_skill, "execute_compound_task"):
|
||
return wechat_skill.execute_compound_task(task)
|
||
else:
|
||
# 降级到普通命令执行
|
||
return self.execute_command(task)
|
||
|
||
def execute_douyin_task(self, task: str) -> Dict[str, Any]:
|
||
"""执行抖音任务"""
|
||
douyin_skill = self.skills.get("douyin")
|
||
if not douyin_skill:
|
||
douyin_skill = SKILL_REGISTRY["douyin"](self.device)
|
||
return self.execute_command(task)
|
||
|
||
def execute_xhs_task(self, task: str) -> Dict[str, Any]:
|
||
"""执行小红书任务"""
|
||
xhs_skill = self.skills.get("xhs")
|
||
if not xhs_skill:
|
||
xhs_skill = SKILL_REGISTRY["xhs"](self.device)
|
||
return self.execute_command(task)
|
||
|
||
def execute_xianyu_task(self, task: str) -> Dict[str, Any]:
|
||
"""执行闲鱼任务"""
|
||
xianyu_skill = self.skills.get("xianyu")
|
||
if not xianyu_skill:
|
||
xianyu_skill = SKILL_REGISTRY["xianyu"](self.device)
|
||
return self.execute_command(task)
|
||
|
||
def execute_soul_task(self, task: str) -> Dict[str, Any]:
|
||
"""执行 Soul 任务"""
|
||
soul_skill = self.skills.get("soul")
|
||
if not soul_skill:
|
||
soul_skill = SKILL_REGISTRY["soul"](self.device)
|
||
return self.execute_command(task)
|
||
|
||
def execute_network_reconnect(self) -> Dict[str, Any]:
|
||
"""执行网络恢复（委托 Hawk：开 WiFi、打开设置、连接已保存网络等合法操作）"""
|
||
net_skill = self.skills.get("network_reconnect")
|
||
if not net_skill:
|
||
try:
|
||
net_skill = SKILL_REGISTRY["network_reconnect"](self.device, self.skill_bus)
|
||
except KeyError:
|
||
return {"success": False, "error": "network_reconnect 技能未注册"}
|
||
return net_skill.try_reconnect()
|