Kris
15fc62d478
- 新增六层UI自动化架构：从Backend到Capabilities的完整分层实现 - 添加WeChat 4.0分辨率适配Profile与图像模板资源 - 实现幂等缓存、熔断器、重试策略、链路追踪与监控指标 - 新增头像下载安全校验、发布朋友圈路径白名单防护 - 优化密钥缓存、DB校验逻辑与初始化流程 - 补充完整错误码体系与启动清场机制
242 lines
7.9 KiB
Python
242 lines
7.9 KiB
Python
"""L3 Actions 原子动作库：find / click / type / wait / screenshot。
|
||
|
||
Actions 是 L1 Backend 与 L2 Locator 的组合层：给定 kind 与窗口几何，
|
||
Locator 解析为 ElementHandle（图像优先、几何兜底），Backend 据此执行点击/输入。
|
||
L4 Flow 通过 Actions 组合出完整业务流程（如 SendTextFlow）。
|
||
|
||
策略：
|
||
- 图像路径（P2 启用）：sel.by_image 非空且 locator.is_circuited 为 False 且 opencv 非 None 时优先
|
||
- 命中：record_image_success，返回 strategy=image
|
||
- 未命中：record_image_failure；require_image=True 抛 ElementNotFoundError；否则回退 _geom_find
|
||
- 几何兜底（P1 默认）：按 GeomSpec 计算绝对坐标
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
import time
|
||
from typing import Optional
|
||
|
||
from woc_bridge.ui.backends.base import BackendProtocol, WindowGeometry
|
||
from woc_bridge.ui.locators.base import ElementHandle, Selector
|
||
from woc_bridge.ui.locators.registry import LocatorRegistry
|
||
|
||
logger = logging.getLogger("woc-bridge")
|
||
|
||
|
||
class ElementNotFoundError(Exception):
|
||
"""元素未找到（图像匹配失败且 require_image=True，或几何规格缺失）。
|
||
|
||
TODO: P3 时统一到 errors.py 的 FlowError(code="ELEMENT_NOT_FOUND")。
|
||
"""
|
||
|
||
pass
|
||
|
||
|
||
class Actions:
|
||
"""L3 原子动作库：组合 Backend（L1）与 Locator（L2）。
|
||
|
||
所有方法均为 async，由 L4 Flow 编排调用。
|
||
"""
|
||
|
||
def __init__(
|
||
self, backend: BackendProtocol, locator: LocatorRegistry
|
||
) -> None:
|
||
"""初始化 Actions。
|
||
|
||
Args:
|
||
backend: L1 Backend 实例（如 XdotoolBackend）
|
||
locator: L2 LocatorRegistry 实例
|
||
"""
|
||
self.backend = backend
|
||
self.locator = locator
|
||
|
||
# ------------------------------------------------------------------
|
||
# 元素查找
|
||
# ------------------------------------------------------------------
|
||
async def find_element(
|
||
self, kind: str, win_geom: WindowGeometry
|
||
) -> ElementHandle:
|
||
"""查找元素，优先图像匹配，失败回退几何兜底。
|
||
|
||
Args:
|
||
kind: 元素类别键（如 search_box / send_button）
|
||
win_geom: 当前窗口几何（绝对坐标 + 宽高）
|
||
|
||
Returns:
|
||
ElementHandle：命中元素（含绝对坐标与策略）
|
||
|
||
Raises:
|
||
ElementNotFoundError: require_image=True 且图像匹配失败，
|
||
或 by_geom 为空且无图像命中
|
||
"""
|
||
sel = self.locator.get(kind)
|
||
|
||
# 图像路径（P2 启用，P1 阶段 opencv=None 直接跳过）
|
||
if (
|
||
sel.by_image
|
||
and not self.locator.is_circuited(kind)
|
||
and self.locator.opencv is not None
|
||
):
|
||
shot = await self.backend.screenshot()
|
||
roi = (win_geom.x, win_geom.y, win_geom.width, win_geom.height)
|
||
try:
|
||
result = await self.locator.opencv.find_template(
|
||
shot,
|
||
sel.by_image,
|
||
threshold=sel.threshold,
|
||
roi=roi,
|
||
)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"[action] opencv.find_template exception kind=%s: %s",
|
||
kind,
|
||
exc,
|
||
)
|
||
result = None
|
||
|
||
if result is not None:
|
||
# result = (x, y, w, h, confidence)（绝对坐标，含 ROI 偏移）
|
||
x, y, w, h, conf = result
|
||
self.locator.record_image_success(kind)
|
||
return ElementHandle(
|
||
selector=sel,
|
||
x=x + w // 2,
|
||
y=y + h // 2,
|
||
w=w,
|
||
h=h,
|
||
confidence=conf,
|
||
strategy="image",
|
||
)
|
||
|
||
# 图像未命中
|
||
self.locator.record_image_failure(kind)
|
||
logger.warning(
|
||
"[action] image match failed kind=%s, fall back to geom", kind
|
||
)
|
||
if sel.require_image:
|
||
raise ElementNotFoundError(
|
||
f"image_match_failed kind={kind}"
|
||
)
|
||
|
||
return self._geom_find(sel, win_geom)
|
||
|
||
async def click_element(
|
||
self, kind: str, win_geom: WindowGeometry
|
||
) -> ElementHandle:
|
||
"""查找并点击元素，返回 ElementHandle。
|
||
|
||
Args:
|
||
kind: 元素类别键
|
||
win_geom: 当前窗口几何
|
||
|
||
Returns:
|
||
命中的 ElementHandle
|
||
"""
|
||
elem = await self.find_element(kind, win_geom)
|
||
await self.backend.click(elem.x, elem.y)
|
||
logger.info(
|
||
"[action] click %s at (%d,%d) strategy=%s conf=%.3f",
|
||
kind,
|
||
elem.x,
|
||
elem.y,
|
||
elem.strategy,
|
||
elem.confidence,
|
||
)
|
||
return elem
|
||
|
||
async def type_into(
|
||
self,
|
||
kind: str,
|
||
text: str,
|
||
win_geom: WindowGeometry,
|
||
clear_first: bool = True,
|
||
) -> None:
|
||
"""点击元素聚焦后输入文本，可选先清空。
|
||
|
||
Args:
|
||
kind: 元素类别键（通常为 input_box）
|
||
text: 要输入的文本
|
||
win_geom: 当前窗口几何
|
||
clear_first: 是否先 BackSpace×30 清空旧内容
|
||
"""
|
||
await self.click_element(kind, win_geom)
|
||
if clear_first:
|
||
await self.backend.key_press("BackSpace", repeat=30)
|
||
await self.backend.type_text(text)
|
||
|
||
async def wait_for(
|
||
self,
|
||
kind: str,
|
||
win_geom: WindowGeometry,
|
||
timeout: float = 5.0,
|
||
interval: float = 0.3,
|
||
) -> Optional[ElementHandle]:
|
||
"""轮询等待元素出现，超时返回 None。
|
||
|
||
Args:
|
||
kind: 元素类别键
|
||
win_geom: 当前窗口几何
|
||
timeout: 最大等待秒数
|
||
interval: 轮询间隔秒数
|
||
|
||
Returns:
|
||
ElementHandle 或 None（超时）
|
||
"""
|
||
deadline = time.monotonic() + timeout
|
||
while time.monotonic() < deadline:
|
||
try:
|
||
elem = await self.find_element(kind, win_geom)
|
||
if elem:
|
||
return elem
|
||
except ElementNotFoundError:
|
||
pass
|
||
await asyncio.sleep(interval)
|
||
return None
|
||
|
||
async def screenshot(self) -> bytes:
|
||
"""截图代理方法。"""
|
||
return await self.backend.screenshot()
|
||
|
||
# ------------------------------------------------------------------
|
||
# 几何兜底
|
||
# ------------------------------------------------------------------
|
||
def _geom_find(
|
||
self, selector: Selector, win_geom: WindowGeometry
|
||
) -> ElementHandle:
|
||
"""按 GeomSpec 计算绝对坐标，返回 ElementHandle。
|
||
|
||
简化策略：所有 relative_to 模式统一用
|
||
base_x = win_geom.x + win_geom.width * g.x_ratio + g.x_offset
|
||
base_y = win_geom.y + win_geom.height * g.y_ratio + g.y_offset
|
||
relative_to 仅作语义占位，offset 已含正负号区分方向
|
||
（如 send_button 用 x_offset=-60 表示距右边缘 60px）。
|
||
|
||
Args:
|
||
selector: 元素选择器（必须含 by_geom）
|
||
win_geom: 当前窗口几何
|
||
|
||
Returns:
|
||
ElementHandle（strategy=geom, confidence=1.0）
|
||
|
||
Raises:
|
||
ElementNotFoundError: selector.by_geom 为空
|
||
"""
|
||
if selector.by_geom is None:
|
||
raise ElementNotFoundError(
|
||
f"no geom spec kind={selector.kind}"
|
||
)
|
||
g = selector.by_geom
|
||
base_x = win_geom.x + win_geom.width * g.x_ratio + g.x_offset
|
||
base_y = win_geom.y + win_geom.height * g.y_ratio + g.y_offset
|
||
return ElementHandle(
|
||
selector=selector,
|
||
x=int(base_x),
|
||
y=int(base_y),
|
||
w=0,
|
||
h=0,
|
||
confidence=1.0,
|
||
strategy="geom",
|
||
)
|