213 lines
6.7 KiB
Python
213 lines
6.7 KiB
Python
"""
|
||
连接优先级管理器
|
||
5种控制模式的可用性检测、优先级路由和智能降级。
|
||
|
||
优先级（高→低）：
|
||
1. Hook/Frida — 深度控制，读写微信内部数据
|
||
2. Agent WebSocket — 实时双向，设备端App常驻
|
||
3. ADB 静默控制 — 无需App，ADB直连兜底
|
||
4. AI Agent — 自然语言编排，依赖以上通道
|
||
5. scrcpy 可视化 — 投屏+人工接管
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import subprocess
|
||
from dataclasses import dataclass, field
|
||
from enum import IntEnum
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class ControlMode(IntEnum):
|
||
HOOK = 1
|
||
AGENT_WS = 2
|
||
ADB = 3
|
||
AI_AGENT = 4
|
||
SCRCPY = 5
|
||
|
||
|
||
MODE_META = {
|
||
ControlMode.HOOK: {
|
||
"id": "hook",
|
||
"name": "Hook / Frida",
|
||
"desc": "读写内部能力强，适合微信深度控制",
|
||
"requires": "Root + Frida Server",
|
||
},
|
||
ControlMode.AGENT_WS: {
|
||
"id": "agent",
|
||
"name": "Agent WebSocket",
|
||
"desc": "实时双向通信，设备端App常驻",
|
||
"requires": "设备端安装工作App",
|
||
},
|
||
ControlMode.ADB: {
|
||
"id": "adb",
|
||
"name": "ADB 静默控制",
|
||
"desc": "无需Root，部署简单，稳定兜底",
|
||
"requires": "USB调试已开启",
|
||
},
|
||
ControlMode.AI_AGENT: {
|
||
"id": "ai",
|
||
"name": "AI Agent",
|
||
"desc": "自然语言编排复杂任务",
|
||
"requires": "需要底层通道(Hook/WS/ADB)之一可用",
|
||
},
|
||
ControlMode.SCRCPY: {
|
||
"id": "scrcpy",
|
||
"name": "scrcpy 可视化",
|
||
"desc": "投屏与人工接管",
|
||
"requires": "ADB连接 + scrcpy已安装",
|
||
},
|
||
}
|
||
|
||
|
||
@dataclass
|
||
class ModeStatus:
|
||
mode: ControlMode
|
||
available: bool
|
||
priority: int
|
||
detail: str = ""
|
||
meta: Dict[str, Any] = field(default_factory=dict)
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
m = MODE_META.get(self.mode, {})
|
||
return {
|
||
"id": m.get("id", str(self.mode)),
|
||
"name": m.get("name", ""),
|
||
"desc": m.get("desc", ""),
|
||
"requires": m.get("requires", ""),
|
||
"priority": self.priority,
|
||
"available": self.available,
|
||
"detail": self.detail,
|
||
**self.meta,
|
||
}
|
||
|
||
|
||
class ConnectionPriorityManager:
|
||
"""对单台设备评估所有控制模式的可用性并按优先级排序。"""
|
||
|
||
def __init__(self):
|
||
self._scrcpy_available: Optional[bool] = None
|
||
|
||
def evaluate(self, device_id: str) -> List[ModeStatus]:
|
||
from services.ws_hub import ws_hub
|
||
from services.adb_device import adb_manager
|
||
from services.hook_module_service import hook_module_service
|
||
|
||
results: List[ModeStatus] = []
|
||
|
||
# 1. Hook / Frida
|
||
hook_ok = self._check_hook(device_id, hook_module_service)
|
||
results.append(ModeStatus(
|
||
mode=ControlMode.HOOK,
|
||
available=hook_ok,
|
||
priority=1,
|
||
detail="Frida Server 运行中" if hook_ok else "未检测到 Hook 服务",
|
||
))
|
||
|
||
# 2. Agent WebSocket
|
||
ws_ok = ws_hub.is_online(device_id)
|
||
ws_detail = "WebSocket 已连接"
|
||
if not ws_ok:
|
||
adb_serials = adb_manager.scan_devices()
|
||
for serial in adb_serials:
|
||
adb_dev = adb_manager.get_device(serial)
|
||
if adb_dev:
|
||
android_id = adb_dev._shell("settings get secure android_id").strip()
|
||
import hashlib
|
||
md5_id = hashlib.md5(android_id.encode()).hexdigest()
|
||
if md5_id == device_id:
|
||
ws_detail = f"设备在线(ADB:{serial})但Agent App未连接WebSocket"
|
||
break
|
||
else:
|
||
ws_detail = "设备Agent未连接"
|
||
results.append(ModeStatus(
|
||
mode=ControlMode.AGENT_WS,
|
||
available=ws_ok,
|
||
priority=2,
|
||
detail=ws_detail,
|
||
))
|
||
|
||
# 3. ADB
|
||
adb_ok = False
|
||
adb_serial = ""
|
||
adb_dev_obj = adb_manager.get_device(device_id)
|
||
if adb_dev_obj and adb_dev_obj.is_online():
|
||
adb_ok = True
|
||
adb_serial = device_id
|
||
else:
|
||
for serial in adb_manager.scan_devices():
|
||
dev = adb_manager.get_device(serial)
|
||
if dev and dev.is_online():
|
||
adb_ok = True
|
||
adb_serial = serial
|
||
break
|
||
results.append(ModeStatus(
|
||
mode=ControlMode.ADB,
|
||
available=adb_ok,
|
||
priority=3,
|
||
detail=f"ADB 已连接 ({adb_serial})" if adb_ok else "ADB 未连接",
|
||
meta={"adb_serial": adb_serial} if adb_ok else {},
|
||
))
|
||
|
||
# 4. AI Agent（依赖底层通道）
|
||
has_base = ws_ok or adb_ok or hook_ok
|
||
results.append(ModeStatus(
|
||
mode=ControlMode.AI_AGENT,
|
||
available=has_base,
|
||
priority=4,
|
||
detail="底层通道就绪" if has_base else "需要至少一个底层通道",
|
||
))
|
||
|
||
# 5. scrcpy
|
||
scrcpy_ok = adb_ok and self._check_scrcpy()
|
||
results.append(ModeStatus(
|
||
mode=ControlMode.SCRCPY,
|
||
available=scrcpy_ok,
|
||
priority=5,
|
||
detail="scrcpy 可用" if scrcpy_ok else ("ADB未连接" if not adb_ok else "scrcpy 未安装"),
|
||
))
|
||
|
||
results.sort(key=lambda m: m.priority)
|
||
return results
|
||
|
||
def get_best_mode(self, device_id: str) -> Optional[ModeStatus]:
|
||
modes = self.evaluate(device_id)
|
||
for m in modes:
|
||
if m.available:
|
||
return m
|
||
return None
|
||
|
||
def choose_execution_channel(self, device_id: str) -> str:
|
||
"""返回最优执行通道标识: hook / agent / adb / ai / offline"""
|
||
best = self.get_best_mode(device_id)
|
||
if not best:
|
||
return "offline"
|
||
return MODE_META[best.mode]["id"]
|
||
|
||
@staticmethod
|
||
def _check_hook(device_id: str, hook_service) -> bool:
|
||
try:
|
||
if not hasattr(hook_service, "_read_json"):
|
||
return False
|
||
state = hook_service._read_json(hook_service.device_state_file, {})
|
||
item = state.get(device_id, {})
|
||
return bool(item.get("supports_hook", False))
|
||
except Exception:
|
||
return False
|
||
|
||
def _check_scrcpy(self) -> bool:
|
||
if self._scrcpy_available is not None:
|
||
return self._scrcpy_available
|
||
try:
|
||
r = subprocess.run(["which", "scrcpy"], capture_output=True, text=True, timeout=3)
|
||
self._scrcpy_available = r.returncode == 0
|
||
except Exception:
|
||
self._scrcpy_available = False
|
||
return self._scrcpy_available
|
||
|
||
|
||
connection_priority = ConnectionPriorityManager()
|