241 lines
9.7 KiB
Python
241 lines
9.7 KiB
Python
"""uiauto2 (uiautodev) 本地服务客户端封装。
|
||
|
||
uiautodev 启动后本地监听 http://localhost:20242，提供设备元素树查询。
|
||
前端步骤编辑器"抓取元素"按钮通过本模块拉取当前设备 UI 树供选择。
|
||
|
||
依赖：
|
||
- 用户需先 `pip install uiautodev` 并运行 `uiauto.dev` 启动本地服务
|
||
- 未运行时本模块函数返回友好错误，不抛异常
|
||
|
||
API 参考（uiautodev 0.14）：
|
||
- GET /api/info — 服务信息（用于探测是否运行）
|
||
- GET /api/android/{serial}/dump_hierarchy — 元素树 JSON
|
||
"""
|
||
import requests
|
||
from collections import Counter
|
||
|
||
from core.logger import get_logger
|
||
|
||
_log = get_logger("core.uiauto")
|
||
|
||
# uiautodev 默认本地端口（固定 20242）
|
||
_UIAUTO_BASE = "http://localhost:20242"
|
||
# 请求超时（秒）。connect 超时短，避免前端等太久
|
||
_TIMEOUT = (1, 8)
|
||
|
||
|
||
def is_running():
|
||
"""探测 uiauto2 本地服务是否在运行。"""
|
||
try:
|
||
r = requests.get(f"{_UIAUTO_BASE}/api/info", timeout=_TIMEOUT)
|
||
return r.status_code == 200
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def list_devices():
|
||
"""获取 uiauto2 已连接的设备列表。
|
||
|
||
返回 (ok, data_or_error)：
|
||
ok=True — data 是设备列表 [{serial, model, product, name, status}, ...]
|
||
ok=False — data 是错误消息字符串
|
||
"""
|
||
try:
|
||
r = requests.get(f"{_UIAUTO_BASE}/api/android/list", timeout=_TIMEOUT)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
return True, data
|
||
except requests.exceptions.ConnectionError:
|
||
return False, "uiauto2 未启动，请运行 `uiauto.dev`"
|
||
except Exception as e:
|
||
_log.warning(f"list_devices 异常: {e}")
|
||
return False, f"获取设备列表失败: {e}"
|
||
|
||
|
||
def get_screenshot(serial):
|
||
"""获取设备截图（通过 uiauto2 服务）。
|
||
|
||
返回 (ok, data_or_error)：
|
||
ok=True — data 是 JPEG 二进制数据
|
||
ok=False — data 是错误消息字符串
|
||
"""
|
||
if not serial:
|
||
return False, "缺少 serial"
|
||
try:
|
||
r = requests.get(
|
||
f"{_UIAUTO_BASE}/api/android/{serial}/screenshot/0",
|
||
timeout=_TIMEOUT,
|
||
)
|
||
r.raise_for_status()
|
||
return True, r.content
|
||
except requests.exceptions.ConnectionError:
|
||
return False, "uiauto2 未启动"
|
||
except Exception as e:
|
||
_log.warning(f"get_screenshot 异常: {e}")
|
||
return False, f"截图失败: {e}"
|
||
|
||
|
||
def get_elements(serial):
|
||
"""获取指定设备的 UI 元素树。
|
||
|
||
返回 (ok, data_or_error)：
|
||
ok=True — data 是元素列表 [{name, attrs..., suggested:{type,value,indexed?,occ?,total?,broad?}}, ...]
|
||
ok=False — data 是错误消息字符串
|
||
|
||
元素树由 uiautodev 的 dump_hierarchy 返回（JSON）。我们递归提取每个节点的
|
||
关键属性（resource-id/text/description/class/bounds...），供前端列表展示和选择。
|
||
|
||
选择器精度策略（精确到具体按钮的关键）：
|
||
1. 有 resource-id/text/content-desc 的元素：
|
||
先预统计该属性在整棵树中的出现次数——
|
||
- 唯一出现：直接用属性选择器 //*[@resource-id="x"]
|
||
- 重复出现（如抖音信息流点赞按钮同 id 几十个）：附加 [k] 位置谓词
|
||
精确到具体实例 //*[@resource-id="x"][k]。uiautomator2 的 d.xpath()
|
||
底层是 lxml 标准 XPath，[k] 与抓取时同一语义，不会误中屏幕外第一个。
|
||
2. 无任何属性的元素：
|
||
用最近一个有属性祖先的选择器限定范围 + 同 class 兄弟序号定位
|
||
（如 //*[@resource-id="x"]/FrameLayout/ImageView[2]）；
|
||
整棵树都没有属性时退化为从根开始的结构路径（标记 broad，前端提示脆弱）。
|
||
"""
|
||
if not serial:
|
||
return False, "缺少 serial"
|
||
try:
|
||
r = requests.get(
|
||
f"{_UIAUTO_BASE}/api/android/{serial}/hierarchy",
|
||
timeout=_TIMEOUT,
|
||
)
|
||
if r.status_code == 404:
|
||
return False, f"设备 {serial} 未连接到 uiauto2"
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
# uiautodev hierarchy 返回 Node 树：{key, name, bounds, rect, properties, children}
|
||
# 提取成扁平的可选列表（保留层级缩进信息）
|
||
elements = []
|
||
_extract(data, elements)
|
||
if not elements:
|
||
return False, "当前界面未抓取到元素"
|
||
return True, elements
|
||
except requests.exceptions.ConnectionError:
|
||
return False, "uiauto2 未启动，请运行 `uiauto.dev`"
|
||
except requests.exceptions.Timeout:
|
||
return False, "uiauto2 请求超时"
|
||
except Exception as e:
|
||
_log.warning(f"get_elements 异常: {e}")
|
||
return False, f"抓取失败: {e}"
|
||
|
||
|
||
def _xpath_q(v):
|
||
"""XPath 字符串字面量：优先双引号，值含双引号时改用单引号包裹（XPath 1.0 无转义）。"""
|
||
if '"' in v:
|
||
return "'" + v + "'"
|
||
return '"' + v + '"'
|
||
|
||
|
||
def _extract(root, out):
|
||
"""把 uiautodev 元素树扁平化为可选列表，并为每个节点生成精确选择器建议。"""
|
||
# 全树属性出现次数（预统计，供重复元素加 [k] 序号消歧）
|
||
id_cnt, text_cnt, desc_cnt = Counter(), Counter(), Counter()
|
||
# 文档顺序已出现次数（决定当前元素是第几个）
|
||
seen_id, seen_text, seen_desc = Counter(), Counter(), Counter()
|
||
|
||
def count_attrs(node):
|
||
if not isinstance(node, dict):
|
||
return
|
||
props = node.get("properties") or {}
|
||
rid, text, desc = (props.get(k, "") for k in ("resource-id", "text", "content-desc"))
|
||
if rid:
|
||
id_cnt[rid] += 1
|
||
if text:
|
||
text_cnt[text] += 1
|
||
if desc:
|
||
desc_cnt[desc] += 1
|
||
for c in node.get("children") or []:
|
||
count_attrs(c)
|
||
|
||
def attr_selector(attr, value, cnt, seen):
|
||
"""属性选择器：唯一直接出，重复加 [k] 位置谓词。返回 (选择器, suggested)。"""
|
||
seen[value] += 1
|
||
val = f'//*[@{attr}={_xpath_q(value)}]'
|
||
if cnt[value] > 1:
|
||
val += f"[{seen[value]}]"
|
||
return val, {"type": "xpath", "value": val,
|
||
"indexed": True, "occ": seen[value], "total": cnt[value]}
|
||
return val, {"type": "xpath", "value": val}
|
||
|
||
def flatten(node, depth=0, path="", ctx=None, tag_path="", tag_index=1):
|
||
"""递归扁平化。
|
||
|
||
ctx: 最近一个有属性祖先的选择器（含序号），无属性元素用它限定范围。
|
||
tag_path: 从根到本节点的完整结构路径（class + 同class兄弟序号），兜底用。
|
||
tag_index: 本节点在父下同 class 兄弟中的序号（1-based）。
|
||
"""
|
||
if not isinstance(node, dict):
|
||
return
|
||
props = node.get("properties") or {}
|
||
name = node.get("name") or props.get("class") or ""
|
||
# bounds：优先取 properties 中的原始字符串 "[x1,y1][x2,y2]"（前端渲染 overlay 用）
|
||
bounds_str = props.get("bounds", "")
|
||
if not bounds_str and node.get("rect"):
|
||
# rect 兜底：只有整数像素坐标才用；uiautodev 归一化浮点坐标无法换算像素
|
||
r = node["rect"]
|
||
try:
|
||
vals = [r["x"], r["y"], r["x"] + r["width"], r["y"] + r["height"]]
|
||
if all(isinstance(v, (int, float)) and float(v).is_integer() for v in vals):
|
||
bounds_str = f"[{int(vals[0])},{int(vals[1])}][{int(vals[2])},{int(vals[3])}]"
|
||
except (TypeError, KeyError):
|
||
pass
|
||
|
||
rid = props.get("resource-id", "")
|
||
text = props.get("text", "")
|
||
desc = props.get("content-desc", "")
|
||
cls = props.get("class", "")
|
||
tag = cls.split(".")[-1] if cls else "*"
|
||
# 始终带同 class 兄弟序号（含 [1]），结构路径才精确无歧义
|
||
seg = f"{tag}[{tag_index}]"
|
||
own_tag_path = f"{tag_path}/{seg}" if tag_path else seg
|
||
|
||
# ---- 推荐选择器：唯一属性 > 锚点祖先限定 > 全结构路径 ----
|
||
if rid:
|
||
child_ctx, suggested = attr_selector("resource-id", rid, id_cnt, seen_id)
|
||
elif text:
|
||
child_ctx, suggested = attr_selector("text", text, text_cnt, seen_text)
|
||
elif desc:
|
||
child_ctx, suggested = attr_selector("content-desc", desc, desc_cnt, seen_desc)
|
||
elif ctx:
|
||
val = f"{ctx}/{seg}"
|
||
suggested = {"type": "xpath", "value": val}
|
||
child_ctx = val
|
||
else:
|
||
# 无唯一属性且无锚点祖先：用从根开始的结构路径（脆弱，标记 broad 让前端提示）
|
||
suggested = {"type": "xpath", "value": f"//{own_tag_path}", "broad": True}
|
||
child_ctx = None
|
||
|
||
out.append({
|
||
"depth": depth,
|
||
"path": path,
|
||
"name": name,
|
||
"resource_id": rid,
|
||
"text": text,
|
||
"description": desc,
|
||
"class": cls,
|
||
"package": props.get("package", ""),
|
||
"clickable": props.get("clickable", ""),
|
||
"bounds": bounds_str,
|
||
"suggested": suggested,
|
||
})
|
||
|
||
# 递归子节点，同时计算每个子节点在父下同 class 兄弟中的序号
|
||
children = node.get("children") or []
|
||
for i, child in enumerate(children):
|
||
cprops = child.get("properties") or {}
|
||
ccls = cprops.get("class", "")
|
||
ctag = ccls.split(".")[-1] if ccls else "*"
|
||
same = 1 + sum(
|
||
1 for prev in children[:i]
|
||
if ((prev.get("properties") or {}).get("class", "")).split(".")[-1] == ctag
|
||
)
|
||
flatten(child, depth + 1, f"{path}/{i}", child_ctx, own_tag_path, same)
|
||
|
||
count_attrs(root)
|
||
flatten(root)
|