feat: Android 重构 service/engine/ui/frida/antiban、AgentForegroundService、APK产品说明、架构图

Made-with: Cursor

sdk/app/data/hook/device_modules.json

@@ -68,11 +68,19 @@
     "modules": []
   },
   "redmi-note11-test": {
+    "supports_hook": false,
+    "frida_version": "",
+    "root_status": false,
+    "hook_framework": "frida-server",
+    "updated_at": "2026-03-16T09:59:18.233188+00:00",
+    "modules": []
+  },
+  "273c1980866992cf800c00ada5cc6375": {
     "supports_hook": false,
     "frida_version": "",
     "root_status": true,
     "hook_framework": "frida-server",
-    "updated_at": "2026-03-16T09:26:13.470654+00:00",
+    "updated_at": "2026-03-16T21:35:57.473693+00:00",
     "modules": []
   }
 }

sdk/app/routers/connection.py

@@ -205,8 +205,10 @@ async def simulate_hook_event(req: SimHookEventRequest) -> Dict[str, Any]:
 # ========== 连接优先级 ==========
 
 @router.get("/connection/modes/{device_id}")
-async def get_device_connection_modes(device_id: str) -> Dict[str, Any]:
-    """获取设备所有控制模式的可用性与优先级。"""
+async def get_device_connection_modes(device_id: str, refresh: bool = False) -> Dict[str, Any]:
+    """获取设备所有控制模式的可用性与优先级。refresh=true 强制刷新缓存。"""
+    if refresh:
+        connection_priority.invalidate_cache()
     modes = await connection_priority.async_evaluate(device_id)
     best = next((m for m in modes if m.available), None)
     return {

sdk/app/routers/qrcode.py

@@ -26,6 +26,7 @@ class ProjectBindConfig(BaseModel):
     project_id: str
     project_name: Optional[str] = ""
     server: Optional[str] = "ws://sdk.quwanzhi.com:8899/ws/device"
+    pwa: Optional[str] = ""
 
 
 @router.post("/generate")
@@ -37,7 +38,8 @@ async def generate_qrcode(config: ProjectBindConfig):
     {
         "project_id": "project_001",
         "project_name": "测试项目",
-        "server": "ws://sdk.quwanzhi.com:8899/ws/device"
+        "server": "ws://sdk.quwanzhi.com:8899/ws/device",
+        "pwa": ""
     }
     
     返回: Base64编码的PNG图片
@@ -48,12 +50,14 @@ async def generate_qrcode(config: ProjectBindConfig):
             "message": "服务器未安装qrcode库，请运行: pip install qrcode[pil]"
         }
     
