500 lines
19 KiB
Python
500 lines
19 KiB
Python
"""
|
||
工作手机SDK v3.0 - ADB设备控制服务
|
||
直接通过ADB命令控制设备（兼容模拟器）
|
||
"""
|
||
|
||
import asyncio
|
||
import subprocess
|
||
import base64
|
||
import logging
|
||
import time
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from typing import Dict, Any, List, Optional
|
||
from datetime import datetime
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_adb_executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="adb")
|
||
|
||
import os as _os
|
||
_adb_env = _os.environ.copy()
|
||
_adb_key = _os.path.expanduser("~/.android/adbkey")
|
||
if _os.path.exists(_adb_key):
|
||
_adb_env["ADB_VENDOR_KEYS"] = _adb_key
|
||
|
||
|
||
class ADBDevice:
|
||
"""ADB设备控制"""
|
||
|
||
def __init__(self, serial: str):
|
||
self.serial = serial
|
||
self.last_active = datetime.now()
|
||
|
||
def _adb(self, *args, timeout: int = 30) -> subprocess.CompletedProcess:
|
||
"""执行ADB命令"""
|
||
cmd = ["adb", "-s", self.serial] + list(args)
|
||
return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, env=_adb_env)
|
||
|
||
def _shell(self, cmd: str, timeout: int = 30) -> str:
|
||
"""执行shell命令"""
|
||
result = self._adb("shell", cmd, timeout=timeout)
|
||
return result.stdout.strip()
|
||
|
||
# ========== 设备信息 ==========
|
||
|
||
def get_info(self) -> Dict[str, Any]:
|
||
"""获取设备信息"""
|
||
try:
|
||
model = self._shell("getprop ro.product.model")
|
||
brand = self._shell("getprop ro.product.brand")
|
||
version = self._shell("getprop ro.build.version.release")
|
||
sdk = self._shell("getprop ro.build.version.sdk")
|
||
size = self._shell("wm size")
|
||
|
||
# 解析分辨率
|
||
width, height = 1080, 2400
|
||
if "Physical size:" in size:
|
||
dims = size.split("Physical size:")[-1].strip()
|
||
if "x" in dims:
|
||
width, height = map(int, dims.split("x"))
|
||
|
||
return {
|
||
"serial": self.serial,
|
||
"model": model or "Unknown",
|
||
"brand": brand or "Unknown",
|
||
"android_version": version or "Unknown",
|
||
"sdk_version": sdk or "Unknown",
|
||
"display": {
|
||
"width": width,
|
||
"height": height
|
||
}
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"获取设备信息失败: {e}")
|
||
return {"serial": self.serial, "error": str(e)}
|
||
|
||
def is_online(self) -> bool:
|
||
"""检查设备是否在线"""
|
||
try:
|
||
result = subprocess.run(
|
||
["adb", "devices"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=5,
|
||
env=_adb_env,
|
||
)
|
||
return self.serial in result.stdout and "device" in result.stdout
|
||
except:
|
||
return False
|
||
|
||
# ========== 基础操作 ==========
|
||
|
||
def screenshot(self) -> Dict[str, Any]:
|
||
"""截图"""
|
||
try:
|
||
# 先在设备上截图保存到临时路径
|
||
self._shell("screencap -p /sdcard/screen.png", timeout=15)
|
||
|
||
# 拉取到本地
|
||
result = subprocess.run(
|
||
["adb", "-s", self.serial, "exec-out", "cat", "/sdcard/screen.png"],
|
||
capture_output=True,
|
||
timeout=15
|
||
)
|
||
|
||
if result.returncode == 0 and result.stdout and len(result.stdout) > 100:
|
||
b64 = base64.b64encode(result.stdout).decode('utf-8')
|
||
info = self.get_info()
|
||
return {
|
||
"code": 200,
|
||
"data": {
|
||
"base64": b64,
|
||
"width": info.get("display", {}).get("width", 1080),
|
||
"height": info.get("display", {}).get("height", 2400),
|
||
"size": len(result.stdout),
|
||
"format": "png"
|
||
}
|
||
}
|
||
return {"code": 500, "message": "截图失败", "detail": f"returncode={result.returncode}, stdout_len={len(result.stdout)}"}
|
||
except Exception as e:
|
||
return {"code": 500, "message": str(e)}
|
||
|
||
def click(self, x: int, y: int) -> Dict[str, Any]:
|
||
"""点击坐标"""
|
||
try:
|
||
self._shell(f"input tap {x} {y}")
|
||
self.last_active = datetime.now()
|
||
return {"code": 200, "data": {"success": True, "x": x, "y": y}}
|
||
except Exception as e:
|
||
return {"code": 500, "message": str(e)}
|
||
|
||
def click_text(self, text: str, timeout: float = 15) -> Dict[str, Any]:
|
||
"""点击文字（通过UI树查找，支持text/content-desc/模糊匹配）"""
|
||
try:
|
||
import re
|
||
try:
|
||
self._shell("uiautomator dump /sdcard/ui_tree.xml", timeout=10)
|
||
xml = self._shell("cat /sdcard/ui_tree.xml", timeout=5)
|
||
except (subprocess.TimeoutExpired, Exception) as te:
|
||
return {"code": 408, "message": f"UI树获取超时，跳过点击: {text}"}
|
||
|
||
if not xml or len(xml) < 50:
|
||
return {"code": 404, "message": f"UI树为空，无法查找: {text}"}
|
||
|
||
# 1. 精确匹配 text 属性
|
||
pattern = f'text="{re.escape(text)}"[^>]*bounds="\\[(\\d+),(\\d+)\\]\\[(\\d+),(\\d+)\\]"'
|
||
match = re.search(pattern, xml)
|
||
if match:
|
||
x1, y1, x2, y2 = map(int, match.groups())
|
||
return self.click((x1 + x2) // 2, (y1 + y2) // 2)
|
||
|
||
# 2. 精确匹配 content-desc 属性
|
||
pattern_desc = f'content-desc="{re.escape(text)}"[^>]*bounds="\\[(\\d+),(\\d+)\\]\\[(\\d+),(\\d+)\\]"'
|
||
match = re.search(pattern_desc, xml)
|
||
if match:
|
||
x1, y1, x2, y2 = map(int, match.groups())
|
||
return self.click((x1 + x2) // 2, (y1 + y2) // 2)
|
||
|
||
# 3. 模糊匹配：包含目标文字
|
||
pattern_fuzzy = f'text="[^"]*{re.escape(text)}[^"]*"[^>]*bounds="\\[(\\d+),(\\d+)\\]\\[(\\d+),(\\d+)\\]"'
|
||
match = re.search(pattern_fuzzy, xml)
|
||
if match:
|
||
x1, y1, x2, y2 = map(int, match.groups())
|
||
return self.click((x1 + x2) // 2, (y1 + y2) // 2)
|
||
|
||
# 4. 不区分大小写匹配
|
||
pattern_ci = f'(?i)text="[^"]*{re.escape(text)}[^"]*"[^>]*bounds="\\[(\\d+),(\\d+)\\]\\[(\\d+),(\\d+)\\]"'
|
||
match = re.search(pattern_ci, xml)
|
||
if match:
|
||
x1, y1, x2, y2 = map(int, match.groups())
|
||
return self.click((x1 + x2) // 2, (y1 + y2) // 2)
|
||
|
||
return {"code": 404, "message": f"未找到文字: {text}"}
|
||
except subprocess.TimeoutExpired:
|
||
return {"code": 408, "message": f"操作超时: {text}"}
|
||
except Exception as e:
|
||
return {"code": 500, "message": str(e)}
|
||
|
||
def input_text(self, text: str, clear: bool = False) -> Dict[str, Any]:
|
||
"""输入文字（默认追加，不清除已有内容）"""
|
||
try:
|
||
if clear:
|
||
# 全选并删除（合并为一条命令减少开销）
|
||
self._shell("input keyevent KEYCODE_CTRL_A && input keyevent KEYCODE_DEL", timeout=10)
|
||
|
||
escaped = text.replace(" ", "%s").replace("'", "\\'").replace('"', '\\"')
|
||
self._shell(f"input text '{escaped}'", timeout=15)
|
||
|
||
self.last_active = datetime.now()
|
||
return {"code": 200, "data": {"success": True, "text": text}}
|
||
except Exception as e:
|
||
return {"code": 500, "message": str(e)}
|
||
|
||
def swipe(self, direction: str, scale: float = 0.5) -> Dict[str, Any]:
|
||
"""滑动"""
|
||
try:
|
||
info = self.get_info()
|
||
w = info.get("display", {}).get("width", 1080)
|
||
h = info.get("display", {}).get("height", 2400)
|
||
|
||
cx, cy = w // 2, h // 2
|
||
dist = int(h * scale * 0.4)
|
||
|
||
if direction == "up":
|
||
x1, y1, x2, y2 = cx, cy + dist, cx, cy - dist
|
||
elif direction == "down":
|
||
x1, y1, x2, y2 = cx, cy - dist, cx, cy + dist
|
||
elif direction == "left":
|
||
dist = int(w * scale * 0.4)
|
||
x1, y1, x2, y2 = cx + dist, cy, cx - dist, cy
|
||
elif direction == "right":
|
||
dist = int(w * scale * 0.4)
|
||
x1, y1, x2, y2 = cx - dist, cy, cx + dist, cy
|
||
else:
|
||
return {"code": 400, "message": f"无效方向: {direction}"}
|
||
|
||
self._shell(f"input swipe {x1} {y1} {x2} {y2} 500")
|
||
self.last_active = datetime.now()
|
||
return {"code": 200, "data": {"success": True, "direction": direction}}
|
||
except Exception as e:
|
||
return {"code": 500, "message": str(e)}
|
||
|
||
def press_key(self, key: str) -> Dict[str, Any]:
|
||
"""按键"""
|
||
key_map = {
|
||
"back": "KEYCODE_BACK",
|
||
"home": "KEYCODE_HOME",
|
||
"recent": "KEYCODE_APP_SWITCH",
|
||
"enter": "KEYCODE_ENTER",
|
||
"search": "KEYCODE_SEARCH",
|
||
"tab": "KEYCODE_TAB",
|
||
"del": "KEYCODE_DEL",
|
||
"delete": "KEYCODE_DEL",
|
||
"volume_up": "KEYCODE_VOLUME_UP",
|
||
"volume_down": "KEYCODE_VOLUME_DOWN",
|
||
"power": "KEYCODE_POWER",
|
||
"menu": "KEYCODE_MENU",
|
||
"space": "KEYCODE_SPACE",
|
||
}
|
||
|
||
keycode = key_map.get(key, key)
|
||
try:
|
||
self._shell(f"input keyevent {keycode}")
|
||
return {"code": 200, "data": {"success": True, "key": key}}
|
||
except Exception as e:
|
||
return {"code": 500, "message": str(e)}
|
||
|
||
# ========== APP操作 ==========
|
||
|
||
def start_app(self, package: str) -> Dict[str, Any]:
|
||
"""启动APP（resolve launcher activity → am start）"""
|
||
try:
|
||
activity = self._shell(
|
||
f"cmd package resolve-activity --brief {package} 2>/dev/null | tail -1",
|
||
timeout=10,
|
||
).strip()
|
||
if activity and "/" in activity:
|
||
result = self._shell(f"am start -n {activity}", timeout=10)
|
||
else:
|
||
result = self._shell(
|
||
f"monkey -p {package} -c android.intent.category.LAUNCHER 1",
|
||
timeout=20,
|
||
)
|
||
self.last_active = datetime.now()
|
||
return {"code": 200, "data": {"success": True, "package": package, "activity": activity}}
|
||
except Exception as e:
|
||
return {"code": 500, "message": str(e)}
|
||
|
||
def stop_app(self, package: str) -> Dict[str, Any]:
|
||
"""停止APP"""
|
||
try:
|
||
self._shell(f"am force-stop {package}")
|
||
return {"code": 200, "data": {"success": True, "package": package}}
|
||
except Exception as e:
|
||
return {"code": 500, "message": str(e)}
|
||
|
||
def current_app(self) -> Dict[str, Any]:
|
||
"""获取当前APP"""
|
||
try:
|
||
import re
|
||
output = self._shell("dumpsys activity activities | grep mResumedActivity", timeout=20)
|
||
match = re.search(r'(\w+\.\w+[\.\w]*)/[\.\w]+', output)
|
||
if not match:
|
||
output = self._shell("dumpsys window | grep -E 'mCurrentFocus'", timeout=20)
|
||
match = re.search(r'(\w+\.\w+[\.\w]*)/[\.\w]+', output)
|
||
package = match.group(1) if match else "unknown"
|
||
return {
|
||
"code": 200,
|
||
"data": {
|
||
"package": package,
|
||
"raw": output
|
||
}
|
||
}
|
||
except Exception as e:
|
||
return {"code": 500, "message": str(e)}
|
||
|
||
def installed_apps(self) -> Dict[str, Any]:
|
||
"""获取已安装APP"""
|
||
try:
|
||
output = self._shell("pm list packages -3")
|
||
packages = [
|
||
line.replace("package:", "")
|
||
for line in output.split("\n")
|
||
if line.startswith("package:")
|
||
]
|
||
return {"code": 200, "data": {"packages": packages}}
|
||
except Exception as e:
|
||
return {"code": 500, "message": str(e)}
|
||
|
||
# ========== UI树 ==========
|
||
|
||
def get_ui_tree(self) -> Dict[str, Any]:
|
||
"""获取UI树"""
|
||
try:
|
||
# 先dump到文件再读取（限时5秒，防止卡住）
|
||
self._shell("uiautomator dump /sdcard/ui_tree.xml", timeout=15)
|
||
xml = self._shell("cat /sdcard/ui_tree.xml", timeout=10)
|
||
|
||
if not xml or len(xml) < 50:
|
||
return {"code": 500, "message": "获取UI树失败"}
|
||
|
||
return {"code": 200, "data": {"xml": xml, "length": len(xml)}}
|
||
except Exception as e:
|
||
return {"code": 500, "message": str(e)}
|
||
|
||
|
||
class ADBDeviceManager:
|
||
"""ADB设备管理器"""
|
||
|
||
_SCAN_CACHE_TTL = 3 # 缓存 3 秒，避免浏览器高频轮询打爆线程池
|
||
|
||
def __init__(self):
|
||
self.devices: Dict[str, ADBDevice] = {}
|
||
self._deviceid_to_serial: Dict[str, str] = {}
|
||
self._scan_cache: Optional[List[str]] = None
|
||
self._scan_cache_time: float = 0
|
||
self._scan_all_cache: Optional[List[Dict[str, str]]] = None
|
||
self._scan_all_cache_time: float = 0
|
||
|
||
def scan_devices(self) -> List[str]:
|
||
"""扫描在线设备（仅已授权），带 TTL 缓存"""
|
||
now = time.time()
|
||
if self._scan_cache is not None and (now - self._scan_cache_time) < self._SCAN_CACHE_TTL:
|
||
return self._scan_cache
|
||
try:
|
||
result = subprocess.run(
|
||
["adb", "devices"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=5,
|
||
env=_adb_env,
|
||
)
|
||
|
||
serials = []
|
||
for line in result.stdout.strip().split("\n")[1:]:
|
||
if "\tdevice" in line:
|
||
serial = line.split("\t")[0]
|
||
serials.append(serial)
|
||
|
||
if serial not in self.devices:
|
||
self.devices[serial] = ADBDevice(serial)
|
||
self._build_deviceid_map(serial)
|
||
|
||
self._scan_cache = serials
|
||
self._scan_cache_time = time.time()
|
||
return serials
|
||
except Exception as e:
|
||
logger.error(f"扫描设备失败: {e}")
|
||
return self._scan_cache or []
|
||
|
||
def scan_all_devices(self) -> List[Dict[str, str]]:
|
||
"""扫描全部 ADB 设备（含 unauthorized / offline），带 TTL 缓存"""
|
||
now = time.time()
|
||
if self._scan_all_cache is not None and (now - self._scan_all_cache_time) < self._SCAN_CACHE_TTL:
|
||
return self._scan_all_cache
|
||
try:
|
||
result = subprocess.run(
|
||
["adb", "devices", "-l"],
|
||
capture_output=True, text=True, timeout=5, env=_adb_env,
|
||
)
|
||
devices = []
|
||
for line in result.stdout.strip().split("\n")[1:]:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
parts = line.split()
|
||
if len(parts) < 2:
|
||
continue
|
||
serial = parts[0]
|
||
adb_status = parts[1]
|
||
props = {}
|
||
for p in parts[2:]:
|
||
if ":" in p:
|
||
k, v = p.split(":", 1)
|
||
props[k] = v
|
||
entry = {
|
||
"serial": serial,
|
||
"adb_status": adb_status,
|
||
"model": props.get("model", ""),
|
||
"device": props.get("device", ""),
|
||
"transport_id": props.get("transport_id", ""),
|
||
"usb": props.get("usb", ""),
|
||
"product": props.get("product", ""),
|
||
"controllable": adb_status == "device",
|
||
}
|
||
if adb_status == "device" and serial not in self.devices:
|
||
self.devices[serial] = ADBDevice(serial)
|
||
self._build_deviceid_map(serial)
|
||
devices.append(entry)
|
||
self._scan_all_cache = devices
|
||
self._scan_all_cache_time = time.time()
|
||
return devices
|
||
except Exception as e:
|
||
logger.error(f"扫描全部设备失败: {e}")
|
||
return self._scan_all_cache or []
|
||
|
||
def _build_deviceid_map(self, serial: str):
|
||
"""为 serial 构建 device_id→serial 映射"""
|
||
import hashlib
|
||
try:
|
||
r = subprocess.run(
|
||
["adb", "-s", serial, "shell", "settings", "get", "secure", "android_id"],
|
||
capture_output=True, text=True, timeout=5, env=_adb_env,
|
||
)
|
||
android_id = r.stdout.strip()
|
||
if android_id:
|
||
device_id = hashlib.md5(android_id.encode()).hexdigest()
|
||
self._deviceid_to_serial[device_id] = serial
|
||
except Exception:
|
||
pass
|
||
|
||
def register_mapping(self, device_id: str, serial: str):
|
||
"""注册 device_id → serial 映射（由 connection_priority 等外部服务调用）"""
|
||
if device_id and serial:
|
||
self._deviceid_to_serial[device_id] = serial
|
||
|
||
def resolve_serial(self, identifier: str) -> Optional[str]:
|
||
"""将 serial 或 device_id 解析为 ADB serial"""
|
||
if identifier in self.devices:
|
||
return identifier
|
||
if identifier in self._deviceid_to_serial:
|
||
return self._deviceid_to_serial[identifier]
|
||
self.scan_devices()
|
||
if identifier in self.devices:
|
||
return identifier
|
||
return self._deviceid_to_serial.get(identifier)
|
||
|
||
def get_device(self, identifier: str) -> Optional[ADBDevice]:
|
||
"""获取设备（同时接受 serial 和 device_id）"""
|
||
serial = self.resolve_serial(identifier)
|
||
if not serial:
|
||
return None
|
||
if serial not in self.devices:
|
||
if self._check_device(serial):
|
||
self.devices[serial] = ADBDevice(serial)
|
||
return self.devices.get(serial)
|
||
|
||
def _check_device(self, serial: str) -> bool:
|
||
"""检查设备是否存在"""
|
||
try:
|
||
result = subprocess.run(
|
||
["adb", "-s", serial, "get-state"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=5,
|
||
env=_adb_env,
|
||
)
|
||
return "device" in result.stdout
|
||
except:
|
||
return False
|
||
|
||
def list_devices(self) -> List[Dict[str, Any]]:
|
||
"""列出所有设备"""
|
||
self.scan_devices()
|
||
|
||
devices = []
|
||
for serial, device in self.devices.items():
|
||
info = device.get_info()
|
||
info["status"] = "online" if device.is_online() else "offline"
|
||
devices.append(info)
|
||
|
||
return devices
|
||
|
||
# ── 异步包装（不阻塞 FastAPI 事件循环） ──
|
||
|
||
async def async_scan_devices(self) -> List[str]:
|
||
loop = asyncio.get_event_loop()
|
||
return await loop.run_in_executor(_adb_executor, self.scan_devices)
|
||
|
||
async def async_scan_all_devices(self) -> List[Dict[str, str]]:
|
||
loop = asyncio.get_event_loop()
|
||
return await loop.run_in_executor(_adb_executor, self.scan_all_devices)
|
||
|
||
async def async_list_devices(self) -> List[Dict[str, Any]]:
|
||
loop = asyncio.get_event_loop()
|
||
return await loop.run_in_executor(_adb_executor, self.list_devices)
|
||
|
||
|
||
# 全局实例
|
||
adb_manager = ADBDeviceManager()
|