-    # 生成二维码内容（JSON格式）
-    qr_content = json.dumps({
+    qr_data = {
         "server": config.server,
-        "project_id": config.project_id,
-        "project_name": config.project_name
-    }, ensure_ascii=False)
+        "project": config.project_id,
+        "project_name": config.project_name,
+    }
+    if config.pwa:
+        qr_data["pwa"] = config.pwa
+    qr_content = json.dumps(qr_data, ensure_ascii=False)
     
     # 生成二维码
     qr = qrcode.QRCode(
@@ -86,7 +90,8 @@ async def generate_qrcode(config: ProjectBindConfig):
 async def get_qrcode_image(
     project_id: str,
     project_name: str = "",
-    server: str = "ws://sdk.quwanzhi.com:8899/ws/device"
+    server: str = "ws://sdk.quwanzhi.com:8899/ws/device",
+    pwa: str = ""
 ):
     """
     直接获取二维码图片
@@ -102,12 +107,10 @@ async def get_qrcode_image(
             status_code=500
         )
     
-    # 生成二维码内容
-    qr_content = json.dumps({
-        "server": server,
-        "project_id": project_id,
-        "project_name": project_name
-    }, ensure_ascii=False)
+    qr_data = {"server": server, "project": project_id, "project_name": project_name}
+    if pwa:
+        qr_data["pwa"] = pwa
+    qr_content = json.dumps(qr_data, ensure_ascii=False)
     
     # 生成二维码
     qr = qrcode.QRCode(
@@ -136,7 +139,8 @@ async def get_qrcode_image(
 async def get_qrcode_html(
     project_id: str,
     project_name: str = "",
-    server: str = "ws://sdk.quwanzhi.com:8899/ws/device"
+    server: str = "ws://sdk.quwanzhi.com:8899/ws/device",
+    pwa: str = ""
 ):
     """
     获取带样式的二维码HTML页面
@@ -152,12 +156,10 @@ async def get_qrcode_html(
             status_code=500
         )
     
-    # 生成二维码内容
-    qr_content = json.dumps({
-        "server": server,
-        "project_id": project_id,
-        "project_name": project_name
-    }, ensure_ascii=False)
+    qr_data = {"server": server, "project": project_id, "project_name": project_name}
+    if pwa:
+        qr_data["pwa"] = pwa
+    qr_content = json.dumps(qr_data, ensure_ascii=False)
     
     # 生成二维码
     qr = qrcode.QRCode(

sdk/app/services/connection_priority.py

@@ -15,12 +15,15 @@ from __future__ import annotations
 import asyncio
 import logging
 import subprocess
+import time
 from dataclasses import dataclass, field
 from enum import IntEnum
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 logger = logging.getLogger(__name__)
 
+_CACHE_TTL = 30  # seconds
+
 
 class ControlMode(IntEnum):
     HOOK = 1
@@ -91,156 +94,99 @@ class ConnectionPriorityManager:
 
     def __init__(self):
         self._scrcpy_available: Optional[bool] = None
-        self._frida_available: Optional[bool] = None
-        self._serial_to_ws_id: Dict[str, str] = {}
-
-    def _resolve_ws_id(self, device_id: str) -> Optional[str]:
-        """将 ADB serial 映射到 WS device_id，带缓存。"""
-        import hashlib
-        if device_id in self._serial_to_ws_id:
-            return self._serial_to_ws_id[device_id]
-
-        from services.adb_device import adb_manager
-        adb_dev = adb_manager.get_device(device_id)
-        if not adb_dev:
-            return None
-        try:
-            android_id = adb_dev._shell("settings get secure android_id").strip()
-            if android_id and android_id != "null":
-                md5_id = hashlib.md5(android_id.encode()).hexdigest()
-                self._serial_to_ws_id[device_id] = md5_id
-                return md5_id
-        except Exception:
-            pass
-        return None
+        self._cache: Dict[str, Tuple[float, List[ModeStatus]]] = {}
 
     def evaluate(self, device_id: str) -> List[ModeStatus]:
+        now = time.time()
+        cached = self._cache.get(device_id)
+        if cached and (now - cached[0]) < _CACHE_TTL:
+            return cached[1]
+
         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 (use cached result, refresh in background)
-        hook_ok = self._check_hook(device_id, hook_module_service)
+        # --- Collect raw signals ---
+        adb_ok, adb_serial = self._check_adb(device_id, adb_manager)
+        ws_ok, ws_detail = self._check_ws(device_id, ws_hub)
+        hook_ok = self._check_hook_fast()
+        scrcpy_ok = adb_ok and self._check_scrcpy()
+        has_base = ws_ok or adb_ok or hook_ok
+
+        # 1. Hook / Frida
         results.append(ModeStatus(
-            mode=ControlMode.HOOK,
-            available=hook_ok,
-            priority=1,
+            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:
-            resolved_ws_id = self._serial_to_ws_id.get(device_id)
-            if resolved_ws_id and resolved_ws_id in ws_hub.connections:
-                ws_ok = True
-                ws_detail = f"WebSocket 已连接 (Agent: {resolved_ws_id[:8]}…)"
-            else:
-                online_ids = list(ws_hub.connections.keys())
-                if online_ids:
-                    ws_ok = True
-                    self._serial_to_ws_id[device_id] = online_ids[0]
-                    ws_detail = f"WebSocket 已连接 ({online_ids[0][:8]}…)"
-                else:
-                    ws_detail = "设备Agent未连接"
         results.append(ModeStatus(
-            mode=ControlMode.AGENT_WS,
-            available=ws_ok,
-            priority=2,
+            mode=ControlMode.AGENT_WS, available=ws_ok, priority=2,
             detail=ws_detail,
         ))
 
-        # 3. ADB
-        adb_ok = False
-        adb_serial = ""
-        resolved = adb_manager.resolve_serial(device_id)
-        if resolved:
-            adb_dev_obj = adb_manager.get_device(resolved)
-            if adb_dev_obj and adb_dev_obj.is_online():
-                adb_ok = True
-                adb_serial = adb_dev_obj.serial
-        if not adb_ok:
-            for serial in adb_manager.scan_devices():
-                dev = adb_manager.get_device(serial)
-                if dev and dev.is_online():
-                    adb_ok = True
-                    adb_serial = dev.serial
-                    break
-        if adb_ok and adb_serial:
-            adb_manager.register_mapping(device_id, adb_serial)
+        # 3. ADB 静默控制
         results.append(ModeStatus(
-            mode=ControlMode.ADB,
-            available=adb_ok,
-            priority=3,
+            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
+        # 4. AI Agent
         results.append(ModeStatus(
-            mode=ControlMode.AI_AGENT,
-            available=has_base,
-            priority=4,
+            mode=ControlMode.AI_AGENT, available=has_base, priority=4,
             detail="底层通道就绪" if has_base else "需要至少一个底层通道",
         ))
 
-        # 5. scrcpy
-        scrcpy_ok = adb_ok and self._check_scrcpy()
+        # 5. scrcpy 可视化
         results.append(ModeStatus(
-            mode=ControlMode.SCRCPY,
-            available=scrcpy_ok,
-            priority=5,
+            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)
+        self._cache[device_id] = (now, results)
         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
+    # ---- individual checks ----
 
-    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_ws(device_id: str, ws_hub) -> Tuple[bool, str]:
+        if ws_hub.is_online(device_id):
+            return True, "WebSocket 已连接"
+        online_ids = list(ws_hub.connections.keys())
+        if online_ids:
+            return True, f"WebSocket 已连接 ({online_ids[0][:8]}…)"
+        return False, "设备Agent未连接"
 
-    def _check_hook(self, device_id: str, hook_service) -> bool:
-        if self._frida_available is True:
-            return True
+    @staticmethod
+    def _check_adb(device_id: str, adb_manager) -> Tuple[bool, str]:
+        resolved = adb_manager.resolve_serial(device_id)
+        if resolved:
+            dev = adb_manager.get_device(resolved)
+            if dev and dev.is_online():
+                adb_manager.register_mapping(device_id, dev.serial)
+                return True, dev.serial
+        for serial in adb_manager.scan_devices():
+            dev = adb_manager.get_device(serial)
+            if dev and dev.is_online():
+                adb_manager.register_mapping(device_id, dev.serial)
+                return True, dev.serial
+        return False, ""
+
+    def _check_hook_fast(self) -> bool:
         try:
-            if hasattr(hook_service, "_read_json"):
-                state = hook_service._read_json(hook_service.device_state_file, {})
-                item = state.get(device_id, {})
-                if item.get("supports_hook", False):
-                    self._frida_available = True
-                    return True
             r = subprocess.run(
                 ["frida-ps", "-U"],
-                capture_output=True, text=True, timeout=5,
+                capture_output=True, text=True, timeout=3,
             )
             if r.returncode == 0 and len(r.stdout.strip().split("\n")) > 2:
-                self._frida_available = True
                 return True
         except Exception:
             pass
         return False
 
-    def invalidate_cache(self):
-        """清除缓存，下次检测重新获取。"""
-        self._frida_available = None
-        self._scrcpy_available = None
-        self._serial_to_ws_id.clear()
-
     def _check_scrcpy(self) -> bool:
         if self._scrcpy_available is not None:
             return self._scrcpy_available
@@ -251,9 +197,25 @@ class ConnectionPriorityManager:
             self._scrcpy_available = False
         return self._scrcpy_available
 
+    # ---- public helpers ----
+
+    def get_best_mode(self, device_id: str) -> Optional[ModeStatus]:
+        for m in self.evaluate(device_id):
+            if m.available:
+                return m
+        return None
+
+    def choose_execution_channel(self, device_id: str) -> str:
+        best = self.get_best_mode(device_id)
+        if not best:
+            return "offline"
+        return MODE_META[best.mode]["id"]
+
+    def invalidate_cache(self):
+        self._cache.clear()
+        self._scrcpy_available = None
 
     async def async_evaluate(self, device_id: str) -> List[ModeStatus]:
-        """异步版 evaluate — 将 ADB 阻塞操作放入线程池"""
         loop = asyncio.get_event_loop()
         return await loop.run_in_executor(None, self.evaluate, device_id)